Blender 원본 검사와 승인 경계를 지키는 로컬 렌더 실행기 추가

This commit is contained in:
NINI
2026-10-09 23:54:55 +09:00
parent 15e84fb7c9
commit 31efdd6681
7 changed files with 1092 additions and 0 deletions
+375
View File
@@ -0,0 +1,375 @@
"""승인 경계·시간·원본 참조 회귀. 생성한 시도는 진단을 위해 남긴다."""
from copy import deepcopy
import json
from pathlib import Path
import unittest
import uuid
import wave
import zlib
import struct
import shutil
import run
ROOT = Path(__file__).resolve().parents[2]
WATER = ROOT / "projects/water-hydrogen-bond-surface-tension"
PROFILE = dict(engine="CYCLES", device="CPU", fps=10, width=90, height=160, samples=4, seed=0)
def write(path, value):
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(json.dumps(value, ensure_ascii=False, indent=2), encoding="utf-8")
def fixture():
folder = ROOT / "tools/blender/.cache" / uuid.uuid4().hex / WATER.name
project = json.loads((WATER / "project.json").read_text(encoding="utf-8"))
project["schema_version"] = "1.1"
project["settings"].pop("image_model", None)
project["settings"]["visual_engine"] = "Blender"
artifacts = {}
for stage in project["stages"]:
if stage["revision"] is None:
continue
data = json.loads((ROOT / stage["artifact_path"]).read_text(encoding="utf-8"))
if stage["stage_id"] in ("08-scenes", "09-entities"):
data.update(schema_version="1.1", revision=2)
stage.update(status="draft", revision=2)
stage["artifact_path"] = (folder / "stages" / stage["stage_id"] / ("r" + str(stage["revision"])) / "artifact.json").relative_to(ROOT).as_posix()
artifacts[stage["stage_id"]] = data
cuts = artifacts["07-cuts"]["payload"]
cuts["narration_duration_ms"] = 2000
cuts["cuts"] = [dict(id="c1", start_ms=0, end_ms=1000, unit_ids=["u1"], narration_text="첫 장면"),
dict(id="c2", start_ms=1000, end_ms=2000, unit_ids=["u1"], narration_text="다음 장면")]
cuts["captions"] = []
source = folder / "stages/08-scenes/r2/sample.blend"
source.parent.mkdir(parents=True, exist_ok=True)
source.write_bytes("검사 전 원본".encode("utf-8"))
narration = folder / "narration.wav"
with wave.open(str(narration), "wb") as audio:
audio.setparams((1, 2, 24000, 0, "NONE", "not compressed"))
audio.writeframes(b"\0\0" * 48000)
artifacts["06-tts"]["payload"]["result"]["narration"].update(
path=narration.relative_to(ROOT).as_posix(), duration_ms=2000)
tts = artifacts["06-tts"]["payload"]
ids = [unit["id"] for unit in tts["request"]["units"]]
tts["result"]["files"] = []
tts["result"]["unit_ranges"] = []
for index, unit_id in enumerate(ids):
path = folder / (unit_id + ".wav")
with wave.open(str(path), "wb") as audio:
audio.setparams((1, 2, 24000, 0, "NONE", "not compressed"))
audio.writeframes(b"\0\0" * 6000)
tts["result"]["files"].append(dict(unit_id=unit_id, media=dict(path=path.relative_to(ROOT).as_posix(), kind="audio", verified=True,
duration_ms=250, width=None, height=None)))
tts["result"]["unit_ranges"].append(dict(unit_id=unit_id, start_ms=index * 250, end_ms=(index + 1) * 250))
for cut in cuts["cuts"]:
cut["unit_ids"] = ids[:4] if cut["id"] == "c1" else ids[4:]
scenes = artifacts["08-scenes"]
scenes["payload"]["scenes"] = [dict(cut_id=cut["id"], description="움직이는 구", start_state="왼쪽", end_state="오른쪽", camera="고정") for cut in cuts["cuts"]]
scenes["payload"]["connections"] = [dict(from_cut_id="c1", to_cut_id="c2", mode="cut", reason="시점 변경")]
scenes["payload"]["preview"] = dict(scene_file=dict(path=source.relative_to(ROOT).as_posix(), verified=False, sha256=None, blender_version=None), render_profile=deepcopy(PROFILE), samples=[])
for stage in project["stages"]:
if stage["stage_id"] in artifacts:
write(ROOT / stage["artifact_path"], artifacts[stage["stage_id"]])
write(folder / "project.json", project)
job = dict(mode="preview", project=(folder / "project.json").relative_to(ROOT).as_posix(),
artifact=next(s["artifact_path"] for s in project["stages"] if s["stage_id"] == "08-scenes"),
output_revision=2, source=source.relative_to(ROOT).as_posix(), scene="SceneA",
frame=None, start=0, end=10, cut_id=["c1"], frame_id=None, reference_id=None,
require_object=["Ball"])
return folder, project, artifacts, job
def production_fixture():
folder, project, artifacts, job = fixture()
for stage in project["stages"]:
if stage["stage_id"] in ("08-scenes", "09-entities"):
stage["status"] = "complete"
artifacts["09-entities"]["inputs"][1]["revision"] = 2
artifacts["09-entities"]["payload"]["appearances"] = [dict(cut_id=c, character_ids=[], background_ids=[]) for c in ("c1", "c2")]
preview = artifacts["08-scenes"]["payload"]["preview"]
preview["scene_file"].update(verified=True, sha256=run.digest(ROOT / job["source"]), blender_version="5.2.2 LTS")
preview["samples"] = [dict(id="sample", cut_ids=["c1"], media=dict(path=job["source"].replace(".blend", ".png"),
kind="image", verified=True, duration_ms=None, width=90, height=160))]
def chunk(kind, body):
return struct.pack(">I", len(body)) + kind + body + struct.pack(">I", zlib.crc32(kind + body))
png = b"\x89PNG\r\n\x1a\n" + chunk(b"IHDR", struct.pack(">IIBBBBB", 90, 160, 8, 2, 0, 0, 0))
png += chunk(b"IDAT", zlib.compress((b"\0" + b"\0" * 270) * 160)) + chunk(b"IEND", b"")
(ROOT / preview["samples"][0]["media"]["path"]).write_bytes(png)
review = next(r for r in project["reviews"] if r["review_id"] == "scenes")
review["artifact_refs"] = [dict(stage_id=s, revision=1 if s == "07-cuts" else 2) for s in ("07-cuts", "08-scenes", "09-entities")]
review["feedback"] = "회귀 검사 전용 승인 기록이며 실제 영상의 사용자 승인이 아닙니다"
reference = json.loads((ROOT / "templates/stages/10-references.json").read_text(encoding="utf-8"))
reference.update(project_id=project["project_id"], revision=2,
inputs=[dict(stage_id=s, revision=1 if s == "02-research" else 2) for s in ("02-research", "08-scenes", "09-entities")])
reference["payload"]["request"]["render_profile"] = deepcopy(PROFILE)
reference["payload"]["request"]["items"] = [dict(id="ball", kind="object", entity_id=None, prompt="검사용 구")]
pointer = next(s for s in project["stages"] if s["stage_id"] == "10-references")
pointer.update(status="draft", revision=2,
artifact_path=(folder / "stages/10-references/r2/artifact.json").relative_to(ROOT).as_posix())
artifacts["10-references"] = reference
source = folder / "stages/10-references/r2/model.blend"
source.parent.mkdir(parents=True, exist_ok=True)
source.write_bytes(b"uninspected fixture")
job.update(mode="production", artifact=pointer["artifact_path"], source=source.relative_to(ROOT).as_posix(),
frame=0, start=None, end=None, reference_id="ball", cut_id=[])
for stage in project["stages"]:
if stage["stage_id"] in artifacts:
write(ROOT / stage["artifact_path"], artifacts[stage["stage_id"]])
write(folder / "project.json", project)
return folder, project, artifacts, job
def endpoint_fixture(motion=False):
folder, project, artifacts, job = production_fixture()
reference = artifacts["10-references"]
reference["payload"]["result"] = dict(scene_file=dict(path=job["source"], verified=True,
sha256=run.digest(ROOT / job["source"]), blender_version="5.2.2 LTS"),
items=[dict(reference_id="ball", object_name="Ball", media=deepcopy(artifacts["08-scenes"]["payload"]["preview"]["samples"][0]["media"]))])
reference_media = reference["payload"]["result"]["items"][0]["media"]
image = folder / "stages/10-references/r2/reference.png"
shutil.copyfile(ROOT / reference_media["path"], image)
reference_media["path"] = image.relative_to(ROOT).as_posix()
frames = [dict(id=cut + "-" + position, cut_id=cut, position=position, prompt="상태 설명", reference_ids=["ball"],
scene_name="SceneA" if cut == "c1" else "SceneB", frame_index=index)
for cut, start in (("c1", 0), ("c2", 10)) for position, index in (("start", start), ("end", start + 10))]
source = folder / "stages/11-images/r2/animation.blend"
source.parent.mkdir(parents=True, exist_ok=True)
source.write_bytes(b"uninspected endpoint fixture")
images = json.loads((ROOT / "templates/stages/11-images.json").read_text(encoding="utf-8"))
images.update(project_id=project["project_id"], revision=2,
inputs=[dict(stage_id=s, revision=1 if s in ("02-research", "03-script", "07-cuts") else 2)
for s in ("02-research", "03-script", "07-cuts", "08-scenes", "09-entities", "10-references")])
images["payload"]["request"].update(frames=frames, scene_path=source.relative_to(ROOT).as_posix(), render_profile=deepcopy(PROFILE))
artifacts["11-images"] = images
for stage in project["stages"]:
if stage["stage_id"] in ("10-references", "11-images"):
stage.update(revision=2, status="complete" if stage["stage_id"] == "10-references" or motion else "draft",
artifact_path=(folder / "stages" / stage["stage_id"] / "r2/artifact.json").relative_to(ROOT).as_posix())
job.update(artifact=(folder / "stages/11-images/r2/artifact.json").relative_to(ROOT).as_posix(),
source=source.relative_to(ROOT).as_posix(), frame_id="c1-start", reference_id=None)
if motion:
images["payload"]["result"] = dict(scene_file=dict(path=job["source"], verified=True,
sha256=run.digest(source), blender_version="5.2.2 LTS"),
frames=[dict(frame_id=f["id"], media=deepcopy(reference["payload"]["result"]["items"][0]["media"])) for f in frames])
for item in images["payload"]["result"]["frames"]:
image = source.parent / (item["frame_id"] + ".png")
shutil.copyfile(ROOT / item["media"]["path"], image)
item["media"]["path"] = image.relative_to(ROOT).as_posix()
clips = [dict(cut_id=c, start_frame_id=c + "-start", end_frame_id=c + "-end", duration_ms=1000,
motion_prompt="구의 움직임", scene_name="SceneA" if c == "c1" else "SceneB", start_frame=start, end_frame=start + 10)
for c, start in (("c1", 0), ("c2", 10))]
artifact = json.loads((ROOT / "templates/stages/12-motion.json").read_text(encoding="utf-8"))
artifact.update(project_id=project["project_id"], revision=2,
inputs=[dict(stage_id=s, revision=1 if s == "07-cuts" else 2) for s in ("07-cuts", "08-scenes", "11-images")])
artifact["payload"]["request"].update(clips=clips, scene_path=job["source"], render_profile=deepcopy(PROFILE))
artifacts["12-motion"] = artifact
pointer = next(s for s in project["stages"] if s["stage_id"] == "12-motion")
pointer.update(revision=2, status="draft", artifact_path=(folder / "stages/12-motion/r2/artifact.json").relative_to(ROOT).as_posix())
job.update(artifact=pointer["artifact_path"], frame=None, start=0, end=10, cut_id=["c1"], frame_id=None)
for stage in project["stages"]:
if stage["stage_id"] in artifacts:
write(ROOT / stage["artifact_path"], artifacts[stage["stage_id"]])
write(folder / "project.json", project)
return folder, project, artifacts, job
class RequestTests(unittest.TestCase):
def setUp(self):
self.folder, self.project, self.artifacts, self.job = fixture()
def persist(self):
write(self.folder / "project.json", self.project)
for stage in self.project["stages"]:
if stage["stage_id"] in self.artifacts:
write(ROOT / stage["artifact_path"], self.artifacts[stage["stage_id"]])
def test_preview_requires_no_new_scenes_approval(self):
validated = run.prepare(ROOT, self.job)
self.assertEqual(validated["profile"], PROFILE)
self.assertEqual(validated["frames"], list(range(10)))
def test_production_rejects_preview_and_old_scenes(self):
self.job["mode"] = "production"
with self.assertRaises(ValueError):
run.prepare(ROOT, self.job)
folder, project, artifacts, job = production_fixture()
self.assertEqual(run.prepare(ROOT, job)["frames"], [0])
review = next(r for r in project["reviews"] if r["review_id"] == "scenes")
for decision in ("pending", "changes_requested", "stale", "old_approved", "no_feedback"):
review.update(decision="approved", feedback="회귀 검사 기록")
review["artifact_refs"][1]["revision"] = 2
if decision == "old_approved":
review["artifact_refs"][1]["revision"] = 1
elif decision == "no_feedback":
review["feedback"] = ""
else:
review["decision"] = decision
write(folder / "project.json", project)
with self.subTest(decision=decision), self.assertRaises(ValueError):
run.prepare(ROOT, job)
def test_bad_pointers_inputs_and_revision(self):
for change in ("revision", "pointer", "stale", "input"):
folder, project, artifacts, job = fixture()
if change == "revision":
job["output_revision"] = 1
elif change == "pointer":
project["stages"][7]["revision"] = 1
elif change == "stale":
project["stages"][2]["status"] = "stale"
else:
artifacts["08-scenes"]["inputs"][0]["revision"] = 99
write(ROOT / job["artifact"], artifacts["08-scenes"])
write(folder / "project.json", project)
with self.subTest(change=change), self.assertRaises(ValueError):
run.prepare(ROOT, job)
def test_ranges_aspect_grid_and_foreign_source(self):
for change in ("long", "negative", "unknown_cut", "foreign", "aspect", "grid"):
folder, project, artifacts, job = fixture()
if change == "long":
job["end"] = 51
elif change == "negative":
job["start"] = -1
elif change == "unknown_cut":
job["cut_id"] = ["foreign"]
elif change == "foreign":
job["source"] = "../foreign.blend"
elif change == "aspect":
artifacts["08-scenes"]["payload"]["preview"]["render_profile"]["width"] = 100
else:
artifacts["07-cuts"]["payload"]["cuts"][0]["end_ms"] = 999
for stage in project["stages"]:
if stage["stage_id"] in artifacts:
write(ROOT / stage["artifact_path"], artifacts[stage["stage_id"]])
with self.subTest(change=change), self.assertRaises(ValueError):
run.prepare(ROOT, job)
def test_source_hash_changed_is_rejected(self):
metadata = dict(path=self.job["source"], verified=True, sha256="0" * 64, blender_version="5.2.2 LTS")
with self.assertRaises(ValueError):
run.check_source(ROOT, metadata)
def test_eevee_does_not_claim_an_unsupported_seed(self):
profile = self.artifacts["08-scenes"]["payload"]["preview"]["render_profile"]
profile.update(engine="BLENDER_EEVEE", device="GPU", seed=5)
self.persist()
with self.assertRaises(ValueError):
run.prepare(ROOT, self.job)
def test_production_missing_or_foreign_preview_is_rejected(self):
for change in ("missing", "foreign", "hash"):
folder, project, artifacts, job = production_fixture()
preview = artifacts["08-scenes"]["payload"]["preview"]
if change == "missing":
preview["samples"][0]["media"]["path"] += ".missing"
elif change == "foreign":
preview["scene_file"]["path"] = self.job["source"]
preview["scene_file"]["sha256"] = run.digest(ROOT / self.job["source"])
else:
preview["scene_file"]["sha256"] = "a" * 64
write(folder / "stages/08-scenes/r2/artifact.json", artifacts["08-scenes"])
with self.subTest(change=change), self.assertRaises(ValueError):
run.prepare(ROOT, job)
def test_attempt_identity_includes_scene_hash_profile_and_range(self):
job = run.prepare(ROOT, self.job)
key = run.identity(job, "a" * 64)
for name, value in (("scene", "SceneB"), ("frames", [10]), ("profile", dict(PROFILE, seed=4))):
other = dict(job, **{name: value})
self.assertNotEqual(key, run.identity(other, "a" * 64))
self.assertNotEqual(key, run.identity(job, "b" * 64))
def test_endpoint_and_motion_requests_match_actual_cuts(self):
for motion in (False, True):
folder, project, artifacts, job = endpoint_fixture(motion)
self.assertEqual(len(run.prepare(ROOT, job)["frames"]), 10 if motion else 1)
for change in ("scene", "endpoint", "reference", "profile", "continuous", "cut_scene"):
folder, project, artifacts, job = endpoint_fixture(motion)
stage = "12-motion" if motion else "11-images"
request = artifacts[stage]["payload"]["request"]
if change == "scene":
job["scene"] = "ForeignScene"
elif change == "endpoint":
artifacts["11-images"]["payload"]["request"]["frames"][3]["frame_index"] = 19
elif change == "reference":
artifacts["11-images"]["payload"]["request"]["frames"][0]["reference_ids"] = ["missing"]
elif change == "profile":
request["render_profile"]["fps"] = 20
elif change == "continuous":
artifacts["08-scenes"]["payload"]["connections"][0]["mode"] = "continuous"
else:
for target in artifacts["11-images"]["payload"]["request"]["frames"]:
target["scene_name"] = "SceneA"
if motion:
artifacts["12-motion"]["payload"]["request"]["clips"][1]["scene_name"] = "SceneA"
for pointer in project["stages"]:
if pointer["stage_id"] in artifacts:
write(ROOT / pointer["artifact_path"], artifacts[pointer["stage_id"]])
with self.subTest(motion=motion, change=change), self.assertRaises(ValueError):
run.prepare(ROOT, job)
def test_current_10_entity_reference_must_exist(self):
folder, project, artifacts, job = production_fixture()
artifacts["10-references"]["payload"]["request"]["items"][0].update(kind="character", entity_id="ABSENT")
write(ROOT / job["artifact"], artifacts["10-references"])
with self.assertRaises(ValueError):
run.prepare(ROOT, job)
def test_new_renderer_rejects_legacy_project_configuration(self):
self.project.update(schema_version="1.0")
self.project["settings"].pop("visual_engine")
self.project["settings"]["image_model"] = "gpt-image-2.5-sunburst"
self.persist()
with self.assertRaises(ValueError):
run.prepare(ROOT, self.job)
def test_complete_06_unit_files_need_actual_metadata(self):
for change in ("missing", "unverified", "duration", "range", "truncated", "id"):
folder, project, artifacts, job = fixture()
tts = artifacts["06-tts"]["payload"]
media = tts["result"]["files"][0]["media"]
if change == "missing":
media["path"] += ".missing"
elif change == "unverified":
media["verified"] = False
elif change == "duration":
media["duration_ms"] = 251
elif change == "range":
tts["result"]["unit_ranges"][0]["end_ms"] = 251
elif change == "id":
tts["result"]["files"][0]["unit_id"] = "foreign"
else:
path = ROOT / media["path"]
path.write_bytes(path.read_bytes()[:-2])
write(folder / "stages/06-tts/r1/artifact.json", artifacts["06-tts"])
with self.subTest(change=change), self.assertRaises(ValueError):
run.prepare(ROOT, job)
def test_old_visual_revision_source_and_media_are_rejected(self):
for input_stage in ("08-scenes", "10-references", "11-images"):
for change in ("source", "media"):
folder, project, artifacts, job = endpoint_fixture(motion=True)
payload = artifacts[input_stage]["payload"]
result = payload["preview"] if input_stage == "08-scenes" else payload["result"]
key = {"08-scenes": "samples", "10-references": "items", "11-images": "frames"}[input_stage]
metadata = result["scene_file"] if change == "source" else result[key][0]["media"]
old = ROOT / metadata["path"].replace("/r2/", "/r1/")
old.parent.mkdir(parents=True, exist_ok=True)
shutil.copyfile(ROOT / metadata["path"], old)
metadata["path"] = old.relative_to(ROOT).as_posix()
if input_stage == "11-images" and change == "source":
artifacts["12-motion"]["payload"]["request"]["scene_path"] = metadata["path"]
job["source"] = metadata["path"]
for pointer in project["stages"]:
if pointer["stage_id"] in artifacts:
write(ROOT / pointer["artifact_path"], artifacts[pointer["stage_id"]])
with self.subTest(stage=input_stage, change=change), self.assertRaises(ValueError):
run.prepare(ROOT, job)
if __name__ == "__main__":
unittest.main()