scripts/discovery/refresh_contacts_from_crawl_html.py

#!/usr/bin/env python3
"""
Re-run structured contact extraction on saved ``_crawl_html/page_*.html`` snapshots.

Useful when extraction rules improve but pages were already fetched (resume skips re-fetch).
Updates ``_manifest.json`` ``structured_contacts`` / ``contact_directory_pages`` and writes
``_contact_images/contacts.json``.
"""
from __future__ import annotations

import argparse
import asyncio
import json
import os
import re
import sys
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Dict, List, Optional, Set
from urllib.parse import urljoin, urlparse

import httpx

_root = Path(__file__).resolve().parents[2]
if str(_root) not in sys.path:
    sys.path.insert(0, str(_root))

from scripts.discovery.contact_directory_heuristics import classify_contact_directory_page
from scrapers.discovery.contact_extract_from_html import (
    extract_contacts_from_page,
    extract_structured_contacts_from_html,
    infer_profile_url_from_source_page,
    merge_contact_manifest_rows,
)
from scripts.discovery.comprehensive_discovery_pipeline_jurisdiction import (
    _merge_contact_directory_pages,
    _merge_prior_extracted_contacts,
    _merge_structured_contact_rows,
)
from scripts.discovery.contact_profile_images import download_profile_images
from scripts.discovery.contacts_bundle import build_contacts_bundle, write_contacts_bundle_json


def _ai_fallback_max_pages_per_jurisdiction() -> int:
    try:
        return max(1, min(20, int((os.getenv("SCRAPED_CONTACT_AI_MAX_PAGES_PER_JURISDICTION") or "3").strip())))
    except ValueError:
        return 3


def _structured_contact_page_quality(rows: List[Dict[str, Any]]) -> float:
    if not rows:
        return 0.0
    total = 0.0
    max_total = float(len(rows) * 7)
    for row in rows:
        if str(row.get("person_name") or "").strip():
            total += 2
        if str(row.get("title_or_role") or "").strip():
            total += 2
        if str(row.get("email") or "").strip():
            total += 1
        if str(row.get("phone") or "").strip():
            total += 1
        if str(row.get("profile_image_url") or "").strip():
            total += 1
    return total / max_total if max_total > 0 else 0.0


def _structured_contact_rows_named_count(rows: List[Dict[str, Any]]) -> int:
    return sum(1 for r in rows if str(r.get("person_name") or "").strip())


