portfolio-hub/scripts/check_docs_sync.py

#!/usr/bin/env python3
"""Guardrail check for implementation-to-spec synchronization.

Rules:
- If files under app/ change, at least one docs/technical/ file must be part of the same change set.
- If files under app/api/ or app/schemas/ change, docs/technical/api-contracts.md must be part of the same change set.
- If files under migrations/versions/ change, docs/technical/persistence-and-migrations.md must be part of the same change set.
- If migrations/versions/ files change, revision IDs found in changed migration files
    must appear in docs/technical/persistence-and-migrations.md.
"""

from __future__ import annotations

import argparse
import re
import subprocess
import sys
from pathlib import Path

ROOT = Path(__file__).resolve().parents[1]
TECHNICAL_DOC_PREFIX = "docs/technical/"
DB_SPEC = "docs/technical/persistence-and-migrations.md"
API_EXAMPLES_SPEC = "docs/technical/api-contracts.md"
APP_PREFIX = "app/"
API_PREFIXES = ("app/api/", "app/schemas/")
MIGRATION_PREFIX = "migrations/versions/"


class CheckError(Exception):
    pass


def _run_git(args: list[str]) -> str:
    result = subprocess.run(
        ["git", *args],
        cwd=ROOT,
        check=True,
        capture_output=True,
        text=True,
    )
    return result.stdout


def _ref_exists(ref: str) -> bool:
    result = subprocess.run(
        ["git", "rev-parse", "--verify", "--quiet", ref],
        cwd=ROOT,
        check=False,
        capture_output=True,
        text=True,
    )
    return result.returncode == 0


def _get_changed_files(*, staged: bool, against: str | None, head: str) -> list[str]:
    if staged:
        output = _run_git(["diff", "--name-only", "--cached", "--diff-filter=ACMR"])
    elif against:
        if not _ref_exists(against):
            raise CheckError(f"base ref not found: {against}")
        if not _ref_exists(head):
            raise CheckError(f"head ref not found: {head}")
        output = _run_git(["diff", "--name-only", "--diff-filter=ACMR", f"{against}...{head}"])
    else:
        if not _ref_exists("HEAD~1"):
            # For a repository with a single commit, there is nothing to compare against.
            return []
        output = _run_git(["diff", "--name-only", "--diff-filter=ACMR", "HEAD~1", "HEAD"])
    return [line.strip() for line in output.splitlines() if line.strip()]


def _extract_revision_id(path: Path) -> str | None:
    text = path.read_text(encoding="utf-8")
    match = re.search(r'^revision\s*=\s*["\']([^"\']+)["\']', text, flags=re.MULTILINE)
    return match.group(1) if match else None


def _collect_changed_revision_ids(changed_files: list[str]) -> list[str]:
    revision_ids: list[str] = []
    for file in changed_files:
        if not file.startswith(MIGRATION_PREFIX) or not file.endswith(".py"):
            continue
        migration_path = ROOT / file
        if not migration_path.exists():
            continue
        revision_id = _extract_revision_id(migration_path)
        if revision_id:
            revision_ids.append(revision_id)
    return revision_ids


def run_check(*, staged: bool, against: str | None, head: str) -> int:
    try:
        changed_files = _get_changed_files(staged=staged, against=against, head=head)
    except CheckError as exc:
        print(f"docs-sync-check: {exc}")
        return 2
    except subprocess.CalledProcessError as exc:
        print("docs-sync-check: failed to read git diff")
        print(exc.stderr.strip())
        return 2

    if not changed_files:
        print("docs-sync-check: no changes detected")
        return 0

    needs_general_spec = any(path.startswith(APP_PREFIX) for path in changed_files)
    needs_api_examples = any(path.startswith(prefix) for prefix in API_PREFIXES for path in changed_files)
    needs_db_spec = any(path.startswith(MIGRATION_PREFIX) for path in changed_files)

    errors: list[str] = []

    if needs_general_spec and not any(path.startswith(TECHNICAL_DOC_PREFIX) for path in changed_files):
        errors.append(
            "Application code changed but no docs/technical/ document was updated in the same commit/PR."
        )

    if needs_api_examples and API_EXAMPLES_SPEC not in changed_files:
        errors.append(
            "API/schema code changed but docs/technical/api-contracts.md was not updated in the same commit/PR."
        )

    if needs_db_spec and DB_SPEC not in changed_files:
        errors.append(
            "Migration changed but docs/technical/persistence-and-migrations.md was not updated in the same commit/PR."
        )

    if needs_db_spec:
        db_spec_path = ROOT / DB_SPEC
        if not db_spec_path.exists():
            errors.append("docs/technical/persistence-and-migrations.md is missing.")
        else:
            db_spec_text = db_spec_path.read_text(encoding="utf-8")
            changed_revisions = _collect_changed_revision_ids(changed_files)
            missing_revisions = [rev for rev in changed_revisions if rev not in db_spec_text]
            if missing_revisions:
                errors.append(
                    "Migration revision IDs missing in docs/technical/persistence-and-migrations.md: "
                    + ", ".join(sorted(set(missing_revisions)))
                )

    if errors:
        print("docs-sync-check: FAILED")
        for error in errors:
            print(f"- {error}")
        print("\nChanged files:")
        for file in changed_files:
            print(f"- {file}")
        return 1

    print("docs-sync-check: OK")
    return 0


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="Verify implementation and docs stay in sync.")
    parser.add_argument(
        "--staged",
        action="store_true",
        help="Check staged files (for pre-commit).",
    )
    parser.add_argument(
        "--against",
        default=None,
        help="Git base ref/sha to diff against (for CI).",
    )
    parser.add_argument(
        "--head",
        default="HEAD",
        help="Git head ref/sha used with --against.",
    )
    return parser.parse_args()


if __name__ == "__main__":
    args = parse_args()
    sys.exit(run_check(staged=args.staged, against=args.against, head=args.head))