Automating GTFS Version Control with Python Scripts

import hashlib
import logging
import subprocess
import tempfile
import zipfile
from datetime import datetime
from pathlib import Path

import pandas as pd
import requests

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

# --- Configure per agency ---
GTFS_REPO      = Path("gtfs-version-control")   # local Git repo root
FEED_URL       = "https://example-transit-agency.com/gtfs.zip"
AGENCY_ID      = "metro-central"                 # used in tag names
# ----------------------------

HASH_STATE_FILE = GTFS_REPO / ".last_hash"


def stream_and_extract(url: str, extract_dir: Path) -> None:
    """Stream GTFS ZIP from url, extract all .txt files to extract_dir."""
    with requests.get(url, stream=True, timeout=60) as resp:
        resp.raise_for_status()
        with tempfile.NamedTemporaryFile(suffix=".zip", delete=False) as tmp:
            for chunk in resp.iter_content(chunk_size=65_536):
                tmp.write(chunk)
            tmp_path = Path(tmp.name)

    with zipfile.ZipFile(tmp_path, "r") as zf:
        txt_members = [m for m in zf.namelist() if m.endswith(".txt")]
        if not txt_members:
            raise ValueError(f"No .txt files found in archive from {url}")
        zf.extractall(extract_dir, members=txt_members)

    tmp_path.unlink(missing_ok=True)
    log.info("Extracted %d GTFS files to %s", len(txt_members), extract_dir)


def compute_normalized_hash(feed_dir: Path) -> str:
    """
    SHA-256 over all .txt files in sorted order, each normalized to a
    canonical byte sequence: dtype=str, whitespace stripped, columns and
    rows sorted alphabetically by first column.
    """
    sha256 = hashlib.sha256()
    files_processed = 0

    for csv_path in sorted(feed_dir.glob("*.txt")):
        try:
            df = pd.read_csv(
                csv_path,
                dtype=str,
                encoding="utf-8-sig",
                keep_default_na=False,
            )
            # Strip whitespace from every string cell
            df = df.apply(
                lambda col: col.str.strip() if col.dtype == "object" else col
            )
            # Canonical ordering: columns alphabetical, rows by first column
            df = df.sort_index(axis=1)
            if not df.empty:
                df = df.sort_values(by=df.columns[0], ignore_index=True)

            canonical = df.to_csv(index=False).encode("utf-8")
            sha256.update(csv_path.name.encode())  # include filename in digest
            sha256.update(canonical)
            files_processed += 1
        except Exception as exc:
            log.warning("Skipping %s: %s", csv_path.name, exc)

    if files_processed == 0:
        raise RuntimeError(f"No valid CSV files hashed in {feed_dir}")

    digest = sha256.hexdigest()
    log.info("Hashed %d files → %s", files_processed, digest[:16] + "…")
    return digest


def git_commit_and_tag(repo_dir: Path, feed_hash: str, agency: str) -> str:
    """
    Initialise repo if needed, stage all changes, commit, and create an
    annotated tag.  Returns the tag name so callers can log or store it.
    """
    def _run(*args: str) -> None:
        subprocess.run(list(args), cwd=repo_dir, check=True, capture_output=True)

    if not (repo_dir / ".git").exists():
        _run("git", "init", "-b", "main")
        _run("git", "config", "user.email", "[email protected]")
        _run("git", "config", "user.name", "GTFS Version Bot")

    _run("git", "add", ".")

    date_str   = datetime.now().strftime("%Y%m%d")
    short_hash = feed_hash[:12]
    tag_name   = f"{agency}/{date_str}/{short_hash}"
    commit_msg = f"feed({agency}): {date_str} hash={short_hash}"

    _run("git", "commit", "-m", commit_msg)
    _run("git", "tag", "-a", tag_name, "-m", commit_msg)
    log.info("Committed and tagged: %s", tag_name)
    return tag_name


def load_previous_hash() -> str:
    if HASH_STATE_FILE.exists():
        return HASH_STATE_FILE.read_text().strip()
    return ""


def save_hash(feed_hash: str) -> None:
    HASH_STATE_FILE.write_text(feed_hash)


def main() -> None:
    GTFS_REPO.mkdir(exist_ok=True)
    extract_dir = GTFS_REPO / "current"
    extract_dir.mkdir(exist_ok=True)

    log.info("Downloading feed from %s", FEED_URL)
    stream_and_extract(FEED_URL, extract_dir)

    feed_hash = compute_normalized_hash(extract_dir)
    previous  = load_previous_hash()

    if feed_hash == previous:
        log.info("Feed unchanged (hash=%s) — skipping commit.", feed_hash[:12])
        return

    log.info("Feed changed: %s → %s", previous[:12] or "none", feed_hash[:12])
    tag = git_commit_and_tag(GTFS_REPO, feed_hash, AGENCY_ID)
    save_hash(feed_hash)
    log.info("Version control complete. Tag: %s", tag)


if __name__ == "__main__":
    main()

import subprocess
from pathlib import Path

GTFS_REPO = Path("gtfs-version-control")
AGENCY_ID = "metro-central"

# List all tags for this agency in chronological order
result = subprocess.run(
    ["git", "tag", "--list", f"{AGENCY_ID}/*", "--sort=creatordate"],
    cwd=GTFS_REPO,
    capture_output=True,
    text=True,
    check=True,
)
tags = result.stdout.strip().splitlines()
print(f"Total snapshots: {len(tags)}")
for tag in tags[-5:]:          # show the five most recent
    print(" ", tag)

# Confirm the stored hash matches a fresh recalculation
import hashlib
import pandas as pd

def _recompute(feed_dir: Path) -> str:
    sha256 = hashlib.sha256()
    for p in sorted(feed_dir.glob("*.txt")):
        df = pd.read_csv(p, dtype=str, encoding="utf-8-sig", keep_default_na=False)
        df = df.apply(lambda c: c.str.strip() if c.dtype == "object" else c)
        df = df.sort_index(axis=1)
        if not df.empty:
            df = df.sort_values(by=df.columns[0], ignore_index=True)
        sha256.update(p.name.encode())
        sha256.update(df.to_csv(index=False).encode("utf-8"))
    return sha256.hexdigest()

stored = (GTFS_REPO / ".last_hash").read_text().strip()
computed = _recompute(GTFS_REPO / "current")
assert stored == computed, f"Hash mismatch: stored={stored[:12]} computed={computed[:12]}"
print("Hash verified:", computed[:12], "…")

Total snapshots: 7
  metro-central/20260610/3f8a1c4e7d2b
  metro-central/20260617/3f8a1c4e7d2b
  metro-central/20260621/3f8a1c4e7d2b
  metro-central/20260622/a1d9f3b82c50
  metro-central/20260624/a1d9f3b82c50
Hash verified: a1d9f3b82c50 …