def _should_try_ai_fallback(
    *,
    page_url: str,
    html: str,
    directory_score: int,
    person_adjacent_image_score: int,
    page_structured: List[Dict[str, Any]],
    low_confidence_score_max: int,
    min_quality: float,
) -> bool:
    quality = _structured_contact_page_quality(page_structured)
    named_count = _structured_contact_rows_named_count(page_structured)
    with_profile_image = sum(1 for r in page_structured if str(r.get("profile_image_url") or "").strip())
    missing_many_images = named_count >= 2 and with_profile_image < max(1, named_count // 2)
    low_confidence = (
        not page_structured
        or quality < min_quality
        or (named_count < 2 and directory_score <= low_confidence_score_max)
        or missing_many_images
    )
    if not low_confidence:
        return False
    if directory_score >= 18 and int(person_adjacent_image_score or 0) >= 6:
        return True
    try:
        from scripts.discovery.contact_extract_crawl4ai import looks_like_contact_roster_page

        return looks_like_contact_roster_page(page_url, html)
    except Exception:
        return False


def _snapshot_stem_to_page_url(homepage: str, snap_stem: str) -> str:
    """``page__220_City-Council`` → ``https://host/220/City-Council``."""
    slug = snap_stem[5:] if snap_stem.startswith("page_") else snap_stem
    slug = slug.lstrip("_")
    eid_m = re.match(r"^directory\.aspx_eid_(\d+)$", slug, re.I) or re.match(
        r"^Directory\.aspx_eid_(\d+)$", slug, re.I
    )
    if eid_m:
        path = f"/directory.aspx?eid={eid_m.group(1)}"
    elif slug == "index":
        path = "/"
    elif slug.isdigit():
        path = f"/{slug}"
    else:
        m = re.match(r"^(\d+)_(.+)$", slug)
        if m:
            path = f"/{m.group(1)}/{m.group(2)}"
        else:
            path = "/" + slug.replace("_", "/")
    base = (homepage or "").strip().rstrip("/")
    if not base:
        return path
    p = urlparse(base)
    return urljoin(f"{p.scheme}://{p.netloc}", path)


async def _download_structured_profile_images(
    jurisdiction_dir: Path,
    structured_contacts: List[Dict[str, Any]],
    *,
    homepage_url: str,
    max_images: int = 48,
) -> List[Dict[str, Any]]:
    jobs: List[Dict[str, Any]] = []
    seen_urls: Set[str] = set()
    for row in structured_contacts:
        img = str(row.get("profile_image_url") or "").strip()
        if not img or img in seen_urls:
            continue
        seen_urls.add(img)
        jobs.append(
            {
                "image_url": img,
                "person_name": str(row.get("person_name") or "").strip(),
                "title_or_role": str(row.get("title_or_role") or "").strip(),
                "source_page_url": str(row.get("source_page_url") or "").strip(),
            }
        )
    if not jobs:
        return []
    out_dir = jurisdiction_dir / "_contact_images"
    referer = (homepage_url or "").strip() or str(jobs[0].get("source_page_url") or "")
    async with httpx.AsyncClient(follow_redirects=True, timeout=60.0) as client:
        return await download_profile_images(
            client,
            jobs,
            out_dir,
            referer=referer,
            max_images=max_images,
            save_as_png=True,
        )


def _cleanup_contact_image_dir(jurisdiction_dir: Path, rows: List[Dict[str, Any]]) -> int:
    """Delete stale image files not referenced by the latest ``contact_profile_images`` rows."""
    img_dir = jurisdiction_dir / "_contact_images"
    if not img_dir.is_dir():
        return 0
    keep = {str(r.get("saved_filename") or "").strip() for r in rows}
    keep.discard("")
    deleted = 0
    for p in img_dir.iterdir():
        if not p.is_file():
            continue
        if p.name == "contacts.json":
            continue
        if p.name in keep:
            continue
        try:
            p.unlink(missing_ok=True)
            deleted += 1
        except OSError:
            continue
    return deleted


def _suggest_similar_jurisdiction_dirs(jurisdiction_dir: Path, limit: int = 8) -> List[str]:
    parts = list(jurisdiction_dir.parts)
    if "scraped_meetings" not in parts:
        return []
    idx = parts.index("scraped_meetings")
    if idx + 1 >= len(parts):
        return []
    state_root = Path(*parts[: idx + 2])
    if not state_root.is_dir():
        return []

    needle = (jurisdiction_dir.name.split("_", 1)[0] or "").strip().lower()
    suggestions: List[tuple[int, str]] = []
    for category_dir in sorted(state_root.iterdir()):
        if not category_dir.is_dir():
            continue
        for cand in sorted(category_dir.iterdir()):
            if not cand.is_dir() or not (cand / "_manifest.json").is_file():
                continue
            name_lower = cand.name.lower()
            score = 2 if needle and name_lower.startswith(needle) else (1 if needle and needle in name_lower else 0)
            rel = f"{category_dir.name}/{cand.name}"
            suggestions.append((score, rel))
    suggestions.sort(key=lambda x: (-x[0], x[1]))
    return [s[1] for s in suggestions[:limit] if s[0] > 0] or [s[1] for s in suggestions[:limit]]


def refresh_jurisdiction_contacts(
    jurisdiction_dir: Path,
    *,
    page_url_contains: Optional[str] = None,
    seed_urls: Optional[List[str]] = None,
    replace_matching_pages: bool = False,
    replace_all_structured_contacts: bool = False,
    download_profile_images_flag: bool = False,
    max_profile_images: int = 48,
    use_ai: bool = False,
    use_ai_fallback: bool = True,
    ai_provider: Optional[str] = None,
    ai_low_confidence_score_max: int = 6,
    ai_min_quality: float = 0.42,
) -> Dict[str, Any]:
    jurisdiction_dir = jurisdiction_dir.expanduser().resolve()
    manifest_path = jurisdiction_dir / "_manifest.json"
    crawl_html = jurisdiction_dir / "_crawl_html"
    if not manifest_path.is_file():
        parent = jurisdiction_dir.parent
        hint_dirs: List[str] = []
        if parent.is_dir():
            hint_dirs = [p.name for p in sorted(parent.iterdir()) if p.is_dir() and (p / "_manifest.json").is_file()][:8]
        hint = f" Available siblings with _manifest.json: {', '.join(hint_dirs)}" if hint_dirs else ""
        cross_hints = _suggest_similar_jurisdiction_dirs(jurisdiction_dir, limit=8)
        has_downloads = (jurisdiction_dir / "_downloads").is_dir()
        msg = f"{manifest_path}."
        if has_downloads:
            msg += " This directory contains _downloads and looks like a source-download cache folder, not a crawl snapshot folder."
        if hint_dirs:
            msg += f" Available siblings with _manifest.json: {', '.join(hint_dirs)}."
        if cross_hints:
            msg += f" Similar jurisdiction dirs in this state: {', '.join(cross_hints)}."
        raise FileNotFoundError(msg)
    if not crawl_html.is_dir():
        raise FileNotFoundError(crawl_html)

    data: Dict[str, Any] = json.loads(manifest_path.read_text(encoding="utf-8"))
    homepage = str(data.get("homepage_url") or "").strip()
    jid = str(data.get("jurisdiction_id") or "").strip()
    st = str(data.get("state") or "").strip()

    seed_norm: Set[str] = set()
    for u in seed_urls or []:
        seed_norm.add(u.split("#")[0].rstrip("/"))

    fresh_structured: List[Dict[str, Any]] = []
    fresh_cdir: List[Dict[str, Any]] = []
    contact_page_rows: List[Dict[str, Any]] = []
    ai_fallback_pages_used = 0
    ai_fallback_pages_limit = _ai_fallback_max_pages_per_jurisdiction()

    filter_sub = (page_url_contains or "").strip().lower()

    for snap in sorted(crawl_html.glob("page_*.html")):
        page_url = _snapshot_stem_to_page_url(homepage, snap.stem)
        if filter_sub and filter_sub not in page_url.lower():
            continue
        html = snap.read_text(encoding="utf-8", errors="replace")
        seed_hit = any(
            page_url.split("#")[0].rstrip("/") == s or page_url.rstrip("/") == s
            for s in seed_norm
        )
        cdir = classify_contact_directory_page(page_url, html)
        flagged = bool(cdir.get("is_directory")) or seed_hit
        if not flagged:
            continue

        rec = {**cdir, "page_url": page_url, "is_directory": True}
        if seed_hit:
            rec["directory_kind"] = str(rec.get("directory_kind") or "seed_url")
            ms = list(rec.get("matched_signals") or [])
            ms.append("refresh_crawl_html")
            rec["matched_signals"] = ms
        fresh_cdir.append(rec)

        page_classification = str(
            cdir.get("directory_kind") or ("seed_url" if seed_hit else "unknown")
        )
        directory_score = int(cdir.get("score") or 0)
        person_adjacent_image_score = int(cdir.get("person_adjacent_image_score") or 0)

        if use_ai:
            from scripts.discovery.contact_extract_crawl4ai import (
                ai_record_to_structured_row,
                extract_contact_directory_sync,
            )

            ai_kwargs = {"provider": ai_provider} if ai_provider else {}
            directory = extract_contact_directory_sync(page_url, **ai_kwargs)
            for rec in directory.contacts:
                prow = ai_record_to_structured_row(
                    rec,
                    source_page_url=page_url,
                    page_classification=page_classification,
                    directory_score=directory_score,
                )
                infer_profile_url_from_source_page(prow)
                fresh_structured.append(prow)
        else:
            page_structured = extract_structured_contacts_from_html(html, page_url)
            if use_ai_fallback and _should_try_ai_fallback(
                page_url=page_url,
                html=html,
                directory_score=directory_score,
                person_adjacent_image_score=person_adjacent_image_score,
                page_structured=page_structured,
                low_confidence_score_max=max(0, ai_low_confidence_score_max),
                min_quality=max(0.0, min(1.0, ai_min_quality)),
            ) and ai_fallback_pages_used < ai_fallback_pages_limit:
                try:
                    from scripts.discovery.contact_extract_crawl4ai import (
                        ai_record_to_structured_row,
                        extract_contact_directory_from_html_sync,
                    )

                    ai_kwargs = {"provider": ai_provider} if ai_provider else {}
                    heuristic_count = len(page_structured)
                    ai_fallback_pages_used += 1
                    ai_directory = extract_contact_directory_from_html_sync(
                        html,
                        page_url,
                        **ai_kwargs,
                    )
                    ai_rows: List[Dict[str, Any]] = []
                    for rec in ai_directory.contacts:
                        ai_rows.append(
                            ai_record_to_structured_row(
                                rec,
                                source_page_url=page_url,
                                page_classification=page_classification,
                                directory_score=directory_score,
                                extraction_method="crawl4ai_llm_fallback",
                            )
                        )
                    if ai_rows:
                        page_structured = _merge_structured_contact_rows(page_structured, ai_rows)
                        print(
                            f"[contact_ai_fallback] page={page_url} heuristic={heuristic_count} ai={len(ai_rows)} merged={len(page_structured)}"
                        )
                except Exception as exc:
                    print(f"[contact_ai_fallback_error] page={page_url} detail={exc!r}")
            elif use_ai_fallback and ai_fallback_pages_used >= ai_fallback_pages_limit:
                print(
                    f"[contact_ai_fallback_skip_limit] page={page_url} "
                    f"used={ai_fallback_pages_used} limit={ai_fallback_pages_limit}"
                )

            for prow in page_structured:
                prow["source_page_url"] = page_url
                prow["page_classification"] = page_classification
                prow["directory_score"] = directory_score
                infer_profile_url_from_source_page(prow)
                fresh_structured.append(prow)

        contact_page_rows.append(extract_contacts_from_page(html, page_url))

    prior_sc = data.get("structured_contacts")
    if replace_all_structured_contacts:
        prior_sc = []
    elif replace_matching_pages and filter_sub and isinstance(prior_sc, list):
        kept = [
            r
            for r in prior_sc
            if isinstance(r, dict)
            and filter_sub not in str(r.get("source_page_url") or "").lower()
        ]
        prior_sc = kept

    if isinstance(prior_sc, list) and prior_sc:
        data["structured_contacts"] = _merge_structured_contact_rows(prior_sc, fresh_structured)
    else:
        data["structured_contacts"] = fresh_structured

    prior_cdp = data.get("contact_directory_pages")
    if isinstance(prior_cdp, list) and prior_cdp:
        data["contact_directory_pages"] = _merge_contact_directory_pages(prior_cdp, fresh_cdir)
    else:
        data["contact_directory_pages"] = fresh_cdir

    pec = data.get("extracted_contacts")
    fresh_flat = merge_contact_manifest_rows(contact_page_rows)
    if isinstance(pec, dict) and pec:
        data["extracted_contacts"] = _merge_prior_extracted_contacts(pec, fresh_flat)
    else:
        data["extracted_contacts"] = fresh_flat

    profile_dl: List[Dict[str, Any]] = []
    if download_profile_images_flag and data["structured_contacts"]:
        profile_dl = asyncio.run(
            _download_structured_profile_images(
                jurisdiction_dir,
                data["structured_contacts"],
                homepage_url=homepage,
                max_images=max_profile_images,
            )
        )
        # Replace old image rows even when empty so stale errors don't survive refresh runs.
        data["contact_profile_images"] = profile_dl
        _cleanup_contact_image_dir(jurisdiction_dir, profile_dl)

    data["contacts_refreshed_at"] = datetime.now(timezone.utc).isoformat()
    manifest_path.write_text(json.dumps(data, indent=2), encoding="utf-8")

    bundle_path = None
    if data["structured_contacts"]:
        bundle = build_contacts_bundle(
            jurisdiction_id=jid,
            state=st,
            homepage_url=homepage,
            scraped_at=data.get("scraped_at"),
            scrape_batch_id=str(data.get("scrape_batch_id") or ""),
            structured_contacts=data["structured_contacts"],
            contact_profile_images=list(data.get("contact_profile_images") or []),
            extracted_contacts=data.get("extracted_contacts"),
        )
        bundle_path = write_contacts_bundle_json(jurisdiction_dir, bundle)

    saved_images = sum(1 for r in profile_dl if r.get("saved_filename"))
    return {
        "jurisdiction_id": jid,
        "structured_contacts": len(data["structured_contacts"]),
        "new_from_snapshots": len(fresh_structured),
        "contacts_json": str(bundle_path) if bundle_path else None,
        "profile_images_saved": saved_images,
    }


def main() -> None:
    ap = argparse.ArgumentParser(description="Refresh structured contacts from _crawl_html snapshots.")
    ap.add_argument(
        "--jurisdiction-dir",
        required=True,
        help="Path to jurisdiction folder (contains _manifest.json and _crawl_html/)",
    )
    ap.add_argument(
        "--page-url-contains",
        default="",
        help="Only process snapshots whose reconstructed URL contains this substring",
    )
    ap.add_argument(
        "--seed-url",
        action="append",
        default=[],
        help="Treat matching URLs as directory seeds (repeatable)",
    )
    ap.add_argument(
        "--replace-matching-pages",
        action="store_true",
        help="Drop prior structured_contacts rows whose source_page_url matches --page-url-contains before merge",
    )
    ap.add_argument(
        "--replace-all-structured-contacts",
        action="store_true",
        help="Replace all prior structured_contacts with freshly extracted rows (full cleanup mode).",
    )
    ap.add_argument(
        "--download-profile-images",
        action="store_true",
        help="Download profile_image_url from structured contacts into _contact_images/",
    )
    ap.add_argument(
        "--max-profile-images",
        type=int,
        default=48,
        help="Cap when using --download-profile-images (default 48)",
    )
    ap.add_argument(
        "--ai",
        action="store_true",
        help=(
            "Use crawl4ai + Groq LLM extraction instead of the heuristic HTML parser. "
            "Requires GROQ_API_KEY in env and `crawl4ai-setup` to have been run."
        ),
    )
    ap.add_argument(
        "--ai-provider",
        default=None,
        help="LiteLLM provider string (default: groq/llama-3.1-8b-instant). Only used with --ai.",
    )
    ap.add_argument(
        "--ai-fallback",
        action="store_true",
        help=(
            "Use fast heuristic extraction first, then invoke AI only on complex/low-confidence "
            "directory pages (board/commissioner layouts). Enabled by default unless --no-ai-fallback."
        ),
    )
    ap.add_argument(
        "--no-ai-fallback",
        action="store_true",
        help="Disable heuristic+AI fallback mode.",
    )
    ap.add_argument(
        "--ai-low-confidence-score-max",
        type=int,
        default=6,
        help="Directory heuristic score threshold for AI fallback (default 6).",
    )
    ap.add_argument(
        "--ai-min-quality",
        type=float,
        default=0.42,
        help="Minimum heuristic row quality before AI fallback is skipped (default 0.42).",
    )
    args = ap.parse_args()
    use_ai_fallback = False if args.no_ai_fallback else (True if not args.ai else bool(args.ai_fallback))
    summary = refresh_jurisdiction_contacts(
        Path(args.jurisdiction_dir),
        page_url_contains=args.page_url_contains or None,
        seed_urls=args.seed_url or None,
        replace_matching_pages=args.replace_matching_pages,
        replace_all_structured_contacts=args.replace_all_structured_contacts,
        download_profile_images_flag=args.download_profile_images,
        max_profile_images=args.max_profile_images,
        use_ai=args.ai,
        use_ai_fallback=use_ai_fallback,
        ai_provider=args.ai_provider,
        ai_low_confidence_score_max=args.ai_low_confidence_score_max,
        ai_min_quality=args.ai_min_quality,
    )
    print(json.dumps(summary, indent=2))


if __name__ == "__main__":
    main()