Evol-1 T05: extraction cascade v0 (rules + local LLM)

.gitignore

@@ -25,8 +25,11 @@ data/raw/*
 !data/raw/local_llm/**
 !data/raw/url_finder/
 !data/raw/url_finder/**
+!data/raw/extraction_cascade/
+!data/raw/extraction_cascade/**
 # Full HTML page dumps are regenerable; keep SERP JSON + identity/labels
 data/raw/url_finder/pages_cache/
+data/raw/extraction_cascade/pages_cache/
 data/models/*.pkl
 data/models/*.pt
 # LightGBM train artifacts (Task #71) — regenerate locally; do not commit binaries

crawlers/extraction_cascade/__init__.py

@@ -0,0 +1,15 @@
+"""Evol-1 T05 — extraction cascade v0 (tier-1 rules + tier-2 local LLM)."""
+
+from __future__ import annotations
+
+from crawlers.extraction_cascade.pipeline import run_firm, run_on_page
+from crawlers.extraction_cascade.schema import FirmCascadeResult, Tier1Indicators
+from crawlers.extraction_cascade.tier1_rules import analyze_page_rules
+
+__all__ = [
+    "FirmCascadeResult",
+    "Tier1Indicators",
+    "analyze_page_rules",
+    "run_firm",
+    "run_on_page",
+]

crawlers/extraction_cascade/__main__.py

@@ -0,0 +1,4 @@
+"""python -m crawlers.extraction_cascade"""
+from crawlers.extraction_cascade.runner import main
+if __name__ == "__main__":
+    raise SystemExit(main())

crawlers/extraction_cascade/cohort.py

@@ -0,0 +1,138 @@
+"""Build the T05 cohort: 28 listed websites + optional frame_pilot URLs."""
+
+from __future__ import annotations
+
+import csv
+import hashlib
+import json
+from dataclasses import asdict, dataclass
+from pathlib import Path
+from typing import Any, Literal
+
+from crawlers.companies.listed_companies import load_seed_companies
+from crawlers.extraction_cascade.paths import COHORT_PATH, RAW_DIR, ROOT
+
+CohortSource = Literal["listed28", "frame_pilot"]
+
+
+@dataclass(frozen=True)
+class CohortFirm:
+    firm_id: str
+    source_cohort: CohortSource
+    website_url: str
+    name: str = ""
+    vsic_code: str | None = None
+    tax_code: str | None = None
+    notes: str = ""
+
+
+def listed28_cohort() -> list[CohortFirm]:
+    rows: list[CohortFirm] = []
+    for company in load_seed_companies():
+        url = (company.get("website_url") or "").strip()
+        ticker = str(company.get("stock_code") or "").strip().upper()
+        if not ticker or not url:
+            continue
+        rows.append(
+            CohortFirm(
+                firm_id=ticker,
+                source_cohort="listed28",
+                website_url=url,
+                name=str(company.get("name") or ""),
+                vsic_code=str(company.get("vsic_code") or "") or None,
+            )
+        )
+    return rows
+
+
+def load_frame_urls_file(path: Path) -> list[CohortFirm]:
+    """Optional JSON list produced after URL-finder on frame_pilot."""
+    if not path.exists():
+        return []
+    payload = json.loads(path.read_text(encoding="utf-8"))
+    if not isinstance(payload, list):
+        raise ValueError(f"frame urls must be a JSON list: {path}")
+    out: list[CohortFirm] = []
+    for row in payload:
+        if not isinstance(row, dict):
+            continue
+        url = str(row.get("website_url") or row.get("url") or "").strip()
+        firm_id = str(row.get("firm_id") or row.get("tax_code") or "").strip()
+        if not url or not firm_id:
+            continue
+        out.append(
+            CohortFirm(
+                firm_id=firm_id,
+                source_cohort="frame_pilot",
+                website_url=url,
+                name=str(row.get("name") or row.get("company_name") or ""),
+                vsic_code=str(row.get("vsic_code") or row.get("vsic_4digit") or "") or None,
+                tax_code=str(row.get("tax_code") or "") or None,
+                notes=str(row.get("notes") or ""),
+            )
+        )
+    return out
+
+
+def sample_frame_for_url_finder(
+    *,
+    frame_csv: Path | None = None,
+    per_division: int = 40,
+    offset: int = 0,
+) -> list[dict[str, Any]]:
+    """Stratified sample from T02 frame (identity only — no website column)."""
+    path = frame_csv or (ROOT / "data" / "raw" / "frame_pilot" / "frame_pilot.csv")
+    if not path.exists():
+        return []
+    by_div: dict[str, list[dict[str, Any]]] = {}
+    with path.open(encoding="utf-8", newline="") as fh:
+        for row in csv.DictReader(fh):
+            div = str(row.get("vsic_division") or "").strip()
+            by_div.setdefault(div, []).append(row)
+    sample: list[dict[str, Any]] = []
+    for div in sorted(by_div):
+        rows = by_div[div]
+        start = max(0, offset)
+        sample.extend(rows[start : start + per_division])
+    return sample
+
+
+def build_cohort(
+    *,
+    frame_urls_path: Path | None = None,
+    include_listed: bool = True,
+) -> list[CohortFirm]:
+    firms: list[CohortFirm] = []
+    if include_listed:
+        firms.extend(listed28_cohort())
+    frame_path = frame_urls_path or (RAW_DIR / "frame_urls.json")
+    firms.extend(load_frame_urls_file(frame_path))
+    # Dedupe by firm_id preferring listed28
+    seen: set[str] = set()
+    out: list[CohortFirm] = []
+    for firm in firms:
+        key = firm.firm_id.lower()
+        if key in seen:
+            continue
+        seen.add(key)
+        out.append(firm)
+    return out
+
+
+def cohort_sha256(firms: list[CohortFirm]) -> str:
+    blob = json.dumps([asdict(f) for f in firms], sort_keys=True, ensure_ascii=False)
+    return hashlib.sha256(blob.encode("utf-8")).hexdigest()
+
+
+def write_cohort(firms: list[CohortFirm], path: Path | None = None) -> Path:
+    target = path or COHORT_PATH
+    target.parent.mkdir(parents=True, exist_ok=True)
+    payload = {
+        "n": len(firms),
+        "listed28": sum(1 for f in firms if f.source_cohort == "listed28"),
+        "frame_pilot": sum(1 for f in firms if f.source_cohort == "frame_pilot"),
+        "sha256": cohort_sha256(firms),
+        "firms": [asdict(f) for f in firms],
+    }
+    target.write_text(json.dumps(payload, indent=2, ensure_ascii=False) + "\n", encoding="utf-8")
+    return target

crawlers/extraction_cascade/compare.py

@@ -0,0 +1,196 @@
+"""Compare tier-1 rules vs tier-2 LLM fields without inventing values."""
+
+from __future__ import annotations
+
+from typing import Any
+
+from crawlers.extraction_cascade.schema import ConflictRow, Tier1Indicators
+
+BOOL_FIELDS = ("has_product_catalog", "has_order_cart")
+LIST_PRESENCE_FIELDS = ("payment_methods", "social_links", "marketplace_links")
+LANG_FIELD = "website_language"
+
+
+def _tier2_field(tier2: dict[str, Any] | None, name: str) -> dict[str, Any]:
+    if not isinstance(tier2, dict):
+        return {"value": None, "abstain": True, "confidence": 0.0, "reason": "missing_tier2"}
+    raw = tier2.get(name)
+    if not isinstance(raw, dict):
+        return {"value": None, "abstain": True, "confidence": 0.0, "reason": "missing_field"}
+    return raw
+
+
+def _boolish(value: Any) -> bool | None:
+    if value is None:
+        return None
+    if isinstance(value, bool):
+        return value
+    return bool(value)
+
+
+def _presence_from_list(value: Any) -> bool:
+    if isinstance(value, list):
+        return len(value) > 0
+    return bool(value)
+
+
+def compare_tiers(
+    firm_id: str,
+    tier1: Tier1Indicators | None,
+    tier2: dict[str, Any] | None,
+    *,
+    fetch_ok: bool,
+) -> list[ConflictRow]:
+    """Return per-field agree / conflict / abstain / skip rows."""
+    if not fetch_ok or tier1 is None:
+        return [
+            ConflictRow(
+                firm_id=firm_id,
+                field="*",
+                kind="skip",
+                tier1_value=None,
+                tier2_value=None,
+                tier2_abstain=True,
+                note="fetch_failed_or_no_tier1",
+            )
+        ]
+
+    rows: list[ConflictRow] = []
+
+    for name in BOOL_FIELDS:
+        t1 = bool(getattr(tier1, name))
+        f2 = _tier2_field(tier2, name)
+        rows.append(_compare_bool(firm_id, name, t1, f2))
+
+    for name in LIST_PRESENCE_FIELDS:
+        t1_list = getattr(tier1, name) or []
+        t1_present = len(t1_list) > 0
+        f2 = _tier2_field(tier2, name)
+        rows.append(_compare_presence(firm_id, name, t1_present, t1_list, f2))
+
+    t1_lang = tier1.website_language
+    f2 = _tier2_field(tier2, LANG_FIELD)
+    rows.append(_compare_lang(firm_id, t1_lang, f2))
+    return rows
+
+
+def _compare_bool(
+    firm_id: str,
+    field: str,
+    t1: bool,
+    f2: dict[str, Any],
+) -> ConflictRow:
+    if f2.get("abstain"):
+        return ConflictRow(
+            firm_id=firm_id,
+            field=field,
+            kind="abstain",
+            tier1_value=t1,
+            tier2_value=f2.get("value"),
+            tier2_abstain=True,
+            note=str(f2.get("reason") or "tier2_abstain"),
+        )
+    t2 = _boolish(f2.get("value"))
+    if t2 is None:
+        return ConflictRow(
+            firm_id=firm_id,
+            field=field,
+            kind="abstain",
+            tier1_value=t1,
+            tier2_value=None,
+            tier2_abstain=True,
+            note="tier2_null_value",
+        )
+    kind = "agree" if t1 == t2 else "conflict"
+    return ConflictRow(
+        firm_id=firm_id,
+        field=field,
+        kind=kind,
+        tier1_value=t1,
+        tier2_value=t2,
+        tier2_abstain=False,
+        note="" if kind == "agree" else "bool_mismatch",
+    )
+
+
+def _compare_presence(
+    firm_id: str,
+    field: str,
+    t1_present: bool,
+    t1_list: list[Any],
+    f2: dict[str, Any],
+) -> ConflictRow:
+    if f2.get("abstain"):
+        return ConflictRow(
+            firm_id=firm_id,
+            field=field,
+            kind="abstain",
+            tier1_value=t1_list,
+            tier2_value=f2.get("value"),
+            tier2_abstain=True,
+            note=str(f2.get("reason") or "tier2_abstain"),
+        )
+    t2_present = _presence_from_list(f2.get("value"))
+    kind = "agree" if t1_present == t2_present else "conflict"
+    return ConflictRow(
+        firm_id=firm_id,
+        field=field,
+        kind=kind,
+        tier1_value=t1_list,
+        tier2_value=f2.get("value"),
+        tier2_abstain=False,
+        note="" if kind == "agree" else "presence_mismatch",
+    )
+
+
+def _compare_lang(
+    firm_id: str,
+    t1_lang: str | None,
+    f2: dict[str, Any],
+) -> ConflictRow:
+    if f2.get("abstain"):
+        return ConflictRow(
+            firm_id=firm_id,
+            field=LANG_FIELD,
+            kind="abstain",
+            tier1_value=t1_lang,
+            tier2_value=f2.get("value"),
+            tier2_abstain=True,
+            note=str(f2.get("reason") or "tier2_abstain"),
+        )
+    t2 = f2.get("value")
+    if t1_lang is None and t2 is None:
+        kind: str = "agree"
+    elif t1_lang == t2:
+        kind = "agree"
+    elif t1_lang in {None, "unknown"} or t2 in {None, "unknown"}:
+        kind = "agree"
+        note = "unknown_compatible"
+        return ConflictRow(
+            firm_id=firm_id,
+            field=LANG_FIELD,
+            kind="agree",
+            tier1_value=t1_lang,
+            tier2_value=t2,
+            tier2_abstain=False,
+            note=note,
+        )
+    else:
+        return ConflictRow(
+            firm_id=firm_id,
+            field=LANG_FIELD,
+            kind="conflict",
+            tier1_value=t1_lang,
+            tier2_value=t2,
+            tier2_abstain=False,
+            note="language_mismatch",
+        )
+    return ConflictRow(
+        firm_id=firm_id,
+        field=LANG_FIELD,
+        kind="agree",
+        tier1_value=t1_lang,
+        tier2_value=t2,
+        tier2_abstain=False,
+        note="",
+    )