Files
workinf_Blender_Wasm/my-tools/prepare_full_parity_module_batch.py
2026-09-02 18:35:58 +08:00

81 lines
3.5 KiB
Python
Executable File

#!/usr/bin/env python3
"""Create a bounded Blender source-inventory batch summary.
WBV2 task cards already define their own module boundary. This helper therefore
never mutates dispositions or claims a task; it only selects unresolved source
units from the canonical dynamic inventory for inspection planning.
"""
from __future__ import annotations
import argparse
import hashlib
import json
from datetime import datetime, timezone
from pathlib import Path
from wbv2_runtime import CONTROL_ROOT, ROOT, queue_pointer
INVENTORY = CONTROL_ROOT / "dynamic-inventory.json"
DEFAULT_OUTPUT = CONTROL_ROOT / "my-tools-module-batch.json"
def main() -> int:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--size", type=int, default=8)
parser.add_argument("--module", help="optional source family filter")
parser.add_argument("--family", help="optional source family filter")
parser.add_argument("--tier", help="accepted for compatibility; WBV2 inventory has no tier field")
parser.add_argument("--output", type=Path, default=DEFAULT_OUTPUT)
args = parser.parse_args()
if args.size < 1:
parser.error("--size must be positive")
try:
inventory = json.loads(INVENTORY.read_text(encoding="utf-8"))
except (OSError, json.JSONDecodeError) as exc:
print(f"WBV2 BATCH FAILED: {exc}")
return 1
units = inventory.get("units", []) if isinstance(inventory, dict) else []
unresolved = []
for unit in units:
if not isinstance(unit, dict):
continue
if unit.get("verified") is True or unit.get("status") in {"verified", "completed"}:
continue
family = unit.get("family") or unit.get("kind")
if args.module and unit.get("module") != args.module:
continue
if args.family and family != args.family:
continue
unresolved.append(unit)
unresolved.sort(key=lambda row: (str(row.get("family", "")), str(row.get("sourcePath", "")), int(row.get("sourceLine", 0) or 0), str(row.get("id", ""))))
selected = unresolved[: args.size]
identity = "|".join(str(row.get("id", row.get("name", ""))) for row in selected)
batch_id = hashlib.sha256(identity.encode("utf-8")).hexdigest()[:24] if identity else None
payload = {
"schemaVersion": 1,
"operation": "BLENDER_WEB_DYNAMIC_SOURCE_BATCH",
"status": "ready" if selected else "no_ready_batch",
"generatedAt": datetime.now(timezone.utc).isoformat(),
"task": queue_pointer().get("task"),
"batchId": batch_id,
"sourceTreeSha256": inventory.get("authority", {}).get("sourceTreeSha256"),
"unitCount": inventory.get("summary", {}).get("unitCount", len(units)),
"unresolvedCount": len(unresolved),
"selected": selected,
"boundaries": [
"This file is planning metadata only and never changes coverage.",
"Only hash-bound Blender source and Chromium evidence may verify a unit.",
],
}
args.output = args.output if args.output.is_absolute() else ROOT / args.output
args.output.parent.mkdir(parents=True, exist_ok=True)
args.output.write_text(json.dumps(payload, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
print(json.dumps({"status": payload["status"], "task": payload["task"], "batchId": batch_id, "records": len(selected), "output": str(args.output.relative_to(ROOT))}, ensure_ascii=False))
return 0 if selected else 2
if __name__ == "__main__":
raise SystemExit(main())