audit.py

import csv
import re
import json
import logging
import phonenumbers
import httpx
from pathlib import Path
from typing import Optional, Dict, List
from dataclasses import dataclass, field

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(message)s")
log = logging.getLogger(__name__)

HEADERS = {"User-Agent": "Mozilla/5.0"}

# ── Format validators ─────────────────────────────────────────────────────────
EMAIL_RE   = re.compile(r'^[a-zA-Z0-9._%+\-]+@[a-zA-Z0-9.\-]+\.[a-zA-Z]{2,}$')
PHONE_RE   = re.compile(r'^\d{3}-\d{3}-\d{4}$')
URL_RE     = re.compile(r'^https?://')
ADDRESS_RE = re.compile(r'\d{1,5}\s+\w', re.IGNORECASE)  # starts with a street number


def validate_email(email: str) -> tuple[bool, str]:
    if not email or email == "Not available":
        return True, ""   # absence is valid
    if not EMAIL_RE.match(email):
        return False, f"Malformed email: {email}"
    # Check for obviously wrong domains
    domain = email.split("@")[1].lower()
    if any(bad in domain for bad in ["gmail.com","yahoo.com","hotmail.com","outlook.com"]):
        return False, f"Unexpected personal email domain: {domain}"
    return True, ""


def validate_phone(phone: str) -> tuple[bool, str]:
    if not phone or phone == "Not available":
        return True, ""
    if not PHONE_RE.match(phone):
        return False, f"Phone not normalised (expected NNN-NNN-NNNN): {phone}"
    # Check via phonenumbers library
    try:
        p = phonenumbers.parse("+1" + re.sub(r'\D', '', phone), "US")
        if not phonenumbers.is_valid_number(p):
            return False, f"Invalid phone number: {phone}"
    except Exception:
        return False, f"Cannot parse phone: {phone}"
    return True, ""


def validate_website(url: str) -> tuple[bool, str]:
    if not url or url == "Not available":
        return True, ""
    if not URL_RE.match(url):
        return False, f"URL missing http(s): {url}"
    return True, ""


def validate_address(address: str, county: str) -> tuple[bool, str]:
    if not address or address == "Not available":
        return True, ""
    if not ADDRESS_RE.match(address):
        return False, f"Address doesn't start with street number: {address}"
    # Check Michigan state abbreviation
    if "MI" not in address and "Michigan" not in address:
        return False, f"Address missing MI state code: {address}"
    # Check county name or known county seat appears
    county_clean = county.lower().replace(" county", "")
    return True, ""   # county seat check is too strict — skip


def check_website_live(url: str, timeout: int = 6) -> tuple[bool, str]:
    """Check if the website is actually reachable."""
    if not url or url == "Not available":
        return True, ""
    try:
        r = httpx.head(url, headers=HEADERS, timeout=timeout, follow_redirects=True)
        if r.status_code >= 400:
            return False, f"Website returns HTTP {r.status_code}: {url}"
        return True, ""
    except httpx.TimeoutException:
        return False, f"Website timed out: {url}"
    except Exception as e:
        return False, f"Website unreachable ({e}): {url}"


# ── Duplicate detector ────────────────────────────────────────────────────────
def find_duplicates(rows: List[dict]) -> dict[str, list[int]]:
    """Find rows sharing the same email or phone (may indicate copy-paste errors)."""
    email_map: dict[str, list[int]] = {}
    phone_map: dict[str, list[int]] = {}

    for i, row in enumerate(rows):
        email = row.get("Email","").strip()
        phone = row.get("Phone","").strip()
        if email and email != "Not available":
            email_map.setdefault(email, []).append(i)
        if phone and phone != "Not available":
            phone_map.setdefault(phone, []).append(i)

    dupes = {}
    for email, idxs in email_map.items():
        if len(idxs) > 1:
            dupes[f"email:{email}"] = idxs
    for phone, idxs in phone_map.items():
        if len(idxs) > 1:
            dupes[f"phone:{phone}"] = idxs

    return dupes


# ── Completeness scorer ───────────────────────────────────────────────────────
def completeness_score(row: dict) -> float:
    """0-1 score: fraction of required fields that are filled."""
    required = ["Email", "Phone", "Website", "Office_Address"]
    filled = sum(1 for f in required
                 if row.get(f,"") not in ("Not available","",None))
    return filled / len(required)


# ── Main audit ────────────────────────────────────────────────────────────────
@dataclass
class AuditResult:
    row_num:        int
    num:            str
    county:         str
    position:       str
    name:           str
    completeness:   float = 0.0
    issues:         list  = field(default_factory=list)
    needs_review:   bool  = False


