Add video metadata and visible watermark removal

src/remove_ai_watermarks/__init__.py

@@ -6,6 +6,9 @@ High-level API (lazy, so ``import remove_ai_watermarks`` stays cheap)::
     raiw.remove_visible("in.png", "out.png")            # clean a file (provenance auto)
     result, removed = raiw.remove_visible(bgr_array)    # array -> array
     raiw.visible_provenance("in.png")                   # -> frozenset of confirmed vendors
+    raiw.inspect_video_metadata("in.mp4")               # -> VideoMetadataReport
+    raiw.remove_video_metadata("in.mp4", "out.mp4")     # verified metadata strip
+    raiw.remove_video_visible("in.mp4", "out.mp4")      # stable Sora or Veo mark removal
 
 For a provenance verdict use the ``identify`` submodule::
 
@@ -27,10 +30,18 @@ _warnings.filterwarnings("ignore", message=r".*ImageProcessorFast.*")
 
 __version__ = "0.20.2"
 
-__all__ = ["__version__", "remove_visible", "visible_provenance"]
+__all__ = [
+    "__version__",
+    "inspect_video_metadata",
+    "remove_video_metadata",
+    "remove_video_visible",
+    "remove_visible",
+    "visible_provenance",
+]
 
 if TYPE_CHECKING:
     from remove_ai_watermarks.api import remove_visible, visible_provenance
+    from remove_ai_watermarks.video import inspect_video_metadata, remove_video_metadata, remove_video_visible
 
 
 def __getattr__(name: str) -> object:
@@ -40,4 +51,8 @@ def __getattr__(name: str) -> object:
         from remove_ai_watermarks import api
 
         return getattr(api, name)
+    if name in ("inspect_video_metadata", "remove_video_metadata", "remove_video_visible"):
+        from remove_ai_watermarks import video
+
+        return getattr(video, name)
     raise AttributeError(f"module {__name__!r} has no attribute {name!r}")

src/remove_ai_watermarks/cli.py

@@ -4,6 +4,7 @@ Provides commands for:
   - Visible watermark removal (Gemini sparkle) - works offline, fast
   - Invisible watermark removal (SynthID etc.) - requires GPU/diffusion models
   - AI metadata stripping - lightweight, no ML deps needed
+  - Experimental video metadata and visible-wordmark removal
 """
 
 from __future__ import annotations
@@ -586,7 +587,7 @@ def _should_skip_invisible_scrub(force: bool, image_path: Path) -> bool:
 @click.option("-v", "--verbose", is_flag=True, help="Enable verbose logging.")
 @click.pass_context
 def main(ctx: click.Context, verbose: bool) -> None:
-    """Remove visible and invisible AI watermarks from images."""
+    """Remove visible and invisible AI watermarks from images, plus provenance metadata from video."""
     from dotenv import load_dotenv
 
     load_dotenv()  # Load .env (e.g. HF_TOKEN)
@@ -1016,6 +1017,23 @@ def cmd_invisible(
 
 
 # ── Metadata operations ──
+def _print_metadata_report(source: Path, has_ai: bool, metadata: dict[str, str]) -> None:
+    """Render one metadata inspection result for the generic and video commands."""
+    if not has_ai:
+        console.print(f"  No AI metadata found in {source.name}")
+        return
+
+    console.print(f"  Warning: AI metadata detected in {source.name}:")
+    if synthid := metadata.get("synthid_watermark"):
+        console.print(f"  Warning: SynthID watermark (inferred from C2PA metadata) {synthid}")
+    table = Table(show_header=True, header_style="bold")
+    table.add_column("Key", style="cyan")
+    table.add_column("Value")
+    for key, value in metadata.items():
+        table.add_row(key, str(value)[:80])
+    console.print(table)
+
+
 @main.command("metadata")
 @click.argument("source", type=click.Path(exists=True, dir_okay=False, path_type=Path))
 @click.option("--check", is_flag=True, help="Check for AI metadata (don't modify).")
@@ -1050,19 +1068,8 @@ def cmd_metadata(
 
     if check or (not remove):
         has_ai = has_ai_metadata(source)
-        if has_ai:
-            console.print(f"  Warning: AI metadata detected in {source.name}:")
-            meta = get_ai_metadata(source)
-            if synthid := meta.get("synthid_watermark"):
-                console.print(f"  Warning: SynthID watermark (inferred from C2PA metadata) {synthid}")
-            table = Table(show_header=True, header_style="bold")
-            table.add_column("Key", style="cyan")
-            table.add_column("Value")
-            for k, v in meta.items():
-                table.add_row(k, str(v)[:80])
-            console.print(table)
-        else:
-            console.print(f"  No AI metadata found in {source.name}")
+        metadata = get_ai_metadata(source) if has_ai else {}
+        _print_metadata_report(source, has_ai, metadata)
 
         if not remove:
             return
@@ -1082,6 +1089,114 @@ def cmd_metadata(
     console.print(f"  AI metadata stripped -> {out}")
 
 
+# ── Experimental video pipeline ──
+@main.group("video")
+def cmd_video() -> None:
+    """Process AI watermarks in video files."""
+
+
+@cmd_video.command("metadata")
+@click.argument("source", type=click.Path(exists=True, dir_okay=False, path_type=Path))
+@click.option("--check", is_flag=True, help="Check for AI metadata (don't modify).")
+@click.option("--remove", is_flag=True, help="Remove AI metadata.")
+@click.option(
+    "-o",
+    "--output",
+    type=click.Path(path_type=Path),
+    default=None,
+    help="Output path (default: <source>_clean with the same container).",
+)
+@click.option("--keep-standard/--remove-all", default=True, help="Keep standard metadata.")
+def cmd_video_metadata(
+    source: Path,
+    check: bool,
+    remove: bool,
+    output: Path | None,
+    keep_standard: bool,
+) -> None:
+    """Check or remove AI metadata without transcoding video streams."""
+    from remove_ai_watermarks.video import inspect_video_metadata, remove_video_metadata
+
+    _banner()
+    try:
+        report = inspect_video_metadata(source)
+    except (OSError, ValueError) as e:
+        raise click.ClickException(str(e)) from e
+
+    if check or not remove:
+        _print_metadata_report(source, report.has_ai_metadata, report.markers)
+
+        if not remove:
+            return
+
+    try:
+        result = remove_video_metadata(source, output, keep_standard=keep_standard)
+    except (OSError, RuntimeError, ValueError) as e:
+        raise click.ClickException(str(e)) from e
+
+    if result.remaining:
+        console.print(f"  FAILED: {len(result.remaining)} AI metadata marker(s) survived in {result.output}")
+        console.print(f"    still present: {', '.join(sorted(result.remaining))}")
+        raise SystemExit(1)
+    console.print(f"  AI metadata stripped -> {result.output}")
+
+
+@cmd_video.command("visible")
+@click.argument("source", type=click.Path(exists=True, dir_okay=False, path_type=Path))
+@click.option(
+    "-o",
+    "--output",
+    type=click.Path(path_type=Path),
+    default=None,
+    help="Output path (default: <source>_clean with the same container).",
+)
+@click.option(
+    "--mark",
+    type=click.Choice(["sora", "veo"]),
+    default="sora",
+    help="Visible AI mark to remove.",
+)
+@click.option(
+    "--backend",
+    type=click.Choice(["auto", "cv2", "migan", "lama"]),
+    default="cv2",
+    help="Per-frame fill backend. cv2 is the fast default; learned backends improve difficult backgrounds.",
+)
+@click.option("--strip-metadata/--keep-metadata", default=True, help="Strip AI metadata from the transcoded output.")
+def cmd_video_visible(
+    source: Path,
+    output: Path | None,
+    mark: str,
+    backend: str,
+    strip_metadata: bool,
+) -> None:
+    """Remove a temporally stable visible AI wordmark from video."""
+    from remove_ai_watermarks.video import remove_video_visible
+
+    _banner()
+    console.print(f"  Scanning {source.name} for a temporally stable {mark} mark...")
+    try:
+        result = remove_video_visible(
+            source,
+            output,
+            mark=mark,
+            backend=backend,
+            strip_metadata=strip_metadata,
+        )
+    except (OSError, RuntimeError, ValueError) as e:
+        raise click.ClickException(str(e)) from e
+
+    if result.output is None:
+        console.print(f"  No stable {mark} watermark detected; no output written")
+        raise SystemExit(EXIT_NO_VISIBLE_MARK)
+    if result.remaining_metadata:
+        console.print(f"  FAILED: {len(result.remaining_metadata)} AI metadata marker(s) survived in {result.output}")
+        raise SystemExit(1)
+    console.print(
+        f"  Removed {mark} watermark from {result.removed_frames}/{result.total_frames} frames -> {result.output}"
+    )
+
+
 # ── Provenance identification ──
 @main.command("identify")
 @click.argument("source", type=click.Path(exists=True, dir_okay=False, path_type=Path))

src/remove_ai_watermarks/metadata.py

@@ -143,7 +143,7 @@ AIGC_MARKERS: tuple[bytes, ...] = (
 # the same object as a PNG ``tEXt`` chunk keyed ``AIGC`` (raw JSON, not XMP), so
 # a JSON object carrying at least one of these is accepted as a valid TC260
 # label even when the namespaced XMP element is absent.
-_TC260_FIELDS: frozenset[str] = frozenset(
+TC260_AIGC_FIELDS: frozenset[str] = frozenset(
     {
         # Producer-side schema (Doubao and most China-served generators).
         "Label",
@@ -359,10 +359,12 @@ def has_ai_metadata(image_path: Path) -> bool:
 def aigc_label(image_path: Path) -> dict[str, str] | None:
     """Parse a China TC260 AI-labeling block, if present.
 
