scripts/demo_plugin_export.py
Rama main · réplica de lectura del repositorio original.
"""Export an external plugin's SDK moments through Clips Kitty's renderer.
This developer demo accepts original synthetic test media only. It uses no
detector, model, or renderer copied from the plugin, and downloads nothing.
"""
from __future__ import annotations
import argparse
import hashlib
import json
import math
import subprocess
import sys
from datetime import datetime, timezone
from pathlib import Path
ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT))
from core.binaries import ffmpeg, ffprobe # noqa: E402
from core.models import ClipCandidate, Segment # noqa: E402
def probe_media(path: Path) -> dict:
"""Read actual exported streams with the app's resolved FFprobe."""
result = subprocess.run(
[ffprobe(), "-v", "error", "-show_streams", "-show_format", "-of", "json", str(path)],
capture_output=True, text=True, check=True, timeout=30,
)
return json.loads(result.stdout)
def sdk_testing():
"""The SDK's test runner from this checkout's own SDK (sdk/python), put on
the path the way the engine does it, so the demo needs no pip-installed
SDK, which could come from another commit than this app."""
import importlib
import plugins._sdk # noqa: F401 (puts sdk/python on the path)
return importlib.import_module("clipskitty_sdk.testing")
def _sha256(path: Path) -> str:
digest = hashlib.sha256()
with path.open("rb") as stream:
for block in iter(lambda: stream.read(1024 * 1024), b""):
digest.update(block)
return digest.hexdigest()
def run_demo(
*, plugin: Path, video: Path, transcript: Path, output: Path,
synthetic_demo: bool = False, max_clips: int = 2,
vertical: bool = False, captions: bool = True,
) -> dict:
"""Run the public SDK plugin process and existing app export functions.
This writes a standalone MP4 for each returned moment (at most two),
caption ASS files, a JSON manifest, log, and isolated StateDB history.
It never opens the user's application database or installs the plugin.
"""
if not synthetic_demo:
raise ValueError("Pass --synthetic-demo to affirm these inputs are original synthetic test assets")
if not 1 <= max_clips <= 2:
raise ValueError("This technical demo exports one or two clips only")
plugin, video, transcript, output = (Path(p).resolve() for p in (plugin, video, transcript, output))
if output in (plugin, ROOT) or plugin in output.parents or ROOT in output.parents:
raise ValueError("Write demo artifacts outside the plugin and app source folders")
if not video.is_file() or not transcript.is_file():
raise ValueError("Supply an existing synthetic MP4 and its authored transcript JSON")
said = json.loads(transcript.read_text(encoding="utf-8"))
if not isinstance(said, dict) or not isinstance(said.get("segments"), list):
raise ValueError("The transcript must contain a segments list")
# Public SDK and app entry points; the external plugin remains a process.
testing = sdk_testing()
from core.state import StateDB
from video.captions import build_captions
from video.cutter import cut_clip
from video.encoding import video_encoder_args
source_probe = probe_media(video)
duration = float(source_probe["format"]["duration"])
source_video = next((s for s in source_probe["streams"] if s["codec_type"] == "video"), None)
if source_video is None or not math.isfinite(duration) or duration <= 0:
raise ValueError("Synthetic input must have a video stream and a positive finite duration")
segments = [Segment(start=float(s["start"]), end=float(s["end"]), text=str(s["text"]),
words=s.get("words")) for s in said["segments"]]
output.mkdir(parents=True, exist_ok=True)
with (output / "demo.log").open("w", encoding="utf-8") as log:
def say(message: str) -> None:
print(message)
log.write(message + "\n")
log.flush()
say("Synthetic media validation only; no real gameplay or human quality review.")
say(f"FFmpeg: {ffmpeg()}; FFprobe: {ffprobe()}")
say(f"Running external plugin through clipskitty_sdk.testing: {plugin}")
run = testing.run_plugin(plugin, video=video, transcript=said, duration=duration,
steps=("find", "understand"),
tmp_path=output / "sdk-jobs", timeout=120)
for event in run.events:
say(f"SDK {event.get('type', 'event')}: {event.get('message', '')}")
if not run.ok:
raise RuntimeError(f"SDK plugin run failed: {run.error}")
if not run.moments:
raise RuntimeError("The plugin returned no moments; no export was invented")
job = json.loads((run.folder / "job.json").read_text(encoding="utf-8"))
plugin_info = job.get("plugin", {})
say(f"Plugin returned {len(run.moments)} moment(s); exporting at most {max_clips}.")
# Use the same CPU selection used by the app, with no hardware/model probes.
video_encoder_args({"video": {"encoder": "cpu"}})
canvas = (1080, 1920) if vertical else (source_video["width"], source_video["height"])
records = []
source_hash = _sha256(video)
video_id = "synthetic-plugin-demo-" + source_hash[:16]
history = StateDB(output / "history.sqlite3")
try:
history.upsert_video(video_id, title="Original synthetic plugin integration test", duration=duration)
for index, moment in enumerate(run.moments[:max_clips], 1):
start, end = float(moment.start), float(moment.end)
if not (math.isfinite(start) and math.isfinite(end) and 0 <= start < end <= duration + 0.05):
raise RuntimeError("SDK returned invalid clip bounds")
candidate = ClipCandidate(
start=start, end=min(duration, end), score=round(moment.score or 0),
hook=moment.title or moment.label, reason=moment.reason,
source=f"plugin:{plugin_info.get('id', '')}@{plugin_info.get('version', '')}",
subscores={"plugin": plugin_info.get("id", ""),
"plugin_version": plugin_info.get("version", ""),
"plugin_label": moment.label, "plugin_why": moment.reason},
)
stem = f"highlight-{index:02d}"
ass = build_captions(segments, candidate, output / f"{stem}.ass", canvas=canvas,
language=str(said.get("language") or "en")) if captions else None
destination = output / f"{stem}.mp4"
if destination == video:
raise ValueError("The demo output must not replace its source video")
say(f"Exporting {start:.2f}–{end:.2f}s via existing app renderer: {destination}")
if vertical:
from video.cropper import render_vertical
staging = output / f"{stem}.source.mp4"
cut_clip(video, candidate, staging, normalize=False)
render_vertical(staging, {"mode": "fit_blur"}, destination, ass_path=ass,
normalize=True)
else:
cut_clip(video, candidate, destination, ass_path=ass, normalize=True, final=True)
exported = probe_media(destination)
exported_duration = float(exported["format"]["duration"])
if abs(exported_duration - candidate.duration) > 0.25:
raise RuntimeError("Exported duration differs from the SDK moment by more than 0.25 seconds")
if not any(s["codec_type"] == "video" for s in exported["streams"]):
raise RuntimeError("Export has no video stream")
if any(s["codec_type"] == "audio" for s in source_probe["streams"]) and not any(
s["codec_type"] == "audio" for s in exported["streams"]
):
raise RuntimeError("Export lost the source audio stream")
clip_id = history.add_clip(
video_id, candidate.start, candidate.end, candidate.score, candidate.hook,
path=str(destination), title="Synthetic plugin integration highlight",
scores=json.dumps(candidate.subscores),
render_opts=json.dumps({"synthetic_demo": True, "layout": "fit_blur" if vertical else "original",
"captions": ass is not None, "encoder": "cpu"}),
)
if clip_id is None:
clip_id = next((row["id"] for row in history.clips_for_video(video_id)
if abs(row["start_s"] - candidate.start) < 0.01
and abs(row["end_s"] - candidate.end) < 0.01), None)
records.append({"path": str(destination), "source_start": candidate.start,
"source_end": candidate.end, "expected_duration": candidate.duration,
"category": moment.label, "detector_score": moment.score,
"reason": moment.reason, "captions": str(ass) if ass else None,
"caption_origin": "authored synthetic transcript; audio contains tones",
"history_clip_id": clip_id, "sha256": _sha256(destination), "probe": exported})
say(f"Verified {exported_duration:.3f}s with video and preserved source audio.")
history.set_outcome(video_id, {"synthetic_demo": True, "plugin": plugin_info,
"exported_clips": len(records)})
history.set_video_status(video_id, "done")
finally:
history.close()
manifest = {
"schema_version": 1, "generated_at_utc": datetime.now(timezone.utc).isoformat(),
"synthetic_demo": True, "consent": "Caller supplied --synthetic-demo for original test assets",
"scope": "Technical SDK-to-existing-app-export validation; no real gameplay or quality claim",
"manual_quality_review": "not_performed", "plugin": {**plugin_info, "path": str(plugin)},
"source": {"video": str(video), "transcript": str(transcript), "sha256": source_hash,
"transcript_sha256": _sha256(transcript), "probe": source_probe},
"sdk_job_folder": str(run.folder), "sdk_notes": run.notes,
"renderer": {"cutter": "video.cutter.cut_clip", "encoder": "cpu", "normalize_audio": True,
"vertical": "video.cropper.render_vertical:fit_blur" if vertical else None,
"captions": "video.captions.build_captions" if captions else None,
"ffmpeg": ffmpeg(), "ffprobe": ffprobe()},
"history_database": str(output / "history.sqlite3"), "clips": records,
}
manifest_path = output / "manifest.json"
manifest_path.write_text(json.dumps(manifest, indent=2) + "\n", encoding="utf-8")
say(f"Saved manifest: {manifest_path}")
return manifest
def main() -> int:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--plugin", type=Path, required=True, help="External MIT plugin checkout")
parser.add_argument("--video", type=Path, required=True, help="Original synthetic MP4")
parser.add_argument("--transcript", type=Path, required=True, help="Authored synthetic transcript JSON")
parser.add_argument("--output", type=Path, default=ROOT.parent / "scratch" / "plugin-export-demo")
parser.add_argument("--synthetic-demo", action="store_true", help="Affirm inputs are original synthetic test assets")
parser.add_argument("--max-clips", type=int, choices=(1, 2), default=2)
parser.add_argument("--vertical", action="store_true", help="Use existing 1080x1920 fit_blur renderer")
parser.add_argument("--captions", action=argparse.BooleanOptionalAction, default=True)
args = parser.parse_args()
if not args.synthetic_demo:
parser.error("--synthetic-demo is required for this synthetic-only integration check")
run_demo(plugin=args.plugin, video=args.video, transcript=args.transcript, output=args.output,
synthetic_demo=args.synthetic_demo, max_clips=args.max_clips,
vertical=args.vertical, captions=args.captions)
return 0
if __name__ == "__main__":
raise SystemExit(main())