372 lines
14 KiB
Python
372 lines
14 KiB
Python
from __future__ import annotations
|
|
|
|
import argparse
|
|
import fnmatch
|
|
import json
|
|
import pathlib
|
|
from collections.abc import Mapping
|
|
from typing import Any
|
|
|
|
import yaml
|
|
|
|
|
|
CONTEXT_MD = "review-context.md"
|
|
MANIFEST_JSON = "review-file-manifest.json"
|
|
DEFAULT_MAX_FILE_BYTES = 1_000_000
|
|
|
|
|
|
def as_list(value: Any) -> list[str]:
|
|
if value is None:
|
|
return []
|
|
if isinstance(value, list):
|
|
return [str(item) for item in value]
|
|
return [str(value)]
|
|
|
|
|
|
def load_metadata(path: pathlib.Path | None) -> dict[str, Any]:
|
|
if path is None:
|
|
return {}
|
|
text = path.read_text(encoding="utf-8")
|
|
if path.suffix.casefold() == ".json":
|
|
parsed = json.loads(text)
|
|
else:
|
|
parsed = yaml.safe_load(text)
|
|
if parsed is None:
|
|
return {}
|
|
if not isinstance(parsed, Mapping):
|
|
raise ValueError("Metadata must be a JSON or YAML object.")
|
|
return dict(parsed)
|
|
|
|
|
|
def rel_path(path: pathlib.Path, project_root: pathlib.Path) -> str:
|
|
return path.relative_to(project_root).as_posix()
|
|
|
|
|
|
def resolve_source_root(raw_source: str, project_root: pathlib.Path) -> pathlib.Path:
|
|
source = pathlib.Path(raw_source)
|
|
if not source.is_absolute():
|
|
source = project_root / source
|
|
return source.resolve()
|
|
|
|
|
|
def normalize_source_root(raw_source: str, project_root: pathlib.Path) -> str:
|
|
source = resolve_source_root(raw_source, project_root)
|
|
try:
|
|
return rel_path(source, project_root)
|
|
except ValueError:
|
|
return str(source)
|
|
|
|
|
|
def path_matches(path: str, patterns: list[str]) -> bool:
|
|
if not patterns:
|
|
return True
|
|
basename = pathlib.PurePosixPath(path).name
|
|
for pattern in patterns:
|
|
root_pattern = pattern[3:] if pattern.startswith("**/") else None
|
|
if fnmatch.fnmatch(path, pattern) or fnmatch.fnmatch(basename, pattern):
|
|
return True
|
|
if root_pattern and (fnmatch.fnmatch(path, root_pattern) or fnmatch.fnmatch(basename, root_pattern)):
|
|
return True
|
|
return False
|
|
|
|
|
|
def iter_source_files(source: pathlib.Path) -> list[pathlib.Path]:
|
|
if source.is_file():
|
|
return [source]
|
|
if source.is_dir():
|
|
return sorted(path for path in source.rglob("*") if path.is_file())
|
|
return []
|
|
|
|
|
|
def empty_manifest(
|
|
project_root: pathlib.Path,
|
|
review_id: str,
|
|
context_profile: str,
|
|
source_roots: list[str],
|
|
metadata: dict[str, Any],
|
|
) -> dict[str, Any]:
|
|
return {
|
|
"review_id": review_id,
|
|
"context_profile": context_profile,
|
|
"project_root": str(project_root),
|
|
"source_roots": source_roots,
|
|
"missing_source_roots": [],
|
|
"files_included": [],
|
|
"files_excluded": [],
|
|
"important_reports": [],
|
|
"test_or_validation_outputs": [],
|
|
"diff_or_changed_file_references": [],
|
|
"prior_decision_references": [],
|
|
"known_non_goals": as_list(metadata.get("known_non_goals", metadata.get("non_goals"))),
|
|
"open_questions_for_reviewer_or_agent": as_list(
|
|
metadata.get("open_questions_for_reviewer_or_agent", metadata.get("open_questions"))
|
|
),
|
|
"metadata": metadata,
|
|
"blocking_errors": [],
|
|
"machine_readable_summary": {},
|
|
}
|
|
|
|
|
|
def classify_paths(included_paths: list[str]) -> dict[str, list[str]]:
|
|
important_reports: list[str] = []
|
|
validation_outputs: list[str] = []
|
|
diffs: list[str] = []
|
|
decisions: list[str] = []
|
|
|
|
for path in included_paths:
|
|
folded = path.casefold()
|
|
name = pathlib.PurePosixPath(path).name.casefold()
|
|
if any(token in folded for token in ("report", "audit", "review")):
|
|
important_reports.append(path)
|
|
if any(token in folded for token in ("test", "validation", "gate", "log")):
|
|
validation_outputs.append(path)
|
|
if any(token in folded for token in ("diff", "changed-file", "changed_files", "changed-files")):
|
|
diffs.append(path)
|
|
if any(token in folded for token in ("decision", "owner-decision", "approval")) or "decision" in name:
|
|
decisions.append(path)
|
|
|
|
return {
|
|
"important_reports": sorted(set(important_reports)),
|
|
"test_or_validation_outputs": sorted(set(validation_outputs)),
|
|
"diff_or_changed_file_references": sorted(set(diffs)),
|
|
"prior_decision_references": sorted(set(decisions)),
|
|
}
|
|
|
|
|
|
def build_manifest(
|
|
project_root: pathlib.Path,
|
|
review_id: str,
|
|
context_profile: str,
|
|
source_roots_raw: list[str],
|
|
include_patterns: list[str],
|
|
exclude_patterns: list[str],
|
|
metadata: dict[str, Any],
|
|
max_file_bytes: int,
|
|
) -> dict[str, Any]:
|
|
normalized_source_roots = [
|
|
normalize_source_root(raw_source, project_root) for raw_source in source_roots_raw
|
|
]
|
|
manifest = empty_manifest(project_root, review_id, context_profile, normalized_source_roots, metadata)
|
|
seen: dict[str, pathlib.Path] = {}
|
|
|
|
for raw_source in source_roots_raw:
|
|
source = resolve_source_root(raw_source, project_root)
|
|
normalized = normalize_source_root(raw_source, project_root)
|
|
if not source.exists():
|
|
manifest["missing_source_roots"].append(normalized)
|
|
manifest["blocking_errors"].append(f"Missing source root: {normalized}")
|
|
continue
|
|
try:
|
|
source.relative_to(project_root)
|
|
except ValueError:
|
|
manifest["missing_source_roots"].append(normalized)
|
|
manifest["blocking_errors"].append(f"Source root is outside project_root: {normalized}")
|
|
continue
|
|
for file_path in iter_source_files(source):
|
|
rel = rel_path(file_path.resolve(), project_root)
|
|
seen[rel] = file_path
|
|
|
|
included: list[dict[str, Any]] = []
|
|
excluded: list[dict[str, Any]] = []
|
|
for file_rel in sorted(seen):
|
|
file_path = seen[file_rel]
|
|
if not path_matches(file_rel, include_patterns):
|
|
continue
|
|
if exclude_patterns and path_matches(file_rel, exclude_patterns):
|
|
excluded.append({"path": file_rel, "reason": "exclude_pattern"})
|
|
continue
|
|
size = file_size_bytes(file_path)
|
|
if size > max_file_bytes:
|
|
excluded.append({"path": file_rel, "reason": "size_exceeds_limit", "size_bytes": size})
|
|
continue
|
|
source_root = source_root_for_file(file_path, source_roots_raw, project_root)
|
|
included.append({"path": file_rel, "size_bytes": size, "source_root": source_root})
|
|
|
|
manifest["files_included"] = included
|
|
manifest["files_excluded"] = sorted(excluded, key=lambda item: item["path"])
|
|
categories = classify_paths([item["path"] for item in included])
|
|
manifest.update(categories)
|
|
finalize_manifest(manifest)
|
|
return manifest
|
|
|
|
|
|
def source_root_for_file(file_path: pathlib.Path, source_roots_raw: list[str], project_root: pathlib.Path) -> str:
|
|
file_path = file_path.resolve()
|
|
best: tuple[int, str] | None = None
|
|
for raw_source in source_roots_raw:
|
|
source = resolve_source_root(raw_source, project_root)
|
|
if source.is_file() and source == file_path:
|
|
root_rel = normalize_source_root(raw_source, project_root)
|
|
score = len(pathlib.PurePosixPath(root_rel).parts)
|
|
if best is None or score > best[0]:
|
|
best = (score, root_rel)
|
|
elif source.is_dir():
|
|
try:
|
|
file_path.relative_to(source)
|
|
except ValueError:
|
|
continue
|
|
root_rel = normalize_source_root(raw_source, project_root)
|
|
score = len(pathlib.PurePosixPath(root_rel).parts)
|
|
if best is None or score > best[0]:
|
|
best = (score, root_rel)
|
|
return "." if best is None or best[1] == "" else best[1]
|
|
|
|
|
|
def file_size_bytes(path: pathlib.Path) -> int:
|
|
return len(path.read_text(encoding="utf-8", errors="replace").encode("utf-8"))
|
|
|
|
|
|
def finalize_manifest(manifest: dict[str, Any]) -> None:
|
|
status = "FAIL" if manifest["blocking_errors"] else "PASS"
|
|
manifest["machine_readable_summary"] = {
|
|
"status": status,
|
|
"source_roots_count": len(manifest["source_roots"]),
|
|
"missing_source_roots_count": len(manifest["missing_source_roots"]),
|
|
"files_included_count": len(manifest["files_included"]),
|
|
"files_excluded_count": len(manifest["files_excluded"]),
|
|
"important_reports_count": len(manifest["important_reports"]),
|
|
"validation_outputs_count": len(manifest["test_or_validation_outputs"]),
|
|
"diff_references_count": len(manifest["diff_or_changed_file_references"]),
|
|
"prior_decisions_count": len(manifest["prior_decision_references"]),
|
|
"blocking_error_count": len(manifest["blocking_errors"]),
|
|
}
|
|
|
|
|
|
def write_manifest(manifest: dict[str, Any], output_dir: pathlib.Path) -> None:
|
|
(output_dir / MANIFEST_JSON).write_text(
|
|
json.dumps(manifest, ensure_ascii=False, indent=2),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
|
|
def write_context(manifest: dict[str, Any], output_dir: pathlib.Path) -> None:
|
|
lines = [
|
|
"# Review Context",
|
|
"",
|
|
"This context is an index. It does not replace reading source files.",
|
|
"",
|
|
"## Metadata",
|
|
"",
|
|
f"- Review ID: `{manifest['review_id']}`",
|
|
f"- Context profile: `{manifest['context_profile']}`",
|
|
f"- Status: `{manifest['machine_readable_summary']['status']}`",
|
|
"",
|
|
"## Source Roots",
|
|
"",
|
|
]
|
|
for source in manifest["source_roots"]:
|
|
lines.append(f"- `{source}`")
|
|
if manifest["missing_source_roots"]:
|
|
lines.extend(["", "## Missing Source Roots", ""])
|
|
for source in manifest["missing_source_roots"]:
|
|
lines.append(f"- `{source}`")
|
|
lines.extend(["", "## Files Included", ""])
|
|
if manifest["files_included"]:
|
|
for item in manifest["files_included"]:
|
|
lines.append(f"- `{item['path']}` ({item['size_bytes']} bytes)")
|
|
else:
|
|
lines.append("- None")
|
|
lines.extend(["", "## Files Excluded", ""])
|
|
if manifest["files_excluded"]:
|
|
for item in manifest["files_excluded"]:
|
|
detail = f", {item['size_bytes']} bytes" if "size_bytes" in item else ""
|
|
lines.append(f"- `{item['path']}`: {item['reason']}{detail}")
|
|
else:
|
|
lines.append("- None")
|
|
for title, key in [
|
|
("Important Reports", "important_reports"),
|
|
("Test Or Validation Outputs", "test_or_validation_outputs"),
|
|
("Diff Or Changed File References", "diff_or_changed_file_references"),
|
|
("Prior Decision References", "prior_decision_references"),
|
|
("Known Non Goals", "known_non_goals"),
|
|
("Open Questions For Reviewer Or Agent", "open_questions_for_reviewer_or_agent"),
|
|
]:
|
|
lines.extend(["", f"## {title}", ""])
|
|
values = manifest[key]
|
|
if values:
|
|
for value in values:
|
|
lines.append(f"- `{value}`" if isinstance(value, str) else f"- {value}")
|
|
else:
|
|
lines.append("- None")
|
|
lines.extend(["", "## Blocking Errors", ""])
|
|
if manifest["blocking_errors"]:
|
|
for error in manifest["blocking_errors"]:
|
|
lines.append(f"- {error}")
|
|
else:
|
|
lines.append("- None")
|
|
lines.extend(["", "## Machine Readable Manifest", "", f"- `{MANIFEST_JSON}`"])
|
|
(output_dir / CONTEXT_MD).write_text("\n".join(lines) + "\n", encoding="utf-8")
|
|
|
|
|
|
def write_outputs(manifest: dict[str, Any], output_dir: pathlib.Path) -> None:
|
|
output_dir.mkdir(parents=True, exist_ok=True)
|
|
write_manifest(manifest, output_dir)
|
|
write_context(manifest, output_dir)
|
|
|
|
|
|
def run(
|
|
project_root: pathlib.Path,
|
|
review_id: str,
|
|
output_dir: pathlib.Path,
|
|
source_roots: list[str],
|
|
include_patterns: list[str],
|
|
exclude_patterns: list[str],
|
|
metadata_path: pathlib.Path | None,
|
|
context_profile: str,
|
|
max_file_bytes: int,
|
|
) -> int:
|
|
project_root = project_root.resolve()
|
|
output_dir = output_dir.resolve()
|
|
output_dir.mkdir(parents=True, exist_ok=True)
|
|
metadata = load_metadata(metadata_path)
|
|
if not project_root.is_dir():
|
|
manifest = empty_manifest(project_root, review_id, context_profile, source_roots, metadata)
|
|
manifest["blocking_errors"].append(f"project_root is not a directory: {project_root}")
|
|
finalize_manifest(manifest)
|
|
write_outputs(manifest, output_dir)
|
|
return 2
|
|
|
|
normalized_sources = source_roots or ["."]
|
|
normalized_includes = include_patterns or ["**/*"]
|
|
manifest = build_manifest(
|
|
project_root=project_root,
|
|
review_id=review_id,
|
|
context_profile=context_profile,
|
|
source_roots_raw=normalized_sources,
|
|
include_patterns=normalized_includes,
|
|
exclude_patterns=exclude_patterns,
|
|
metadata=metadata,
|
|
max_file_bytes=max_file_bytes,
|
|
)
|
|
write_outputs(manifest, output_dir)
|
|
return 1 if manifest["blocking_errors"] else 0
|
|
|
|
|
|
def main(argv: list[str] | None = None) -> int:
|
|
parser = argparse.ArgumentParser(description="Build a file-first review context index.")
|
|
parser.add_argument("--project-root", required=True, type=pathlib.Path)
|
|
parser.add_argument("--review-id", required=True)
|
|
parser.add_argument("--output-dir", required=True, type=pathlib.Path)
|
|
parser.add_argument("--source-root", action="append", default=[])
|
|
parser.add_argument("--include-pattern", action="append", default=[])
|
|
parser.add_argument("--exclude-pattern", action="append", default=[])
|
|
parser.add_argument("--metadata", type=pathlib.Path)
|
|
parser.add_argument("--context-profile", default="generic")
|
|
parser.add_argument("--max-file-bytes", type=int, default=DEFAULT_MAX_FILE_BYTES)
|
|
args = parser.parse_args(argv)
|
|
return run(
|
|
project_root=args.project_root,
|
|
review_id=args.review_id,
|
|
output_dir=args.output_dir,
|
|
source_roots=args.source_root,
|
|
include_patterns=args.include_pattern,
|
|
exclude_patterns=args.exclude_pattern,
|
|
metadata_path=args.metadata,
|
|
context_profile=args.context_profile,
|
|
max_file_bytes=args.max_file_bytes,
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|