load_durable_queue.py

"""Bulk-load verified missing URLs into the durable ingest queue store
(ingest_queue.db) as status='queued'.

This is the persistence-first path: instead of POSTing to the volatile
in-memory API queue (lost on restart), we write straight to the SQLite
store the patched server recovers from on boot. On the next server start
`ingest_persist.load_recoverable()` re-enqueues every 'queued' row.

Ordering: phase-1 (creators) get earlier `queued_at` than phase-2
(galleries/videos), and the durable recovery loads ascending by
`queued_at` — so creators are processed first across the whole backlog,
regardless of how many workers run.

Dedup: skips any URL already present in the store in a blocking state
(queued/downloading/completed, plus failed unless INGEST_RETRY_FAILED=1)
AND any URL already represented in the coomer DB have-set / source_url /
known-creator sets (same rules as dedup_and_queue_tabs.py). Idempotent —
safe to re-run as agents append verified slices.

Usage:
  .venv/bin/python load_durable_queue.py --phase1 /tmp/q/phase1_creators.txt \\
      --phase2 /tmp/q/agent*_verified.txt            # load
  .venv/bin/python load_durable_queue.py --phase1 ... --phase2 ... --dry-run
"""
from __future__ import annotations

import argparse
import glob
import sys
import time
import uuid
from pathlib import Path

sys.path.insert(0, str(Path(__file__).parent))

import ingest_persist
from gap_analysis import PG_DSN
from gap_analysis.core import load_have_set, normalize_slug
from gap_analysis.known import load_known_creators
from ingest_router import detect_route
from dedup_and_queue_tabs import (
    is_duplicate,
    load_source_urls,
    load_creator_slugs_by_service,
)


_HR_SLUG_RE = __import__("re").compile(r"hentairead\.com/hentai/([^/\s?#]+)", __import__("re").I)


def load_hentairead_have_slugs(pg_dsn: str) -> set[str]:
    """hentairead posts store NO source_url; the real source is embedded in
    `content` as `Source: https://hentairead.com/hentai/<slug>/`. The generic
    source_url dedup is therefore blind to ~all of them (Agent 1 found 98.8%
    of a 'missing' slice was already present). Build the true have-set from
    the content field."""
    import psycopg2
    out: set[str] = set()
    conn = psycopg2.connect(pg_dsn)
    try:
        with conn.cursor() as cur:
            cur.execute(
                "SELECT content FROM posts p JOIN creators c ON c.id=p.creator_id "
                "WHERE c.service='hentairead' AND content ~* 'hentairead\\.com/hentai/'"
            )
            for (content,) in cur:
                if not content:
                    continue
                m = _HR_SLUG_RE.search(content)
                if m:
                    out.add(m.group(1).lower())
    finally:
        conn.close()
    return out


def _read(path_globs: list[str]) -> list[str]:
    seen: set[str] = set()
    out: list[str] = []
    for g in path_globs:
        for p in sorted(glob.glob(g)):
            for line in Path(p).read_text(encoding="utf-8", errors="replace").splitlines():
                u = line.strip()
                if u.lower().startswith(("http://", "https://")) and u not in seen:
                    seen.add(u)
                    out.append(u)
    return out


def main() -> int:
    ap = argparse.ArgumentParser()
    ap.add_argument("--phase1", nargs="*", default=[], help="creator URL list(s) — queued first")
    ap.add_argument("--phase2", nargs="*", default=[], help="gallery/video URL list(s)")
    ap.add_argument("--dry-run", action="store_true")
    args = ap.parse_args()

    ingest_persist.init_db()

    print("[1/3] loading DB dedup sets", file=sys.stderr)
    nh_have = load_have_set(PG_DSN)
    here = Path(__file__).parent
    nh_known = load_known_creators(str(here / "tabs-organized.txt"), PG_DSN)
    src_urls = load_source_urls(PG_DSN)
    creators_by_svc = load_creator_slugs_by_service(PG_DSN)
    hr_have = load_hentairead_have_slugs(PG_DSN)
    print(f"      have={len(nh_have)} known={len(nh_known)} "
          f"source_urls={len(src_urls)} svc={len(creators_by_svc)} "
          f"hentairead_slugs={len(hr_have)}", file=sys.stderr)

    phase1 = _read(args.phase1)
    phase2 = _read(args.phase2)
    # A URL queued as a creator must not also be queued as a phase-2 item.
    p1set = set(phase1)
    phase2 = [u for u in phase2 if u not in p1set]
    print(f"[2/3] phase1={len(phase1)} phase2={len(phase2)}", file=sys.stderr)

    base = time.time()
    queued = skipped_db = skipped_store = skipped_hr = 0
    plan: list[tuple[str, float]] = []

    for idx, u in enumerate(phase1):
        plan.append((u, base + idx * 1e-3))
    p2base = base + len(phase1) * 1e-3 + 1.0
    for idx, u in enumerate(phase2):
        plan.append((u, p2base + idx * 1e-3))

    for u, qa in plan:
        route = detect_route(u)
        if route is None:
            continue
        dup, _reason = is_duplicate(u, nh_have, nh_known, src_urls, creators_by_svc)
        if dup:
            skipped_db += 1
            continue
        # hentairead galleries: source_url is always empty; dedup on the
        # slug embedded in posts.content (the generic check can't see these).
        if route.handler == "hentairead_gallery":
            _m = _HR_SLUG_RE.search(u)
            if _m and _m.group(1).lower() in hr_have:
                skipped_hr += 1
                continue
        if ingest_persist.find_duplicate(route.handler, route.content_id, u):
            skipped_store += 1
            continue
        if args.dry_run:
            queued += 1
            continue
        task = type("T", (), {})()
        task.task_id = str(uuid.uuid4())
        task.url = u
        task.handler = route.handler
        task.service = route.service
        task.content_id = route.content_id
        task.status = "queued"
        task.queued_at = qa
        task.started_at = None
        task.completed_at = None
        task.error = None
        task.result = None
        ingest_persist.record(task)
        queued += 1

    print(f"[3/3] queued={queued} skipped_db={skipped_db} "
          f"skipped_hentairead={skipped_hr} skipped_store={skipped_store}"
          f"{'  (DRY RUN)' if args.dry_run else ''}",
          file=sys.stderr)
    print("durable store stats:", ingest_persist.stats(), file=sys.stderr)
    return 0


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