-    Three serializations are recognized:
+    Supported serializations are:
 
     - a PNG ``tEXt``/``iTXt`` chunk keyed ``AIGC`` carrying the raw JSON object
       (as written by Doubao / ByteDance), read via PIL;
+    - a native MP4/MOV ``AIGC`` key in ``moov.udta.meta.keys`` whose matching
+      ``ilst`` item carries the raw JSON object;
     - an XMP ``<TC260:AIGC>{...}</TC260:AIGC>`` block (HTML-entity encoded text),
       found by a container-agnostic raw-byte scan (PNG/JPEG/WebP alike); and
     - a raw-JSON ``{"AIGC":{...}}`` block with no namespace, as embedded in JPEG
@@ -375,7 +377,7 @@ def aigc_label(image_path: Path) -> dict[str, str] | None:
     Returns the decoded JSON (e.g. ``{"Label": "1", "ContentProducer": ...}``)
     or None. The generic forms (the PNG-chunk key ``AIGC``, the bare
     ``{"AIGC":...}`` object, and the bare ``AIGC{...}`` blob) are accepted only
-    if they carry at least one known TC260 field (``_TC260_FIELDS``); the
+    if they carry at least one known TC260 field (``TC260_AIGC_FIELDS``); the
     namespaced XMP element is unambiguous, so any JSON object is accepted.
     """
     import html
@@ -390,7 +392,7 @@ def aigc_label(image_path: Path) -> dict[str, str] | None:
         if not isinstance(parsed, dict):
             return None
         fields = {str(k): str(v) for k, v in cast("dict[object, object]", parsed).items()}
-        if require_tc260_field and not (_TC260_FIELDS & fields.keys()):
+        if require_tc260_field and not (TC260_AIGC_FIELDS & fields.keys()):
             return None
         return fields
 
@@ -407,6 +409,25 @@ def aigc_label(image_path: Path) -> dict[str, str] | None:
     if isinstance(value, str) and (result := _parse(value, require_tc260_field=True)):
         return result
 
+    # Native MP4/MOV TC260 metadata (TC260-PG-20257A): the ``AIGC`` key lives
+    # in ``moov.udta.meta.keys`` and points to a raw JSON value in ``ilst``.
+    # Read it through the bounded box walker so a tail ``moov`` after a large
+    # ``mdat`` is found without loading or scanning the media payload.
+    from remove_ai_watermarks.noai.isobmff import tc260_aigc_payloads
+
+    for payload in tc260_aigc_payloads(image_path):
+        if result := _parse(payload.decode("utf-8", "replace"), require_tc260_field=True):
+            return result
+
+    # Native MKV/WebM TC260 metadata: ``Segment.Tags.Tag.SimpleTag`` carries
+    # ``TagName=AIGC`` and the raw JSON in ``TagString``. The EBML walker seeks
+    # over clusters and reads only bounded metadata values.
+    from remove_ai_watermarks.noai.ebml import tc260_aigc_payloads as ebml_tc260_aigc_payloads
+
+    for payload in ebml_tc260_aigc_payloads(image_path):
+        if result := _parse(payload.decode("utf-8", "replace"), require_tc260_field=True):
+            return result
+
     # XMP TC260:AIGC, namespaced (unambiguous) in either serialization RDF allows:
     # an element  <TC260:AIGC>{...}</TC260:AIGC>  or an attribute  TC260:AIGC="{...}"
     # (the attribute form is what PicWish writes). Both are HTML-entity encoded.
@@ -750,7 +771,7 @@ def _is_aigc_exif_value(raw: object) -> bool:
     Mirrors ``aigc_label``'s EXIF path: the ``{"AIGC":{...}}`` wrapper embedded in
     ``UserComment`` / ``ImageDescription`` by China-served generators (Doubao's
     producer schema AND Tencent Cloud's service-provider schema, both keyed under
-    ``_TC260_FIELDS``). Gated on both the ``AIGC`` marker and a TC260 field so a
+    ``TC260_AIGC_FIELDS``). Gated on both the ``AIGC`` marker and a TC260 field so a
     coincidental token cannot false-drop a genuine caption/comment. Accepts a ``str``
     too (a PNG ``tEXt``/``iTXt`` value), not only EXIF bytes.
     """
@@ -761,7 +782,7 @@ def _is_aigc_exif_value(raw: object) -> bool:
     if b"AIGC" not in raw:
         return False
     text = bytes(raw).decode("latin-1", "ignore")
-    return any(field in text for field in _TC260_FIELDS)
+    return any(field in text for field in TC260_AIGC_FIELDS)
 
 
 def _ai_exif_targets(loaded: dict[str, Any]) -> list[tuple[str, int, bytes, str]]:
