diff --git a/app/services/alternative_songs.py b/app/services/alternative_songs.py index c208ac0f..12cc2efa 100644 --- a/app/services/alternative_songs.py +++ b/app/services/alternative_songs.py @@ -15,6 +15,7 @@ import uuid from typing import Any, Callable +from services.asset_manifest import SCHEMA_NAME, publish_generation_sidecar, read_asset_manifest from services.mix_concat import concatenate_multi_clip_videos, probe_duration_seconds from services.output_result_kind import classify_output_result_kind from services.video_editor import probe_audio @@ -45,15 +46,66 @@ def load_sidecar(video_path: str) -> dict[str, Any]: return payload -def save_sidecar(video_path: str, sidecar: dict[str, Any]) -> None: - path = sidecar_path(video_path) - os.makedirs(os.path.dirname(path) or ".", exist_ok=True) - tmp = path + ".tmp" - with open(tmp, "w", encoding="utf-8") as handle: - json.dump(sidecar, handle, indent=2, ensure_ascii=False) - handle.flush() - os.fsync(handle.fileno()) - os.replace(tmp, path) +def _clean_text(value: Any) -> str | None: + text = str(value or "").strip() + return text or None + + +def _v1_origin(document: Any) -> dict[str, Any] | None: + if not isinstance(document, dict) or document.get("schema") != SCHEMA_NAME: + return None + origin = document.get("origin") + return origin if isinstance(origin, dict) else None + + +def _has_director_pipeline(payload: dict[str, Any]) -> bool: + params = payload.get("params") if isinstance(payload.get("params"), dict) else {} + return any( + _clean_text(value) + for value in ( + payload.get("pipeline_id"), + payload.get("director_pipeline_id"), + params.get("director_pipeline_id"), + params.get("_director_pipeline_id"), + ) + ) + + +def save_sidecar(video_path: str, sidecar: dict[str, Any], *, tool: str | None = None) -> None: + """Publish gallery sidecar fields through the v1 asset-manifest writer. + + An explicit ``tool`` wins. Otherwise a canonical ``origin.tool`` on the + payload or existing v1 file is preserved so series-assembly/director/studio + are not rewritten. If only a director pipeline id is present, tool is + omitted so publish can attribute director. Actor is never invented as + ``user``; missing workspace stays unset instead of ``default``. + """ + payload = sidecar if isinstance(sidecar, dict) else {} + payload_origin = _v1_origin(payload) + disk_origin = None + existing = load_sidecar(video_path) + if _v1_origin(existing) is not None: + manifest = read_asset_manifest(video_path) + origin = None if manifest is None else manifest.get("origin") + disk_origin = origin if isinstance(origin, dict) else None + resolved_tool = ( + _clean_text(tool) + or _clean_text((payload_origin or {}).get("tool")) + or _clean_text((disk_origin or {}).get("tool")) + ) + if resolved_tool is None and _has_director_pipeline(payload): + resolved_tool = None + workspace_id = ( + _clean_text(payload.get("workspace")) + or _clean_text((payload_origin or {}).get("workspace_id")) + or _clean_text((disk_origin or {}).get("workspace_id")) + ) + publish_generation_sidecar( + video_path, + payload, + workspace_id=workspace_id, + tool=resolved_tool, + ) def _song_list(sidecar: dict[str, Any]) -> list[dict[str, Any]]: @@ -323,7 +375,7 @@ def write_mounted_sidecar( "created_at": time.time(), "parent_output": parent_name, } - save_sidecar(output_path, payload) + save_sidecar(output_path, payload, tool="alternative-songs") song["status"] = "mounted" song["mounted_output"] = os.path.basename(output_path) song["extra_clip_count"] = extra_count @@ -335,7 +387,7 @@ def write_mounted_sidecar( ) if classify != "music_video": payload["params"]["result_kind"] = "music_video" - save_sidecar(output_path, payload) + save_sidecar(output_path, payload, tool="alternative-songs") def remount_clips( diff --git a/tests/test_alternative_songs.py b/tests/test_alternative_songs.py index 1199c5ae..afc0e50a 100644 --- a/tests/test_alternative_songs.py +++ b/tests/test_alternative_songs.py @@ -1,3 +1,4 @@ +import json import os import random import sys @@ -20,8 +21,15 @@ remove_song, resolve_remount_sources, save_sidecar, + sidecar_path, source_clip_names, unique_mounted_name, + write_mounted_sidecar, +) +from services.asset_manifest import ( # noqa: E402 + SCHEMA_NAME, + publish_generation_sidecar, + read_asset_manifest, ) from services.output_result_kind import classify_output_result_kind # noqa: E402 @@ -149,5 +157,109 @@ def test_sidecar_roundtrip_keeps_songs(self): self.assertTrue(songs[0]["id"].startswith("song-")) +class AlternativeSongSidecarTests(unittest.TestCase): + def test_save_sidecar_publishes_v1_and_roundtrips_songs(self): + with tempfile.TemporaryDirectory() as tmp: + video = os.path.join(tmp, "clip.mp4") + Path(video).write_bytes(b"not-a-real-mp4") + sidecar = load_sidecar(video) + attach_song(sidecar, audio_name="en.mp3", duration_seconds=9) + sidecar["params"]["api_key"] = "secret-key" + save_sidecar(video, sidecar) + raw = json.loads(Path(sidecar_path(video)).read_text(encoding="utf-8")) + loaded = read_asset_manifest(video) + text = Path(sidecar_path(video)).read_text(encoding="utf-8") + self.assertEqual(raw["schema"], SCHEMA_NAME) + self.assertIsNotNone(loaded) + self.assertEqual(loaded["origin"]["actor"], "unknown") + self.assertEqual(raw["params"]["alternative_songs"][0]["audio_name"], "en.mp3") + self.assertTrue(raw["params"]["alternative_songs"][0]["id"].startswith("song-")) + self.assertNotIn("secret-key", text) + self.assertNotEqual(loaded["origin"].get("actor"), "user") + + def test_save_sidecar_keeps_existing_v1_asset_id(self): + with tempfile.TemporaryDirectory() as tmp: + video = os.path.join(tmp, "clip.mp4") + Path(video).write_bytes(b"not-a-real-mp4") + sidecar = load_sidecar(video) + attach_song(sidecar, audio_name="en.mp3", duration_seconds=9) + save_sidecar(video, sidecar) + first = read_asset_manifest(video) + self.assertIsNotNone(first) + asset_id = first["asset"]["id"] + mutated = load_sidecar(video) + attach_song(mutated, audio_name="pt.mp3", duration_seconds=8) + save_sidecar(video, mutated) + again = read_asset_manifest(video) + raw = json.loads(Path(sidecar_path(video)).read_text(encoding="utf-8")) + self.assertIsNotNone(again) + self.assertEqual(again["asset"]["id"], asset_id) + self.assertEqual(len(raw["params"]["alternative_songs"]), 2) + + def test_save_sidecar_keeps_series_assembly_origin_tool(self): + with tempfile.TemporaryDirectory() as tmp: + video = os.path.join(tmp, "episode.mp4") + Path(video).write_bytes(b"video") + publish_generation_sidecar( + video, + { + "generation_mode": "video", + "params": {"pipeline_type": "series_episode"}, + }, + workspace_id="lab", + tool="series-assembly", + ) + before = read_asset_manifest(video) + self.assertIsNotNone(before) + sidecar = load_sidecar(video) + attach_song(sidecar, audio_name="en.mp3", duration_seconds=9) + save_sidecar(video, sidecar) + loaded = read_asset_manifest(video) + raw = json.loads(Path(sidecar_path(video)).read_text(encoding="utf-8")) + self.assertIsNotNone(loaded) + self.assertEqual(loaded["origin"]["tool"], "series-assembly") + self.assertEqual(loaded["asset"]["id"], before["asset"]["id"]) + self.assertEqual(raw["params"]["alternative_songs"][0]["audio_name"], "en.mp3") + + def test_write_mounted_sidecar_publishes_alternative_songs_origin(self): + with tempfile.TemporaryDirectory() as tmp: + output = os.path.join(tmp, "clip_en_mv.mp4") + Path(output).write_bytes(b"video") + parent_sidecar = { + "params": { + "pipeline_type": "music_video", + "model_type": "ffmpeg_remount", + "resolution": "720p", + }, + } + song = attach_song({"params": {}}, audio_name="en.mp3", duration_seconds=9) + planned = [{ + "name": "shot.mp4", + "path": os.path.join(tmp, "shot.mp4"), + "duration": 4, + "used": 4, + "extra": False, + }] + write_mounted_sidecar( + output_path=output, + parent_name="clip.mp4", + parent_sidecar=parent_sidecar, + song=song, + planned=planned, + job_id="alt-song-1", + workspace="night-shift", + ) + raw = json.loads(Path(sidecar_path(output)).read_text(encoding="utf-8")) + loaded = read_asset_manifest(output, workspace_id="night-shift") + self.assertEqual(raw["schema"], SCHEMA_NAME) + self.assertEqual(raw["parent_output"], "clip.mp4") + self.assertEqual(raw["params"]["parent_output"], "clip.mp4") + self.assertEqual(song["status"], "mounted") + self.assertIsNotNone(loaded) + self.assertEqual(loaded["origin"]["tool"], "alternative-songs") + self.assertEqual(loaded["origin"]["workspace_id"], "night-shift") + self.assertEqual(loaded["origin"]["actor"], "unknown") + + if __name__ == "__main__": unittest.main()