diff --git a/tests/test_utils_video_metadata.py b/tests/test_utils_video_metadata.py index 1d562677..21335c1a 100644 --- a/tests/test_utils_video_metadata.py +++ b/tests/test_utils_video_metadata.py @@ -1,8 +1,11 @@ """Tests for utils/video_metadata.py: prompt metadata from video containers. -ComfyUI's video savers store ``{"prompt": ..., "workflow": ...}`` as JSON in -the container's ``comment`` tag. ffprobe reads it; when ffprobe is missing -videos are skipped without noise. +Current ComfyUI savers (core SaveVideo/SaveWEBM, VideoHelperSuite) write one +container tag per key (``prompt``, ``workflow``, ``parameters``, +``comfyassets_prompt_pairs``); older VideoHelperSuite wrote one ``comment`` +JSON envelope, with ``description`` as a fallback. One selector picks a single +carrier for the inspector and the scan. ffprobe reads the tags; when ffprobe +is missing videos are skipped without noise. """ import json @@ -19,9 +22,11 @@ sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from utils import video_metadata # noqa: E402 from utils.video_metadata import ( # noqa: E402 VIDEO_EXTENSIONS, + DuplicateVideoTagError, is_video_path, metadata_from_tags, read_video_metadata, + select_video_carrier, ) HAVE_FFMPEG = bool(shutil.which("ffmpeg") and shutil.which("ffprobe")) @@ -69,6 +74,149 @@ class TestMetadataFromTags(unittest.TestCase): self.assertEqual(metadata_from_tags(tags), {"prompt": GRAPH}) +class TestSelectVideoCarrier(unittest.TestCase): + """Spec 2026-10-19 resolution parity section 8: one carrier, never merged.""" + + WORKFLOW = json.dumps({"nodes": [{"id": "per-key"}]}) + + def test_per_key_tags_win_over_comment(self): + tags = { + "prompt": GRAPH, + "workflow": self.WORKFLOW, + "comment": json.dumps({"prompt": json.dumps({"9": {}})}), + } + metadata, formats, carrier = select_video_carrier(tags) + self.assertEqual(metadata, {"prompt": GRAPH, "workflow": self.WORKFLOW}) + self.assertEqual( + formats, {"prompt": "original-text", "workflow": "original-text"} + ) + self.assertIsNone(carrier) + + def test_matroska_uppercase_keys_fold(self): + tags = {"PROMPT": GRAPH, "Workflow": self.WORKFLOW, "ENCODER": "Lavf"} + metadata, formats, carrier = select_video_carrier(tags) + self.assertEqual(metadata, {"prompt": GRAPH, "workflow": self.WORKFLOW}) + self.assertEqual( + formats, {"prompt": "original-text", "workflow": "original-text"} + ) + self.assertIsNone(carrier) + folded = {"COMMENT": json.dumps({"prompt": GRAPH})} + self.assertEqual(select_video_carrier(folded)[0], {"prompt": GRAPH}) + + def test_folded_duplicate_keys_fail(self): + for tags in ( + {"PROMPT": GRAPH, "prompt": json.dumps({"9": {}})}, + {"Workflow": self.WORKFLOW, "WORKFLOW": "{}", "prompt": GRAPH}, + {"comment": json.dumps({"prompt": GRAPH}), "COMMENT": "{}"}, + ): + with self.assertRaises(DuplicateVideoTagError): + select_video_carrier(tags) + # Byte-identical duplicates are one record, as for PNG text chunks. + metadata, _, _ = select_video_carrier({"PROMPT": GRAPH, "prompt": GRAPH}) + self.assertEqual(metadata, {"prompt": GRAPH}) + + def test_comment_envelope_still_read(self): + envelope = json.dumps({"prompt": GRAPH, "workflow": {"nodes": []}}) + metadata, formats, carrier = select_video_carrier( + {"comment": envelope, "encoder": "Lavf"} + ) + self.assertEqual(metadata["prompt"], GRAPH) + self.assertEqual(json.loads(metadata["workflow"]), {"nodes": []}) + self.assertEqual( + formats, {"prompt": "original-text", "workflow": "reserialized-json"} + ) + self.assertEqual(carrier, envelope) + + def test_description_fallback(self): + envelope = json.dumps({"prompt": GRAPH}) + tags = {"comment": "not json", "description": envelope} + self.assertEqual( + select_video_carrier(tags), + ({"prompt": GRAPH}, {"prompt": "original-text"}, envelope), + ) + self.assertIsNone(select_video_carrier({"comment": "note", "encoder": "x"})) + self.assertIsNone(select_video_carrier({"comment": "[1]"})) + self.assertIsNone(select_video_carrier({})) + self.assertIsNone(select_video_carrier(None)) + self.assertIsNone(select_video_carrier({"comment": 7})) + + def test_role_carrier_beats_workflow_only_carrier(self): + envelope = json.dumps({"prompt": GRAPH, "workflow": {"nodes": ["c"]}}) + metadata, formats, carrier = select_video_carrier( + {"workflow": self.WORKFLOW, "comment": envelope} + ) + self.assertEqual(metadata["prompt"], GRAPH) + self.assertEqual(json.loads(metadata["workflow"]), {"nodes": ["c"]}) + self.assertEqual(carrier, envelope) + for role in ("parameters", "comfyassets_prompt_pairs"): + metadata, _, carrier = select_video_carrier( + {"comment": json.dumps({"workflow": {}}), role: "x"} + ) + self.assertEqual(metadata, {role: "x"}) + self.assertIsNone(carrier) + # Without a role anywhere, the first workflow-only carrier wins. + metadata, _, carrier = select_video_carrier( + {"workflow": self.WORKFLOW, "comment": json.dumps({"workflow": {}})} + ) + self.assertEqual(metadata, {"workflow": self.WORKFLOW}) + self.assertIsNone(carrier) + + def test_carriers_never_merge(self): + metadata, _, carrier = select_video_carrier( + { + "prompt": GRAPH, + "comment": json.dumps({"workflow": {"nodes": ["comment"]}}), + "description": json.dumps({"parameters": "Steps: 1"}), + } + ) + self.assertEqual(metadata, {"prompt": GRAPH}) + self.assertIsNone(carrier) + metadata, _, carrier = select_video_carrier( + { + "workflow": self.WORKFLOW, + "comment": json.dumps({"prompt": GRAPH}), + "description": json.dumps({"parameters": "Steps: 1"}), + } + ) + self.assertEqual(metadata, {"prompt": GRAPH}) + self.assertEqual(carrier, json.dumps({"prompt": GRAPH})) + + def test_per_key_values_are_original_text_and_envelope_objects_reserialized( + self, + ): + exact = '{"3":{"inputs":{"cfg":6.5000000000000000001}}}' + metadata, formats, _ = select_video_carrier({"prompt": exact}) + self.assertEqual(metadata, {"prompt": exact}) + self.assertEqual(formats, {"prompt": "original-text"}) + envelope = json.dumps( + {"prompt": {"3": {}}, "workflow": {"nodes": []}, "parameters": "Steps: 1"} + ) + metadata, formats, carrier = select_video_carrier({"comment": envelope}) + self.assertEqual(json.loads(metadata["prompt"]), {"3": {}}) + self.assertEqual(json.loads(metadata["workflow"]), {"nodes": []}) + self.assertEqual(metadata["parameters"], "Steps: 1") + self.assertEqual( + formats, + { + "prompt": "reserialized-json", + "workflow": "reserialized-json", + "parameters": "original-text", + }, + ) + self.assertEqual(carrier, envelope) + + def test_null_role_markers_and_invalid_workflow_keep_their_shape(self): + # The inspector refuses a null role marker rather than falling back. + metadata, formats, _ = select_video_carrier( + {"comment": json.dumps({"prompt": None, "workflow": [1]})} + ) + self.assertEqual(metadata, {"prompt": None}) + self.assertEqual(formats, {"workflow": "invalid"}) + metadata, formats, _ = select_video_carrier({"prompt": 7, "workflow": 1}) + self.assertEqual(metadata, {"prompt": None}) + self.assertEqual(formats, {"workflow": "invalid"}) + + class TestIsVideoPath(unittest.TestCase): def test_extensions(self): diff --git a/utils/video_metadata.py b/utils/video_metadata.py index a7884dcf..f3fe6d5b 100644 --- a/utils/video_metadata.py +++ b/utils/video_metadata.py @@ -1,14 +1,18 @@ """Prompt metadata from video containers. -ComfyUI's video savers (core SaveVideo, VideoHelperSuite, ...) write the -queued graph into the container's ``comment`` tag as JSON: -``{"prompt": "", "workflow": {...}}``. ffprobe reads it on every -platform; when ffprobe is not installed, videos are skipped with a single -notice instead of one error per file. +Current ComfyUI video savers (core SaveVideo/SaveWEBM, VideoHelperSuite since +``c67ff3e``) write one container tag per key, each a JSON string: +``prompt``, ``workflow``, ``parameters``, ``comfyassets_prompt_pairs``. Older +VideoHelperSuite wrote one ``comment`` tag holding a JSON envelope +``{"prompt": "", "workflow": {...}}``; ``description`` is read as a +fallback envelope. ffprobe reads the tags on every platform; when ffprobe is +not installed, videos are skipped with a single notice instead of one error +per file. +``select_video_carrier`` picks exactly one carrier for both the inspector and +the scan, so Images and ingestion read the same prompts from the same file. The result uses the same shape as PNG text chunks (string values keyed -``prompt`` / ``workflow`` / ``parameters``) so the existing prompt parser -handles both. +``prompt`` / ``workflow`` / ...) so the existing prompt parser handles both. """ import functools @@ -20,10 +24,15 @@ import shutil # Fixed argv, absolute binary, no shell. import subprocess # nosec B404 +from .pair_execution import EXECUTION_KEY + VIDEO_EXTENSIONS = (".mp4", ".mov", ".m4v", ".webm", ".mkv", ".avi", ".wmv") FFPROBE_TIMEOUT = 20 # seconds; a stuck probe must not hang a worker thread -_COMMENT_TAGS = ("comment", "description") -_METADATA_KEYS = ("prompt", "workflow", "parameters") +# Carrier order: per-key tags, then the ``comment`` envelope, then ``description``. +ROLE_TAG_KEYS = ("prompt", "parameters", EXECUTION_KEY) +METADATA_TAG_KEYS = ("prompt", "workflow", "parameters", EXECUTION_KEY) +ENVELOPE_TAG_KEYS = ("comment", "description") +VIDEO_TAG_KEYS = METADATA_TAG_KEYS + ENVELOPE_TAG_KEYS _NO_WINDOW = getattr(subprocess, "CREATE_NO_WINDOW", 0) # no console flash on Windows logger = logging.getLogger("prompt_manager.video") @@ -46,32 +55,118 @@ def reset_missing_notice(): _missing_notice_shown = False -def metadata_from_tags(tags) -> dict: - """Normalise container tags into PNG-text-chunk shaped metadata. +class DuplicateVideoTagError(ValueError): + """Two container tags fold to the same metadata key with different text.""" - Returns ``{}`` when no tag carries a JSON object with a known key. + +def _folded_tags(tags: dict) -> dict: + """Known tags keyed by ASCII-lowercased name. + + Matroska muxers upper-case tag names and ffprobe reports them as written, + so the fold happens here. Two tags folding to one key are a duplicate + record: byte-identical duplicates are one record (as for PNG text chunks), + differing ones fail the read. """ - if not isinstance(tags, dict): - return {} - for tag in _COMMENT_TAGS: - raw = tags.get(tag) + folded = {} + for key, value in tags.items(): + if not isinstance(key, str): + continue + name = key.lower() + if name not in VIDEO_TAG_KEYS: + continue + if name in folded and folded[name] != value: + raise DuplicateVideoTagError(f"Duplicate video metadata tag: {name}") + folded[name] = value + return folded + + +def _record(value, key): + """One carrier field as (metadata text or None, record format or None). + + A null marker (or a non-text role value) has no record format: it is kept + in the metadata as ``None`` so the inspector refuses it instead of letting + fallback evidence win, while scans drop it. + """ + if key == "workflow" and not isinstance(value, (str, dict)): + return None, "invalid" + if isinstance(value, str): + return value, "original-text" + if isinstance(value, dict): + return json.dumps(value), "reserialized-json" + return None, None + + +def _carriers(folded: dict): + """Candidate carriers in order: (fields, raw envelope text or None).""" + per_key = {key: folded[key] for key in METADATA_TAG_KEYS if key in folded} + if per_key: + yield per_key, None + for tag in ENVELOPE_TAG_KEYS: + raw = folded.get(tag) if not isinstance(raw, str): continue try: - data = json.loads(raw) + envelope = json.loads(raw) except ValueError: continue - if not isinstance(data, dict): + if isinstance(envelope, dict): + yield envelope, raw + + +def select_video_carrier(tags): + """Choose the one container carrier of prompt metadata. + + Returns ``(metadata, record_formats, carrier_text)`` or ``None`` when no + carrier holds a known key. ``metadata`` maps known keys to their text, or + to ``None`` for an explicit null marker that the inspector must refuse; + ``record_formats`` is ``original-text``, ``reserialized-json`` (an envelope + object) or ``invalid`` (a workflow that is neither text nor an object), and + ``carrier_text`` is the exact envelope for reserialized values (``None`` + for per-key tags). The first carrier holding a role key wins; otherwise the + first holding ``workflow``. Carriers are never merged. + Raises :class:`DuplicateVideoTagError` for differing fold-duplicate tags. + """ + if not isinstance(tags, dict): + return None + candidates = list(_carriers(_folded_tags(tags))) + selected = next( + (item for item in candidates if any(key in item[0] for key in ROLE_TAG_KEYS)), + None, + ) + if selected is None: + selected = next((item for item in candidates if "workflow" in item[0]), None) + if selected is None: + return None + fields, carrier_text = selected + metadata, formats = {}, {} + for key in METADATA_TAG_KEYS: + if key not in fields: continue - found = {} - for key in _METADATA_KEYS: - value = data.get(key) - if value is None: - continue - found[key] = value if isinstance(value, str) else json.dumps(value) - if found: - return found - return {} + text, record_format = _record(fields[key], key) + if record_format == "invalid": + formats[key] = record_format + continue + metadata[key] = text + if record_format is not None: + formats[key] = record_format + return metadata, formats, carrier_text + + +def metadata_from_tags(tags) -> dict: + """Normalise container tags into PNG-text-chunk shaped metadata for scans. + + Returns ``{}`` when no carrier holds a known key, and when fold-duplicate + tags make the file's evidence untrustworthy. + """ + try: + selected = select_video_carrier(tags) + except DuplicateVideoTagError as e: + logger.debug(f"Video tags skipped: {e}") + return {} + if selected is None: + return {} + metadata, _, _ = selected + return {key: value for key, value in metadata.items() if isinstance(value, str)} def read_video_metadata(path, timeout=FFPROBE_TIMEOUT) -> dict: