from __future__ import annotations import argparse import csv import hashlib import json from dataclasses import dataclass from datetime import datetime, timezone from pathlib import Path from typing import Any @dataclass class FileCheck: path: str status: str error_code: str expected_sha256: str actual_sha256: str @dataclass class CaseResult: case_id: str package_id: str expected_status: str status: str file_count: int checked_file_count: int pass_file_count: int fail_file_count: int error_codes: list[str] 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 load_manifest(path: Path) -> dict[str, Any]: with path.open("r", encoding="utf-8") as handle: payload = json.load(handle) if not isinstance(payload, dict): raise ValueError("MANIFEST_NOT_OBJECT") files = payload.get("files") if not isinstance(files, list): raise ValueError("MANIFEST_FILES_NOT_LIST") return payload def _safe_resolve_case_file(case_dir: Path, relative_path: str) -> tuple[Path | None, str]: raw_path = Path(relative_path) if raw_path.is_absolute(): return None, "ABSOLUTE_PATH_NOT_ALLOWED" resolved_case_dir = case_dir.resolve() resolved_file = (case_dir / raw_path).resolve() try: resolved_file.relative_to(resolved_case_dir) except ValueError: return None, "PATH_OUTSIDE_CASE" return resolved_file, "" def _expected_status(case_id: str) -> str: return "FAIL" if case_id.startswith("invalid_") else "PASS" def validate_case(case_dir: Path) -> tuple[CaseResult, list[FileCheck]]: case_id = case_dir.name expected_status = _expected_status(case_id) manifest_path = case_dir / "manifest.json" file_checks: list[FileCheck] = [] error_codes: list[str] = [] if not manifest_path.exists(): return ( CaseResult( case_id=case_id, package_id="", expected_status=expected_status, status="FAIL", file_count=0, checked_file_count=0, pass_file_count=0, fail_file_count=0, error_codes=["MANIFEST_MISSING"], ), file_checks, ) try: manifest = load_manifest(manifest_path) except (json.JSONDecodeError, ValueError) as exc: return ( CaseResult( case_id=case_id, package_id="", expected_status=expected_status, status="FAIL", file_count=0, checked_file_count=0, pass_file_count=0, fail_file_count=0, error_codes=[str(exc) or exc.__class__.__name__], ), file_checks, ) package_id = str(manifest.get("package_id") or "") files = manifest["files"] for item in files: if not isinstance(item, dict): error_codes.append("FILE_ITEM_NOT_OBJECT") continue rel_path = str(item.get("path") or "") expected_sha = str(item.get("sha256") or "").lower() target_path, path_error = _safe_resolve_case_file(case_dir, rel_path) if path_error: error_codes.append(path_error) file_checks.append(FileCheck(rel_path, "FAIL", path_error, expected_sha, "")) continue if target_path is None or not target_path.exists(): error_codes.append("FILE_MISSING") file_checks.append(FileCheck(rel_path, "FAIL", "FILE_MISSING", expected_sha, "")) continue actual_sha = sha256_file(target_path) if actual_sha.lower() != expected_sha: error_codes.append("SHA256_MISMATCH") file_checks.append(FileCheck(rel_path, "FAIL", "SHA256_MISMATCH", expected_sha, actual_sha)) continue file_checks.append(FileCheck(rel_path, "PASS", "", expected_sha, actual_sha)) fail_file_count = sum(1 for row in file_checks if row.status == "FAIL") pass_file_count = sum(1 for row in file_checks if row.status == "PASS") status = "PASS" if not error_codes and len(file_checks) == len(files) else "FAIL" return ( CaseResult( case_id=case_id, package_id=package_id, expected_status=expected_status, status=status, file_count=len(files), checked_file_count=len(file_checks), pass_file_count=pass_file_count, fail_file_count=fail_file_count, error_codes=sorted(set(error_codes)), ), file_checks, ) def write_outputs(input_root: Path, output_dir: Path) -> dict[str, Any]: output_dir.mkdir(parents=True, exist_ok=True) case_dirs = sorted(path for path in input_root.iterdir() if path.is_dir()) case_results: list[CaseResult] = [] file_rows: list[dict[str, str]] = [] for case_dir in case_dirs: result, checks = validate_case(case_dir) case_results.append(result) for check in checks: file_rows.append( { "case_id": result.case_id, "package_id": result.package_id, "file_path": check.path, "status": check.status, "error_code": check.error_code, "expected_sha256": check.expected_sha256, "actual_sha256": check.actual_sha256, } ) summary = { "experiment_id": input_root.name, "run_id": output_dir.name, "created_at_utc": datetime.now(timezone.utc).isoformat(), "input_root": str(input_root), "output_dir": str(output_dir), "total_cases": len(case_results), "pass_count": sum(1 for row in case_results if row.status == "PASS"), "fail_count": sum(1 for row in case_results if row.status == "FAIL"), "expected_matrix_match_count": sum(1 for row in case_results if row.status == row.expected_status), "quality_pass": all(row.status == row.expected_status for row in case_results), } with (output_dir / "case_results.csv").open("w", newline="", encoding="utf-8-sig") as handle: writer = csv.DictWriter( handle, fieldnames=[ "case_id", "package_id", "expected_status", "status", "file_count", "checked_file_count", "pass_file_count", "fail_file_count", "error_codes", ], ) writer.writeheader() for row in case_results: writer.writerow( { "case_id": row.case_id, "package_id": row.package_id, "expected_status": row.expected_status, "status": row.status, "file_count": row.file_count, "checked_file_count": row.checked_file_count, "pass_file_count": row.pass_file_count, "fail_file_count": row.fail_file_count, "error_codes": ";".join(row.error_codes), } ) with (output_dir / "file_check_details.csv").open("w", newline="", encoding="utf-8-sig") as handle: writer = csv.DictWriter( handle, fieldnames=[ "case_id", "package_id", "file_path", "status", "error_code", "expected_sha256", "actual_sha256", ], ) writer.writeheader() writer.writerows(file_rows) with (output_dir / "summary.json").open("w", encoding="utf-8") as handle: json.dump(summary, handle, ensure_ascii=False, indent=2) readout = [ f"# {summary['run_id']} 结果导读", "", "- 创建人员: ai-codex", "- 文件职责: 记录 manifest 包校验实验的可读结论，不作为核心业务主表。", "- 管理规范/模板: project-d/exp-doc/实验规范.md; project-d/dev-doc/编码规范.md", "- 引用文件: summary.json; case_results.csv; file_check_details.csv", "- 记录方式: append-only 实验日志 + 开发日志 + 审计报告交叉引用。", "", "## 结论", "", f"- total_cases: {summary['total_cases']}", f"- pass_count: {summary['pass_count']}", f"- fail_count: {summary['fail_count']}", f"- expected_matrix_match_count: {summary['expected_matrix_match_count']}", f"- quality_pass: {summary['quality_pass']}", ] with (output_dir / "readout.md").open("w", encoding="utf-8") as handle: handle.write("\n".join(readout) + "\n") return summary def main() -> int: parser = argparse.ArgumentParser(description="Validate manifest packages for EXP-D-HEAVY-001.") parser.add_argument("--input-root", required=True, type=Path) parser.add_argument("--output-dir", required=True, type=Path) args = parser.parse_args() summary = write_outputs(args.input_root, args.output_dir) return 0 if summary["quality_pass"] else 1 if __name__ == "__main__": raise SystemExit(main())