pipeline_miami/verify/common.py

"""Shared helpers for the Miami validation harness (mirrors pipeline_philly/verify/common.py).

Read-only outward calls; the served-DB connection forces session read_only so
the server rejects writes regardless of role permissions.
"""
from __future__ import annotations

import os
import pathlib

REPO_ROOT = pathlib.Path(__file__).resolve().parents[2]
SOURCES_YAML = REPO_ROOT / "docs" / "cities" / "miami" / "sources.yaml"
COMPONENTS_YAML = REPO_ROOT / "pipeline_miami" / "verify" / "components.yaml"
QA_REPORTS_DIR = REPO_ROOT / "docs" / "qa_reports" / "miami"
SOURCES_DIR = REPO_ROOT / "data" / "cities" / "miami" / "sources"
PARSED_DIR = REPO_ROOT / "data" / "cities" / "miami" / "reference" / "parsed"


def load_env_local() -> None:
    p = REPO_ROOT / ".env.local"
    if not p.exists():
        return
    for line in p.read_text().splitlines():
        if not line or line.startswith("#") or "=" not in line:
            continue
        k, _, v = line.partition("=")
        os.environ.setdefault(k.strip(), v.strip().strip('"').strip("'"))


def served_db_dsn() -> str:
    load_env_local()
    dsn = os.environ.get("VALIDATE_READONLY_DATABASE_URL")
    if not dsn:
        raise SystemExit("VALIDATE_READONLY_DATABASE_URL not set in .env.local (a read-only role; see verify/README.md)")
    return dsn


def latest_source_path(source_id: str) -> pathlib.Path:
    d = SOURCES_DIR / source_id
    if not d.exists():
        raise SystemExit(f"no fetched source for {source_id}; run python -m pipeline_miami.discover --source {source_id}")
    for day in sorted([p for p in d.iterdir() if p.is_dir() and p.name[:4].isdigit()], reverse=True):
        for f in day.iterdir():
            if f.name != "provenance.json":
                return f
    raise SystemExit(f"no data file under {d}")


def pad4(v) -> str | None:
    if v is None:
        return None
    s = str(v).strip()
    if s.endswith(".0"):
        s = s[:-2]
    if not s.isdigit() or len(s) > 4:
        return None
    return s.zfill(4)


def ensure_qa_dir() -> None:
    QA_REPORTS_DIR.mkdir(parents=True, exist_ok=True)