Blame for src/weekly_analyst/cli.py

"""Command line: watch, run, approve (spec 6.5, 6.7; ticket 023).

approve is the only path that can send. It needs an interactive terminal or a local token file that the
local user creates by hand. No model tool reaches it (plan.STATE_ALLOWLIST has no approve entry).
Run as: python -m weekly_analyst.cli <command> ...
"""

import argparse
import getpass
import logging
import os
import sys
import tempfile
import time
from datetime import UTC, datetime
from pathlib import Path

from weekly_analyst.analysis.checks import check_report
from weekly_analyst.analysis.model import LlamaCppClient
from weekly_analyst.core.config import Settings
from weekly_analyst.core.scheduler import file_sha256
from weekly_analyst.core.store import StateStore
from weekly_analyst.core.text import numerals
from weekly_analyst.data.extract import ExtractError, extract
from weekly_analyst.data.metrics_file import load_metrics_file
from weekly_analyst.delivery import watch
from weekly_analyst.delivery.outbox import DeliveryConfig, DeliveryRefused, Outbox
from weekly_analyst.report import render_pdf
from weekly_analyst.report.layout import stamp_label
from weekly_analyst.report.model import ReportModel, ReportTooLong, ShownNumber
from weekly_analyst.runner import RunConfig, RunLocked, run
from weekly_analyst.systems.check import pdf_text

TOKEN_FILE = "approve.token"
STATE_DB = "state.db"
LOCAL_SENT = "local-sent-folder"  # the local sent folder is the only transport; approve sends to it through the gates
POLL_S = 30.0  # watch polls the folder this often (seconds)
N_CTX = 2048  # llama.cpp context length; a library-style default, not measured (spike ticket 028 decides)


class ApproveRefused(RuntimeError):
    """approve ran without an interactive terminal or a local token file."""


def approver_for(token_path: Path, *, interactive: bool) -> str:
    token = Path(token_path)
    local_token = token.is_file() and not token.is_symlink() and token.stat().st_size > 0
    if not (interactive or local_token):
        raise ApproveRefused("approve needs an interactive terminal or a local token file; nothing approved")
    return getpass.getuser()


def _is_interactive() -> bool:
    # isatty() is True for the Windows NUL device; a real console also answers the terminal-size query.
    try:
        os.get_terminal_size(sys.stdin.fileno())
    except (OSError, ValueError, AttributeError):
        return False
    return True


class _Approval:
    """What approve re-renders and re-checks: the stored report bundle, the metrics file and the latest CSV."""

    def __init__(self, report: dict, approver: str, approved_at: str, metrics, csv: Path, work: Path) -> None:
        self._report = report
        self._approver = approver
        self._approved_at = approved_at
        self._metrics = metrics
        self._csv = csv
        self._work = work

    def render_stamped(self, draft: Path, approver: str) -> Path:
        """Re-render the stored ReportModel as approved: the same model, so the same text and numbers as the draft."""
        out = self._work / Path(draft).name
        try:
            render_pdf(
                ReportModel.from_dict(self._report["model"]),
                out,
                status="approved",
                approver=approver,
                paper=self._report["paper"],
                approved_at=self._approved_at,
            )
        except (ReportTooLong, ValueError) as err:
            raise DeliveryRefused(f"stamp: {err}") from None
        return out

    def recheck(self, path: Path) -> list[str]:
        """Check the stamped text against the stored numbers, on a snapshot rebuilt from the same CSV."""
        snapshot_dir = self._work / "snapshot"
        snapshot_dir.mkdir()
        try:
            snapshot = extract(self._csv, self._metrics, snapshot_dir, Settings())
        except ExtractError as err:
            return [f"snapshot: {err}"]
        if snapshot.snapshot_hash != self._report["model"]["snapshot_hash"]:
            return ["snapshot: the source changed during approve"]
        shown = [ShownNumber.from_dict(d) for d in self._report["shown_numbers"]]
        # The approved stamp has numerals of its own (the time), so they are exempt on top of the draft's list.
        stamp = stamp_label("approved", self._approver, self._approved_at)
        exempt = [*self._report["structural_numbers"], *numerals(stamp)]
        failures = check_report(
            pdf_text(path),
            shown,
            self._metrics,
            snapshot.parquet_path,
            exempt=exempt,
            model_text=self._report["model_text"],
        )
        return [f"{failure.kind}: {failure.detail}" for failure in failures]


def _stored_payload(store: StateStore, run_id: str) -> dict:
    row = store.load_run(run_id)
    if row is None or row["state"] != "outbox" or not (row["payload"].get("report") or {}).get("model"):
        raise DeliveryRefused(f"no pending brief with a stored report for run {run_id}; nothing approved")
    return row["payload"]


