geointel/scripts/run_accuracy_phase1_baseline.py

#!/usr/bin/env python3
"""Create a bounded, read-only GeoIntel Phase-1 accuracy baseline."""

from __future__ import annotations

import argparse
import hashlib
import json
import re
import subprocess
from collections import Counter, defaultdict
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Iterable

SCHEMA_VERSION = 1
SKIP_DIRS = {".git", ".pytest_cache", ".ruff_cache", ".venv", "__pycache__", "dist", "node_modules"}
CODE_ROOTS = {
    "backend": "backend/app",
    "backend_tests": "backend/tests",
    "frontend": "frontend/src",
    "frontend_e2e": "frontend/e2e",
    "root_tests": "tests",
    "scripts": "scripts",
    "migrations": "backend/alembic/versions",
}
CODE_SUFFIXES = {".py", ".ts", ".tsx", ".js", ".mjs", ".sh", ".ps1"}
ARTIFACT_ROOTS = ("models", "datasets", "data", "storage", "artifacts", "output")
MODEL_SUFFIXES = {".pt", ".pth", ".onnx", ".engine", ".safetensors"}
RASTER_SUFFIXES = {".tif", ".tiff", ".vrt", ".jp2"}
VECTOR_SUFFIXES = {".geojson", ".gpkg", ".shp", ".fgb"}
HASH_SUFFIXES = MODEL_SUFFIXES | RASTER_SUFFIXES | VECTOR_SUFFIXES | {
    ".json", ".yaml", ".yml", ".csv", ".txt", ".md", ".lock"
}
MARKERS = {
    "fixture": re.compile(r"\bfixture\b", re.IGNORECASE),
    "mock": re.compile(r"\bmock(?:ed|ing|s)?\b", re.IGNORECASE),
    "placeholder": re.compile(r"\bplaceholder\b", re.IGNORECASE),
    "heuristic": re.compile(r"\bheuristic(?:s)?\b", re.IGNORECASE),
    "not_configured": re.compile(r"\bnot_configured\b", re.IGNORECASE),
    "todo": re.compile(r"\bTODO\b"),
    "fallback": re.compile(r"\b(?:fallback|fall back)\b", re.IGNORECASE),
}


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--repo-root", type=Path, default=Path(__file__).resolve().parents[1])
    parser.add_argument("--output-dir", type=Path, required=True)
    parser.add_argument("--max-hash-bytes", type=int, default=64 * 1024 * 1024)
    return parser.parse_args()


def now() -> str:
    return datetime.now(timezone.utc).isoformat()


def sha256_file(path: Path) -> str:
    digest = hashlib.sha256()
    with path.open("rb") as handle:
        for chunk in iter(lambda: handle.read(1024 * 1024), b""):
            digest.update(chunk)
    return digest.hexdigest()


def write_json(path: Path, payload: Any) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(
        json.dumps(payload, ensure_ascii=False, indent=2, sort_keys=True) + "\n",
        encoding="utf-8",
    )


def git(repo: Path, *arguments: str) -> str:
    result = subprocess.run(
        ["git", *arguments],
        cwd=repo,
        check=True,
        capture_output=True,
        text=True,
        encoding="utf-8",
        errors="replace",
    )
    return result.stdout


def iter_files(root: Path) -> Iterable[Path]:
    if not root.is_dir():
        return
    for path in sorted(root.rglob("*")):
        if path.is_file() and not any(part in SKIP_DIRS for part in path.parts):
            yield path


def line_count(path: Path) -> int:
    with path.open("r", encoding="utf-8", errors="replace") as handle:
        return sum(1 for _ in handle)


def code_inventory(repo: Path) -> dict[str, Any]:
    groups: dict[str, dict[str, int]] = {}
    for name, relative in CODE_ROOTS.items():
        files = [path for path in iter_files(repo / relative) if path.suffix.lower() in CODE_SUFFIXES]
        groups[name] = {
            "file_count": len(files),
            "line_count": sum(line_count(path) for path in files),
        }

    pytest_pattern = re.compile(r"^\s*(?:async\s+)?def\s+test_", re.MULTILINE)
    route_pattern = re.compile(r"@router\.(?:get|post|put|patch|delete)\s*\(")
    pytest_count = 0
    route_count = 0
    frontend_test_count = 0
    for base in (repo / "backend/tests", repo / "tests"):
        for path in iter_files(base):
            if path.suffix == ".py":
                pytest_count += len(pytest_pattern.findall(path.read_text(encoding="utf-8", errors="replace")))
    for path in iter_files(repo / "backend/app/api/routes"):
        if path.suffix == ".py":
            route_count += len(route_pattern.findall(path.read_text(encoding="utf-8", errors="replace")))
    for path in iter_files(repo / "frontend"):
        if ".test." in path.name.lower() or ".spec." in path.name.lower():
            frontend_test_count += 1
    return {
        "groups": groups,
        "pytest_test_function_count": pytest_count,
        "frontend_test_file_count": frontend_test_count,
        "api_route_decorator_count": route_count,
    }


