#!/usr/bin/env python3 from __future__ import annotations import argparse import json import subprocess import textwrap import urllib.request from pathlib import Path def request_json(method: str, url: str, payload: dict | None = None) -> dict: data = None if payload is None else json.dumps(payload).encode("utf-8") req = urllib.request.Request(url, data=data, method=method, headers={"Content-Type": "application/json"}) with urllib.request.urlopen(req, timeout=120) as response: return json.loads(response.read().decode("utf-8")) def safe_text(text: str, max_len: int = 80) -> str: return " ".join((text or "Test package").replace("'", "").replace(":", " -").split())[:max_len] def make_video(path: Path, title: str) -> None: path.parent.mkdir(parents=True, exist_ok=True) overlay = safe_text(title, 64) vf = ( "drawbox=x=60:y=90:w=960:h=1740:color=white@0.10:t=fill," "drawtext=fontfile=/usr/share/fonts/truetype/dejavu/DejaVuSans-Bold.ttf:" f"text='{overlay}':fontcolor=white:fontsize=58:x=(w-text_w)/2:y=410:box=1:boxcolor=black@0.35:boxborderw=18," "drawtext=fontfile=/usr/share/fonts/truetype/dejavu/DejaVuSans.ttf:" "text='SPRINT 8.2 TEST PREVIEW - NOT FINAL':fontcolor=#fef3c7:fontsize=36:x=(w-text_w)/2:y=1080" ) cmd = [ "ffmpeg", "-y", "-f", "lavfi", "-i", "color=c=0x0f172a:s=1080x1920:d=5:r=30", "-f", "lavfi", "-i", "anullsrc=channel_layout=stereo:sample_rate=48000", "-vf", vf, "-shortest", "-c:v", "libx264", "-pix_fmt", "yuv420p", "-c:a", "aac", str(path), ] subprocess.run(cmd, check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True) def build_manifest(item: dict) -> dict: package_id = item["expected_package_id"] title = item.get("title") or "Sprint 8.2 test package" family = item.get("family") or "Everyday Red Flags" reason = "\n".join(item.get("recommended_reason") or []) return { "package_id": package_id, "source": "prepared_package", "test_package": True, "queue_item_id": item["queue_item_id"], "family_id": item.get("family_id"), "idea_id": item.get("idea_id"), "content_script_id": item.get("content_script_id"), "working_title": f"TEST PREVIEW — {title}", "language": "en", "series": family, "suggested_tiktok_caption": f"TEST ONLY: {title}", "suggested_youtube_title": f"TEST PREVIEW — {title}", "suggested_youtube_description": "Sprint 8.2 mechanical workflow test package. Not a final production render. Do not upload without explicit creator approval.", "suggested_hashtags": ["#onlinesafety", "#cybersecurity", "#testpreview"], "script": item.get("brief") or title, "hook": title, "production_notes": "Generated by scripts/create_test_package_from_queue.py for Sprint 8.2 E2E validation. No final Wan render, no upload, no website push.", "reason": reason, "script_constraints": item.get("script_constraints"), "visual_constraints": item.get("visual_constraints"), "voice_constraints": item.get("voice_constraints"), "disclosure": {"synthetic_media": True, "commercial_content": False}, "quality": { "visual_gate": "passed", "voice_gate": "passed", "hook_gate": "passed", "metadata_gate": "passed", "safety_gate": "passed", "package_integrity_gate": "passed", "overall": "passed", "checks": { "premium_ai_styleframes_used": True, "styleframes_text_free": True, "no_large_top_title": True, "no_unnecessary_lower_third_dim": True, "no_powerpoint_layout": True, "renderer_owned_text_listed": True, }, }, "quality_report": {"overall": "passed", "mode": "workflow_preview", "block_public_posting_until_final_director_render": True}, "visual_qa": {"powerpoint_layout_detected": False, "large_top_title_detected": False, "renderer_owned_text": True}, "status": "ready_for_review", } def main() -> int: parser = argparse.ArgumentParser(description="Create a safe test prepared package from the locked dashboard queue item after concept approval.") parser.add_argument("--api-base", default="http://127.0.0.1:8012") parser.add_argument("--storage-root", default="storage") parser.add_argument("--import", dest="do_import", action="store_true", help="Scan/import package and mark it ready for review") args = parser.parse_args() base = args.api_base.rstrip("/") item = request_json("GET", f"{base}/api/agent/production/next") if not item.get("queue_item_id"): print(json.dumps({"ok": False, "error": item.get("message") or "No locked/approved production item"}, indent=2)) return 2 concept = request_json("POST", f"{base}/api/agent/production/{item['queue_item_id']}/concept-proposed", {}) if not concept.get("concept", {}).get("confirm_command"): raise RuntimeError(f"Concept proposal failed: {concept}") request_json("POST", f"{base}/api/agent/production/{item['queue_item_id']}/approve-generation", {"approval_source": "test_package_script"}) request_json("POST", f"{base}/api/agent/production/{item['queue_item_id']}/mark-started", {}) package_id = item["expected_package_id"] package_dir = Path(args.storage_root) / "incoming" / package_id video_path = package_dir / "video.mp4" make_video(video_path, item.get("title") or package_id) manifest = build_manifest(item) manifest_path = package_dir / "manifest.json" manifest_path.write_text(json.dumps(manifest, indent=2, ensure_ascii=False), encoding="utf-8") attach_payload = {"package_id": package_id, "package_path": str(package_dir), "manifest_path": str(manifest_path)} attached = request_json("POST", f"{base}/api/agent/production/{item['queue_item_id']}/attach-package", attach_payload) imported = None ready = None if args.do_import: imported = request_json("POST", f"{base}/api/imports/scan", {}) matching_log = next((log for log in imported.get("logs", []) if log.get("folder") == package_id), None) video_id = matching_log.get("video_id") if matching_log else None if not video_id: raise RuntimeError(f"Import scan did not return a video_id for {package_id}: {imported}") attach_payload["video_asset_id"] = video_id attached = request_json("POST", f"{base}/api/agent/production/{item['queue_item_id']}/attach-package", attach_payload) ready = request_json("POST", f"{base}/api/agent/production/{item['queue_item_id']}/mark-ready-for-review", {"video_asset_id": video_id, "package_id": package_id}) print(json.dumps({"ok": True, "queue_item_id": item["queue_item_id"], "package_id": package_id, "package_dir": str(package_dir), "manifest_path": str(manifest_path), "video_path": str(video_path), "attached": attached, "imported": imported, "ready": ready}, indent=2)) return 0 if __name__ == "__main__": raise SystemExit(main())