| """Bharat Courts adapter for Supreme Court PDF recovery. |
| |
| The fast path in :mod:`pdf_sources` opens a verified individual object from the |
| public SCI AWS archive. Some otherwise valid judgments are absent from that |
| object map. Bharat Courts can resolve the same archive metadata and extract the |
| PDF from the official per-year tar bundle, so this module is deliberately used |
| only as the slower fallback. |
| """ |
|
|
| from __future__ import annotations |
|
|
| import asyncio |
| from difflib import SequenceMatcher |
| import os |
| import re |
| from typing import Any, Iterable |
|
|
|
|
| class BharatCourtsPdfError(RuntimeError): |
| pass |
|
|
|
|
| _CLIENT: Any | None = None |
| _CLIENT_LOCK: asyncio.Lock | None = None |
| _YEAR_LOCKS: dict[int, asyncio.Lock] = {} |
| _YEAR_ROWS: dict[int, list[Any]] = {} |
|
|
|
|
| def _norm(value: object) -> str: |
| return re.sub(r"[^a-z0-9]+", " ", str(value or "").lower()).strip() |
|
|
|
|
| def _year_from(*values: object) -> int | None: |
| for value in values: |
| match = re.search(r"\b(19|20)\d{2}\b", str(value or "")) |
| if match: |
| return int(match.group(0)) |
| return None |
|
|
|
|
| def _cache_bytes() -> int: |
| try: |
| gib = max(1.0, float(os.environ.get("THEMIS_BHARAT_CACHE_GB", "5"))) |
| except ValueError: |
| gib = 5.0 |
| return int(gib * 1024**3) |
|
|
|
|
| async def _client(): |
| global _CLIENT, _CLIENT_LOCK |
| if _CLIENT is not None: |
| return _CLIENT |
| if _CLIENT_LOCK is None: |
| _CLIENT_LOCK = asyncio.Lock() |
| async with _CLIENT_LOCK: |
| if _CLIENT is None: |
| try: |
| from bharat_courts import ArchiveClient |
| except ImportError as exc: |
| raise BharatCourtsPdfError( |
| "Bharat Courts archive support is not installed" |
| ) from exc |
| _CLIENT = ArchiveClient( |
| cache_dir=os.environ.get("THEMIS_PDF_CACHE", "/tmp/pdf_cache"), |
| cache_max_bytes=_cache_bytes(), |
| metadata_cache=False, |
| ) |
| return _CLIENT |
|
|
|
|
| async def _year_judgments(year: int) -> list[Any]: |
| if year in _YEAR_ROWS: |
| return _YEAR_ROWS[year] |
| lock = _YEAR_LOCKS.setdefault(year, asyncio.Lock()) |
| async with lock: |
| if year in _YEAR_ROWS: |
| return _YEAR_ROWS[year] |
| client = await _client() |
| rows = [] |
| async for judgment in client.iter_judgments( |
| court="sci", year=year, batch_size=500, max_results=5000 |
| ): |
| rows.append(judgment) |
| _YEAR_ROWS[year] = rows |
| return rows |
|
|
|
|
| def _best_match( |
| rows: Iterable[Any], |
| *, |
| case_name: str, |
| neutral_citation: str, |
| equivalent_citations: Iterable[object], |
| decision_date: str, |
| ) -> Any | None: |
| neutral = _norm(neutral_citation) |
| equivalents = {_norm(value) for value in equivalent_citations if _norm(value)} |
| title = _norm(case_name) |
| wanted_date = str(decision_date or "")[:10] |
| ranked = [] |
| for row in rows: |
| case_id = _norm(getattr(row, "case_id", "")) |
| citation = _norm(getattr(row, "citation", "")) |
| row_title = _norm(getattr(row, "title", "")) |
| score = 0.0 |
| exact_identity = bool(neutral and case_id == neutral) |
| exact_reporter = bool(citation and citation in equivalents) |
| if exact_identity: |
| score += 200.0 |
| if exact_reporter: |
| score += 170.0 |
| title_ratio = SequenceMatcher(None, title, row_title).ratio() if title and row_title else 0.0 |
| score += 100.0 * title_ratio |
| row_date = str(getattr(row, "decision_date", "") or "")[:10] |
| if wanted_date and row_date == wanted_date: |
| score += 25.0 |
| if getattr(row, "pdf_path", None): |
| score += 5.0 |
| ranked.append((score, exact_identity or exact_reporter, title_ratio, row)) |
| if not ranked: |
| return None |
| ranked.sort(key=lambda item: item[0], reverse=True) |
| score, exact, title_ratio, row = ranked[0] |
| |
| |
| return row if exact or (title_ratio >= 0.78 and score >= 88.0) else None |
|
|
|
|
| async def resolve_and_fetch_pdf( |
| *, |
| year: int | str | None, |
| path: str | None, |
| case_name: str, |
| neutral_citation: str, |
| equivalent_citations: Iterable[object], |
| decision_date: str, |
| ) -> tuple[bytes, dict[str, Any]]: |
| """Resolve one SCI judgment and return verified PDF bytes plus provenance.""" |
| resolved_year = int(year) if str(year or "").isdigit() else _year_from( |
| decision_date, neutral_citation, *equivalent_citations |
| ) |
| if not resolved_year: |
| raise BharatCourtsPdfError("judgment year is unavailable") |
| client = await _client() |
| judgment = None |
| if path: |
| try: |
| from bharat_courts import Judgment, SUPREME_COURT |
| except ImportError as exc: |
| raise BharatCourtsPdfError( |
| "Bharat Courts archive support is not installed" |
| ) from exc |
| judgment = Judgment( |
| case_id=neutral_citation or None, |
| title=case_name or None, |
| court=SUPREME_COURT, |
| pdf_path=str(path), |
| source="archive", |
| year=resolved_year, |
| ) |
| else: |
| rows = await _year_judgments(resolved_year) |
| judgment = _best_match( |
| rows, |
| case_name=case_name, |
| neutral_citation=neutral_citation, |
| equivalent_citations=equivalent_citations, |
| decision_date=decision_date, |
| ) |
| if judgment is None or not getattr(judgment, "pdf_path", None): |
| raise BharatCourtsPdfError("no unambiguous Bharat Courts PDF match") |
| try: |
| data = await client.fetch_pdf(judgment, language="english") |
| except Exception as exc: |
| raise BharatCourtsPdfError(str(exc)[:300]) from exc |
| if not data.startswith(b"%PDF"): |
| raise BharatCourtsPdfError("archive returned a non-PDF payload") |
| return data, { |
| "provider": "bharat_courts", |
| "source_name": "Bharat Courts public archive", |
| "case_id": getattr(judgment, "case_id", None), |
| "year": resolved_year, |
| "path": getattr(judgment, "pdf_path", None), |
| } |
|
|
|
|
| __all__ = ["BharatCourtsPdfError", "resolve_and_fetch_pdf"] |
|
|
