PlantAI-rag/tools/audit_functional_surface_parity.py

#!/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())