Agency Metadata and Feed Versioning Practices

pip install pydantic

import re
from typing import Optional
import pydantic
from pydantic import Field, field_validator

VALID_IANA_TIMEZONES: set[str] = set()
try:
    import zoneinfo
    VALID_IANA_TIMEZONES = zoneinfo.available_timezones()
except ImportError:
    pass  # Python 3.8 fallback: skip IANA check

class AgencyMetadata(pydantic.BaseModel):
    model_config = pydantic.ConfigDict(str_strip_whitespace=True)

    agency_id: str
    agency_name: str
    agency_url: pydantic.AnyHttpUrl
    agency_timezone: str
    agency_lang: str = Field(pattern=r"^[a-z]{2,3}$")
    agency_phone: Optional[str] = None
    agency_fare_url: Optional[pydantic.AnyHttpUrl] = None
    agency_email: Optional[pydantic.EmailStr] = None

    @field_validator("agency_timezone")
    @classmethod
    def validate_iana_timezone(cls, v: str) -> str:
        if VALID_IANA_TIMEZONES and v not in VALID_IANA_TIMEZONES:
            raise ValueError(
                f"agency_timezone '{v}' is not a valid IANA timezone. "
                "Common mistake: use 'America/New_York', not 'US/Eastern'."
            )
        return v

class FeedInfoMetadata(pydantic.BaseModel):
    model_config = pydantic.ConfigDict(str_strip_whitespace=True)

    feed_publisher_name: str
    feed_publisher_url: pydantic.AnyHttpUrl
    feed_lang: str = Field(pattern=r"^[a-z]{2,3}$")
    feed_start_date: str = Field(pattern=r"^\d{8}$")
    feed_end_date: str = Field(pattern=r"^\d{8}$")
    feed_version: str
    default_lang: Optional[str] = Field(default=None, pattern=r"^[a-z]{2,3}$")

import hashlib
from pathlib import Path

def compute_sha256(file_path: Path) -> str:
    """Stream the raw archive in 8 kB chunks to avoid loading it fully into memory."""
    sha256 = hashlib.sha256()
    with open(file_path, "rb") as fh:
        for chunk in iter(lambda: fh.read(8192), b""):
            sha256.update(chunk)
    return sha256.hexdigest()

import csv
import zipfile
import logging
from datetime import datetime, timezone
from pydantic import ValidationError

logging.basicConfig(level=logging.INFO, format="%(levelname)s %(name)s: %(message)s")
log = logging.getLogger("gtfs.metadata")

REQUIRED_FILES = {"agency.txt", "feed_info.txt"}

def parse_csv_from_zip(zf: zipfile.ZipFile, filename: str) -> list[dict]:
    """Read a UTF-8-BOM-safe CSV from an open ZipFile and return rows as dicts."""
    with zf.open(filename) as raw:
        text = raw.read().decode("utf-8-sig")
    reader = csv.DictReader(text.splitlines(), skipinitialspace=True)
    return [
        {k.strip(): v.strip() for k, v in row.items() if k}
        for row in reader
    ]

def extract_and_validate_feed(feed_path: Path) -> dict:
    """
    Extract agency.txt and feed_info.txt from a GTFS zip, validate both,
    and return a structured result with a SHA-256 checksum.
    """
    if not feed_path.exists():
        raise FileNotFoundError(f"GTFS archive not found: {feed_path}")

    archive_hash = compute_sha256(feed_path)

    with zipfile.ZipFile(feed_path, "r") as zf:
        present = set(zf.namelist())
        missing = REQUIRED_FILES - present
        if missing:
            raise ValueError(
                f"GTFS archive is missing required metadata files: {sorted(missing)}"
            )

        # --- agency.txt ---
        raw_agencies = parse_csv_from_zip(zf, "agency.txt")
        agencies: list[AgencyMetadata] = []
        for i, row in enumerate(raw_agencies):
            try:
                agencies.append(AgencyMetadata(**row))
            except ValidationError as exc:
                log.error("agency.txt row %d failed validation: %s", i, exc)
                raise

        # --- feed_info.txt ---
        raw_feed_info = parse_csv_from_zip(zf, "feed_info.txt")
        try:
            feed_info = FeedInfoMetadata(**raw_feed_info[0])
        except (IndexError, ValidationError) as exc:
            log.error("feed_info.txt validation failed: %s", exc)
            raise

    # Validate date window
    start_dt = datetime.strptime(feed_info.feed_start_date, "%Y%m%d")
    end_dt = datetime.strptime(feed_info.feed_end_date, "%Y%m%d")
    if start_dt > end_dt:
        raise ValueError(
            f"feed_start_date ({feed_info.feed_start_date}) is after "
            f"feed_end_date ({feed_info.feed_end_date})."
        )

    log.info(
        "Feed validated: version=%s agencies=%d hash=%s...",
        feed_info.feed_version,
        len(agencies),
        archive_hash[:12],
    )

    return {
        "agencies": [a.model_dump() for a in agencies],
        "feed_info": feed_info.model_dump(),
        "checksum": archive_hash,
        "validated_at": datetime.now(timezone.utc).isoformat(),
    }

