feat(oracle): prepare first capture packets

scripts/oracle_collection_queue.py

@@ -103,6 +103,8 @@ REQUIRED_EVIDENCE_METADATA_FIELDS = [
     "timezone",
     "operator_note",
 ]
+SHADBALA_DRAFT_PLANETS = ["Sun", "Moon", "Mars", "Mercury", "Jupiter", "Venus", "Saturn"]
+SHADBALA_DRAFT_COMPONENTS = ["sthana", "dig", "kala", "chesta", "naisargika", "drik", "total_rupa"]
 
 
 def _resolve_path(path: str) -> str:
@@ -182,6 +184,13 @@ def _evidence_packet(case: dict[str, Any], target_fields: list[str]) -> dict[str
         field: _target_value(target, field)
         for field in target_fields
     }
+    if "target.shadbala_components" in target_placeholders and _is_blank_target(
+        target_placeholders["target.shadbala_components"]
+    ):
+        target_placeholders["target.shadbala_components"] = {
+            planet: {component: None for component in SHADBALA_DRAFT_COMPONENTS}
+            for planet in SHADBALA_DRAFT_PLANETS
+        }
     existing_placeholders = existing.get("target_placeholders", {})
     if isinstance(existing_placeholders, dict):
         target_placeholders.update(existing_placeholders)
@@ -217,6 +226,10 @@ def _evidence_packet(case: dict[str, Any], target_fields: list[str]) -> dict[str
     }
 
 
+def _is_blank_target(value: Any) -> bool:
+    return value is None or value == "" or value == [] or value == {}
+
+
 def _task_from_template(case: dict[str, Any]) -> dict[str, Any]:
     case_id = case.get("id") or case.get("case_id")
     target = case.get("target", {})

scripts/prepare_oracle_capture_packets.py

@@ -0,0 +1,135 @@
+#!/usr/bin/env python3
+"""Prepare fillable external-oracle evidence packets and operator notes."""
+
+from __future__ import annotations
+
+import argparse
+import json
+import os
+import sys
+from pathlib import Path
+from typing import Any
+
+
+ROOT_DIR = Path(__file__).resolve().parents[1]
+if str(ROOT_DIR / "scripts") not in sys.path:
+    sys.path.insert(0, str(ROOT_DIR / "scripts"))
+
+from oracle_collection_queue import build_queue, write_evidence_packets  # noqa: E402
+from oracle_evidence_validator import build_report  # noqa: E402
+
+
+FIRST_PRIORITY_CAPTURE_ID = "external_template_steve_jobs_dasha_lahiri"
+
+
+def _resolve(path: str) -> Path:
+    candidate = Path(path)
+    if candidate.is_absolute():
+        return candidate
+    return ROOT_DIR / candidate
+
+
+def _load_json(path: str) -> dict[str, Any]:
+    with _resolve(path).open("r", encoding="utf-8") as fh:
+        return json.load(fh)
+
+
+def _write_json(path: Path, value: dict[str, Any]) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    with path.open("w", encoding="utf-8") as fh:
+        json.dump(value, fh, ensure_ascii=False, indent=2)
+        fh.write("\n")
+
+
+def _write_next_steps(path: Path, oracle_file: str, output_dir: str, first_packet: str) -> None:
+    text = f"""# External Oracle Capture Next Steps
+
+First priority packet:
+
+`{os.path.basename(first_packet)}`
+
+## Fill The Packet
+
+1. Open JHora, PyJHora, VedAstro, or another documented external source.
+2. Use the exact birth data, ayanamsa, node mode, timezone, and settings in the packet.
+3. Save a redacted screenshot or stdout snippet under `references/oracle/artifacts/`.
+4. Fill all metadata fields and all `target_placeholders`.
+5. Set `status` to `external_verified` only after the artifact and target values are filled.
+
+不得把本仓库本地输出当作 external oracle。
+
+## Apply The Packet
+
+```bash
+python3 scripts/oracle_collection_queue.py \\
+  --oracle-file {oracle_file} \\
+  --apply-packet {first_packet} \\
+  --format json
+```
+
+## Validate Again
+
+```bash
+python3 scripts/oracle_collection_queue.py \\
+  --oracle-file {oracle_file} \\
+  --format json > /tmp/jyotish_oracle_queue_filled.json
+
+python3 scripts/oracle_evidence_validator.py \\
+  --queue-file /tmp/jyotish_oracle_queue_filled.json
+```
+
+Draft packets in `{output_dir}` must remain `valid_packets: 0` until real external evidence is filled.
+"""
+    path.write_text(text, encoding="utf-8")
+
+
+def prepare_packets(oracle_file: str, output_dir: str) -> dict[str, Any]:
+    output_path = _resolve(output_dir)
+    output_path.mkdir(parents=True, exist_ok=True)
+    oracle = _load_json(oracle_file)
+    queue = build_queue(oracle)
+    packet_count = write_evidence_packets(queue, str(output_path))
+    validator_report = build_report(queue)
+
+    packets = sorted(path.name for path in output_path.glob("external_*.json"))
+    first_priority = output_path / f"{FIRST_PRIORITY_CAPTURE_ID}.json"
+    first_priority_path = first_priority if first_priority.exists() else output_path / packets[0]
+
+    manifest = {
+        "scope": "external_oracle_capture_packet_manifest",
+        "oracle_file": oracle_file,
+        "packet_count": packet_count,
+        "packets": packets,
+        "first_priority_packet": str(first_priority_path),
+        "validator_summary": validator_report["summary"],
+    }
+    _write_json(output_path / "capture_manifest.json", manifest)
+    _write_next_steps(output_path / "OPERATOR_NEXT_STEPS.md", oracle_file, output_dir, str(first_priority_path))
+
+    return {
+        "scope": "external_oracle_capture_packet_preparation",
+        "oracle_file": oracle_file,
+        "output_dir": str(output_path),
+        "packet_count": packet_count,
+        "first_priority_packet": str(first_priority_path),
+        "validator_summary": validator_report["summary"],
+        "manifest": str(output_path / "capture_manifest.json"),
+        "operator_next_steps": str(output_path / "OPERATOR_NEXT_STEPS.md"),
+    }
+
+
+def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description="Prepare external oracle capture packets")
+    parser.add_argument("--oracle-file", required=True)
+    parser.add_argument("--output-dir", required=True)
+    return parser.parse_args(argv)
+
+
+def main(argv: list[str] | None = None) -> int:
+    args = parse_args(argv)
+    print(json.dumps(prepare_packets(args.oracle_file, args.output_dir), ensure_ascii=False, indent=2))
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())