youtube-automate/youtube_automate/doctor.py

"""Preflight checks.

`doctor` is the first acceptance criterion and the thing to run when something
breaks. Every check returns a row rather than raising, so one failure doesn't
hide the others.
"""

from __future__ import annotations

import grp
import os
import stat
import subprocess
from dataclasses import dataclass
from pathlib import Path

from . import config, jellyfin, ytdlp
from .settings import Settings

MEDIA_GROUP = "mediaserver"


@dataclass
class Check:
    name: str
    ok: bool
    detail: str
    fatal: bool = True


def _deno() -> Check:
    binary = config.VENV_BIN / "deno"
    if not binary.exists():
        return Check(
            "js runtime",
            False,
            f"deno not found at {binary} — yt-dlp cannot solve n challenges (specs.md §3)",
        )
    try:
        result = subprocess.run(
            [str(binary), "--version"], capture_output=True, text=True, timeout=30
        )
    except OSError as exc:
        return Check("js runtime", False, f"deno unusable: {exc}")
    if result.returncode != 0:
        return Check("js runtime", False, "deno --version failed")
    return Check("js runtime", True, result.stdout.splitlines()[0])


def _ytdlp() -> Check:
    try:
        return Check("yt-dlp", True, ytdlp.version())
    except (OSError, ytdlp.YtdlpError) as exc:
        return Check("yt-dlp", False, str(exc))


def _ejs() -> Check:
    try:
        from importlib.metadata import version as pkg_version

        return Check("yt-dlp-ejs", True, pkg_version("yt-dlp-ejs"))
    except Exception:
        return Check(
            "yt-dlp-ejs",
            False,
            "not installed — reinstall with the yt-dlp[default] extra (specs.md §3)",
        )


def _pot(settings: Settings) -> Check:
    url = settings.get_str("pot_provider_url")
    try:
        info = ytdlp.pot_provider_ping(url)
    except Exception as exc:
        return Check("pot provider", False, f"{url}/ping unreachable: {exc}")

    server_version = str(info.get("version", "?"))
    installed = ytdlp.plugin_version()
    if installed and installed != server_version:
        return Check(
            "pot provider",
            False,
            f"version skew: server {server_version} vs plugin {installed}",
        )
    return Check("pot provider", True, f"up, version {server_version}")


def _database() -> Check:
    try:
        from . import db

        conn = db.connect()
        version = conn.execute("PRAGMA user_version").fetchone()[0]
        mode = conn.execute("PRAGMA journal_mode").fetchone()[0]
        conn.close()
    except Exception as exc:
        return Check("database", False, str(exc))
    if str(mode).lower() != "wal":
        return Check("database", False, f"journal_mode is {mode}, expected wal")
    return Check("database", True, f"{config.DB_PATH} (schema v{version}, {mode})")


def _media_root() -> Check:
    root = config.MEDIA_ROOT
    if not root.is_dir():
        return Check("media root", False, f"{root} does not exist")
    if not os.access(root, os.W_OK | os.X_OK):
        return Check("media root", False, f"{root} is not writable")

    info = root.stat()
    problems = []
    try:
        group = grp.getgrgid(info.st_gid).gr_name
    except KeyError:
        group = str(info.st_gid)
    if group != MEDIA_GROUP:
        problems.append(f"group is {group}, expected {MEDIA_GROUP}")
    if not info.st_mode & stat.S_ISGID:
        problems.append("setgid bit not set (new dirs won't inherit the group)")

    if problems:
        return Check("media root", False, f"{root}: " + "; ".join(problems))
    return Check("media root", True, f"{root} ({group}, setgid)")


def _work_dir() -> Check:
    work = config.WORK_DIR
    if not work.is_dir():
        return Check("work dir", False, f"{work} does not exist")
    if work.stat().st_dev != config.MEDIA_ROOT.stat().st_dev:
        return Check(
            "work dir",
            False,
            "not on the same filesystem as the media root — moves would be copies",
        )
    if not (work / ".ignore").exists():
        return Check("work dir", False, f"{work}/.ignore missing", fatal=False)
    return Check("work dir", True, f"{work} (same fs, .ignore present)")


def _jellyfin(settings: Settings) -> Check:
    client = jellyfin.from_settings(settings)
    if not client.base_url:
        return Check("jellyfin", False, "jellyfin_url not set", fatal=False)
    try:
        info = client.public_info()
    except jellyfin.JellyfinError as exc:
        return Check("jellyfin", False, str(exc))

    label = f"{info.get('ServerName', '?')} {info.get('Version', '?')}"
    if not client.api_key:
        return Check(
            "jellyfin",
            False,
            f"{label} reachable but no API key set (run set-jellyfin-key)",
            fatal=False,
        )
    try:
        library = client.find_library(config.MEDIA_ROOT)
    except jellyfin.JellyfinError as exc:
        return Check("jellyfin", False, f"API key rejected: {exc}")
    if library is None:
        return Check(
            "jellyfin",
            False,
            f"{label}, no library for {config.MEDIA_ROOT} (run setup-jellyfin-library)",
            fatal=False,
        )
    return Check("jellyfin", True, f"{label}, library '{library.get('Name')}'")


def run_checks(settings: Settings) -> list[Check]:
    return [
        _ytdlp(),
        _ejs(),
        _deno(),
        _pot(settings),
        _database(),
        _media_root(),
        _work_dir(),
        _jellyfin(settings),
    ]


def report(checks: list[Check]) -> tuple[str, int]:
    """Render the checks and return (text, exit_code)."""
    lines = []
    failed_fatal = 0
    for check in checks:
        if check.ok:
            mark = "ok  "
        elif check.fatal:
            mark = "FAIL"
            failed_fatal += 1
        else:
            mark = "warn"
        lines.append(f"  [{mark}] {check.name:<14} {check.detail}")

    if failed_fatal:
        lines.append(f"\n{failed_fatal} fatal problem(s).")
    else:
        lines.append("\nAll fatal checks passed.")
    return "\n".join(lines), (1 if failed_fatal else 0)