from datetime import date

def build_version_tag(
    feed_info: FeedInfoMetadata,
    archive_hash: str,
    scheme: str = "date",
) -> str:
    """
    Return a canonical version tag for artifact registry storage.

    scheme='date'  → '20240315'   (aligns with feed_start_date)
    scheme='hash'  → 'sha256:abc123...'  (cryptographic immutability)
    scheme='feed'  → passthrough of feed_version from feed_info.txt
    """
    if scheme == "date":
        return feed_info.feed_start_date
    if scheme == "hash":
        return f"sha256:{archive_hash}"
    if scheme == "feed":
        return feed_info.feed_version
    raise ValueError(f"Unknown versioning scheme: {scheme!r}")

def verify_extraction_result(result: dict, feed_path: Path) -> None:
    agencies = result["agencies"]
    feed_info = result["feed_info"]

    # At least one agency must be present
    assert len(agencies) >= 1, "No agencies extracted from agency.txt"

    # Every agency must have a non-empty agency_id
    for agency in agencies:
        assert agency["agency_id"], f"Empty agency_id found: {agency}"

    # agency_id values must be unique within the feed
    ids = [a["agency_id"] for a in agencies]
    assert len(ids) == len(set(ids)), f"Duplicate agency_id values: {ids}"

    # Date window must be valid and non-expired relative to today
    today = date.today().strftime("%Y%m%d")
    assert feed_info["feed_end_date"] >= today, (
        f"Feed has expired: feed_end_date={feed_info['feed_end_date']}"
    )

    # Checksum must match the file on disk (guards against race conditions)
    on_disk_hash = compute_sha256(feed_path)
    assert result["checksum"] == on_disk_hash, (
        "Checksum mismatch — archive was modified between extraction and verification"
    )

    print(
        f"Verification passed: {len(agencies)} agency record(s), "
        f"valid through {feed_info['feed_end_date']}"
    )

from concurrent.futures import ThreadPoolExecutor, as_completed
from pathlib import Path

def process_feed_batch(
    feed_paths: list[Path],
    max_workers: int = 8,
) -> list[dict]:
    """
    Validate and checksum a batch of GTFS archives in parallel.
    Returns a list of result dicts; logs and skips feeds that fail validation.
    """
    results = []
    with ThreadPoolExecutor(max_workers=max_workers) as pool:
        future_to_path = {
            pool.submit(extract_and_validate_feed, p): p
            for p in feed_paths
        }
        for future in as_completed(future_to_path):
            path = future_to_path[future]
            try:
                result = future.result()
                results.append(result)
            except Exception as exc:
                log.error("Feed %s failed: %s", path.name, exc)
    return results

import pandas as pd

def results_to_parquet(results: list[dict], output_path: Path) -> None:
    records = []
    for r in results:
        for agency in r["agencies"]:
            records.append({
                "agency_id": agency["agency_id"],
                "agency_name": agency["agency_name"],
                "agency_timezone": agency["agency_timezone"],
                "feed_version": r["feed_info"]["feed_version"],
                "feed_start_date": r["feed_info"]["feed_start_date"],
                "feed_end_date": r["feed_info"]["feed_end_date"],
                "archive_sha256": r["checksum"],
                "validated_at": r["validated_at"],
            })
    df = pd.DataFrame(records).astype({
        "agency_id": "string",
        "agency_name": "string",
        "agency_timezone": "string",
        "feed_version": "string",
        "archive_sha256": "string",
        "validated_at": "string",
    })
    df.to_parquet(output_path, index=False, compression="snappy")
    log.info("Wrote %d agency records to %s", len(df), output_path)