def migration_inventory(repo: Path) -> dict[str, Any]:
    revision_re = re.compile(r'^revision\s*(?::[^=]+)?=\s*["\x27]([^"\x27]+)["\x27]', re.MULTILINE)
    down_re = re.compile(
        r'^down_revision\s*(?::[^=]+)?=\s*(?:["\x27]([^"\x27]+)["\x27]|None)',
        re.MULTILINE,
    )
    rows = []
    for path in iter_files(repo / "backend/alembic/versions"):
        if path.suffix != ".py":
            continue
        text = path.read_text(encoding="utf-8", errors="replace")
        revision = revision_re.search(text)
        down = down_re.search(text)
        rows.append({
            "path": path.relative_to(repo).as_posix(),
            "revision": revision.group(1) if revision else None,
            "down_revision": down.group(1) if down and down.group(1) else None,
        })
    revisions = {row["revision"] for row in rows if row["revision"]}
    parents = {row["down_revision"] for row in rows if row["down_revision"]}
    return {
        "count": len(rows),
        "records": rows,
        "heads_from_static_chain": sorted(revisions - parents),
        "missing_revision_identifiers": sum(row["revision"] is None for row in rows),
    }


def mirror_inventory(repo: Path, tracked: set[str]) -> dict[str, Any]:
    identical = 0
    different = []
    for relative in sorted(item for item in tracked if not item.startswith("geointel/")):
        mirror_relative = f"geointel/{relative}"
        if mirror_relative not in tracked:
            continue
        source = repo / relative
        mirror = repo / mirror_relative
        if not source.is_file() or not mirror.is_file():
            continue
        source_hash = sha256_file(source)
        mirror_hash = sha256_file(mirror)
        if source_hash == mirror_hash:
            identical += 1
        else:
            different.append({
                "path": relative,
                "root_sha256": source_hash,
                "mirror_sha256": mirror_hash,
                "root_size_bytes": source.stat().st_size,
                "mirror_size_bytes": mirror.stat().st_size,
            })
    return {
        "tracked_mirror_file_count": sum(item.startswith("geointel/") for item in tracked),
        "paired_identical_file_count": identical,
        "paired_different_file_count": len(different),
        "different_files": different,
        "risk": (
            "The tracked geointel/ repository mirror can create ambiguous imports, stale tests "
            "and local/deployment drift; Docker excludes it but local tools may not."
        ),
    }


def marker_inventory(repo: Path) -> dict[str, Any]:
    counts: Counter[str] = Counter()
    examples: dict[str, list[dict[str, Any]]] = defaultdict(list)
    for base in (repo / "backend/app", repo / "frontend/src", repo / "scripts"):
        for path in iter_files(base):
            if path.suffix.lower() not in CODE_SUFFIXES:
                continue
            relative = path.relative_to(repo).as_posix()
            for number, text in enumerate(
                path.read_text(encoding="utf-8", errors="replace").splitlines(), start=1
            ):
                for name, pattern in MARKERS.items():
                    if pattern.search(text):
                        counts[name] += 1
                        if len(examples[name]) < 25:
                            examples[name].append({
                                "path": relative,
                                "line": number,
                                "text": text.strip()[:240],
                            })
    return {
        "counts": dict(sorted(counts.items())),
        "examples": dict(sorted(examples.items())),
        "interpretation": "Triage signals only; production impact requires a traced contract/runtime path.",
    }


def artifact_role(path: Path) -> str:
    suffix = path.suffix.lower()
    name = path.name.lower()
    if suffix in MODEL_SUFFIXES:
        return "model_checkpoint"
    if "manifest" in name or name in {"dataset.yaml", "data.yaml"}:
        return "manifest"
    if any(token in name for token in ("audit", "evaluation", "assessment", "metric", "report")):
        return "evaluation_or_audit"
    if any(token in name for token in ("contact_sheet", "review")) or suffix in {".png", ".jpg", ".jpeg"}:
        return "visual_review"
    if suffix in RASTER_SUFFIXES:
        return "raster"
    if suffix in VECTOR_SUFFIXES:
        return "vector"
    if suffix in {".db", ".sqlite", ".sqlite3", ".wal", ".shm"} or ".db-" in name:
        return "database_runtime_state"
    if suffix == ".txt" and "label" in path.as_posix().lower():
        return "label"
    return "other"


