| """Deterministic old/new Indian criminal-code section correspondences. |
| |
| The source dataset records verified direct correspondences from the new codes to |
| the repealed codes. This module builds the reverse index as well, without |
| turning a one-to-many correspondence into a claim of legal equivalence. |
| """ |
|
|
| from __future__ import annotations |
|
|
| import json |
| import re |
| from pathlib import Path |
| from typing import Any |
|
|
|
|
| ACT_NAMES = { |
| "IPC": "Indian Penal Code, 1860", |
| "BNS": "Bharatiya Nyaya Sanhita, 2023", |
| "CRPC": "Code of Criminal Procedure, 1973", |
| "BNSS": "Bharatiya Nagarik Suraksha Sanhita, 2023", |
| "IEA": "Indian Evidence Act, 1872", |
| "BSA": "Bharatiya Sakshya Adhiniyam, 2023", |
| } |
|
|
| ACT_ALIASES = { |
| "IPC": "IPC", |
| "INDIAN PENAL CODE": "IPC", |
| "BNS": "BNS", |
| "BHARATIYA NYAYA SANHITA": "BNS", |
| "CRPC": "CRPC", |
| "CRIMINAL PROCEDURE CODE": "CRPC", |
| "CODE OF CRIMINAL PROCEDURE": "CRPC", |
| "BNSS": "BNSS", |
| "BHARATIYA NAGARIK SURAKSHA SANHITA": "BNSS", |
| "IEA": "IEA", |
| "INDIAN EVIDENCE ACT": "IEA", |
| "BSA": "BSA", |
| "BHARATIYA SAKSHYA ADHINIYAM": "BSA", |
| } |
|
|
| PAIRS = { |
| "IPC": "BNS", |
| "BNS": "IPC", |
| "CRPC": "BNSS", |
| "BNSS": "CRPC", |
| "IEA": "BSA", |
| "BSA": "IEA", |
| } |
|
|
|
|
| def normalise_act(value: object) -> str: |
| text = str(value or "").upper() |
| text = re.sub(r"\bI\.?\s*P\.?\s*C\.?", "IPC", text) |
| text = re.sub(r"\bCR\.?\s*P\.?\s*C\.?", "CRPC", text) |
| text = re.sub(r"\b(?:18|19|20)\d{2}\b", " ", text) |
| text = re.sub(r"[^A-Z0-9]+", " ", text) |
| return ACT_ALIASES.get(re.sub(r"\s+", " ", text).strip(), "") |
|
|
|
|
| def normalise_section(value: object) -> str: |
| text = re.sub( |
| r"^\s*(?:sections?|secs?\.?|ss?\.?)\s*[-:]*\s*", |
| "", |
| str(value or ""), |
| flags=re.IGNORECASE, |
| ) |
| text = re.sub(r"\s+", "", text).upper().strip(".,;:") |
| return text if re.fullmatch(r"\d+[A-Z]*", text) else "" |
|
|
|
|
| class StatuteCrosswalk: |
| """Load and query a verified crosswalk in either direction.""" |
|
|
| def __init__(self, payload: dict[str, Any], *, source_path: Path | None = None): |
| mappings = payload.get("mappings") |
| if not isinstance(mappings, list): |
| raise ValueError("Crosswalk must contain a mappings list.") |
| self.schema_version = payload.get("schema_version") |
| self.description = str(payload.get("description") or "") |
| self.source = payload.get("source") if isinstance(payload.get("source"), dict) else {} |
| self.source_path = source_path |
| self._index: dict[tuple[str, str], list[dict[str, str]]] = {} |
| for item in mappings: |
| if not isinstance(item, dict): |
| continue |
| source = item.get("from") if isinstance(item.get("from"), dict) else {} |
| from_act = normalise_act(source.get("act")) |
| from_section = normalise_section(source.get("section")) |
| if not from_act or not from_section: |
| continue |
| for target in item.get("to") or []: |
| if not isinstance(target, dict): |
| continue |
| to_act = normalise_act(target.get("act")) |
| to_section = normalise_section(target.get("section")) |
| if not to_act or not to_section or PAIRS.get(from_act) != to_act: |
| continue |
| self._append(from_act, from_section, to_act, to_section) |
| self._append(to_act, to_section, from_act, from_section) |
|
|
| @classmethod |
| def from_file(cls, path: str | Path) -> "StatuteCrosswalk": |
| resolved = Path(path).expanduser().resolve() |
| return cls(json.loads(resolved.read_text(encoding="utf-8")), source_path=resolved) |
|
|
| def _append(self, from_act: str, from_section: str, to_act: str, to_section: str) -> None: |
| values = self._index.setdefault((from_act, from_section), []) |
| record = {"act": to_act, "section": to_section} |
| if record not in values: |
| values.append(record) |
|
|
| @property |
| def mapping_count(self) -> int: |
| return len(self._index) |
|
|
| def lookup(self, act: object, section: object) -> dict[str, Any]: |
| code = normalise_act(act) |
| number = normalise_section(section) |
| corresponding = [ |
| { |
| **item, |
| "act_name": ACT_NAMES[item["act"]], |
| "label": f'{item["act"]} section {item["section"]}', |
| } |
| for item in self._index.get((code, number), []) |
| ] |
| source_label = f"{code} {number}".strip() |
| target_label = ", ".join(f'{item["act"]} {item["section"]}' for item in corresponding) |
| table_key = "_".join(sorted((code, PAIRS.get(code, "")))) |
| source_tables = self.source.get("tables") if isinstance(self.source.get("tables"), dict) else {} |
| source_url = next( |
| ( |
| value |
| for key, value in source_tables.items() |
| if set(key.split("_")) == {code, PAIRS.get(code, "")} |
| ), |
| None, |
| ) |
| return { |
| "found": bool(corresponding), |
| "from": source_label, |
| "to": target_label or None, |
| "query": { |
| "act": code, |
| "act_name": ACT_NAMES.get(code), |
| "section": number, |
| "label": f"{code} section {number}" if code and number else source_label, |
| }, |
| "corresponding": corresponding, |
| "one_to_many": len(corresponding) > 1, |
| "direction": f"{code.lower()}_to_{PAIRS.get(code, '').lower()}" if code else None, |
| "source": { |
| "publisher": self.source.get("publisher"), |
| "url": source_url, |
| "table": table_key, |
| }, |
| "notice": ( |
| "These are direct statutory correspondences from the published crosswalk. " |
| "They are not a finding that the provisions are legally identical; compare the text and case law." |
| ), |
| } |
|
|
|
|
| def default_crosswalk_path() -> Path: |
| return Path(__file__).resolve().parents[1] / "section_crosswalk.json" |
|
|
|
|
| def load_default_crosswalk(path: str | Path | None = None) -> StatuteCrosswalk: |
| return StatuteCrosswalk.from_file(path or default_crosswalk_path()) |
|
|
