366 lines
14 KiB
Python
366 lines
14 KiB
Python
#!/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())
|