anicli/cli/utils.py - master - dig/anicli_ru

anicli/cli/utils.py
88 строк3 KB
An0nX
refactor: modernize type hints, improve aniskip merging and history fallback
09 янв 2026, 00:43
09 янв 2026, 00:43925b7f8
import asyncio
from typing import Any

import questionary
from rich.live import Live
from rich.spinner import Spinner
from rich.table import Table

from anicli.cli import app_context
from anicli.config import config
from anicli.db.service import db_service


def generate_status_table_health(statuses: dict[str, str]) -> Table:
    """Generates a Rich Table for health check status."""
    table = Table("Источник", "Тест", "Статус")
    for name, status in statuses.items():
        table.add_row(name, "Ping", status)
    return table


async def check_provider_safe(prov: Any, statuses: dict[str, Any], live: Live) -> None:
    """Safely checks a provider's health."""
    try:
        await asyncio.to_thread(
            prov.search, "test", timeout=config.settings.network.connect_timeout
        )
        statuses[prov.name] = "[green]OK[/green]"
    except Exception:
        statuses[prov.name] = "[red]Error[/red]"
    live.update(generate_status_table_health(statuses))


async def health_check() -> None:
    """Performs a health check on all providers and integrations."""
    providers = app_context.loader.get_all()
    statuses = {p.name: Spinner("dots", style="cyan") for p in providers}
    if config.settings.shikimori.enabled:
        statuses["Shikimori"] = Spinner("dots", style="magenta")

    with Live(generate_status_table_health(statuses), refresh_per_second=10) as live:
        tasks = [check_provider_safe(p, statuses, live) for p in providers]

        async def check_shiki() -> None:
            if "Shikimori" in statuses:
                uid = await app_context.shikimori.get_user_id()
                statuses["Shikimori"] = (
                    f"[green]OK (User: {uid})[/green]"
                    if uid
                    else "[red]Auth Failed[/red]"
                )
                live.update(generate_status_table_health(statuses))

        tasks.append(check_shiki())
        await asyncio.gather(*tasks)


async def database_menu() -> None:
    """Provides options for database management."""
    choices = [
        "Очистить предсказания таймкодов",
        "Очистить всю историю просмотров",
        "Полная очистка БД (история + таймкоды)",
        "Назад",
    ]
    action = await questionary.select(
        "Управление базой данных:", choices=choices
    ).ask_async()

    if action == "Назад" or not action:
        return

    if not await questionary.confirm(
        f"Вы уверены, что хотите выполнить '{action}'? Это действие необратимо."
    ).ask_async():
        return

    if action == "Очистить предсказания таймкодов":
        count = await db_service.clear_predicted_skips()
        app_context.console.print(f"[green]Удалено {count} записей о таймкодах.[/green]")
    elif action == "Очистить всю историю просмотров":
        count = await db_service.clear_history()
        app_context.console.print(f"[green]Удалено {count} записей истории.[/green]")
    elif action == "Полная очистка БД (история + таймкоды)":
        h_count, s_count = await db_service.clear_database()
        app_context.console.print(
            f"[green]Удалено {h_count} записей истории и {s_count} записей о таймкодах.[/green]"
        )