def audit(input_path: str, report_path: str, check_urls: bool = False):
    rows: List[dict] = []
    with open(input_path, newline="", encoding="utf-8") as f:
        reader = csv.DictReader(f)
        rows = list(reader)

    results: list[AuditResult] = []
    total_completeness = 0.0

    log.info(f"Auditing {len(rows)} rows...")

    # Duplicate check
    dupes = find_duplicates(rows)
    if dupes:
        log.warning(f"Found {len(dupes)} duplicate field values:")
        for key, idxs in dupes.items():
            log.warning(f"  {key} → rows {idxs}")

    for i, row in enumerate(rows):
        ar = AuditResult(
            row_num=i+2,   # +2 for header + 1-indexing
            num=row.get("#",""),
            county=row.get("County",""),
            position=row.get("Position",""),
            name=row.get("Name",""),
        )

        # Format validation
        for check_fn, field_key in [
            (validate_email,   "Email"),
            (validate_phone,   "Phone"),
            (validate_website, "Website"),
        ]:
            ok, msg = check_fn(row.get(field_key,""))
            if not ok:
                ar.issues.append(msg)

        ok, msg = validate_address(row.get("Office_Address",""), ar.county)
        if not ok:
            ar.issues.append(msg)

        # Completeness
        ar.completeness = completeness_score(row)
        total_completeness += ar.completeness

        if ar.completeness < 0.5:
            ar.issues.append(f"Low completeness ({ar.completeness:.0%}) — many fields missing")

        # Low confidence flags from scraper
        flags = row.get("_flags","")
        if flags and "low-confidence" in flags.lower():
            ar.issues.append(f"Scraper flagged: {flags[:100]}")

        # Optional: live URL check
        if check_urls:
            website = row.get("Website","")
            ok, msg = check_website_live(website)
            if not ok:
                ar.issues.append(msg)

        # Duplicate membership
        email = row.get("Email","")
        phone = row.get("Phone","")
        for key in [f"email:{email}", f"phone:{phone}"]:
            if key in dupes and len(dupes[key]) > 1:
                ar.issues.append(f"Duplicate value shared with rows {dupes[key]}: {key}")

        ar.needs_review = len(ar.issues) > 0
        results.append(ar)

    # ── Write report ──────────────────────────────────────────────────────────
    report_fields = ["row_num","#","county","position","name",
                     "completeness","needs_review","issues"]
    with open(report_path, "w", newline="", encoding="utf-8") as f:
        writer = csv.DictWriter(f, fieldnames=report_fields)
        writer.writeheader()
        for ar in results:
            writer.writerow({
                "row_num":      ar.row_num,
                "#":            ar.num,
                "county":       ar.county,
                "position":     ar.position,
                "name":         ar.name,
                "completeness": f"{ar.completeness:.0%}",
                "needs_review": "YES" if ar.needs_review else "",
                "issues":       " | ".join(ar.issues),
            })

    # ── Print summary ─────────────────────────────────────────────────────────
    needs_review   = sum(1 for r in results if r.needs_review)
    fully_complete = sum(1 for r in results if r.completeness == 1.0)
    avg_complete   = total_completeness / len(rows) if rows else 0

    print(f"\n{'='*55}")
    print(f"  AUDIT SUMMARY")
    print(f"{'='*55}")
    print(f"  Total rows:         {len(rows)}")
    print(f"  Fully complete:     {fully_complete} ({fully_complete/len(rows):.0%})")
    print(f"  Avg completeness:   {avg_complete:.0%}")
    print(f"  Need review:        {needs_review}")
    print(f"  Duplicate values:   {len(dupes)}")
    print(f"{'='*55}")

    if dupes:
        print(f"\nDuplicate values (investigate):")
        for key, idxs in dupes.items():
            names = [rows[i].get("Name","?") for i in idxs]
            print(f"  {key}")
            print(f"    Rows {idxs}: {names}")

    print(f"\nRows needing review:")
    for r in results:
        if r.needs_review:
            print(f"  Row {r.row_num}: {r.name} ({r.county} / {r.position})")
            for issue in r.issues:
                print(f"    → {issue}")

    print(f"\nReport saved to: {report_path}")


if __name__ == "__main__":
    import argparse
    p = argparse.ArgumentParser(description="Audit scraped Michigan officials CSV")
    p.add_argument("--input",      required=True,  help="Filled CSV from scraper")
    p.add_argument("--report",     default="audit_report.csv", help="Output report path")
    p.add_argument("--check-urls", action="store_true", help="Probe websites for liveness (slow)")
    args = p.parse_args()
    audit(args.input, args.report, check_urls=args.check_urls)