#!/usr/bin/env python3
"""Run the pinned upstream planner offline. Never submit tasks or quote live prices."""
import argparse
import hashlib
import html
import json
from pathlib import Path
import shutil
import subprocess
import sys
import tempfile
HERE = Path(__file__).resolve().parent
UPSTREAM_SHA = "5fa9b19f52fb8920c83567468fa30e8391b8c503"
def planned_tasks(script, persona):
source = HERE / "upstream/scripts/build_tasks.py"
expected = json.loads((HERE / "provenance.json").read_text(encoding="utf-8"))["builderSha256"]
if hashlib.sha256(source.read_bytes()).hexdigest() != expected:
raise ValueError("Pinned upstream builder changed")
if not isinstance(script, list) or not script or len(script) > 10:
raise ValueError("Use 1–10 reviewed script segments")
ids = set()
for item in script:
if (not isinstance(item, dict) or not isinstance(item.get("id"), str)
or not item["id"].isalnum() or item["id"] in ids
or not isinstance(item.get("t"), str) or not item["t"].strip()
or item.get("shot") not in {"full", "pip"}):
raise ValueError("Invalid or duplicate segment")
ids.add(item["id"])
with tempfile.TemporaryDirectory(prefix="opcboost-intro-") as temporary:
root = Path(temporary)
(root / "scripts").mkdir()
shutil.copyfile(source, root / "scripts/build_tasks.py")
(root / "persona.json").write_text(json.dumps(persona, ensure_ascii=False), encoding="utf-8")
(root / "script.json").write_text(json.dumps(script, ensure_ascii=False), encoding="utf-8")
result = subprocess.run([sys.executable, "-X", "utf8", str(root / "scripts/build_tasks.py"),
str(root / "script.json"), str(root / "tasks.json"),
"--purpose", "introducing the OPCBoost reproducible case library"],
text=True, encoding="utf-8", capture_output=True, timeout=15)
if result.returncode:
raise ValueError(result.stderr.strip() or "Upstream planning failed")
tasks = json.loads((root / "tasks.json").read_text(encoding="utf-8"))
if len(tasks) != len(script) or any(not 4 <= task["duration"] <= 15 for task in tasks):
raise ValueError("Unexpected upstream task structure")
# Its internal fixture-unit cost is deliberately not published as a service quote.
return {"status": "offline-preparation", "readyToSubmit": False,
"estimatedCredits": None, "priceStatus": "not-quoted",
"generatedVideo": False, "upstreamCommit": UPSTREAM_SHA,
"durationBasis": "upstream text heuristic, not recorded audio",
"tasks": tasks}
def timestamp(seconds):
return f"{seconds//3600:02}:{seconds//60%60:02}:{seconds%60:02},000"
def outputs(plan, script):
seconds, subtitles, rows = 0, [], []
for index, (task, item) in enumerate(zip(plan["tasks"], script), 1):
end = seconds + task["duration"]
subtitle_text = " ".join(item["t"].split())
subtitles.append(f"{index}\n{timestamp(seconds)} --> {timestamp(end)}\n{subtitle_text}\n")
rows.append(f'{seconds:02}–{end:02}s · {html.escape(item["shot"])}'
f'
{html.escape(task["label"])}
{html.escape(item["t"])}
'
f'{html.escape(task["resolution"])} · 计划 {task["duration"]} 秒')
seconds = end
body = '''
OPCBoost 数字人口播 · 离线制作台
'''.replace("TOTAL", str(seconds)).replace("COUNT", str(len(script))).replace("ROWS", "".join(rows))
strips=[]
offset=0
for i,(task,item) in enumerate(zip(plan['tasks'][:3],script[:3])):
y=220+i*100
strips.append(f'{offset:02}–{offset+task["duration"]:02}s · {task["resolution"]} · {html.escape(task["label"])}{html.escape(item["t"])}')
offset+=task['duration']
preview='\n'
return {"draft-tasks.json": json.dumps(plan, ensure_ascii=False, indent=2) + "\n",
"planned-subtitles.srt": "\n".join(subtitles), "index.html": body, "preview.svg":preview}
def main():
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--output-dir", type=Path, default=HERE)
args = parser.parse_args()
standalone = args.output_dir.resolve() != HERE.resolve()
if standalone and args.output_dir.exists():
parser.error("--output-dir must name a new directory")
script = json.loads((HERE / "script.json").read_text(encoding="utf-8"))
persona = json.loads((HERE / "persona.offline.json").read_text(encoding="utf-8"))
try:
plan = planned_tasks(script, persona)
rendered = outputs(plan, script)
except (ValueError, subprocess.TimeoutExpired) as error:
parser.error(str(error))
args.output_dir.mkdir(parents=True, exist_ok=True)
if standalone:
# Explicit public dependencies keep every local link and the replay runnable.
for name in ("script.json", "persona.offline.json", "provenance.json", "reproduce.py",
"LICENSE.upstream.txt", "NOTICE.upstream.md", "THIRD_PARTY.upstream.md",
"upstream/scripts/build_tasks.py"):
target = args.output_dir / name
target.parent.mkdir(parents=True, exist_ok=True)
shutil.copyfile(HERE / name, target)
for name, content in rendered.items():
(args.output_dir / name).write_text(content, encoding="utf-8")
print(f"Offline preparation: {len(plan['tasks'])} segments / {sum(t['duration'] for t in plan['tasks'])} planned seconds; price unknown; video not generated.")
if __name__ == "__main__":
main()