def artifact_inventory(
    repo: Path,
    tracked: set[str],
    output: Path,
    max_hash_bytes: int,
) -> dict[str, Any]:
    records = []
    output = output.resolve()
    for root_name in ARTIFACT_ROOTS:
        for path in iter_files(repo / root_name):
            resolved = path.resolve()
            if resolved == output or output in resolved.parents:
                continue
            stat = path.stat()
            relative = path.relative_to(repo).as_posix()
            can_hash = stat.st_size <= max_hash_bytes and path.suffix.lower() in HASH_SUFFIXES
            records.append({
                "path": relative,
                "role": artifact_role(path),
                "size_bytes": stat.st_size,
                "modified_at": datetime.fromtimestamp(stat.st_mtime, timezone.utc).isoformat(),
                "tracked": relative in tracked,
                "sha256": sha256_file(path) if can_hash else None,
                "hash_omission_reason": None if can_hash else "suffix_or_size_limit",
            })
    roles = Counter(row["role"] for row in records)
    roots = Counter(row["path"].split("/", 1)[0] for row in records)
    return {
        "roots": list(ARTIFACT_ROOTS),
        "file_count": len(records),
        "total_size_bytes": sum(row["size_bytes"] for row in records),
        "role_counts": dict(sorted(roles.items())),
        "root_counts": dict(sorted(roots.items())),
        "model_checkpoint_count": roles.get("model_checkpoint", 0),
        "records": records,
        "limitations": [
            "Ignored Tower corpora and mounted model volumes can be absent locally.",
            "Large/non-evidence files are inventoried without a SHA-256 above the configured ceiling.",
            "Near-duplicate imagery and semantic label quality need dedicated corpus checks.",
        ],
    }


def main() -> int:
    args = parse_args()
    repo = args.repo_root.expanduser().resolve()
    output = args.output_dir.expanduser()
    output = output.resolve() if output.is_absolute() else (repo / output).resolve()
    if not (repo / ".git").exists():
        raise SystemExit(f"Not a Git repository root: {repo}")
    if output == repo:
        raise SystemExit("Output directory must not equal the repository root")

    tracked = set(filter(None, git(repo, "ls-files").splitlines()))
    repository = {
        "schema_version": SCHEMA_VERSION,
        "generated_at": now(),
        "repo_root": str(repo),
        "git": {
            "branch": git(repo, "branch", "--show-current").strip(),
            "head": git(repo, "rev-parse", "HEAD").strip(),
            "status_porcelain": [line for line in git(repo, "status", "--short").splitlines() if line],
            "tracked_file_count": len(tracked),
            "top_level_tracked_counts": dict(sorted(Counter(
                item.split("/", 1)[0] for item in tracked
            ).items())),
        },
        "code": code_inventory(repo),
        "migrations": migration_inventory(repo),
        "tracked_mirror": mirror_inventory(repo, tracked),
    }
    signals = marker_inventory(repo)
    artifacts = artifact_inventory(repo, tracked, output, args.max_hash_bytes)
    findings = []
    mirror = repository["tracked_mirror"]
    if mirror["tracked_mirror_file_count"]:
        findings.append({
            "id": "P1-REPO-001",
            "severity": "high",
            "title": "Tracked nested repository mirror creates ambiguous source state",
            "evidence": {
                "tracked_mirror_file_count": mirror["tracked_mirror_file_count"],
                "paired_different_file_count": mirror["paired_different_file_count"],
            },
        })
    if artifacts["model_checkpoint_count"] == 0:
        findings.append({
            "id": "P1-ML-LOCAL-001",
            "severity": "info",
            "title": "No local checkpoint is available in the repository checkout",
            "interpretation": "Production model truth must be verified on the mounted Tower volume.",
        })
    summary = {
        "schema_version": SCHEMA_VERSION,
        "generated_at": now(),
        "status": "findings_present" if findings else "no_static_findings",
        "finding_count": len(findings),
        "findings": findings,
        "baseline_scope": [
            "tracked repository state",
            "static code/test/migration inventory",
            "tracked mirror comparison",
            "local artifact inventory",
            "mock/fixture/placeholder/fallback triage signals",
        ],
        "separate_required_evidence": [
            "Tower database/storage audit",
            "Tower CUDA/model preflight and representative inference",
            "corpus leakage/duplicate/label/time review",
            "independent human visual review",
        ],
    }
    write_json(output / "repository-inventory.json", repository)
    write_json(output / "local-artifact-inventory.json", artifacts)
    write_json(output / "static-risk-signals.json", signals)
    write_json(output / "phase1-baseline-summary.json", summary)
    print(json.dumps(summary, ensure_ascii=False, indent=2, sort_keys=True))
    return 0


if __name__ == "__main__":
    raise SystemExit(main())