mirror of
https://github.com/magnus919/agent-skills.git
synced 2026-09-11 19:47:12 +03:00
100 lines
3.6 KiB
Python
Executable File
100 lines
3.6 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Link attributed, reviewed visual observations to existing EDL events."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
from pathlib import Path
|
|
|
|
ALLOWED_CLASSES = {
|
|
"technical_observation",
|
|
"human_or_vision_observation",
|
|
"heuristic",
|
|
"unresolved_claim",
|
|
}
|
|
|
|
|
|
def fail(code: str, message: str) -> int:
|
|
print(json.dumps({"ok": False, "error": {"code": code, "message": message}}))
|
|
return 2
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("manifest")
|
|
parser.add_argument("edl")
|
|
parser.add_argument("--output", required=True)
|
|
parser.add_argument("--json", action="store_true")
|
|
args = parser.parse_args()
|
|
try:
|
|
manifest = json.loads(Path(args.manifest).read_text())
|
|
edl = json.loads(Path(args.edl).read_text())
|
|
output = Path(args.output)
|
|
except (OSError, json.JSONDecodeError) as exc:
|
|
return fail("invalid_input", str(exc))
|
|
if output.exists():
|
|
return fail("output_exists", "output already exists")
|
|
review = manifest.get("review", {})
|
|
if review.get("status") != "reviewed" or not review.get("reviewer"):
|
|
return fail(
|
|
"review_evidence_missing",
|
|
"manifest needs an attributed reviewed status before EDL import",
|
|
)
|
|
observations = review.get("observations")
|
|
if not isinstance(observations, list) or not observations:
|
|
return fail("review_evidence_missing", "manifest has no reviewed observations")
|
|
events = {event.get("id"): event for event in edl.get("events", [])}
|
|
imported = 0
|
|
for observation in observations:
|
|
event_id = observation.get("edl_event_id")
|
|
evidence_class = observation.get("evidence_class")
|
|
if event_id not in events:
|
|
return fail("event_not_found", f"observation targets unknown EDL event: {event_id}")
|
|
if evidence_class not in ALLOWED_CLASSES:
|
|
return fail("invalid_evidence_class", f"unsupported evidence class: {evidence_class}")
|
|
if observation.get("coverage_scope") in {"whole_asset", "all_unsampled_intervals"}:
|
|
return fail(
|
|
"unsupported_coverage_claim",
|
|
"sparse samples cannot establish whole-asset or unsampled absence",
|
|
)
|
|
if (
|
|
not observation.get("id")
|
|
or not observation.get("observation")
|
|
or not observation.get("artifact_refs")
|
|
):
|
|
return fail(
|
|
"invalid_observation", "each observation needs id, text, and artifact references"
|
|
)
|
|
events[event_id].setdefault("evidence", []).append(
|
|
{
|
|
"type": "visual_review",
|
|
"packet_id": manifest.get("packet_id"),
|
|
"observation_id": observation["id"],
|
|
"asset_id": manifest.get("asset_id"),
|
|
"reviewer": review["reviewer"],
|
|
"evidence_class": evidence_class,
|
|
"confidence": observation.get("confidence"),
|
|
"editorial_consequence": observation.get("editorial_consequence"),
|
|
"artifact_refs": observation["artifact_refs"],
|
|
"coverage_scope": observation.get("coverage_scope", "sampled_artifacts_only"),
|
|
}
|
|
)
|
|
imported += 1
|
|
output.write_text(json.dumps(edl, indent=2) + "\n")
|
|
print(
|
|
json.dumps(
|
|
{
|
|
"ok": True,
|
|
"output": str(output),
|
|
"imported_observations": imported,
|
|
"executed_media_change": False,
|
|
}
|
|
)
|
|
)
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|