pipeline_philly/verify/common.py

"""Shared helpers for the Philly validation harness.

Mirrors verify/common.py (NYC) in spirit. Read-only outward calls; the served
DB connection forces session-level read_only=True so the server rejects writes
regardless of role permissions.
"""
from __future__ import annotations

import os
import pathlib
import sys

REPO_ROOT = pathlib.Path(__file__).resolve().parents[2]

SOURCES_YAML = REPO_ROOT / "docs" / "cities" / "philly" / "sources.yaml"
COMPONENTS_YAML = REPO_ROOT / "pipeline_philly" / "verify" / "components.yaml"
QA_REPORTS_DIR = REPO_ROOT / "docs" / "qa_reports" / "philly"
SOURCES_DIR = REPO_ROOT / "data" / "cities" / "philly" / "sources"


def load_env_local() -> None:
    """Manually parse .env.local so we don't need python-dotenv."""
    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:
    """Read-only DSN, same convention as verify/common.py."""
    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 — see verify/README.md.",
        )
    return dsn


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


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