""" SCR (Supreme Court Reports) Judgment Scraper scraper with complete metadata extraction for RAG pipelines. """ import os import sys import requests from bs4 import BeautifulSoup import json import re import logging import random import time from pathlib import Path from datetime import datetime from dataclasses import dataclass, field, asdict from typing import Optional # --------------------------------------------------------------------------- # Logging # --------------------------------------------------------------------------- Path("data").mkdir(exist_ok=True) logging.basicConfig( level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s", handlers=[ logging.StreamHandler(), logging.FileHandler("data/scraper.log", encoding="utf-8"), ], ) log = logging.getLogger(__name__) # --------------------------------------------------------------------------- # Data model # --------------------------------------------------------------------------- @dataclass class CaseCited: name: str citation: str treatment: str # "relied on" | "referred to" | "overruled" | "distinguished" @dataclass class SectionRef: act: str provision: str # e.g. "s.61(2)", "r.22" number: str # e.g. "61(2)", "22" @dataclass class JudgmentMetadata: # --- identifiers --- case_name: str = "" appeal_no: str = "" citation: str = "" neutral_citation: str = "" # --- court info --- court: str = "Supreme Court" lower_court: str = "" jurisdiction: str = "India" state: Optional[str] = None # --- date --- date: Optional[str] = None # ISO-8601 YYYY-MM-DD # --- bench --- bench: list = field(default_factory=list) # ["Sanjay Kumar, J", ...] author_judge: str = "" # --- outcome --- outcome: str = "" case_type: str = "" # --- statutes --- acts: list = field(default_factory=list) sections: list = field(default_factory=list) # --- case law --- cases_cited: list = field(default_factory=list) # --- text fields --- keywords: list = field(default_factory=list) issue: str = "" short_summary: str = "" # headnote-derived, for metadata filtering full_headnote: str = "" # full headnote text, used as RAG chunk content # --- source --- source_url: str = "" scraped_at: str = field(default_factory=lambda: datetime.utcnow().isoformat()) pdf_path: str = "" # --------------------------------------------------------------------------- # Constants # --------------------------------------------------------------------------- MONTHS = { "jan": 1, "feb": 2, "mar": 3, "apr": 4, "may": 5, "jun": 6, "jul": 7, "aug": 8, "sep": 9, "oct": 10, "nov": 11, "dec": 12, "january": 1, "february": 2, "march": 3, "april": 4, "june": 6, "july": 7, "august": 8, "september": 9, "october": 10, "november": 11, "december": 12, } # Canonical act names (keyed by lowercase fragment). # All variants of the same act must map to the SAME canonical string — # this prevents duplicate section entries under different name spellings. ACT_ALIASES = { "insolvency and bankruptcy code": "Insolvency and Bankruptcy Code, 2016", "ibc": "Insolvency and Bankruptcy Code, 2016", # Both the full name and short name resolve to the same canonical string "national company law appellate tribunal rules": "NCLAT Rules, 2016", "nclat rules": "NCLAT Rules, 2016", "nclat rule": "NCLAT Rules, 2016", "constitution of india": "Constitution of India", "code of criminal procedure": "Code of Criminal Procedure, 1973", "crpc": "Code of Criminal Procedure, 1973", "indian penal code": "Indian Penal Code, 1860", "ipc": "Indian Penal Code, 1860", "civil procedure code": "Code of Civil Procedure, 1908", "cpc": "Code of Civil Procedure, 1908", "arbitration and conciliation": "Arbitration and Conciliation Act, 1996", "right of children": "Right of Children to Free and Compulsory Education Act, 2009", "companies act": "Companies Act, 2013", } # Acts list normaliser — applied after scraping meta.acts so that # "National Company Law Appellate Tribunal Rules, 2016" and # "NCLAT Rules, 2016" are unified before section attribution. def _normalise_acts(acts: list[str]) -> list[str]: """Deduplicate acts list by resolving all entries through ACT_ALIASES.""" seen: set[str] = set() result: list[str] = [] for act in acts: canonical = act # default: keep as-is act_lower = act.lower() for key, canon in ACT_ALIASES.items(): if key in act_lower: canonical = canon break if canonical not in seen: seen.add(canonical) result.append(canonical) return result # Prefix type → which kind of act it belongs to # "section" prefixes → statutory acts (IBC, IPC, CPC…) # "rule" prefixes → rules/regulations (NCLAT Rules, etc.) SECTION_PREFIXES = {"s", "ss", "sec", "section"} RULE_PREFIXES = {"r", "rr", "rule"} # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def parse_date(raw: str) -> Optional[str]: """Return ISO-8601 date or None.""" if not raw: return None raw = raw.strip() for fmt in ("%d %B %Y", "%d %b %Y", "%d-%m-%Y", "%Y-%m-%d"): try: return datetime.strptime(raw, fmt).strftime("%Y-%m-%d") except ValueError: pass m = re.search(r"(\d{1,2})\s+([A-Za-z]+)\s+(\d{4})", raw) if m: day, mon, year = m.groups() month_num = MONTHS.get(mon.lower()) if month_num: return f"{int(year):04d}-{month_num:02d}-{int(day):02d}" return None def clean_text(text: str) -> str: return re.sub(r"\s+", " ", text).strip() def _resolve_act(prefix: str, sentence_act: Optional[str], known_acts: list[str]) -> str: """ Determine the most appropriate act for a provision reference. Priority: sentence-level act mention > prefix-type hint > first known act. """ if sentence_act: return sentence_act p = prefix.lower().rstrip(".") if p in RULE_PREFIXES: # Find the first rules/regulations act in known_acts for a in known_acts: if "rules" in a.lower() or "regulations" in a.lower(): return a return "NCLAT Rules, 2016" # safe default for SCR judgments if p in SECTION_PREFIXES: # Find the first non-rules act in known_acts for a in known_acts: if "rules" not in a.lower() and "regulations" not in a.lower(): return a return known_acts[0] if known_acts else "Unknown Act" return known_acts[0] if known_acts else "Unknown Act" # --------------------------------------------------------------------------- # Core parser # --------------------------------------------------------------------------- def parse_judgment_html(html_content: str, source_url: str = "") -> dict: """ Parse a judgment HTML fragment (from the SCR splitview endpoint) and return a fully-populated JudgmentMetadata dict. """ soup = BeautifulSoup(html_content, "html.parser") meta = JudgmentMetadata(source_url=source_url) full_text = soup.get_text(separator="\n") # ------------------------------------------------------------------ # 1. Case name (FIX: was returning file path slug) # ------------------------------------------------------------------ # Try known CSS classes first for cls in ["Case-Title", "CaseTitle", "case-title", "Parties", "Party-Name"]: el = soup.find(class_=cls) if el: meta.case_name = clean_text(el.get_text()) break # Fallback: extract "Appellant v. Respondent" pattern from full text if not meta.case_name: m = re.search( r"([A-Z][A-Za-z\s,\.&]+)\s+[vV][sS]?\.?\s+([A-Z][A-Za-z\s,\.&]+)", full_text, ) if m: meta.case_name = clean_text(m.group(0)) # Fallback: use citation-derived name (strip underscores, page range) if not meta.case_name and source_url: path_part = source_url.split("path=")[-1] # e.g. "2026_5_577_583" → not a useful name, skip if not re.match(r"^\d{4}_\d+_\d+", path_part): meta.case_name = path_part.replace("_", " ").strip() # ------------------------------------------------------------------ # 2. Appeal / case number (FIX: broader regex patterns) # ------------------------------------------------------------------ for cls in ["Appeal-No", "AppealNo", "CaseNo", "case-no", "Appeal-Number"]: el = soup.find(class_=cls) if el: meta.appeal_no = clean_text(el.get_text()) break if not meta.appeal_no: # Covers all variants: # "Civil Appeal No. 7458 of 2026" # "Civil Appeal No(s). 14439-14440 of 2025" # "Criminal Appeal Nos. 123-124 of 2025" # "Special Leave Petition No. 456 of 2024" m = re.search( r"((?:Civil|Criminal|Special Leave|Writ)\s+(?:Appeal|Petition)\s+" r"No(?:s|\(s\))?\.?\s*[\d\-]+(?:\s*(?:and|&)\s*[\d\-]+)?\s*of\s*\d{4})", full_text, re.I, ) if m: meta.appeal_no = clean_text(m.group(1)) # ------------------------------------------------------------------ # 3. Citation + neutral citation # ------------------------------------------------------------------ cit_el = soup.find(class_="Citation") if cit_el: raw_cit = clean_text(cit_el.get_text(separator=" ")) meta.citation = raw_cit nc_m = re.search(r"\d{4}\s+INSC\s+\d+", raw_cit) if nc_m: meta.neutral_citation = nc_m.group(0) # ------------------------------------------------------------------ # 4. Date # ------------------------------------------------------------------ date_el = soup.find(class_="Date-of-Decision") meta.date = parse_date(date_el.get_text(strip=True) if date_el else "") # ------------------------------------------------------------------ # 5. Bench & author judge (FIX: "JJ." suffix leaking as a judge name) # ------------------------------------------------------------------ coram_el = soup.find(class_="Coram") if coram_el: raw_bench = clean_text(coram_el.get_text()).strip("[]") # Remove trailing "JJ." / "J." designations before splitting # e.g. "Sanjay Kumar* and K. Vinod Chandran, JJ." raw_bench = re.sub(r",?\s*JJ?\.$", "", raw_bench, flags=re.I).strip() # Split on " and " or ", " judges_raw = re.split(r"\s+and\s+|,\s*", raw_bench, flags=re.I) judges_raw = [j.strip() for j in judges_raw if j.strip()] bench_clean = [] for j in judges_raw: is_author = "*" in j name = j.replace("*", "").strip() if not name: continue # Append ", J" suffix if not already present if not re.search(r",?\s*J\.?$", name, re.I): name = name + ", J" bench_clean.append(name) if is_author and not meta.author_judge: meta.author_judge = name meta.bench = bench_clean # If author not marked by asterisk, use first judge as author if not meta.author_judge and bench_clean: meta.author_judge = bench_clean[0] # ------------------------------------------------------------------ # 6. Acts — normalised to canonical names to prevent duplicates # ------------------------------------------------------------------ acts_el = soup.find(class_="Acts") if acts_el: raw_acts = acts_el.get_text(strip=True) raw_list = [a.strip().rstrip(".") for a in raw_acts.split(";") if a.strip()] meta.acts = _normalise_acts(raw_list) # ------------------------------------------------------------------ # 7. Sections (structured, act-aware) (FIX: wrong act attribution) # ------------------------------------------------------------------ meta.sections = _extract_sections_structured(soup, meta.acts) # ------------------------------------------------------------------ # 8. Lower court # ------------------------------------------------------------------ m = re.search( r"From the (?:Judgment and )?Order dated[^o]+of the\s+(.+?)\s+in\s+", full_text, re.I | re.S, ) if m: meta.lower_court = clean_text(m.group(1)) elif "NCLAT" in full_text: nm = re.search(r"(National Company Law Appellate Tribunal[^,\n]*)", full_text) if nm: meta.lower_court = clean_text(nm.group(1)) elif "High Court" in full_text: hm = re.search(r"(High Court of[^,\n]+)", full_text) if hm: meta.lower_court = clean_text(hm.group(1)) # ------------------------------------------------------------------ # 9. Case type # ------------------------------------------------------------------ acts_lower = " ".join(meta.acts).lower() if "insolvency" in acts_lower or "ibc" in acts_lower: meta.case_type = "Insolvency / IBC" elif "constitution" in acts_lower: meta.case_type = "Constitutional" elif "criminal procedure" in acts_lower or "ipc" in acts_lower: meta.case_type = "Criminal" elif "civil procedure" in acts_lower: meta.case_type = "Civil" elif "arbitration" in acts_lower: meta.case_type = "Arbitration" elif "tax" in acts_lower or "income" in acts_lower: meta.case_type = "Tax" elif "labour" in acts_lower or "industrial" in acts_lower: meta.case_type = "Labour" # ------------------------------------------------------------------ # 10. Outcome # ------------------------------------------------------------------ result_el = soup.find(class_="Result") if result_el: meta.outcome = clean_text(result_el.get_text()) else: tail = full_text[-600:] for phrase in [ "appeals allowed", "appeal allowed", "appeals dismissed", "appeal dismissed", "petition allowed", "petition dismissed", "partly allowed", "disposed of", "remanded back", "set aside", ]: if phrase in tail.lower(): meta.outcome = phrase.title() break # ------------------------------------------------------------------ # 11. Cases cited (FIX: citations were empty) # ------------------------------------------------------------------ meta.cases_cited = _extract_cases_cited(soup) # ------------------------------------------------------------------ # 12. Keywords # ------------------------------------------------------------------ kw_el = soup.find(class_="Keywords") if kw_el: raw_kw = kw_el.get_text(strip=True) meta.keywords = [k.strip().rstrip(".") for k in raw_kw.split(";") if k.strip()] # ------------------------------------------------------------------ # 13. Issue + headnote + short summary (FIX: summary was same as issue) # ------------------------------------------------------------------ issue_el = soup.find(class_="Issues-for-Consideration") if issue_el: meta.issue = clean_text(issue_el.get_text()) headnote_els = soup.find_all(class_="Headnote") meta.full_headnote = " ".join( clean_text(h.get_text(separator=" ")) for h in headnote_els ) # short_summary = first 2 sentences of headnote "Held:" portion meta.short_summary = _make_short_summary(meta.full_headnote, meta.issue) return asdict(meta) # --------------------------------------------------------------------------- # Section extraction (FIX: s.61(2) was being attributed to NCLAT Rules) # --------------------------------------------------------------------------- _PROVISION_RE = re.compile( r"\b(section|sec|ss?|rule|rr?)\s*\.?\s*([0-9]+(?:\([0-9A-Za-z]+\))?[A-Za-z]?)", re.I, ) _ACT_MENTION_RE = re.compile( r"(insolvency\s+and\s+bankruptcy\s+code" r"|nclat\s+rules?" r"|national\s+company\s+law\s+appellate\s+tribunal\s+rules?" r"|constitution\s+of\s+india" r"|code\s+of\s+criminal\s+procedure" r"|indian\s+penal\s+code" r"|civil\s+procedure\s+code" r"|arbitration\s+and\s+conciliation)", re.I, ) def _extract_sections_structured(soup: BeautifulSoup, known_acts: list[str]) -> list[dict]: """ Extract section/rule references with correct act attribution. Strategy: - Scan headnote + keywords text sentence by sentence. - If a sentence explicitly names an act, attribute all provisions in that sentence to that act. - Otherwise use prefix type (s/sec → statutory act, r/rule → rules act) to pick the right act from known_acts. - Deduplicate by (act, number) pair. """ search_els = ( soup.find_all(class_="Headnote") + soup.find_all(class_="Keywords") + soup.find_all(class_="Judgment-Body") ) text = " ".join(el.get_text(separator=" ") for el in search_els) if search_els else soup.get_text() sentences = re.split(r"[.;–]\s+", text) seen: set[tuple] = set() results: list[dict] = [] for sentence in sentences: # Resolve act context for this sentence act_m = _ACT_MENTION_RE.search(sentence) sentence_act: Optional[str] = None if act_m: alias_key = re.sub(r"\s+", " ", act_m.group(0).lower()) for key, canonical in ACT_ALIASES.items(): if key in alias_key: sentence_act = canonical break for m in _PROVISION_RE.finditer(sentence): prefix, number = m.group(1), m.group(2) act = _resolve_act(prefix, sentence_act, known_acts) provision = f"{prefix.lower().rstrip('.')}.{number}" key = (act, number) if key not in seen: seen.add(key) results.append(asdict(SectionRef(act=act, provision=provision, number=number))) return results # --------------------------------------------------------------------------- # Cases cited (FIX: citation regex wasn't matching inline SCR citations) # --------------------------------------------------------------------------- _TREATMENT_RE = re.compile( r"\b(relied\s+on|referred\s+to|overruled|distinguished|followed|approved|dissented)\b", re.I, ) # Matches: (2022) 2 SCC 244 | [2021] 14 SCR 736 | 2026 INSC 479 _CITATION_RE = re.compile( r"(?:\((\d{4})\)\s*\d+\s+SCC\s+\d+" r"|\[(\d{4})\]\s*\d+\s+SCR\s+\d+" r"|\d{4}\s+INSC\s+\d+)", re.I, ) # Matches the full citation string for capture _CITATION_FULL_RE = re.compile( r"(?:\(\d{4}\)\s*\d+\s+SCC\s+\d+" r"|\[\d{4}\]\s*\d+\s+SCR\s+\d+" r"|\d{4}\s+INSC\s+\d+)", re.I, ) def _extract_cases_cited(soup: BeautifulSoup) -> list[dict]: """ Parse the 'Case Law Cited' section. Handles bold/italic case names followed by citations and treatment labels. """ results: list[dict] = [] seen: set[str] = set() # Strategy 1: find by CSS class case_law_el = soup.find(class_=re.compile(r"Case.?Law|CaseLaw|Cases.?Cited", re.I)) if case_law_el: raw_text = case_law_el.get_text(separator="\n") else: # Strategy 2: heuristic heading search full = soup.get_text(separator="\n") m = re.search( r"Case Law Cited\s*\n(.*?)(?:\n(?:List of Acts|List of Keywords|Appearances|Judgment)\s*\n|\Z)", full, re.S | re.I, ) raw_text = m.group(1) if m else "" if not raw_text: return results # Each case entry is typically on 1-2 lines; split on blank lines or clear separators blocks = re.split(r"\n{2,}|\n(?=[A-Z])", raw_text.strip()) for block in blocks: block = clean_text(block) if not block or len(block) < 10: continue # Skip section/list headers if re.match(r"^(List of|Case Law|Appearances|Judgment)", block, re.I): continue # Find all citations in the block all_citations = _CITATION_FULL_RE.findall(block) citation_str = " : ".join(all_citations) if all_citations else "" # Treatment treatment_m = _TREATMENT_RE.search(block) treatment = clean_text(treatment_m.group(0)).lower() if treatment_m else "cited" # Case name: text before the first citation, or before " – relied on" etc. name = block first_cit_m = _CITATION_FULL_RE.search(block) if first_cit_m: name = block[: first_cit_m.start()] # Also cut at treatment label treatment_pos = _TREATMENT_RE.search(name) if treatment_pos: name = name[: treatment_pos.start()] # Clean up trailing punctuation / dashes / SCR refs name = re.sub(r"\s*[–\-:]+\s*$", "", name) name = re.sub(r"\s*\[?\d{4}\]?\s*\d*\s*S[CR]{2}.*", "", name) name = clean_text(name) if not name or name in seen or len(name) < 5: continue seen.add(name) results.append(asdict(CaseCited(name=name, citation=citation_str, treatment=treatment))) return results # --------------------------------------------------------------------------- # Short summary (FIX: was identical to issue; now derived from headnote) # --------------------------------------------------------------------------- def _make_short_summary(headnote: str, issue: str) -> str: """ Extract a 2-3 sentence summary from the 'Held:' portion of the headnote. Falls back to the first 2 sentences of the headnote, then to the issue. """ if headnote: # Prefer the "Held:" conclusion held_m = re.search(r"\bHeld\s*:\s*(.+?)(?:\[Para|\Z)", headnote, re.S | re.I) base = held_m.group(1) if held_m else headnote sentences = re.split(r"(?<=[.!?])\s+–?\s*", base.strip()) summary = " ".join(sentences[:3]).strip() if len(summary) > 500: summary = summary[:500].rsplit(" ", 1)[0] + "…" if summary: return summary # Final fallback: issue return issue[:400] + "…" if len(issue) > 400 else issue # --------------------------------------------------------------------------- # HTTP layer # --------------------------------------------------------------------------- BASE_URL = "https://scr.sci.gov.in/scrsearch/" HEADERS = { "User-Agent": ( "Mozilla/5.0 (Windows NT 10.0; Win64; x64) " "AppleWebKit/537.36 (KHTML, like Gecko) " "Chrome/124.0.0.0 Safari/537.36" ), "Accept-Language": "en-US,en;q=0.9", } def build_session() -> requests.Session: s = requests.Session() s.headers.update(HEADERS) return s def fetch_homepage(session: requests.Session) -> BeautifulSoup: log.info("Fetching SCR homepage...") r = session.get(BASE_URL, timeout=30) r.raise_for_status() return BeautifulSoup(r.text, "html.parser") def download_captcha(session: requests.Session, soup: BeautifulSoup, out_path: Path) -> None: captcha_img = soup.find(id="captcha_image") if not captcha_img: raise RuntimeError("CAPTCHA image element not found on homepage.") url = f"https://scr.sci.gov.in{captcha_img.get('src', '')}" r = session.get(url, timeout=15) r.raise_for_status() out_path.write_bytes(r.content) log.info(f"CAPTCHA saved → {out_path.resolve()}") def verify_captcha(session: requests.Session, captcha_code: str, search_text: str) -> str: payload = { "captcha": captcha_code, "search_text": search_text, "search_opt": "PHRASE", "escr_flag": "", "proximity": "", "sel_lang": "", "neu_cit_year": "", "neu_no": "", "ncn": "", "citation_vol": "", "citation_year": "", "citation_supl": "", "citation_page": "", "ajax_req": "true", "app_token": "", } r = session.post( f"{BASE_URL}?p=pdf_search/checkCaptcha", data=payload, headers={"X-Requested-With": "XMLHttpRequest"}, timeout=30, ) r.raise_for_status() data = r.json() if data.get("captcha_status") != "Y": raise ValueError("CAPTCHA verification failed.") return data.get("app_token", "") def init_search_session(session, search_text, captcha_code, app_token): params = { "p": "pdf_search/home", "text": search_text, "captcha": captcha_code, "search_opt": "PHRASE", "fcourt_type": "3", "escr_flag": "", "app_token": app_token, } session.get(BASE_URL, params=params, timeout=30).raise_for_status() log.info("Search session initialized.") def fetch_results_list(session, app_token, start=0, length=50): """Fetch a specific paginated batch of results.""" payload = { "p": "pdf_search/home/", "sEcho": "1", "iColumns": "2", "sColumns": ",", "iDisplayStart": str(start), "iDisplayLength": str(length), "fcourt_type": "3", "search_opt": "PHRASE", "ajax_req": "true", "app_token": app_token, } r = session.post( f"{BASE_URL}?p=pdf_search/home/", data=payload, headers={"X-Requested-With": "XMLHttpRequest"}, timeout=30, ) r.raise_for_status() return r.json().get("reportrow", {}).get("aaData", []) def fetch_splitview(session, args, app_token): payload = { "val": args[0], "citation_year": args[1], "path": args[2], "fcourt_type": "3", "nc_display": args[3], "flag": args[4], "ajax_req": "true", "app_token": app_token, } r = session.post( f"{BASE_URL}?p=pdf_search/splitview", data=payload, headers={"X-Requested-With": "XMLHttpRequest"}, timeout=30, ) r.raise_for_status() return r.json().get("outputfile", "") def parse_splitview_args(row_html: str) -> Optional[list]: soup = BeautifulSoup(row_html, "html.parser") btn = soup.find("a", onclick=lambda x: x and "open_splitview" in x and "'H'" in x) if not btn: return None m = re.search(r"open_splitview\(([^)]+)\)", btn.get("onclick", "")) if not m: return None args = [a.strip().strip("'\"") for a in m.group(1).split(",")] return args if len(args) >= 5 else None def fetch_pdf_from_splitview(session, html_content, path, app_token, pdf_dir): from pathlib import Path import re pdf_dir = Path(pdf_dir) pdf_dir.mkdir(parents=True, exist_ok=True) # Extract hidden field values from splitview HTML year_m = re.search(r"name='year'[^>]*value='(\d+)'", html_content) vol_m = re.search(r"name='volume'[^>]*value='(\d+)'", html_content) part_m = re.search(r"name='partno'[^>]*value='(\d+)'", html_content) year = year_m.group(1) if year_m else path.split("_")[0] volume = vol_m.group(1) if vol_m else path.split("_")[1] part = part_m.group(1) if part_m else path.split("_")[2] # path = "2026_5_577_583" → pages = "577_583" parts = path.split("_") pages = f"{parts[2]}_{parts[3]}" if len(parts) >= 4 else path candidate_urls = [ f"https://scr.sci.gov.in/scrsearch/pdfs/{year}/{volume}/{pages}.pdf", f"https://scr.sci.gov.in/scrsearch/pdfs/{path}.pdf", f"https://scr.sci.gov.in/scrsearch/pdfs/{year}_{volume}_{pages}.pdf", f"https://scr.sci.gov.in/scrsearch/?p=pdf_search/viewpdf&year={year}&volume={volume}&partno={part}&app_token={app_token}", ] for url in candidate_urls: try: r = session.get(url, timeout=30) if r.status_code == 200 and r.content[:4] == b"%PDF": pdf_path = pdf_dir / f"{path}.pdf" pdf_path.write_bytes(r.content) log.info(f" ✓ PDF saved ({len(r.content) // 1024} KB) from: {url}") return str(pdf_path) else: log.info(f" Not a PDF at: {url} (status={r.status_code})") except Exception as e: log.info(f" Failed: {url} → {e}") log.warning(f" No PDF found for: {path}") return "" # --------------------------------------------------------------------------- # Rewritten Main Function # --------------------------------------------------------------------------- def main(): Path("data/html").mkdir(parents=True, exist_ok=True) session = build_session() # Switch to JSON Lines (.jsonl) for incremental, crash-proof saving out_path = Path("data/html/extracted_judgments.jsonl") err_path = Path("data/html/errors.jsonl") # 1. Auto-Resume Check start_offset = 0 if out_path.exists(): with open(out_path, "r", encoding="utf-8") as f: start_offset = sum(1 for _ in f) if start_offset > 0: log.info(f"Found {start_offset} existing records. Resuming from there.") # 2. Homepage + CAPTCHA try: homepage_soup = fetch_homepage(session) except Exception as e: log.error(f"Could not reach SCR homepage: {e}") return captcha_path = Path("data/html/captcha.png") try: download_captcha(session, homepage_soup, captcha_path) if sys.platform == "win32": os.startfile(str(captcha_path.resolve())) except Exception as e: log.error(f"CAPTCHA download failed: {e}") return # 3. Collect inputs search_text = input("Enter search keyword [insolvency]: ").strip() or "insolvency" max_results_raw = input("How many total judgments to scrape? [1000]: ").strip() max_results = int(max_results_raw) if max_results_raw.isdigit() else 1000 if start_offset >= max_results: log.info("Target number of judgments already reached in previous runs. Exiting.") return # 4. CAPTCHA verification loop app_token = None for attempt in range(1, 4): captcha_code = input(f"Enter CAPTCHA code (attempt {attempt}/3): ").strip() if not captcha_code: continue try: app_token = verify_captcha(session, captcha_code, search_text) log.info("CAPTCHA verified successfully.") break except ValueError as e: log.warning(f"Attempt {attempt} failed: {e}") if attempt < 3: try: homepage_soup = fetch_homepage(session) download_captcha(session, homepage_soup, captcha_path) if sys.platform == "win32": os.startfile(str(captcha_path.resolve())) except Exception as ref_e: log.error(f"CAPTCHA refresh failed: {ref_e}") if app_token is None: log.error("All CAPTCHA attempts failed. Exiting.") return # 5. Init search session try: init_search_session(session, search_text, captcha_code, app_token) except Exception as e: log.error(f"Search session init failed: {e}") return # 6. Paginated Fetch & Incremental Save batch_size = 50 records_fetched = start_offset log.info(f"Targeting {max_results} judgments. Starting from index {records_fetched}...") while records_fetched < max_results: fetch_count = min(batch_size, max_results - records_fetched) log.info(f"\n--- Fetching batch: {records_fetched} to {records_fetched + fetch_count - 1} ---") # Fetch the page chunk try: rows = fetch_results_list(session, app_token, start=records_fetched, length=fetch_count) except Exception as e: log.error(f"Failed to fetch results batch at offset {records_fetched}: {e}") log.info("Session may have timed out. Restart the script; it will auto-resume where it left off.") break if not rows: log.info("No more results returned by the server. Search exhausted.") break # Process the chunk for idx, row in enumerate(rows, 1): row_html = row[1] if len(row) > 1 else "" args = parse_splitview_args(row_html) current_global_idx = records_fetched + idx if not args: log.warning(f"[{current_global_idx}] Could not parse splitview args. Skipping.") with open(err_path, "a", encoding="utf-8") as ef: ef.write(json.dumps({"index": current_global_idx, "reason": "no splitview args"}) + "\n") continue citation_path = args[2] log.info(f"[{current_global_idx}/{max_results}] Parsing: {citation_path}") try: html_content = fetch_splitview(session, args, app_token) source_url = f"{BASE_URL}?p=pdf_search/splitview&path={citation_path}" metadata = parse_judgment_html(html_content, source_url=source_url) metadata["pdf_path"] = fetch_pdf_from_splitview( session, html_content, citation_path, app_token, pdf_dir="data/pdfs" ) # Incremental Save: Append to JSONL immediately with open(out_path, "a", encoding="utf-8") as f: f.write(json.dumps(metadata, ensure_ascii=False) + "\n") except Exception as e: log.error(f"[{current_global_idx}] Failed: {e}") with open(err_path, "a", encoding="utf-8") as ef: ef.write(json.dumps({"index": current_global_idx, "path": citation_path, "reason": str(e)}) + "\n") # Dynamic human-like delay to prevent IP blocking time.sleep(random.uniform(1.0, 2.5)) records_fetched += len(rows) log.info(f"\n✓ Process stopped. Data securely saved to {out_path.resolve()}") if __name__ == "__main__": main()