368 lines
14 KiB
Python
368 lines
14 KiB
Python
from __future__ import annotations
|
|
|
|
import argparse
|
|
import fnmatch
|
|
import json
|
|
import pathlib
|
|
import sys
|
|
import zipfile
|
|
from collections.abc import Iterable, Mapping
|
|
from typing import Any
|
|
|
|
import yaml
|
|
|
|
|
|
REPORT_JSON = "bundle-audit.json"
|
|
REPORT_MD = "bundle-audit.md"
|
|
|
|
|
|
BUILTIN_PROFILES: dict[str, dict[str, Any]] = {
|
|
"generic": {
|
|
"required_files": {
|
|
"brief": ["*brief*.md", "*brief*.markdown"],
|
|
"manifest": ["*manifest*.json", "*manifest*.yaml", "*manifest*.yml"],
|
|
"validation_sidecar": [
|
|
"*validation*sidecar*.json",
|
|
"*validation*sidecar*.yaml",
|
|
"*validation*sidecar*.yml",
|
|
],
|
|
},
|
|
"optional_files": {
|
|
"report": ["*report*.md", "*report*.markdown"],
|
|
"zip": ["*.zip"],
|
|
"changed_files": ["*changed*files*.md", "*changed-files*.json", "*changed*files*.json"],
|
|
},
|
|
"allowed_extra_patterns": [],
|
|
}
|
|
}
|
|
|
|
|
|
def as_list(value: Any) -> list[str]:
|
|
if value is None:
|
|
return []
|
|
if isinstance(value, str):
|
|
return [value]
|
|
if isinstance(value, Iterable):
|
|
return [str(item) for item in value if item is not None]
|
|
return [str(value)]
|
|
|
|
|
|
def load_structured(path: pathlib.Path) -> Any:
|
|
text = path.read_text(encoding="utf-8")
|
|
suffix = path.suffix.casefold()
|
|
if suffix == ".json":
|
|
return json.loads(text)
|
|
if suffix in {".yaml", ".yml"}:
|
|
return yaml.safe_load(text)
|
|
return text
|
|
|
|
|
|
def load_profile_config(path: pathlib.Path | None) -> dict[str, Any]:
|
|
if path is None:
|
|
return {}
|
|
parsed = load_structured(path)
|
|
if not isinstance(parsed, Mapping):
|
|
raise ValueError("Profile config must be a JSON or YAML object.")
|
|
return dict(parsed)
|
|
|
|
|
|
def get_profile(profile_name: str, config: Mapping[str, Any]) -> dict[str, Any]:
|
|
profiles = dict(BUILTIN_PROFILES)
|
|
configured_profiles = config.get("profiles") if isinstance(config, Mapping) else None
|
|
if isinstance(configured_profiles, Mapping):
|
|
for name, profile in configured_profiles.items():
|
|
if isinstance(profile, Mapping):
|
|
profiles[str(name)] = dict(profile)
|
|
if profile_name not in profiles:
|
|
raise ValueError(f"Unknown profile: {profile_name}")
|
|
return normalize_profile(profiles[profile_name])
|
|
|
|
|
|
def normalize_pattern_map(raw: Any) -> dict[str, list[str]]:
|
|
if not isinstance(raw, Mapping):
|
|
return {}
|
|
return {str(key): as_list(value) for key, value in raw.items()}
|
|
|
|
|
|
def normalize_profile(raw_profile: Mapping[str, Any]) -> dict[str, Any]:
|
|
return {
|
|
"required_files": normalize_pattern_map(raw_profile.get("required_files")),
|
|
"optional_files": normalize_pattern_map(raw_profile.get("optional_files")),
|
|
"allowed_extra_patterns": as_list(raw_profile.get("allowed_extra_patterns")),
|
|
}
|
|
|
|
|
|
def rel_path(path: pathlib.Path, root: pathlib.Path) -> str:
|
|
return path.relative_to(root).as_posix()
|
|
|
|
|
|
def discover_files(bundle_root: pathlib.Path) -> list[pathlib.Path]:
|
|
return sorted(path for path in bundle_root.rglob("*") if path.is_file())
|
|
|
|
|
|
def match_patterns(file_rel: str, patterns: list[str]) -> bool:
|
|
basename = pathlib.PurePosixPath(file_rel).name
|
|
return any(fnmatch.fnmatch(file_rel, pattern) or fnmatch.fnmatch(basename, pattern) for pattern in patterns)
|
|
|
|
|
|
def files_for_patterns(files_rel: list[str], patterns: list[str]) -> list[str]:
|
|
return sorted(file_rel for file_rel in files_rel if match_patterns(file_rel, patterns))
|
|
|
|
|
|
def status_for_category(files: list[str], missing_status: str = "missing") -> dict[str, Any]:
|
|
return {"status": "present" if files else missing_status, "files": files}
|
|
|
|
|
|
def inspect_structured_files(
|
|
bundle_root: pathlib.Path,
|
|
file_rels: list[str],
|
|
missing_status: str = "missing",
|
|
) -> dict[str, Any]:
|
|
if not file_rels:
|
|
return {"status": missing_status, "files": []}
|
|
errors: list[str] = []
|
|
for file_rel in file_rels:
|
|
try:
|
|
load_structured(bundle_root / file_rel)
|
|
except Exception as exc:
|
|
errors.append(f"{file_rel}: {exc}")
|
|
if errors:
|
|
return {"status": "unreadable", "files": file_rels, "errors": errors}
|
|
return {"status": "present", "files": file_rels}
|
|
|
|
|
|
def inspect_zip_files(bundle_root: pathlib.Path, zip_rels: list[str]) -> dict[str, Any]:
|
|
if not zip_rels:
|
|
return {"status": "missing", "files": []}
|
|
errors: list[str] = []
|
|
entry_counts: dict[str, int] = {}
|
|
for file_rel in zip_rels:
|
|
path = bundle_root / file_rel
|
|
try:
|
|
with zipfile.ZipFile(path) as archive:
|
|
bad_member = archive.testzip()
|
|
if bad_member is not None:
|
|
errors.append(f"{file_rel}: unreadable member {bad_member}")
|
|
entry_counts[file_rel] = len(archive.namelist())
|
|
except Exception as exc:
|
|
errors.append(f"{file_rel}: {exc}")
|
|
if errors:
|
|
return {"status": "unreadable", "files": zip_rels, "errors": errors, "entry_counts": entry_counts}
|
|
return {"status": "readable", "files": zip_rels, "entry_counts": entry_counts}
|
|
|
|
|
|
def category_matches(profile: Mapping[str, Any], files_rel: list[str]) -> tuple[dict[str, list[str]], dict[str, list[str]]]:
|
|
required_matches = {
|
|
category: files_for_patterns(files_rel, patterns)
|
|
for category, patterns in profile["required_files"].items()
|
|
}
|
|
optional_matches = {
|
|
category: files_for_patterns(files_rel, patterns)
|
|
for category, patterns in profile["optional_files"].items()
|
|
}
|
|
return required_matches, optional_matches
|
|
|
|
|
|
def allowed_patterns(profile: Mapping[str, Any]) -> list[str]:
|
|
patterns: list[str] = []
|
|
for pattern_map_name in ("required_files", "optional_files"):
|
|
for category_patterns in profile[pattern_map_name].values():
|
|
patterns.extend(category_patterns)
|
|
patterns.extend(profile.get("allowed_extra_patterns", []))
|
|
return patterns
|
|
|
|
|
|
def extra_files(files_rel: list[str], profile: Mapping[str, Any]) -> list[str]:
|
|
patterns = allowed_patterns(profile)
|
|
return [file_rel for file_rel in files_rel if not match_patterns(file_rel, patterns)]
|
|
|
|
|
|
def empty_report(bundle_root: pathlib.Path, profile_name: str) -> dict[str, Any]:
|
|
return {
|
|
"bundle_root": str(bundle_root),
|
|
"profile": profile_name,
|
|
"files_discovered": [],
|
|
"required_files_present": {},
|
|
"required_files_missing": [],
|
|
"optional_files_present": {},
|
|
"manifest_status": {"status": "missing", "files": []},
|
|
"zip_status": {"status": "missing", "files": []},
|
|
"validation_sidecar_status": {"status": "missing", "files": []},
|
|
"report_status": {"status": "missing", "files": []},
|
|
"warnings": [],
|
|
"blocking_errors": [],
|
|
"machine_readable_summary": {},
|
|
}
|
|
|
|
|
|
def audit_bundle(bundle_root: pathlib.Path, profile_name: str, profile: Mapping[str, Any]) -> dict[str, Any]:
|
|
report = empty_report(bundle_root, profile_name)
|
|
files = discover_files(bundle_root)
|
|
files_rel = [rel_path(path, bundle_root) for path in files]
|
|
report["files_discovered"] = files_rel
|
|
|
|
required_matches, optional_matches = category_matches(profile, files_rel)
|
|
report["required_files_present"] = {
|
|
category: matches for category, matches in required_matches.items() if matches
|
|
}
|
|
report["optional_files_present"] = {
|
|
category: matches for category, matches in optional_matches.items() if matches
|
|
}
|
|
|
|
for category, matches in required_matches.items():
|
|
if not matches:
|
|
report["required_files_missing"].append(category)
|
|
report["blocking_errors"].append(f"Missing required {category} file.")
|
|
|
|
manifest_files = required_matches.get("manifest", [])
|
|
sidecar_files = required_matches.get("validation_sidecar", [])
|
|
report_files = optional_matches.get("report", [])
|
|
zip_files = optional_matches.get("zip", [])
|
|
|
|
report["manifest_status"] = inspect_structured_files(bundle_root, manifest_files)
|
|
report["validation_sidecar_status"] = inspect_structured_files(bundle_root, sidecar_files)
|
|
report["report_status"] = status_for_category(report_files)
|
|
report["zip_status"] = inspect_zip_files(bundle_root, zip_files)
|
|
|
|
if report["manifest_status"]["status"] == "unreadable":
|
|
report["blocking_errors"].append("Manifest file is unreadable.")
|
|
if report["validation_sidecar_status"]["status"] == "unreadable":
|
|
report["blocking_errors"].append("Validation sidecar file is unreadable.")
|
|
if report["zip_status"]["status"] == "unreadable":
|
|
report["blocking_errors"].append("Zip file is unreadable.")
|
|
if not zip_files:
|
|
report["warnings"].append("No zip file matched the profile.")
|
|
if not report_files:
|
|
report["warnings"].append("No review report file matched the profile.")
|
|
|
|
for file_rel in extra_files(files_rel, profile):
|
|
report["warnings"].append(f"Extra file not matched by profile: {file_rel}")
|
|
|
|
report["required_files_missing"] = sorted(report["required_files_missing"])
|
|
report["warnings"] = sorted(report["warnings"])
|
|
report["blocking_errors"] = sorted(set(report["blocking_errors"]))
|
|
finalize_report(report)
|
|
return report
|
|
|
|
|
|
def finalize_report(report: dict[str, Any]) -> None:
|
|
status = "FAIL" if report["blocking_errors"] else "PASS"
|
|
report["machine_readable_summary"] = {
|
|
"status": status,
|
|
"files_discovered_count": len(report["files_discovered"]),
|
|
"required_missing_count": len(report["required_files_missing"]),
|
|
"warning_count": len(report["warnings"]),
|
|
"blocking_error_count": len(report["blocking_errors"]),
|
|
"manifest_status": report["manifest_status"]["status"],
|
|
"zip_status": report["zip_status"]["status"],
|
|
"validation_sidecar_status": report["validation_sidecar_status"]["status"],
|
|
"report_status": report["report_status"]["status"],
|
|
}
|
|
|
|
|
|
def write_json_report(report: Mapping[str, Any], output_dir: pathlib.Path) -> None:
|
|
(output_dir / REPORT_JSON).write_text(
|
|
json.dumps(report, ensure_ascii=False, indent=2),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
|
|
def write_markdown_report(report: Mapping[str, Any], output_dir: pathlib.Path) -> None:
|
|
summary = report["machine_readable_summary"]
|
|
lines = [
|
|
"# Review Bundle Audit",
|
|
"",
|
|
"## Summary",
|
|
"",
|
|
f"- Status: `{summary['status']}`",
|
|
f"- Bundle root: `{report['bundle_root']}`",
|
|
f"- Profile: `{report['profile']}`",
|
|
f"- Files discovered: {summary['files_discovered_count']}",
|
|
f"- Required files missing: {summary['required_missing_count']}",
|
|
f"- Warnings: {summary['warning_count']}",
|
|
f"- Blocking errors: {summary['blocking_error_count']}",
|
|
"",
|
|
"## Required Files",
|
|
"",
|
|
]
|
|
for category, files in report["required_files_present"].items():
|
|
lines.append(f"- `{category}`: {', '.join(f'`{file}`' for file in files)}")
|
|
if report["required_files_missing"]:
|
|
lines.append("")
|
|
lines.append("Missing:")
|
|
for category in report["required_files_missing"]:
|
|
lines.append(f"- `{category}`")
|
|
lines.extend(["", "## Package Status", ""])
|
|
for key in ("manifest_status", "validation_sidecar_status", "zip_status", "report_status"):
|
|
status = report[key]["status"]
|
|
files = report[key].get("files", [])
|
|
rendered_files = ", ".join(f"`{file}`" for file in files) if files else "none"
|
|
lines.append(f"- `{key}`: `{status}` ({rendered_files})")
|
|
lines.extend(["", "## Blocking Errors", ""])
|
|
if report["blocking_errors"]:
|
|
for error in report["blocking_errors"]:
|
|
lines.append(f"- {error}")
|
|
else:
|
|
lines.append("- None")
|
|
lines.extend(["", "## Warnings", ""])
|
|
if report["warnings"]:
|
|
for warning in report["warnings"]:
|
|
lines.append(f"- {warning}")
|
|
else:
|
|
lines.append("- None")
|
|
lines.extend(["", "## Files Discovered", ""])
|
|
for file_rel in report["files_discovered"]:
|
|
lines.append(f"- `{file_rel}`")
|
|
(output_dir / REPORT_MD).write_text("\n".join(lines) + "\n", encoding="utf-8")
|
|
|
|
|
|
def write_reports(report: Mapping[str, Any], output_dir: pathlib.Path) -> None:
|
|
output_dir.mkdir(parents=True, exist_ok=True)
|
|
write_json_report(report, output_dir)
|
|
write_markdown_report(report, output_dir)
|
|
|
|
|
|
def run(
|
|
bundle_root: pathlib.Path,
|
|
output_dir: pathlib.Path,
|
|
profile_name: str,
|
|
profile_config: pathlib.Path | None,
|
|
) -> int:
|
|
bundle_root = bundle_root.resolve()
|
|
output_dir = output_dir.resolve()
|
|
output_dir.mkdir(parents=True, exist_ok=True)
|
|
report = empty_report(bundle_root, profile_name)
|
|
|
|
if not bundle_root.is_dir():
|
|
report["blocking_errors"].append(f"Bundle root is not a directory: {bundle_root}")
|
|
finalize_report(report)
|
|
write_reports(report, output_dir)
|
|
return 2
|
|
|
|
try:
|
|
config = load_profile_config(profile_config)
|
|
profile = get_profile(profile_name, config)
|
|
report = audit_bundle(bundle_root, profile_name, profile)
|
|
except Exception as exc:
|
|
report["blocking_errors"].append(str(exc))
|
|
finalize_report(report)
|
|
write_reports(report, output_dir)
|
|
return 2
|
|
|
|
write_reports(report, output_dir)
|
|
return 1 if report["blocking_errors"] else 0
|
|
|
|
|
|
def main(argv: list[str] | None = None) -> int:
|
|
parser = argparse.ArgumentParser(description="Audit a review bundle directory.")
|
|
parser.add_argument("--bundle-root", required=True, type=pathlib.Path)
|
|
parser.add_argument("--output-dir", required=True, type=pathlib.Path)
|
|
parser.add_argument("--profile", default="generic")
|
|
parser.add_argument("--profile-config", type=pathlib.Path)
|
|
args = parser.parse_args(argv)
|
|
return run(args.bundle_root, args.output_dir, args.profile, args.profile_config)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|