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())
|
