testsAndMisc/python_pkg/wsg_grabber/cli.py

"""Command-line entry point for the /wsg/ grabber.

Usage:
    PYTHONPATH=~/testsAndMisc python3 -m python_pkg.wsg_grabber
    PYTHONPATH=~/testsAndMisc python3 -m python_pkg.wsg_grabber scrape --seconds 60
    PYTHONPATH=~/testsAndMisc python3 -m python_pkg.wsg_grabber stats
"""

from __future__ import annotations

import argparse
import sys
import time
from typing import TYPE_CHECKING

from python_pkg.wsg_grabber import (
    app,
    db,
    logs,
    paths,
    player,
    store,
    store_threads,
    ui,
)

# Re-exported so `cli.show_logs` / `cli.summarise` keep working for callers and
# tests; the implementation lives in _logs_report to keep this module under the
# 250-line cap. __all__ is what marks them as re-exports -- the `X as X` alias
# form is rejected here (ruff PLC0414) and then stripped as unused.
from python_pkg.wsg_grabber._logs_report import show_logs, summarise
from python_pkg.wsg_grabber.states import FileState

if TYPE_CHECKING:
    from collections.abc import Callable

__all__ = [
    "build_parser",
    "main",
    "open_reviewer",
    "progress",
    "scrape",
    "show_logs",
    "stats",
    "summarise",
]

_REPORT_EVERY_S = 15.0

_DESCRIPTION = (
    "Scrape 4chan /wsg/ for videos and triage them with a keep/pass reviewer. "
    "Nothing is ever deleted: a pass moves the file into the trash directory "
    "for you to clear by hand."
)


def _emit(message: str) -> None:
    """Write a line to stdout.

    Args:
        message: Line to write, without a trailing newline.
    """
    sys.stdout.write(f"{message}\n")


def build_parser() -> argparse.ArgumentParser:
    """Construct the argument parser.

    Returns:
        argparse.ArgumentParser: Parser with the ``review``, ``scrape`` and
        ``stats`` subcommands. ``review`` is the default.
    """
    parser = argparse.ArgumentParser(
        prog="python -m python_pkg.wsg_grabber",
        description=_DESCRIPTION,
    )
    parser.add_argument(
        "--log-level",
        choices=logs.LEVELS,
        default="info",
        help="detail written to the session log (default: info)",
    )
    parser.add_argument(
        "--echo-log",
        action="store_true",
        help="also print log records to stderr as they happen",
    )
    subparsers = parser.add_subparsers(dest="command")
    subparsers.add_parser(
        "review",
        help="download in the background and review on screen (default)",
    )
    subparsers.add_parser("stats", help="print index counts and exit")
    subparsers.add_parser(
        "logs",
        help="show where session logs live and summarise the newest one",
    )
    scrape_parser = subparsers.add_parser(
        "scrape",
        help="index and download without opening the reviewer",
    )
    scrape_parser.add_argument(
        "--seconds",
        type=float,
        default=0.0,
        help="stop after this long (0 means run until interrupted)",
    )
    return parser


def main(argv: list[str] | None = None) -> int:
    """Run the requested subcommand.

    Args:
        argv: Argument list, defaulting to ``sys.argv[1:]``.

    Returns:
        int: Process exit status.
    """
    args = build_parser().parse_args(argv)
    logs.start(args.log_level, echo=args.echo_log)
    try:
        return _dispatch(args)
    finally:
        logs.stop()


def _dispatch(args: argparse.Namespace) -> int:
    """Run the subcommand named in *args*.

    Args:
        args: Parsed arguments.

    Returns:
        int: Process exit status.
    """
    if args.command == "logs":
        return show_logs()
    if args.command == "stats":
        return stats()
    if args.command == "scrape":
        return scrape(args.seconds)
    return open_reviewer()


def stats() -> int:
    """Print what the index knows.

    Returns:
        int: Exit status.
    """
    paths.ensure_dirs()
    conn = db.open_index(paths.db_path())
    try:
        counts = store.counts(conn)
        _emit(f"index        {paths.db_path()}")
        _emit(f"known files  {counts.get('total', 0)}")
        for state in FileState:
            _emit(f"  {state.value:<12} {counts.get(state.value, 0)}")
        _emit(f"threads      {len(store_threads.known_threads(conn))}")
        _emit(f"keep dir     {paths.keep_dir()}")
        _emit(f"trash dir    {paths.trash_dir()}  (never emptied automatically)")
    finally:
        conn.close()
    return 0


def scrape(seconds: float, sleeper: Callable[[float], None] | None = None) -> int:
    """Index and download without opening a window.

    Args:
        seconds: Stop after this long; 0 runs until interrupted.
        sleeper: Optional stand-in for :func:`time.sleep`, used by tests to
            avoid actually waiting.

    Returns:
        int: Exit status.
    """
    rest = time.sleep if sleeper is None else sleeper
    session = app.open_session()
    started = time.monotonic()
    reported = started
    try:
        _emit("scraping /wsg/ — Ctrl-C to stop")
        while True:
            rest(1.0)
            now = time.monotonic()
            if now - reported >= _REPORT_EVERY_S:
                reported = now
                _emit(progress(session))
            if seconds and now - started >= seconds:
                break
    except KeyboardInterrupt:
        _emit("stopping")
    finally:
        _emit(progress(session))
        session.shutdown()
    return 0


def progress(session: app.Session) -> str:
    """Summarise index progress.

    Args:
        session: The running session.

    Returns:
        str: One-line summary.
    """
    counts = store.counts(session.conn)
    return (
        f"known {counts.get('total', 0)} · "
        f"ready {counts.get(FileState.READY.value, 0)} · "
        f"pending {store.pending_downloads(session.conn)}"
    )


def open_reviewer() -> int:
    """Open the reviewer window and hand control to it.

    Returns:
        int: Exit status.
    """
    session = app.open_session()
    try:
        window = ui.ReviewWindow(
            app.initial_state(session),
            session.events,
            ui.Callbacks(
                commit=session.commit,
                missing=session.missing,
                undo=session.undo,
                shutdown=session.shutdown,
            ),
        )
        window.attach_player(
            player.MpvPlayer(window.video_wid(), paths.ipc_socket_path()),
        )
        window.run()
    except BaseException:
        # The worker is a non-daemon thread, so anything raised between opening
        # the session and entering the Tk loop -- mpv missing, a stale socket --
        # would otherwise leave the process alive forever with no window.
        logs.error("review.startup_failed")
        session.shutdown()
        raise
    return 0