File size: 6,295 Bytes
1d9bd9b | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 | """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]
# An identity/reporter match is conclusive. A title-only match must be
# strong enough that a same-year namesake cannot silently supply a PDF.
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"]
|