diff --git a/pipeline/__main__.py b/pipeline/__main__.py index 464e051..3d63569 100644 --- a/pipeline/__main__.py +++ b/pipeline/__main__.py @@ -197,6 +197,17 @@ async def _go() -> tuple[list[RawDocument], list[dict[str, Any]]]: return asyncio.run(_go()) +def _already_handled(ledger: Ledger) -> Callable[[str], bool]: + """Tell a paid source (Indian Kanoon) which documents not to download again. + + That's every document an earlier run settled, and also one whose case is still waiting + to publish: its record is carried over from data-staging, so downloading it again buys + nothing. While publishing was blocked (2026-10-12..14) we paid for the same 38 judgments + three days running. A document whose extraction failed is still retried. + """ + return lambda url: ledger.is_pending(url) or not ledger.should_process(url) + + def _fetch_article_bodies(docs: list[RawDocument]) -> list[RawDocument]: # pragma: no cover async def _go() -> list[RawDocument]: async with PoliteClient() as client: @@ -1137,9 +1148,8 @@ def run( if docs is not None: raw_docs, court_prefilter = docs, [] else: - settled = ledger raw_docs, court_prefilter = _fetch_documents( - run_date, court_only=is_refresh, seen=lambda url: not settled.should_process(url) + run_date, court_only=is_refresh, seen=_already_handled(ledger) ) report.source_alerts = source_alerts(court_prefilter) for alert in report.source_alerts: diff --git a/pipeline/tests/test_orchestrator.py b/pipeline/tests/test_orchestrator.py index 78f0d78..6cfc386 100644 --- a/pipeline/tests/test_orchestrator.py +++ b/pipeline/tests/test_orchestrator.py @@ -3060,3 +3060,19 @@ def test_refresh_keeps_cases_still_waiting_on_data_staging( ) cnrs = {r.get("cnr") for r in json.loads((main / "2026" / "TG.json").read_text())} assert {"C-EXISTING", "C-BRAND-NEW"} <= cnrs + + +def test_paid_source_skips_settled_and_waiting_documents_but_retries_failures() -> None: + """Indian Kanoon must not download a judgment again once it is settled, or while its case + is waiting to publish (carried over from data-staging); a failed one is still retried.""" + from pipeline.ledger import Ledger + + ledger = Ledger() + ledger.record("https://ex.invalid/settled", "published", "2026-10-12") + ledger.record("https://ex.invalid/waiting", "staged_pending", "2026-10-12") + ledger.record("https://ex.invalid/failed-once", "failed", "2026-10-12") + seen = orchestrator._already_handled(ledger) + assert seen("https://ex.invalid/settled") is True + assert seen("https://ex.invalid/waiting") is True + assert seen("https://ex.invalid/failed-once") is False + assert seen("https://ex.invalid/new") is False