@@ -1154,6 +1175,7 @@ def remove_ai_metadata(
     from remove_ai_watermarks.noai.isobmff import (
         blank_ai_exif_tokens,
         blank_ai_xmp_packets,
+        blank_tc260_aigc_tags,
         is_isobmff,
         strip_c2pa_boxes,
     )
@@ -1164,17 +1186,20 @@ def remove_ai_metadata(
         data = source_path.read_bytes()
         # Top-level uuid/jumb boxes (C2PA + AI-label XMP), then the meta-box items
         # the top-level stripper can't reach (HEIF/AVIF store them in mdat/idat):
-        # AI-label XMP packets and AI-generator tokens in an Exif item -- both
-        # blanked in place (same length) so box sizes and iloc offsets stay valid
-        # and the coded image is untouched.
+        # Native TC260 tags, AI-label XMP packets, and AI-generator tokens in an
+        # Exif item are blanked in place (same length) so box sizes and iloc /
+        # media offsets stay valid and the coded content is untouched.
         cleaned, stripped = strip_c2pa_boxes(data)
+        cleaned, tc260_blanked = blank_tc260_aigc_tags(cleaned)
         cleaned, blanked = blank_ai_xmp_packets(cleaned)
         cleaned, exif_blanked = blank_ai_exif_tokens(cleaned)
         output_path.parent.mkdir(parents=True, exist_ok=True)
         output_path.write_bytes(cleaned)
         logger.info(
-            "Stripped %d AI-provenance box(es), blanked %d meta-box XMP packet(s) + %d EXIF token(s) → %s",
+            "Stripped %d AI-provenance box(es), blanked %d native TC260 tag(s) + "
+            "%d meta-box XMP packet(s) + %d EXIF token(s) → %s",
             stripped,
+            tc260_blanked,
             blanked,
             exif_blanked,
             output_path,

src/remove_ai_watermarks/noai/ebml.py

@@ -0,0 +1,167 @@
+"""Bounded Matroska/WebM metadata reader for native TC260 AIGC labels.
+
+TC260-PG-20257A stores the label as a Matroska ``SimpleTag`` whose
+``TagName`` is ``AIGC`` and whose ``TagString`` is the normative JSON object.
+The walker seeks over unrelated elements such as clusters instead of reading
+their payloads.
+"""
+
+from __future__ import annotations
+
+import json
+from typing import TYPE_CHECKING, BinaryIO, cast
+
+if TYPE_CHECKING:
+    from collections.abc import Iterator
+    from pathlib import Path
+
+from remove_ai_watermarks.metadata import TC260_AIGC_FIELDS
+
+_EBML_MAGIC = b"\x1aE\xdf\xa3"
+_SEGMENT_ID = 0x18538067
+_TAGS_ID = 0x1254C367
+_TAG_ID = 0x7373
+_SIMPLE_TAG_ID = 0x67C8
+_TAG_NAME_ID = 0x45A3
+_TAG_STRING_ID = 0x4487
+_MAX_TAG_NAME_BYTES = 256
+_MAX_TC260_VALUE_BYTES = 1024 * 1024
+
+
+def _vint_length(first: int, *, maximum: int) -> int | None:
+    """Return an EBML variable-integer length from its first byte."""
+    mask = 0x80
+    for length in range(1, maximum + 1):
+        if first & mask:
+            return length
+        mask >>= 1
+    return None
+
+
+def _read_element_header(
+    stream: BinaryIO,
+    pos: int,
+    limit: int,
+) -> tuple[int, int, int] | None:
+    """Return ``(element_id, payload_start, element_end)`` inside ``limit``."""
+    if pos < 0 or pos >= limit:
+        return None
+    stream.seek(pos)
+    first_raw = stream.read(1)
+    if not first_raw:
+        return None
+    id_length = _vint_length(first_raw[0], maximum=4)
+    if id_length is None or pos + id_length >= limit:
+        return None
+    element_id_raw = first_raw + stream.read(id_length - 1)
+    if len(element_id_raw) != id_length:
+        return None
+    element_id = int.from_bytes(element_id_raw, "big")
+
+    size_first_raw = stream.read(1)
+    if not size_first_raw:
+        return None
+    size_length = _vint_length(size_first_raw[0], maximum=8)
+    if size_length is None:
+        return None
+    size_rest = stream.read(size_length - 1)
+    if len(size_rest) != size_length - 1:
+        return None
+    marker = 1 << (8 - size_length)
+    size_value = int.from_bytes(bytes([size_first_raw[0] & (marker - 1)]) + size_rest, "big")
+    payload_start = pos + id_length + size_length
+    if payload_start > limit:
+        return None
+    unknown_size = size_value == (1 << (7 * size_length)) - 1
+    element_end = limit if unknown_size else payload_start + size_value
+    if element_end > limit:
+        return None
+    return element_id, payload_start, element_end
+
+
+def _iter_elements(
+    stream: BinaryIO,
+    start: int,
+    end: int,
+) -> Iterator[tuple[int, int, int]]:
+    """Yield valid direct children from one bounded EBML region."""
+    pos = start
+    while pos < end:
+        header = _read_element_header(stream, pos, end)
+        if header is None:
+            return
+        element_id, payload_start, element_end = header
+        yield element_id, payload_start, element_end
+        if element_end <= pos:
+            return
+        pos = element_end
+
+
+def _read_bounded(
+    stream: BinaryIO,
+    start: int,
+    end: int,
+    maximum: int,
+) -> bytes | None:
+    size = end - start
+    if size < 0 or size > maximum:
+        return None
+    stream.seek(start)
+    value = stream.read(size)
+    return value if len(value) == size else None
+
+
+def _is_tc260_aigc_json(value: bytes) -> bool:
+    try:
+        parsed = json.loads(value.decode("utf-8"))
+    except (UnicodeDecodeError, ValueError):
+        return False
+    if not isinstance(parsed, dict):
+        return False
+    fields = cast("dict[object, object]", parsed)
+    return bool(TC260_AIGC_FIELDS & {str(key) for key in fields})
+
+
+def _simple_tag_payloads(
+    stream: BinaryIO,
+    start: int,
+    end: int,
+) -> tuple[bytes, ...]:
+    name: bytes | None = None
+    values: list[bytes] = []
+    for element_id, payload_start, element_end in _iter_elements(stream, start, end):
+        if element_id == _TAG_NAME_ID:
+            name = _read_bounded(stream, payload_start, element_end, _MAX_TAG_NAME_BYTES)
+        elif element_id == _TAG_STRING_ID:
+            value = _read_bounded(stream, payload_start, element_end, _MAX_TC260_VALUE_BYTES)
+            if value is not None:
+                values.append(value)
+    if name != b"AIGC":
+        return ()
+    return tuple(value for value in values if _is_tc260_aigc_json(value))
+
+
+def tc260_aigc_payloads(path: str | Path) -> tuple[bytes, ...]:
+    """Read validated TC260 values from Matroska/WebM ``SimpleTag`` entries."""
+    found: list[bytes] = []
+    try:
+        with open(path, "rb") as stream:
+            if stream.read(4) != _EBML_MAGIC:
+                return ()
+            stream.seek(0, 2)
+            file_size = stream.tell()
+            for element_id, payload_start, element_end in _iter_elements(stream, 0, file_size):
+                if element_id != _SEGMENT_ID:
+                    continue
+                for child_id, child_start, child_end in _iter_elements(stream, payload_start, element_end):
+                    if child_id != _TAGS_ID:
+                        continue
+                    for tag_id, tag_start, tag_end in _iter_elements(stream, child_start, child_end):
+                        if tag_id != _TAG_ID:
+                            continue
+                        for simple_id, simple_start, simple_end in _iter_elements(stream, tag_start, tag_end):
+                            if simple_id == _SIMPLE_TAG_ID:
+                                found.extend(_simple_tag_payloads(stream, simple_start, simple_end))
+    except OSError:
+        return ()
+    return tuple(found)

src/remove_ai_watermarks/noai/isobmff.py

@@ -1,4 +1,4 @@
-"""Minimal ISOBMFF box walker for stripping C2PA from AVIF / HEIF / MP4 / JPEG-XL.
+"""Minimal ISOBMFF box walker for AI provenance in AVIF / HEIF / MP4 / JPEG-XL.
 
 The ISO Base Media File Format wraps content in nested ``[size:4][type:4][...]``
 boxes. C2PA stores its manifest in a top-level ``uuid`` box keyed by the
@@ -8,6 +8,11 @@ carry C2PA, and emit the rest verbatim. The codestream (``mdat`` for ISOBMFF,
 ``jxlc`` / ``jxlp`` for JPEG-XL) is untouched, so pixel data is preserved
 bit-for-bit.
 
+TC260-PG-20257A video metadata is nested instead:
+``moov.udta.meta.keys/ilst``. Its detector seeks through those boxes without
+reading media payloads, and its stripper blanks the validated key/value in
+place so fast-start media offsets remain valid.
+
 This file intentionally avoids dependencies on format-specific libraries
 (pillow-heif, pillow-jxl, pymp4) so it works on systems where they aren't
 installed.
@@ -17,10 +22,12 @@ Reference: ISO/IEC 14496-12 (ISOBMFF) and C2PA 2.1 spec §11.
 
 from __future__ import annotations
 
+import io
+import json
 import logging
 import re
 import struct
-from typing import TYPE_CHECKING, Any
+from typing import TYPE_CHECKING, Any, BinaryIO, cast
 
 if TYPE_CHECKING:
     from collections.abc import Iterator
@@ -31,6 +38,7 @@ from remove_ai_watermarks.metadata import (
     C2PA_UUID,
     IPTC_AI_FIELD_MARKERS,
     IPTC_AI_MARKERS,
+    TC260_AIGC_FIELDS,
 )
 
 logger = logging.getLogger(__name__)
@@ -53,6 +61,13 @@ _AI_LABEL_MARKERS: tuple[bytes, ...] = AIGC_MARKERS + IPTC_AI_MARKERS + IPTC_AI_
 # blanked in place (see ``blank_ai_xmp_packets``).
 _XMP_PACKET_RE = re.compile(rb"<\?xpacket begin=.*?<\?xpacket end=[^>]*?\?>", re.DOTALL)
 
+# TC260-PG-20257A stores an MP4/MOV label as an ``AIGC`` key in
+# ``moov.udta.meta.keys`` and its JSON value in the corresponding
+# ``moov.udta.meta.ilst`` item. The value is intentionally bounded before it is
+# read: the normative object is tiny, and a corrupt size must not allocate an
+# arbitrary amount of memory during an inspection.
+_MAX_TC260_VALUE_BYTES = 1024 * 1024
+
 
 def _iter_top_level_boxes(data: bytes) -> Iterator[tuple[int, int, bytes, int]]:
     """Yield ``(start, end, type, payload_offset)`` for each top-level box.
@@ -84,6 +99,199 @@ def _iter_top_level_boxes(data: bytes) -> Iterator[tuple[int, int, bytes, int]]:
         pos += size
 
 
+def _read_box_header(
+    stream: BinaryIO,
+    pos: int,
+    limit: int,
+) -> tuple[int, bytes, int] | None:
+    """Return ``(end, type, payload_offset)`` for one box inside ``limit``."""
+    if pos < 0 or pos + 8 > limit:
+        return None
+    stream.seek(pos)
+    header = stream.read(8)
+    if len(header) != 8:
+        return None
+    size32 = struct.unpack(">I", header[:4])[0]
+    box_type = header[4:8]
+    payload_off = pos + 8
+    if size32 == 1:
+        extended = stream.read(8)
+        if len(extended) != 8:
+            return None
+        size = struct.unpack(">Q", extended)[0]
+        payload_off = pos + 16
+    elif size32 == 0:
+        size = limit - pos
+    else:
+        size = size32
+    end = pos + size
+    if size < payload_off - pos or end > limit:
+        return None
+    return end, box_type, payload_off
+
+
+def _iter_file_boxes(
+    stream: BinaryIO,
+    start: int,
+    end: int,
+) -> Iterator[tuple[int, int, bytes, int]]:
+    """Yield valid boxes from one bounded container region."""
+    pos = start
+    while pos + 8 <= end:
+        header = _read_box_header(stream, pos, end)
+        if header is None:
+            return
+        box_end, box_type, payload_off = header
+        yield pos, box_end, box_type, payload_off
+        pos = box_end
+
+
+def _tc260_key_indices(
+    stream: BinaryIO,
+    payload_off: int,
+    box_end: int,
+) -> dict[int, tuple[int, int]]:
+    """Map every exact ``AIGC`` key index to its byte span."""
+    if payload_off + 8 > box_end:
+        return {}
+    stream.seek(payload_off)
+    prefix = stream.read(8)
+    if len(prefix) != 8:
+        return {}
+    entry_count = struct.unpack(">I", prefix[4:8])[0]
+    pos = payload_off + 8
+    found: dict[int, tuple[int, int]] = {}
+    for index in range(1, entry_count + 1):
+        if pos + 8 > box_end:
+            return {}
+        stream.seek(pos)
+        header = stream.read(8)
+        if len(header) != 8:
+            return {}
+        entry_size = struct.unpack(">I", header[:4])[0]
+        entry_end = pos + entry_size
+        if entry_size < 8 or entry_end > box_end:
+            return {}
+        name_start = pos + 8
+        if entry_end - name_start == 4:
+            stream.seek(name_start)
+            if stream.read(4) == b"AIGC":
+                found[index] = (name_start, entry_end)
+        pos = entry_end
+    return found
+
+
+def _is_tc260_aigc_json(value: bytes) -> bool:
+    """Require a JSON object carrying at least one normative TC260 field."""
+    try:
+        parsed = json.loads(value.decode("utf-8"))
+    except (UnicodeDecodeError, ValueError):
+        return False
+    if not isinstance(parsed, dict):
+        return False
+    fields = cast("dict[object, object]", parsed)
+    return bool(TC260_AIGC_FIELDS & {str(key) for key in fields})
+
+
+def _tc260_aigc_regions(
+    stream: BinaryIO,
+    file_size: int,
+) -> list[tuple[int, int, int, int, bytes]]:
+    """Locate validated native TC260 entries without reading media payloads.
+
+    Each tuple is ``(key_start, key_end, value_start, value_end, value)``.
+    """
+    regions: list[tuple[int, int, int, int, bytes]] = []
+    for _moov_start, moov_end, moov_type, moov_payload in _iter_file_boxes(stream, 0, file_size):
+        if moov_type != b"moov":
+            continue
+        for _udta_start, udta_end, udta_type, udta_payload in _iter_file_boxes(
+            stream,
+            moov_payload,
+            moov_end,
+        ):
+            if udta_type != b"udta":
+                continue
+            for _meta_start, meta_end, meta_type, meta_payload in _iter_file_boxes(
+                stream,
+                udta_payload,
+                udta_end,
+            ):
+                if meta_type != b"meta" or meta_payload + 4 > meta_end:
+                    continue
+                keys: dict[int, tuple[int, int]] = {}
+                ilst_boxes: list[tuple[int, int]] = []
+                for _child_start, child_end, child_type, child_payload in _iter_file_boxes(
+                    stream,
+                    meta_payload + 4,
+                    meta_end,
+                ):
+                    if child_type == b"keys":
+                        keys.update(_tc260_key_indices(stream, child_payload, child_end))
+                    elif child_type == b"ilst":
+                        ilst_boxes.append((child_payload, child_end))
+                if not keys:
+                    continue
+                for ilst_payload, ilst_end in ilst_boxes:
+                    for _item_start, item_end, item_type, item_payload in _iter_file_boxes(
+                        stream,
+                        ilst_payload,
+                        ilst_end,
+                    ):
+                        index = int.from_bytes(item_type, "big")
+                        key_span = keys.get(index)
+                        if key_span is None:
+                            continue
+                        for _data_start, data_end, data_type, data_payload in _iter_file_boxes(
+                            stream,
+                            item_payload,
+                            item_end,
+                        ):
+                            value_start = data_payload + 8
+                            value_size = data_end - value_start
+                            if data_type != b"data" or value_size < 0 or value_size > _MAX_TC260_VALUE_BYTES:
+                                continue
+                            stream.seek(value_start)
+                            value = stream.read(value_size)
+                            if len(value) == value_size and _is_tc260_aigc_json(value):
+                                regions.append((*key_span, value_start, data_end, value))
+    return regions
+
+
+def tc260_aigc_payloads(path: str | Path) -> tuple[bytes, ...]:
+    """Read native TC260 ``AIGC`` JSON values from an MP4/MOV container."""
+    try:
+        with open(path, "rb") as stream:
+            if not is_isobmff(stream.read(8)):
+                return ()
+            stream.seek(0, 2)
+            file_size = stream.tell()
+            return tuple(region[4] for region in _tc260_aigc_regions(stream, file_size))
+    except OSError:
+        return ()
+
+
+def blank_tc260_aigc_tags(data: bytes) -> tuple[bytes, int]:
+    """Blank native TC260 values in place while preserving every box offset.
+
+    Removing a nested ``ilst`` item would shift ``mdat`` in a fast-start MP4 and
+    invalidate its chunk offsets. Replacing the four-byte key with ``free`` and
+    the JSON value with spaces keeps every box size and media offset unchanged.
+    """
+    if not is_isobmff(data):
+        return data, 0
+    regions = _tc260_aigc_regions(io.BytesIO(data), len(data))
+    if not regions:
+        return data, 0
+    out = bytearray(data)
+    key_spans: set[tuple[int, int]] = set()
+    for key_start, key_end, value_start, value_end, _value in regions:
+        key_spans.add((key_start, key_end))
+        out[key_start:key_end] = b"free"
+        out[value_start:value_end] = b" " * (value_end - value_start)
+    return bytes(out), len(key_spans)
+
+
 def is_isobmff(data: bytes) -> bool:
     """Cheap sniff: ISOBMFF files start with an ``ftyp`` box."""
     return len(data) >= 8 and data[4:8] == b"ftyp"

src/remove_ai_watermarks/video.py

@@ -0,0 +1,212 @@
+"""High-level video processing API.
+
+Supported experimental stages are container-level AI metadata inspection and
+removal plus temporally stabilized visible Sora and Veo removal. The pixel path
+reuses the image package's shared fill backends.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from pathlib import Path
+
+VIDEO_EXTENSIONS: frozenset[str] = frozenset({".mp4", ".mov", ".m4v", ".webm", ".mkv"})
+_ISOBMFF_VIDEO_EXTENSIONS: frozenset[str] = frozenset({".mp4", ".mov", ".m4v"})
+_EBML_VIDEO_EXTENSIONS: frozenset[str] = frozenset({".webm", ".mkv"})
+_EBML_MAGIC = b"\x1aE\xdf\xa3"
+
+
+@dataclass(frozen=True)
+class VideoMetadataReport:
+    """AI metadata found in one supported video container."""
+
+    source: Path
+    has_ai_metadata: bool
+    markers: dict[str, str]
+
+
+@dataclass(frozen=True)
+class VideoMetadataResult:
+    """Result of a verified video metadata-removal operation."""
+
+    source: Path
+    output: Path
+    detected: dict[str, str]
+    remaining: dict[str, str]
+
+
+@dataclass(frozen=True)
+class VideoVisibleResult:
+    """Result of visible AI-watermark removal from a video."""
+
+    source: Path
+    output: Path | None
+    mark: str
+    total_frames: int
+    detected_frames: int
+    removed_frames: int
+    remaining_metadata: dict[str, str]
+
+
+def _video_source(source: str | Path) -> Path:
+    path = Path(source)
+    if not path.exists():
+        raise FileNotFoundError(f"Video does not exist: {path}")
+    if not path.is_file():
+        raise ValueError(f"Video source must be a file: {path}")
+    if path.suffix.lower() not in VIDEO_EXTENSIONS:
+        supported = ", ".join(sorted(VIDEO_EXTENSIONS))
+        raise ValueError(f"Unsupported video format {path.suffix or '<none>'}; expected one of: {supported}")
+    with path.open("rb") as stream:
+        head = stream.read(12)
+    suffix = path.suffix.lower()
+    matches_container = (suffix in _ISOBMFF_VIDEO_EXTENSIONS and len(head) >= 8 and head[4:8] == b"ftyp") or (
+        suffix in _EBML_VIDEO_EXTENSIONS and head.startswith(_EBML_MAGIC)
+    )
+    if not matches_container:
+        raise ValueError(f"Video content does not match its {suffix} extension: {path}")
+    return path
+
+
+def _video_output(
+    source: Path,
+    output: str | Path | None,
+    *,
+    operation: str = "metadata removal",
+) -> Path:
+    path = Path(output) if output is not None else source.with_stem(source.stem + "_clean")
+    if path.suffix.lower() != source.suffix.lower():
+        raise ValueError(
+            f"Video output container must match the source ({source.suffix}); "
+            f"{operation} does not change containers to {path.suffix or '<none>'}"
+        )
+    if path.resolve() == source.resolve():
+        raise ValueError(f"Video {operation} requires a distinct output path")
+    return path
+
+
+def inspect_video_metadata(source: str | Path) -> VideoMetadataReport:
+    """Inspect supported AI-provenance metadata in a video container."""
+    from remove_ai_watermarks.metadata import get_ai_metadata, has_ai_metadata
+
+    source_path = _video_source(source)
+    return VideoMetadataReport(
+        source=source_path,
+        has_ai_metadata=has_ai_metadata(source_path),
+        markers=get_ai_metadata(source_path),
+    )
+
+
+def remove_video_metadata(
+    source: str | Path,
+    output: str | Path | None = None,
+    *,
+    keep_standard: bool = True,
+) -> VideoMetadataResult:
+    """Remove AI metadata without transcoding video or audio streams.
+
+    The default output is ``<source_stem>_clean<source_suffix>``. A separate
+    output is required so an experimental video operation never overwrites the
+    original file.
+    """
+    from remove_ai_watermarks.metadata import get_ai_metadata, strip_and_verify
+
+    source_path = _video_source(source)
+    output_path = _video_output(source_path, output)
+    detected = get_ai_metadata(source_path)
+    written, remaining = strip_and_verify(source_path, output_path, keep_standard=keep_standard)
+    return VideoMetadataResult(
+        source=source_path,
+        output=written,
+        detected=detected,
+        remaining=remaining,
+    )
+
+
+def remove_video_visible(
+    source: str | Path,
+    output: str | Path | None = None,
+    *,
+    mark: str = "sora",
+    backend: str = "cv2",
+    strip_metadata: bool = True,
+) -> VideoVisibleResult:
+    """Remove a supported visible AI wordmark from a video.
+
+    Supported marks are ``sora`` and ``veo``. The full sequence is scanned before
+    pixels change, and only recurring candidates are accepted. Audio is copied
+    without re-encoding; video is transcoded because the pixels change. When no
+    stable mark is found, no output is written and ``output`` in the result is
+    ``None``.
+    """
+    from remove_ai_watermarks.metadata import get_ai_metadata
+    from remove_ai_watermarks.video_visible import (
+        encode_clean_video,
+        has_sora_provenance,
+        has_veo_provenance,
+        scan_sora_video,
+        scan_veo_video,
+        stabilize_sora_localizations,
+        stabilize_veo_localizations,
+    )
+    from remove_ai_watermarks.watermark_registry import resolve_backend
+
+    if mark not in {"sora", "veo"}:
+        raise ValueError("Unsupported visible video mark; expected 'sora' or 'veo'")
+    if backend not in {"auto", "cv2", "migan", "lama"}:
+        raise ValueError("Unsupported fill backend; expected auto, cv2, migan, or lama")
+
+    source_path = _video_source(source)
+    output_path = _video_output(source_path, output, operation="visible watermark removal")
+    markers = get_ai_metadata(source_path)
+    if mark == "sora":
+        scan = scan_sora_video(source_path)
+        regions = stabilize_sora_localizations(
+            scan.detections,
+            provenance=has_sora_provenance(markers),
+        )
+        padding_fraction = 0.28
+        mask_style = "box"
+    else:
+        scan = scan_veo_video(source_path)
+        regions = stabilize_veo_localizations(
+            scan.detections,
+            provenance=has_veo_provenance(markers),
+        )
+        padding_fraction = 0.18
+        mask_style = "veo"
+    detected_frames = sum(region is not None for region in regions)
+    if detected_frames == 0:
+        return VideoVisibleResult(
+            source=source_path,
+            output=None,
+            mark=mark,
+            total_frames=len(scan.detections),
+            detected_frames=0,
+            removed_frames=0,
+            remaining_metadata=markers if strip_metadata else {},
+        )
+
+    # Validate optional model availability before ffmpeg creates or overwrites
+    # the requested output.
+    resolve_backend(backend)  # type: ignore[arg-type]
+    removed_frames = encode_clean_video(
+        source_path,
+        output_path,
+        scan,
+        regions,
+        backend=backend,  # type: ignore[arg-type]
+        strip_metadata=strip_metadata,
+        padding_fraction=padding_fraction,
+        mask_style=mask_style,
+    )
+    remaining_metadata = get_ai_metadata(output_path) if strip_metadata else {}
+    return VideoVisibleResult(
+        source=source_path,
+        output=output_path,
+        mark=mark,
+        total_frames=len(scan.detections),
+        detected_frames=detected_frames,
+        removed_frames=removed_frames,
+        remaining_metadata=remaining_metadata,
+    )

src/remove_ai_watermarks/video_visible.py

@@ -0,0 +1,706 @@
+"""Visible AI-watermark localization and removal for video.
+
+Supported marks use fully synthetic silhouettes made from geometric primitives
+and Pillow's bundled font. Sora detection searches the full frame because the
+wordmark moves. Veo detection covers both the current four-point diamond and the
+legacy ``Veo`` text in the bottom-right corner. A single frame is never enough
+to authorize removal: the temporal arbiter requires the candidate to recur at
+the same location across adjacent frames. This keeps isolated lookalikes in
+clean videos from becoming removal masks.
+
+Video pixels are decoded with OpenCV and encoded with the system ``ffmpeg``.
+Audio is stream-copied from the source. The video stream must be transcoded
+because visible-mark removal changes pixels.
+"""
+
+# cv2/numpy boundary: these packages do not expose usable types for many array
+# operations. Public signatures remain annotated while unknown third-party types
+# are relaxed only in this module.
+# pyright: reportUnknownMemberType=false, reportUnknownArgumentType=false, reportUnknownVariableType=false, reportUnknownParameterType=false, reportMissingTypeArgument=false, reportMissingTypeStubs=false, reportMissingImports=false, reportArgumentType=false, reportAssignmentType=false, reportReturnType=false, reportCallIssue=false, reportIndexIssue=false, reportOperatorIssue=false, reportOptionalMemberAccess=false, reportOptionalCall=false, reportOptionalSubscript=false, reportOptionalOperand=false, reportAttributeAccessIssue=false, reportPrivateImportUsage=false, reportPrivateUsage=false, reportInvalidTypeForm=false
+
+from __future__ import annotations
+
+import logging
+import shutil
+import subprocess
+from dataclasses import dataclass
+from functools import lru_cache
+from itertools import pairwise
+from typing import TYPE_CHECKING, Any, Literal
+
+import cv2
+import numpy as np
+from PIL import Image, ImageDraw, ImageFont
+
+if TYPE_CHECKING:
+    from pathlib import Path
+
+    from numpy.typing import NDArray
+
+    from remove_ai_watermarks.watermark_registry import Backend
+
+
+log = logging.getLogger(__name__)
+
+Region = tuple[int, int, int, int]
+
+_NORMALIZED_SHORT_SIDE = 480
+_SORA_TEMPLATE_SIZE = (180, 64)
+_SORA_RELATIVE_HEIGHTS = (0.065, 0.075, 0.085, 0.095, 0.105)
+_SORA_PROVENANCE_WEAK_CONFIDENCE = 0.58
+_SORA_STRICT_WEAK_CONFIDENCE = 0.60
+_SORA_STRONG_CONFIDENCE = 0.65
+_VEO_PROVENANCE_WEAK_CONFIDENCE = 0.45
+_VEO_STRICT_WEAK_CONFIDENCE = 0.50
+_VEO_STRONG_CONFIDENCE = 0.55
+_MIN_STABLE_FRAMES = 5
+_MIN_VEO_STABLE_FRAMES = 12
+_MAX_STABLE_GAP = 2
+_STABLE_IOU = 0.55
+_VEO_REFERENCE_SHORT_SIDE = 720
+_VEO_DIAMOND_PROFILES = (
+    (56, 92, 92),
+    (48, 72, 72),
+    (44, 29, 40),
+)
+
+
+@dataclass(frozen=True)
+class FrameLocalization:
+    """Best untrusted visible-mark candidate found in one decoded frame."""
+
+    frame_index: int
+    confidence: float
+    region: Region | None
+
+
+@dataclass(frozen=True)
+class VideoScan:
+    """Decoded video geometry plus one localization candidate per frame."""
+
+    width: int
+    height: int
+    fps: float
+    detections: tuple[FrameLocalization, ...]
+
+
+def _scalable_default_font(size: int) -> ImageFont.ImageFont | ImageFont.FreeTypeFont:
+    """Load Pillow's bundled scalable font, with a Pillow 10.0 fallback."""
+    try:
+        return ImageFont.load_default(size=size)
+    except TypeError:
+        # ``size=`` was added after Pillow 10.0, which is still within the
+        # package's supported dependency range. Resize that bundled bitmap font
+        # before compositing it into the synthetic template.
+        return ImageFont.load_default()
+
+
+@lru_cache(maxsize=1)
+def _sora_templates() -> tuple[NDArray[Any], NDArray[Any]]:
+    """Return synthetic full-wordmark and mascot-only silhouettes.
+
+    No source frame or provider logo asset contributes pixels to these templates.
+    The cloud-like mascot is assembled from primitive shapes and the word is
+    rendered with Pillow's bundled font.
+    """
+    width, height = _SORA_TEMPLATE_SIZE
+    canvas = Image.new("L", (width, height), 0)
+    draw = ImageDraw.Draw(canvas)
+    draw.rounded_rectangle((4, 8, 58, 57), radius=22, fill=255)
+    draw.ellipse((0, 18, 26, 50), fill=255)
+    draw.ellipse((38, 16, 64, 51), fill=255)
+    draw.ellipse((16, 18, 29, 43), fill=0)
+    draw.ellipse((35, 18, 48, 43), fill=0)
+
+    font = _scalable_default_font(50)
+    if isinstance(font, ImageFont.FreeTypeFont):
+        draw.text((67, 0), "Sora", font=font, fill=255, stroke_width=1, stroke_fill=255)
+    else:
+        text_box = font.getbbox("Sora")
+        text = Image.new("L", (max(1, text_box[2]), max(1, text_box[3])), 0)
+        ImageDraw.Draw(text).text((0, 0), "Sora", font=font, fill=255)
+        text = text.resize((102, 50), Image.Resampling.NEAREST)
+        canvas.paste(text, (67, 0), text)
+
+    full = np.asarray(canvas, dtype=np.uint8)
+    return full, full[:, :64]
+
+
+@lru_cache(maxsize=1)
+def _veo_templates() -> tuple[NDArray[Any], NDArray[Any]]:
+    """Return synthetic current-diamond and legacy-text Veo silhouettes."""
+    size = 256
+    diamond_canvas = Image.new("L", (size, size), 0)
+    diamond_points = (
+        (0.50, 0.02),
+        (0.60, 0.39),
+        (0.98, 0.50),
+        (0.60, 0.61),
+        (0.50, 0.98),
+        (0.40, 0.61),
+        (0.02, 0.50),
+        (0.40, 0.39),
+    )
+    ImageDraw.Draw(diamond_canvas).polygon(
+        [(round(x * size), round(y * size)) for x, y in diamond_points],
+        fill=255,
+    )
+
+    text_canvas = Image.new("L", (140, 60), 0)
+    text_draw = ImageDraw.Draw(text_canvas)
+    text_draw.text((2, 0), "Veo", font=_scalable_default_font(48), fill=255)
+    text = np.asarray(text_canvas, dtype=np.uint8)
+    ys, xs = np.where(text > 0)
+    text = text[ys.min() : ys.max() + 1, xs.min() : xs.max() + 1]
+    return np.asarray(diamond_canvas, dtype=np.uint8), text
+
+
+def _top_hat(gray: NDArray[Any]) -> NDArray[Any]:
+    kernel = np.ones((7, 7), dtype=np.uint8)
+    return cv2.morphologyEx(gray, cv2.MORPH_TOPHAT, kernel)
+
+
+def _normalized_gray(image_bgr: NDArray[Any]) -> tuple[NDArray[Any], float]:
+    gray = image_bgr if image_bgr.ndim == 2 else cv2.cvtColor(image_bgr, cv2.COLOR_BGR2GRAY)
+    height, width = gray.shape[:2]
+    short_side = min(height, width)
+    if short_side <= 0:
+        return gray, 1.0
+    scale = min(1.0, _NORMALIZED_SHORT_SIDE / short_side)
+    if scale < 1.0:
+        gray = cv2.resize(
+            gray,
+            (max(1, round(width * scale)), max(1, round(height * scale))),
+            interpolation=cv2.INTER_AREA,
+        )
+    return gray, scale
+
+
+def _expanded_region(
+    location: tuple[int, int],
+    template_width: int,
+    template_height: int,
+    *,
+    icon_only: bool,
+    scale: float,
+    frame_width: int,
+    frame_height: int,
+) -> Region:
+    x = round(location[0] / scale)
+    y = round(location[1] / scale)
+    height = max(1, round(template_height / scale))
+    width = max(1, round(template_width / scale))
+    if icon_only:
+        width = round(height * _SORA_TEMPLATE_SIZE[0] / _SORA_TEMPLATE_SIZE[1])
+    width = min(width, frame_width - x)
+    height = min(height, frame_height - y)
+    return x, y, max(1, width), max(1, height)
+
+
+def detect_sora_frame(image_bgr: NDArray[Any], *, frame_index: int = 0) -> FrameLocalization:
+    """Locate the strongest synthetic Sora-wordmark match in one frame.
+
+    The returned candidate is intentionally untrusted. Call
+    :func:`stabilize_sora_localizations` across the full sequence before building
+    any removal mask.
+    """
+    if image_bgr.size == 0:
+        return FrameLocalization(frame_index, 0.0, None)
+
+    frame_height, frame_width = image_bgr.shape[:2]
+    gray, scale = _normalized_gray(image_bgr)
+    normalized_height, normalized_width = gray.shape[:2]
+    feature = _top_hat(gray)
+    best_confidence = 0.0
+    best_region: Region | None = None
+
+    for template_index, base_template in enumerate(_sora_templates()):
+        icon_only = template_index == 1
+        for relative_height in _SORA_RELATIVE_HEIGHTS:
+            template_height = max(16, round(min(normalized_height, normalized_width) * relative_height))
+            template_width = max(1, round(base_template.shape[1] * template_height / base_template.shape[0]))
+            if template_height >= normalized_height or template_width >= normalized_width:
+                continue
+            template = cv2.resize(
+                base_template,
+                (template_width, template_height),
+                interpolation=cv2.INTER_AREA,
+            )
+            scores = cv2.matchTemplate(feature, _top_hat(template), cv2.TM_CCOEFF_NORMED)
+            _, confidence, _, location = cv2.minMaxLoc(scores)
+            if confidence <= best_confidence:
+                continue
+            best_confidence = float(confidence)
+            best_region = _expanded_region(
+                location,
+                template_width,
+                template_height,
+                icon_only=icon_only,
+                scale=scale,
+                frame_width=frame_width,
+                frame_height=frame_height,
+            )
+
+    return FrameLocalization(frame_index, best_confidence, best_region)
+
+
+def _match_template(
+    gray: NDArray[Any],
+    template: NDArray[Any],
+    *,
+    region: Region,
+    kernel_size: int,
+) -> tuple[float, Region | None]:
+    """Match one synthetic silhouette inside a bounded frame region."""
+    x, y, width, height = region
+    roi = gray[y : y + height, x : x + width]
+    template_height, template_width = template.shape[:2]
+    if roi.size == 0 or template_height >= roi.shape[0] or template_width >= roi.shape[1]:
+        return 0.0, None
+    kernel = np.ones((kernel_size, kernel_size), dtype=np.uint8)
+    feature = cv2.morphologyEx(roi, cv2.MORPH_TOPHAT, kernel)
+    template_feature = cv2.morphologyEx(template, cv2.MORPH_TOPHAT, kernel)
+    scores = cv2.matchTemplate(feature, template_feature, cv2.TM_CCOEFF_NORMED)
+    _, confidence, _, location = cv2.minMaxLoc(scores)
+    return float(confidence), (
+        x + location[0],
+        y + location[1],
+        template_width,
+        template_height,
+    )
+
+
+def detect_veo_frame(image_bgr: NDArray[Any], *, frame_index: int = 0) -> FrameLocalization:
+    """Locate the strongest current-diamond or legacy-text Veo candidate."""
+    if image_bgr.size == 0:
+        return FrameLocalization(frame_index, 0.0, None)
+
+    frame_height, frame_width = image_bgr.shape[:2]
+    gray = image_bgr if image_bgr.ndim == 2 else cv2.cvtColor(image_bgr, cv2.COLOR_BGR2GRAY)
+    short_scale = min(frame_height, frame_width) / _VEO_REFERENCE_SHORT_SIDE
+    diamond_base, text_base = _veo_templates()
+    best_confidence = 0.0
+    best_region: Region | None = None
+
+    diamond_sizes: set[int] = set()
+    for base_size, right_margin, bottom_margin in _VEO_DIAMOND_PROFILES:
+        diamond_size = max(16, round(base_size * short_scale))
+        diamond_sizes.add(diamond_size)
+        template = cv2.resize(
+            diamond_base,
+            (diamond_size, diamond_size),
+            interpolation=cv2.INTER_AREA,
+        )
+        expected_x = round(frame_width - (right_margin + base_size) * short_scale)
+        expected_y = round(frame_height - (bottom_margin + base_size) * short_scale)
+        search_padding = max(6, round(diamond_size * 0.25))
+        search_x = max(0, expected_x - search_padding)
+        search_y = max(0, expected_y - search_padding)
+        search_width = min(frame_width - search_x, diamond_size + search_padding * 2)
+        search_height = min(frame_height - search_y, diamond_size + search_padding * 2)
+        confidence, candidate = _match_template(
+            gray,
+            template,
+            region=(search_x, search_y, search_width, search_height),
+            kernel_size=max(3, round(7 * short_scale) | 1),
+        )
+        if confidence > best_confidence:
+            best_confidence = confidence
+            best_region = candidate
+
+    # Provider layouts have moved before. A bounded corner search is a safety
+    # net for a relocated diamond, but it is admitted only at a much stronger
+    # per-frame score than the known-profile path. Without this gate, recurring
+    # bright scene details in clean API exports can become stable false matches.
+    corner_x = round(frame_width * 0.65)
+    corner_y = round(frame_height * 0.65)
+    corner_region = (corner_x, corner_y, frame_width - corner_x, frame_height - corner_y)
+    for diamond_size in diamond_sizes:
+        template = cv2.resize(
+            diamond_base,
+            (diamond_size, diamond_size),
+            interpolation=cv2.INTER_AREA,
+        )
+        confidence, candidate = _match_template(
+            gray,
+            template,
+            region=corner_region,
+            kernel_size=max(3, round(7 * short_scale) | 1),
+        )
+        if confidence >= 0.70 and confidence > best_confidence:
+            best_confidence = confidence
+            best_region = candidate
+
+    text_region_width = min(frame_width, max(32, round(180 * short_scale)))
+    text_region_height = min(frame_height, max(24, round(120 * short_scale)))
+    text_region = (
+        frame_width - text_region_width,
+        frame_height - text_region_height,
+        text_region_width,
+        text_region_height,
+    )
+    text_heights = sorted({max(5, round(height * short_scale)) for height in range(8, 22)})
+    for text_height in text_heights:
+        text_width = max(1, round(text_base.shape[1] * text_height / text_base.shape[0]))
+        template = cv2.resize(
+            text_base,
+            (text_width, text_height),
+            interpolation=cv2.INTER_AREA,
+        )
+        confidence, candidate = _match_template(
+            gray,
+            template,
+            region=text_region,
+            kernel_size=max(3, round(3 * short_scale) | 1),
+        )
+        if confidence > best_confidence:
+            best_confidence = confidence
+            best_region = candidate
+
+    return FrameLocalization(frame_index, best_confidence, best_region)
+
+
+def _region_iou(left: Region, right: Region) -> float:
+    lx, ly, lw, lh = left
+    rx, ry, rw, rh = right
+    x0 = max(lx, rx)
+    y0 = max(ly, ry)
+    x1 = min(lx + lw, rx + rw)
+    y1 = min(ly + lh, ry + rh)
+    intersection = max(0, x1 - x0) * max(0, y1 - y0)
+    union = lw * lh + rw * rh - intersection
+    return intersection / union if union > 0 else 0.0
+
+
+def stabilize_sora_localizations(
+    detections: tuple[FrameLocalization, ...] | list[FrameLocalization],
+    *,
+    provenance: bool,
+) -> list[Region | None]:
+    """Accept only spatially recurring Sora candidates and bridge short dropouts.
+
+    Metadata never creates a detection. It only allows a stable visual run whose
+    scores remain below the strict confidence floor, which covers low-contrast
+    Sora marks while clean metadata-bearing exports stay untouched.
+    """
+    weak_floor = _SORA_PROVENANCE_WEAK_CONFIDENCE if provenance else _SORA_STRICT_WEAK_CONFIDENCE
+    return _stabilize_localizations(
+        detections,
+        provenance=provenance,
+        weak_floor=weak_floor,
+        strong_floor=_SORA_STRONG_CONFIDENCE,
+        transition_floor=0.45,
+        min_stable_frames=_MIN_STABLE_FRAMES,
+        cover_after_confirmation=False,
+    )
+
+
+def stabilize_veo_localizations(
+    detections: tuple[FrameLocalization, ...] | list[FrameLocalization],
+    *,
+    provenance: bool,
+) -> list[Region | None]:
+    """Accept temporally recurring current or legacy Veo candidates."""
+    weak_floor = _VEO_PROVENANCE_WEAK_CONFIDENCE if provenance else _VEO_STRICT_WEAK_CONFIDENCE
+    return _stabilize_localizations(
+        detections,
+        provenance=provenance,
+        weak_floor=weak_floor,
+        strong_floor=_VEO_STRONG_CONFIDENCE,
+        transition_floor=0.35,
+        min_stable_frames=_MIN_VEO_STABLE_FRAMES,
+        cover_after_confirmation=True,
+    )
+
+
+def _stabilize_localizations(
+    detections: tuple[FrameLocalization, ...] | list[FrameLocalization],
+    *,
+    provenance: bool,
+    weak_floor: float,
+    strong_floor: float,
+    transition_floor: float,
+    min_stable_frames: int,
+    cover_after_confirmation: bool,
+) -> list[Region | None]:
+    """Apply the shared recurrence policy to provider-specific candidates."""
+    accepted: list[Region | None] = [None] * len(detections)
+    runs: list[list[int]] = []
+    current: list[int] = []
+
+    for position, detection in enumerate(detections):
+        if detection.region is None or detection.confidence < weak_floor:
+            continue
+        if current:
+            previous = detections[current[-1]]
+            frame_gap = detection.frame_index - previous.frame_index
+            if (
+                previous.region is None
+                or frame_gap > _MAX_STABLE_GAP + 1
+                or _region_iou(previous.region, detection.region) < _STABLE_IOU
+            ):
+                runs.append(current)
+                current = []
+        current.append(position)
+    if current:
+        runs.append(current)
+
+    for run in runs:
+        strong = max(detections[position].confidence for position in run) >= strong_floor
+        if len(run) < min_stable_frames or (not provenance and not strong):
+            continue
+        for position in run:
+            accepted[position] = detections[position].region
+        for left_position, right_position in pairwise(run):
+            if right_position - left_position <= 1:
+                continue
+            left = detections[left_position]
+            right = detections[right_position]
+            if left.region is None or right.region is None or _region_iou(left.region, right.region) < _STABLE_IOU:
+                continue
+            for missing_position in range(left_position + 1, right_position):
+                distance_left = missing_position - left_position
+                distance_right = right_position - missing_position
+                accepted[missing_position] = left.region if distance_left <= distance_right else right.region
+
+    # Provider provenance plus a confirmed run establishes a continuously
+    # watermarked app export rather than a clean API export that merely shares
+    # the generator name. Veo may also cover the sequence without provenance
+    # after its longer, strong fixed-position run. Cover low-contrast transition
+    # frames with the nearest confirmed position.
+    confirmed_positions = [position for position, region in enumerate(accepted) if region is not None]
+    if (provenance or cover_after_confirmation) and confirmed_positions:
+        confirmed_regions = [accepted[position] for position in confirmed_positions]
+        carry_position = confirmed_positions[0]
+        for position, region in enumerate(accepted):
+            if region is not None:
+                carry_position = position
+                continue
+            raw = detections[position]
+            if (
+                raw.region is not None
+                and raw.confidence >= transition_floor
+                and any(
+                    confirmed_region is not None and _region_iou(raw.region, confirmed_region) >= _STABLE_IOU
+                    for confirmed_region in confirmed_regions
+                )
+            ):
+                accepted[position] = raw.region
+                continue
+            accepted[position] = accepted[carry_position]
+
+    return accepted
+
+
+def _scan_video(
+    source: Path,
+    detector: Any,
+) -> VideoScan:
+    """Decode a video once and collect one untrusted candidate per frame."""
+    capture = cv2.VideoCapture(str(source))
+    if not capture.isOpened():
+        raise RuntimeError(f"OpenCV could not decode video: {source}")
+    width = int(capture.get(cv2.CAP_PROP_FRAME_WIDTH))
+    height = int(capture.get(cv2.CAP_PROP_FRAME_HEIGHT))
+    fps = float(capture.get(cv2.CAP_PROP_FPS))
+    if width <= 0 or height <= 0 or fps <= 0:
+        capture.release()
+        raise RuntimeError(f"Video has invalid stream geometry or frame rate: {source}")
+
+    detections: list[FrameLocalization] = []
+    frame_index = 0
+    while True:
+        ok, frame = capture.read()
+        if not ok:
+            break
+        if frame.shape[:2] != (height, width):
+            capture.release()
+            raise RuntimeError(f"Video changes frame dimensions at frame {frame_index}: {source}")
+        detections.append(detector(frame, frame_index=frame_index))
+        frame_index += 1
+    capture.release()
+    if not detections:
+        raise RuntimeError(f"Video contains no decodable frames: {source}")
+    return VideoScan(width, height, fps, tuple(detections))
+
+
+def scan_sora_video(source: Path) -> VideoScan:
+    """Decode a video once and collect one untrusted Sora candidate per frame."""
+    return _scan_video(source, detect_sora_frame)
+
+
+def scan_veo_video(source: Path) -> VideoScan:
+    """Decode a video once and collect one untrusted Veo candidate per frame."""
+    return _scan_video(source, detect_veo_frame)
+
+
+def _ffmpeg_video_args(suffix: str) -> list[str]:
+    if suffix == ".webm":
+        return ["-c:v", "libvpx-vp9", "-crf", "18", "-b:v", "0"]
+    return ["-c:v", "libx264", "-preset", "medium", "-crf", "14"]
+
+
+def _mask_for_region(
+    frame_bgr: NDArray[Any],
+    region: Region,
+    *,
+    padding_fraction: float,
+    mask_style: Literal["box", "veo"],
+) -> NDArray[Any]:
+    height, width = frame_bgr.shape[:2]
+    x, y, region_width, region_height = region
+    mask = np.zeros((height, width), dtype=np.uint8)
+    if mask_style == "veo" and 0.80 <= region_width / region_height <= 1.25:
+        diamond_base, _ = _veo_templates()
+        diamond = cv2.resize(
+            diamond_base,
+            (region_width, region_height),
+            interpolation=cv2.INTER_AREA,
+        )
+        diamond = np.where(diamond >= 24, 255, 0).astype(np.uint8)
+        dilation = max(2, round(region_height * 0.08))
+        kernel_size = dilation * 2 + 1
+        diamond = cv2.dilate(
+            diamond,
+            cv2.getStructuringElement(cv2.MORPH_ELLIPSE, (kernel_size, kernel_size)),
+        )
+        x1 = min(width, x + region_width)
+        y1 = min(height, y + region_height)
+        mask[y:y1, x:x1] = diamond[: y1 - y, : x1 - x]
+        return mask
+
+    # A glyph-shaped mask leaves a thin translucent rim outside the approximate
+    # synthetic silhouette. Classical inpainting then pulls that white rim back
+    # into the hole, recreating the mascot as a bright blob. The measured clean
+    # floor on real Sora frames is a full box with roughly 0.28 mark-heights of
+    # context on every side.
+    padding = max(4, round(region_height * padding_fraction))
+    x0 = max(0, x - padding)
+    y0 = max(0, y - padding)
+    x1 = min(width, x + region_width + padding)
+    y1 = min(height, y + region_height + padding)
+    mask[y0:y1, x0:x1] = 255
+    return mask
+
+
+def encode_clean_video(
+    source: Path,
+    output: Path,
+    scan: VideoScan,
+    regions: list[Region | None],
+    *,
+    backend: Backend,
+    strip_metadata: bool,
+    padding_fraction: float = 0.28,
+    mask_style: Literal["box", "veo"] = "box",
+) -> int:
+    """Decode again, fill accepted regions, and encode video while copying audio."""
+    from remove_ai_watermarks.watermark_registry import fill, resolve_backend
+
+    ffmpeg = shutil.which("ffmpeg")
+    if ffmpeg is None:
+        raise RuntimeError("Visible video removal requires ffmpeg on PATH")
+    if len(regions) != len(scan.detections):
+        raise ValueError("Temporal localization count does not match the scanned frame count")
+
+    output.parent.mkdir(parents=True, exist_ok=True)
+    command = [
+        ffmpeg,
+        "-y",
+        "-loglevel",
+        "error",
+        "-f",
+        "rawvideo",
+        "-pix_fmt",
+        "bgr24",
+        "-s:v",
+        f"{scan.width}x{scan.height}",
+        "-r",
+        f"{scan.fps:.12g}",
+        "-i",
+        "pipe:0",
+        "-i",
+        str(source),
+        "-map",
+        "0:v:0",
+        "-map",
+        "1:a?",
+        *_ffmpeg_video_args(output.suffix.lower()),
+        "-c:a",
+        "copy",
+        "-map_metadata",
+        "-1" if strip_metadata else "1",
+        "-map_chapters",
+        "-1" if strip_metadata else "1",
+        "-shortest",
+    ]
+    if output.suffix.lower() in {".mp4", ".mov", ".m4v"}:
+        command.extend(["-movflags", "+faststart"])
+    command.append(str(output))
+    log.info("Encoding visible-watermark removal with ffmpeg: command=%s", command)
+
+    process = subprocess.Popen(  # noqa: S603
+        command,
+        stdin=subprocess.PIPE,
+        stdout=subprocess.DEVNULL,
+        stderr=subprocess.PIPE,
+    )
+    if process.stdin is None or process.stderr is None:
+        process.kill()
+        raise RuntimeError("Could not open ffmpeg pipes")
+
+    capture = cv2.VideoCapture(str(source))
+    if not capture.isOpened():
+        process.kill()
+        raise RuntimeError(f"OpenCV could not reopen video for removal: {source}")
+
+    removed_frames = 0
+    resolved_backend: Literal["cv2", "migan", "lama"] = resolve_backend(backend)
+    try:
+        for frame_index, region in enumerate(regions):
+            ok, frame = capture.read()
+            if not ok:
+                raise RuntimeError(f"Video ended while reading frame {frame_index}: {source}")
+            if region is not None:
+                frame = fill(
+                    frame,
+                    _mask_for_region(
+                        frame,
+                        region,
+                        padding_fraction=padding_fraction,
+                        mask_style=mask_style,
+                    ),
+                    backend=resolved_backend,
+                )
+                removed_frames += 1
+            process.stdin.write(frame.tobytes())
+        process.stdin.close()
+        stderr = process.stderr.read().decode("utf-8", errors="replace")
+        return_code = process.wait()
+    except Exception:
+        process.kill()
+        process.wait()
+        raise
+    finally:
+        capture.release()
+
+    log.info("ffmpeg visible-watermark encode finished: status=%s stderr=%s", return_code, stderr)
+    if return_code != 0:
+        raise RuntimeError(f"ffmpeg failed to encode {output}: {stderr.strip()[:500]}")
+    return removed_frames
+
+
+def has_sora_provenance(markers: dict[str, str]) -> bool:
+    """Whether container provenance specifically names the Sora generator."""
+    return "sora" in markers.get("claim_generator", "").lower()
+
+
+def has_veo_provenance(markers: dict[str, str]) -> bool:
+    """Whether container provenance names Google as the AI-video generator."""
+    identity = " ".join(
+        (
+            markers.get("claim_generator", ""),
+            markers.get("issuer", ""),
+        )
+    ).lower()
+    return "google" in identity and "trainedalgorithmicmedia" in markers.get("source_type", "").lower()