def _latest_csv(folder: Path) -> Path:
    files = watch.scan(folder)
    if not files:
        raise DeliveryRefused("no accepted CSV in the watched folder; nothing approved")
    return max(files, key=lambda path: files[path][0])


def _model(args) -> LlamaCppClient:
    return LlamaCppClient(args.model, threads=os.cpu_count() or 1, n_ctx=N_CTX)


def _run_config(args) -> RunConfig:
    return RunConfig(
        metrics_path=args.metrics,
        work_dir=args.work_dir,
        out_dir=args.out_dir,
        delivery_mode="approval",
        recipients=list(args.recipient),
        paper=args.paper,
        settings=Settings(),
    )


def _print_outcome(outcome) -> None:
    print(f"{outcome.kind} {outcome.run_id}")
    if outcome.report_path is not None:
        print(f"report {outcome.report_path}")
    for reason in outcome.reasons:
        print(f"reason {reason}")


def _run(args) -> int:
    store = StateStore(args.work_dir / STATE_DB)
    outcome = run(args.csv, _run_config(args), _model(args), store)
    _print_outcome(outcome)
    return 1 if outcome.kind == "halted" else 0


def _watch(args) -> int:
    folder = args.folder.resolve()
    cfg = _run_config(args)
    store = StateStore(args.work_dir / STATE_DB)
    client = _model(args)
    seen = watch.scan(folder)  # files already in the folder at start are baselined, not run
    while True:
        time.sleep(POLL_S)
        current = watch.scan(folder)
        for path, signature in current.items():
            if seen.get(path) == signature:
                continue
            try:
                outcome = run(path, cfg, client, store)
            except RunLocked:
                continue  # not marked seen, so the next poll retries it
            seen[path] = signature
            _print_outcome(outcome)


def _approve(args) -> int:
    approver = approver_for(args.work_dir / TOKEN_FILE, interactive=_is_interactive())
    store = StateStore(args.work_dir / STATE_DB)
    payload = _stored_payload(store, args.run_id)
    metrics = load_metrics_file(args.metrics)
    if metrics.checksum != payload["metrics_checksum"]:
        raise DeliveryRefused("metrics file changed since the brief was built; nothing approved")
    csv = _latest_csv(args.folder)
    approved_at = datetime.now(UTC).strftime("%Y-%m-%d %H:%M UTC")
    outbox = Outbox(args.out_dir, store, Settings())
    with tempfile.TemporaryDirectory() as tmp:
        approval = _Approval(payload["report"], approver, approved_at, metrics, csv, Path(tmp))
        stamped = outbox.approve(args.run_id, approver, file_sha256(csv), approval.render_stamped, approval.recheck)
        outbox.send(stamped, LOCAL_SENT, DeliveryConfig(recipients={LOCAL_SENT}, max_bytes=Settings().max_bytes))
    return 0


def _parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(prog="weekly-analyst")
    sub = parser.add_subparsers(dest="command", required=True)

    def run_options(p: argparse.ArgumentParser) -> None:
        p.add_argument("--metrics", type=Path, required=True, help="metrics file (YAML or JSON)")
        p.add_argument("--model", type=Path, required=True, help="GGUF model path")
        p.add_argument("--work-dir", type=Path, default=Path("work"))
        p.add_argument("--out-dir", type=Path, default=Path("out"))
        p.add_argument("--recipient", action="append", default=[], help="allowlisted recipient (repeatable)")
        p.add_argument("--paper", choices=["A4", "Letter"], default="A4")

    p_watch = sub.add_parser("watch", help="run each new or replaced CSV in a folder")
    p_watch.add_argument("folder", type=Path)
    run_options(p_watch)
    p_watch.set_defaults(handler=_watch)

    p_run = sub.add_parser("run", help="run one CSV snapshot")
    p_run.add_argument("csv", type=Path)
    run_options(p_run)
    p_run.set_defaults(handler=_run)

    p_approve = sub.add_parser("approve", help="approve a pending brief (local user only)")
    p_approve.add_argument("run_id")
    p_approve.add_argument("--folder", type=Path, required=True, help="watched folder holding the latest CSV")
    p_approve.add_argument("--metrics", type=Path, required=True, help="metrics file the brief was built with")
    p_approve.add_argument("--work-dir", type=Path, default=Path("work"))
    p_approve.add_argument("--out-dir", type=Path, default=Path("out"))
    p_approve.set_defaults(handler=_approve)
    return parser


def main(argv: list[str] | None = None) -> int:
    args = _parser().parse_args(argv)
    try:
        return args.handler(args)
    except (ApproveRefused, DeliveryRefused, RunLocked) as exc:
        print(f"refused: {exc}", file=sys.stderr)
        return 1


if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
    raise SystemExit(main())