monitor_queue.py

"""Snapshot the URL ingest queue, persist failures, print a status report.

Run periodically (e.g. once an hour) while a bulk queue is in flight.

Outputs (under coomer_viewer/queue_progress/):
  failed_urls.tsv    — full current set of failed URLs (overwritten each run)
  history.jsonl      — append-only one-line-per-run history (counts + deltas)
  last_snapshot.json — last raw task list, used to compute deltas next run
"""

from __future__ import annotations

import json
import sys
import time
import urllib.error
import urllib.request
from collections import Counter
from datetime import datetime, timezone
from pathlib import Path

TASKS_URL = "http://127.0.0.1:8090/api/ingest/url/tasks"
OUT_DIR = Path(__file__).parent / "queue_progress"


def fetch_tasks(timeout: float = 15.0) -> list[dict]:
    req = urllib.request.Request(TASKS_URL, headers={"Accept": "application/json"})
    with urllib.request.urlopen(req, timeout=timeout) as resp:
        return json.loads(resp.read().decode("utf-8"))


def write_failed_tsv(failed: list[dict], dest: Path) -> None:
    cols = ["task_id", "status", "service", "handler", "url", "queued_at", "completed_at", "elapsed_seconds", "error"]
    lines = ["\t".join(cols)]
    for t in failed:
        row = []
        for c in cols:
            v = t.get(c, "")
            if v is None:
                v = ""
            v = str(v).replace("\t", " ").replace("\n", " ").strip()
            if c == "error":
                v = v[:300]
            row.append(v)
        lines.append("\t".join(row))
    dest.write_text("\n".join(lines) + "\n", encoding="utf-8")


def main() -> int:
    OUT_DIR.mkdir(exist_ok=True)
    failed_path = OUT_DIR / "failed_urls.tsv"
    history_path = OUT_DIR / "history.jsonl"
    last_snapshot_path = OUT_DIR / "last_snapshot.json"

    try:
        tasks = fetch_tasks()
    except urllib.error.URLError as e:
        print(f"ERROR: cannot reach {TASKS_URL}: {e}", file=sys.stderr)
        return 2
    except Exception as e:  # noqa: BLE001
        print(f"ERROR: fetching tasks: {e}", file=sys.stderr)
        return 2

    now = time.time()
    now_iso = datetime.fromtimestamp(now, tz=timezone.utc).isoformat(timespec="seconds")

    by_status = Counter(t.get("status", "?") for t in tasks)
    by_handler = Counter(t.get("handler", "?") for t in tasks)

    failed = [t for t in tasks if t.get("status") == "failed"]
    write_failed_tsv(failed, failed_path)

    # delta vs previous snapshot
    prev_status: Counter[str] = Counter()
    prev_failed_ids: set[str] = set()
    if last_snapshot_path.exists():
        try:
            prev = json.loads(last_snapshot_path.read_text())
            prev_status = Counter(prev.get("by_status", {}))
            prev_failed_ids = set(prev.get("failed_task_ids", []))
        except Exception:
            pass

    new_failed = [t for t in failed if t.get("task_id") not in prev_failed_ids]

    delta = {
        s: int(by_status.get(s, 0)) - int(prev_status.get(s, 0))
        for s in set(by_status) | set(prev_status)
    }

    snapshot = {
        "ts": now,
        "ts_iso": now_iso,
        "total": len(tasks),
        "by_status": dict(by_status),
        "by_handler": dict(by_handler),
        "failed_task_ids": [t.get("task_id") for t in failed],
        "delta_vs_last": delta,
        "new_failed_count": len(new_failed),
    }

    last_snapshot_path.write_text(json.dumps(snapshot, indent=2), encoding="utf-8")
    with history_path.open("a", encoding="utf-8") as fh:
        fh.write(
            json.dumps({k: snapshot[k] for k in ("ts_iso", "total", "by_status", "delta_vs_last", "new_failed_count")})
            + "\n"
        )

    print(f"queue snapshot @ {now_iso}")
    print(f"  total tasks: {len(tasks)}")
    print("  by status:")
    for s in ("queued", "downloading", "completed", "failed"):
        n = by_status.get(s, 0)
        d = delta.get(s, 0)
        sign = f" ({d:+d})" if d != 0 else ""
        print(f"    {s:13s} {n:6d}{sign}")
    other = sum(v for k, v in by_status.items() if k not in ("queued", "downloading", "completed", "failed"))
    if other:
        print(f"    other        {other:6d}")
    print(f"  failed file: {failed_path} ({len(failed)} rows, {len(new_failed)} new since last run)")
    if new_failed:
        print("  new failures this run (first 10):")
        for t in new_failed[:10]:
            err = (t.get("error") or "").replace("\n", " ")[:120]
            print(f"    [{t.get('handler','?'):20s}] {t.get('url','')}\n        {err}")
    return 0


if __name__ == "__main__":
    sys.exit(main())