#!/usr/bin/env python3 """Read-only source/target functional surface parity audit.""" from __future__ import annotations import argparse import fnmatch import json import sys from dataclasses import asdict, dataclass from pathlib import Path from typing import Iterable, Sequence PROJECT_ROOT = Path(__file__).resolve().parents[1] if str(PROJECT_ROOT) not in sys.path: sys.path.insert(0, str(PROJECT_ROOT)) from src.rag.project_config import env # noqa: E402 FUNCTIONAL_SUFFIXES = { ".cfg", ".ini", ".json", ".py", ".sh", ".toml", ".yaml", ".yml", } FUNCTIONAL_NAMES = { ".env.example", "Dockerfile", "Makefile", "requirements.txt", "requirements-rag.txt", } OPTIONAL_DOC_SUFFIXES = { ".md", } EXCLUDED_DIRS = { ".codegraph", ".git", ".pytest_cache", ".ruff_cache", ".runtime", ".vscode", "_sandboxes", "__pycache__", "backups", "batch_results", "benchmark_results", "bge-reranker-base", "data", "experiments", "frontend/node_modules", "graphify-out", "logs", "methodology_batch_Ft_spl", "methodology_batch_pecan_spl", "methodology_batch_sunflowers_spl", "mineru", "node_modules", "output", "paper", "tmp", "uploads", "test_supp_output", "test_v14_minimal_output", "test_v14_phase2_output", "v14_learning", "v14_output", "v14_2_test_output", "v15_output", } EXCLUDED_FILE_PATTERNS = ( "*.bak", "*.bak.*", "*.b64", "*.log", "*.pid", "*.pyc", "*.tmp", "api_keys.sh", ) ENTRYPOINT_PATTERNS = ( "api/*.py", "run*.py", "run*.sh", "start*.py", "start*.sh", "scripts/run*.py", "scripts/run*.sh", "scripts/start*.py", "scripts/start*.sh", "tools/*.py", ) @dataclass(frozen=True) class ParityReport: source_root: str target_root: str source_count: int target_count: int missing_count: int extra_count: int missing_entrypoint_count: int missing: list[str] extra: list[str] missing_entrypoints: list[str] def resolve_source_root(source_root: Path | None = None) -> Path: if source_root is not None: return source_root value = env("PLANTAI_SOURCE_ROOT") if value: return Path(value) raise SystemExit("PLANTAI_SOURCE_ROOT is required or pass --source-root.") def parse_args(argv: Iterable[str] | None = None) -> argparse.Namespace: parser = argparse.ArgumentParser( description="Compare source and PlantAI-rag code/config surfaces without editing files." ) parser.add_argument("--source-root", type=Path, default=None) parser.add_argument("--target-root", type=Path, default=PROJECT_ROOT) parser.add_argument("--include-docs", action="store_true", help="Include markdown documents.") parser.add_argument("--json", action="store_true", help="Emit JSON instead of text.") parser.add_argument( "--fail-on-missing-entrypoints", action="store_true", help="Exit non-zero when source entrypoint files are missing from target.", ) parser.add_argument("--max-items", type=int, default=80) return parser.parse_args(list(argv) if argv is not None else None) def normalize_rel(path: Path) -> str: return path.as_posix().removeprefix("./") def is_excluded_dir(rel_dir: str) -> bool: rel_dir = rel_dir.removeprefix("./") if not rel_dir or rel_dir == ".": return False parts = rel_dir.split("/") for index in range(len(parts)): candidate = "/".join(parts[: index + 1]) if candidate in EXCLUDED_DIRS: return True return False def is_excluded_file(rel_path: str) -> bool: name = Path(rel_path).name return any(fnmatch.fnmatch(name, pattern) for pattern in EXCLUDED_FILE_PATTERNS) def is_functional_file(rel_path: str, include_docs: bool = False) -> bool: path = Path(rel_path) if path.name in FUNCTIONAL_NAMES: return True suffixes = set(FUNCTIONAL_SUFFIXES) if include_docs: suffixes.update(OPTIONAL_DOC_SUFFIXES) return path.suffix.lower() in suffixes def discover_surface(root: Path, include_docs: bool = False) -> set[str]: files: set[str] = set() for path in root.rglob("*"): if not path.is_file(): continue rel_path = normalize_rel(path.relative_to(root)) if is_excluded_dir(str(Path(rel_path).parent)): continue if is_excluded_file(rel_path): continue if not is_functional_file(rel_path, include_docs=include_docs): continue files.add(rel_path) return files def is_entrypoint(rel_path: str) -> bool: return any(fnmatch.fnmatch(rel_path, pattern) for pattern in ENTRYPOINT_PATTERNS) def build_report(source_root: Path, target_root: Path, include_docs: bool = False) -> ParityReport: source_files = discover_surface(source_root, include_docs=include_docs) target_files = discover_surface(target_root, include_docs=include_docs) missing = sorted(source_files - target_files) extra = sorted(target_files - source_files) missing_entrypoints = sorted(path for path in missing if is_entrypoint(path)) return ParityReport( source_root=str(source_root), target_root=str(target_root), source_count=len(source_files), target_count=len(target_files), missing_count=len(missing), extra_count=len(extra), missing_entrypoint_count=len(missing_entrypoints), missing=missing, extra=extra, missing_entrypoints=missing_entrypoints, ) def print_report(report: ParityReport, max_items: int) -> None: print("Functional surface parity audit") summary = { "source_count": report.source_count, "target_count": report.target_count, "missing_count": report.missing_count, "extra_count": report.extra_count, "missing_entrypoint_count": report.missing_entrypoint_count, } print(json.dumps(summary, ensure_ascii=False, indent=2)) if report.missing_entrypoints: print("\n[missing_entrypoints]") for path in report.missing_entrypoints[:max_items]: print(f"- {path}") if len(report.missing_entrypoints) > max_items: print(f"- ... {len(report.missing_entrypoints) - max_items} more") if report.missing: print("\n[missing]") for path in report.missing[:max_items]: print(f"- {path}") if len(report.missing) > max_items: print(f"- ... {len(report.missing) - max_items} more") if report.extra: print("\n[extra]") for path in report.extra[:max_items]: print(f"- {path}") if len(report.extra) > max_items: print(f"- ... {len(report.extra) - max_items} more") def main(argv: Iterable[str] | None = None) -> int: args = parse_args(argv) source_root = resolve_source_root(args.source_root) report = build_report( source_root.resolve(), args.target_root.resolve(), include_docs=args.include_docs, ) if args.json: print(json.dumps(asdict(report), ensure_ascii=False, indent=2)) else: print_report(report, max_items=args.max_items) if args.fail_on_missing_entrypoints and report.missing_entrypoint_count: return 2 return 0 if __name__ == "__main__": raise SystemExit(main())