From f330dd9c94f38f8f28a0aa4bb5e092287536d0a7 Mon Sep 17 00:00:00 2001 From: Victor Kuznetsov Date: Wed, 29 Jul 2026 17:28:11 -0700 Subject: [PATCH] Add video metadata and visible watermark removal --- CLAUDE.md | 6 +- README.md | 54 +- docs/cli.md | 54 ++ docs/index.md | 2 +- docs/known-limitations.md | 34 ++ docs/module-internals.md | 68 +++ docs/python-api.md | 59 ++ docs/supported-signals.md | 14 +- docs/watermarking-landscape.md | 27 +- pyproject.toml | 2 +- src/remove_ai_watermarks/__init__.py | 17 +- src/remove_ai_watermarks/cli.py | 143 ++++- src/remove_ai_watermarks/metadata.py | 45 +- src/remove_ai_watermarks/noai/ebml.py | 167 +++++ src/remove_ai_watermarks/noai/isobmff.py | 212 ++++++- src/remove_ai_watermarks/video.py | 212 +++++++ src/remove_ai_watermarks/video_visible.py | 706 ++++++++++++++++++++++ tests/test_metadata.py | 2 +- tests/test_video.py | 701 +++++++++++++++++++++ 19 files changed, 2487 insertions(+), 38 deletions(-) create mode 100644 src/remove_ai_watermarks/noai/ebml.py create mode 100644 src/remove_ai_watermarks/video.py create mode 100644 src/remove_ai_watermarks/video_visible.py create mode 100644 tests/test_video.py diff --git a/CLAUDE.md b/CLAUDE.md index 4223a78..01b7bea 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -1,6 +1,6 @@ # Remove-AI-Watermarks -You are a **principal Python engineer** maintaining a CLI tool and library for removing visible and invisible AI watermarks from images. +You are a **principal Python engineer** maintaining a CLI tool and library for removing AI watermarks from images and provenance metadata from video. ## Scope and non-goals @@ -26,6 +26,9 @@ Per-command exit-code semantics (the no-signal / GPU-missing skip branches), tes - `uv run remove-ai-watermarks identify ` — provenance verdict (platform + watermark inventory + confidence); `--json` for machine output, `--no-visible` to skip both registered visible detectors and the optional open invisible-watermark decoder - `uv run remove-ai-watermarks metadata --check` — inspect AI metadata (C2PA, EXIF, PNG chunks) - `uv run remove-ai-watermarks metadata --remove -o ` — strip all AI metadata +- `uv run remove-ai-watermarks video metadata --check` — inspect AI metadata in MP4/MOV/M4V/WebM/MKV +- `uv run remove-ai-watermarks video metadata --remove -o ` — strip verified video metadata without transcoding streams; the experimental video path requires a separate same-container output and defaults to `_clean` +- `uv run remove-ai-watermarks video visible -o ` — remove a temporally recurring Sora mark, or pass `--mark veo` for the current Veo diamond and legacy `Veo` text. It scans the full sequence first, transcodes video through ffmpeg, copies audio, strips AI metadata by default, and writes no output when no stable mark is found. `cv2` is the fast default; `migan`/`lama` improve difficult backgrounds. - `uv run remove-ai-watermarks batch ` — process every supported image in a directory (output defaults to `_clean/`, set with `-o`). `--mode visible|invisible|metadata|all` (default `visible`); the invisible/all path reuses the full `invisible` knob set above, plus `--backend` and `--sensitivity` for the visible localize -> fill pass. Applies the same no-signal skip per image; see the module doc. **Exit code:** non-zero when any image errored OR (mirroring single `all`) a `--mode invisible`/`all` image carried an invisible signal but the GPU extra was absent, so its SynthID scrub was skipped — it emits a loud warning and copies the input through (invisible mode) so the output dir stays complete; a wrapping service can then detect the incomplete run instead of trusting a silent exit 0. ## Test and lint @@ -79,6 +82,7 @@ Compact map. The full per-module detail (design decisions, tuned thresholds, cal - `upscaler.py` — optional Real-ESRGAN pre-diffusion super-resolution for small inputs (extra `esrgan`, spandrel only). Manual opt-in; the default `--upscaler` stays `lanczos` and the engine always falls back to Lanczos on absence/error. ESRGAN can degrade faces and thin text. - `image_io.py` — centralizes Unicode-safe image IO, alpha preservation, content-based format sniffing, and HEIC/AVIF fallbacks. Callers must check `imwrite` success. No-op visible removal preserves original bytes when the output format is unchanged. - `api.py` — the high-level convenience API, re-exported lazily at the package top level via `__init__.__getattr__` (PEP 562, so `import remove_ai_watermarks` stays cheap): `remove_visible(source, output=None, *, sensitivity="auto", backend="auto", strip_metadata=True, write_noop=True) -> (result_bgr, [labels])` (source = path OR BGR ndarray; a PATH auto-reads metadata provenance and preserves alpha, an ARRAY does neither; `write_noop=True` writes a clean passthrough copy when nothing is removed, `False` leaves `output` untouched so a "no mark = produce nothing" caller like the CLI `visible` command does not clobber a pre-existing file there) and `visible_provenance(path) -> frozenset[str]` (the single metadata→vendor-keys mapper; `cli._visible_provenance` is a thin None-guarded wrapper over it). **`remove_visible` is the ONE path the CLI and library share** — `cli.cmd_visible`'s `--mark auto` branch delegates entirely to it (read → provenance → `remove_auto_marks` → write → `strip_metadata`), so there is no CLI-vs-library drift; `strip_metadata` defaults True to match `visible --strip-metadata`. This is where a library caller should start — NOT the engines directly (`GeminiEngine`/`TextMarkEngine` have no `remove_watermark` any more; removal is registry `remove_auto_marks`/`KnownMark.remove`; the old single-strongest `best_auto_mark` is gone — removal takes EVERY mark). `identify` is NOT top-level re-exported (it collides with the `identify` submodule); use `from remove_ai_watermarks.identify import identify`. +- `video.py` — the experimental high-level video API, also lazy at the package root: `inspect_video_metadata(source) -> VideoMetadataReport`, `remove_video_metadata(source, output=None, *, keep_standard=True) -> VideoMetadataResult`, and `remove_video_visible(source, output=None, *, mark="sora", backend="cv2", strip_metadata=True) -> VideoVisibleResult`. It validates the extension and container signature for MP4/MOV/M4V/WebM/MKV and requires a distinct same-container output. The metadata path never transcodes streams: native MP4/MOV TC260 is read from `moov.udta.meta.keys/ilst`, including a tail `moov` after a large `mdat`, and removal blanks the key/value in place; MKV/WebM TC260 is read by `noai/ebml.py` and stripped through ffmpeg stream copy. The visible path delegates to `video_visible.py`: fully synthetic Sora and Veo silhouettes propose frame boxes, provider-specific temporal recurrence authorizes them, the shared fill backends remove accepted masks in a second pass, and ffmpeg transcodes video while copying audio. Veo covers the current four-point diamond and legacy text; its diamond uses a shape mask rather than erasing the transparent corners of a full box. Metadata can relax a recurring low-contrast match but cannot create one. The inherited ISOBMFF metadata removal path still reads the complete container into memory, so a streaming box copier is required before large-video use. Other visible video labels and invisible video watermarks are not built yet. For the Doubao alpha-distillation history (why content-image reverse-alpha distillation fails by physics and controlled captures were required), see `docs/research-doubao-distillation.md`. diff --git a/README.md b/README.md index 99ffea4..cde4b17 100644 --- a/README.md +++ b/README.md @@ -1,11 +1,15 @@ # Remove AI Watermarks -Remove AI provenance marks from images you generated yourself: +Remove AI provenance marks from images and video you generated yourself: - known visible labels such as the Gemini sparkle and vendor text marks; - invisible pixel watermarks through diffusion regeneration; - C2PA, EXIF, XMP, IPTC, and related AI metadata. +Video support covers metadata inspection and removal plus experimental visible +Sora-wordmark removal. Invisible video-watermark removal remains a follow-up +stage. + > Try it online at [raiw.cc](https://raiw.cc) if you do not want to install Python > or run diffusion models locally. @@ -28,6 +32,8 @@ Remove AI provenance marks from images you generated yourself: | Remove known visible AI marks | `visible` | No | | Erase a region you select | `erase` | No | | Strip AI metadata | `metadata` | No | +| Strip AI metadata from video | `video metadata` | No | +| Remove a known Sora or Veo mark from video | `video visible` | No | | Regenerate an image to disrupt invisible watermarks | `invisible` | Recommended | | Run visible, invisible, and metadata removal | `all` | Recommended | | Process a directory | `batch` | Depends on mode | @@ -58,6 +64,35 @@ Strip metadata without running visible inpainting or diffusion: remove-ai-watermarks metadata image.png --remove -o clean.png ``` +Inspect or remove AI metadata from an MP4, MOV, M4V, WebM, or MKV file: + +```bash +remove-ai-watermarks video metadata input.mp4 --check +remove-ai-watermarks video metadata input.mp4 --remove -o clean.mp4 +``` + +The metadata command does not transcode video or audio streams. When `-o` is +omitted it writes `_clean` and preserves the original. MP4 and MOV +inspection includes the native TC260 `AIGC` tag in +`moov.udta.meta.keys/ilst`, including a `moov` placed after the media payload. +MKV and WebM inspection reads the normative +`Segment.Tags.Tag.SimpleTag` placement. + +Remove a moving Sora wordmark or a Veo corner mark: + +```bash +remove-ai-watermarks video visible input.mp4 -o clean.mp4 +remove-ai-watermarks video visible veo.mp4 --mark veo -o veo_clean.mp4 +``` + +This path scans the complete sequence before changing pixels. It accepts only a +mark that repeats at a stable position across adjacent frames, then reuses the +same OpenCV, MI-GAN, or LaMa fill backends as image removal. Audio is copied +without re-encoding; the video stream is transcoded because its pixels change. +Sora covers the moving Sora 2 mascot and wordmark. Veo covers both the current +four-point diamond and the legacy `Veo` text in the bottom-right corner. No +output is written when no stable mark is found. + For invisible watermark removal, install the diffusion dependencies: ```bash @@ -184,8 +219,9 @@ Visible removal follows three steps: 3. Fill only the masked region with OpenCV, MI-GAN, or LaMa. Metadata removal uses format aware stripping. JPEG metadata removal preserves -the encoded image scan instead of recompressing it. Other supported containers -use their corresponding metadata path. +the encoded image scan instead of recompressing it. Native MP4/MOV TC260 values +are blanked without changing box sizes or media offsets. Other supported +containers use their corresponding metadata path. Invisible removal is different. It regenerates the image through a diffusion pipeline to disrupt pixel and frequency domain watermarks. This changes the @@ -202,6 +238,11 @@ import remove_ai_watermarks as raiw result, removed = raiw.remove_visible("watermarked.png", "clean.png") print(removed) + +report = raiw.inspect_video_metadata("input.mp4") +cleaned = raiw.remove_video_metadata("input.mp4") +visible = raiw.remove_video_visible("sora.mp4", "sora_clean.mp4") +veo = raiw.remove_video_visible("veo.mp4", "veo_clean.mp4", mark="veo") ``` The high level API accepts a file path or a BGR NumPy array. For path inputs it @@ -226,6 +267,13 @@ invisible removal. and selected fill backend. - Invisible removal changes the whole image and may alter faces, text, or fine detail. +- Visible video removal recognizes the moving Sora 2 wordmark and the current + Veo diamond plus legacy `Veo` text. It does not recognize the older Sora Turbo + corner swirl. + The classical OpenCV backend can smear structured backgrounds; use MI-GAN or + LaMa when recovery quality matters. Video SynthID is not removed yet. + MP4/MOV/M4V metadata stripping currently reads the full container into + memory, so very large videos are not an intended metadata input yet. - `qwen-zimage` requires CUDA. The other diffusion profiles also support the devices listed by `remove-ai-watermarks invisible --help`. - Provider watermark systems can change. Validate important outputs with the diff --git a/docs/cli.md b/docs/cli.md index da76b8e..c8ba68a 100644 --- a/docs/cli.md +++ b/docs/cli.md @@ -125,6 +125,60 @@ The command also supports the audio and video containers listed in [supported signals](supported-signals.md). ffmpeg must be available for the non-ISOBMFF audio and video path. +## Strip AI metadata from video + +The experimental video namespace starts with metadata inspection and removal: + +```bash +remove-ai-watermarks video metadata input.mp4 --check +remove-ai-watermarks video metadata input.mp4 --remove -o clean.mp4 +``` + +Supported containers are MP4, MOV, M4V, WebM, and MKV. The operation delegates +to the same verified metadata scanner and stripper as the generic `metadata` +command, so detection and removal stay in parity. Video and audio streams are +not transcoded. For MP4 and MOV, this includes the native TC260 `AIGC` key and +JSON value stored in `moov.udta.meta.keys/ilst`. The inspector seeks past a +large `mdat` to find a tail `moov`; removal blanks the key and value in place so +box sizes and media offsets do not move. + +For MKV and WebM, the inspector reads the native TC260 +`Segment.Tags.Tag.SimpleTag` entry. Removal uses ffmpeg stream copying to +discard container tags and chapters without transcoding the streams. + +When `-o` is omitted, the command writes `_clean` with the same +extension. It never overwrites the source, and it rejects an output with a +different container extension. + +Visible video labels and invisible video watermarks are not handled by this +command. + +## Remove a visible Sora or Veo video mark + +```bash +remove-ai-watermarks video visible input.mp4 -o clean.mp4 +remove-ai-watermarks video visible veo.mp4 --mark veo -o veo_clean.mp4 +``` + +The experimental command supports the moving Sora mascot and wordmark plus two +Veo corner variants: the current four-point diamond and the legacy `Veo` text. +Sora searches the whole frame at multiple scales. Veo searches the +bottom-right corner using separate synthetic silhouettes for the two variants. +Both require a spatially recurring candidate across adjacent frames. Matching +provider provenance may relax the visual score, but metadata alone never +creates a detection. Clean API exports therefore remain untouched. + +The video stream is transcoded and the original audio stream is copied. +Supported input and output containers are MP4, MOV, M4V, WebM, and MKV; the +output extension must match the input. The default `cv2` backend is fast but can +smear structured backgrounds. Select `--backend migan` or `--backend lama` for +a learned fill, or `--backend auto` to choose the best installed backend. + +AI metadata is stripped from the encoded output by default. Use +`--keep-metadata` to retain mapped container metadata. When no temporally stable +mark is found, the command writes no output and exits with the no-visible-mark +status. + ## Remove invisible watermarks Install the diffusion dependencies first: diff --git a/docs/index.md b/docs/index.md index 9128ba7..f096e33 100644 --- a/docs/index.md +++ b/docs/index.md @@ -8,7 +8,7 @@ to run the tool. Use the maintainer references only when changing the code. | Page | Use it when | | --- | --- | | [Installation](installation.md) | You need the CLI, an optional model backend, or a development environment. | -| [CLI guide](cli.md) | You want a command for one image, a directory, or a specific watermark type. | +| [CLI guide](cli.md) | You want a command for an image, video metadata or visible marks, a directory, or a specific watermark type. | | [Python API](python-api.md) | You want to call the package from Python. | | [Supported signals](supported-signals.md) | You need to know which visible marks, metadata formats, and invisible signals are covered. | | [Known limitations](known-limitations.md) | You need the quality, device, format, or verification boundaries. | diff --git a/docs/known-limitations.md b/docs/known-limitations.md index 72c9e7d..35b4a93 100644 --- a/docs/known-limitations.md +++ b/docs/known-limitations.md @@ -174,6 +174,40 @@ WebM, Matroska, MP3, WAV, FLAC, OGG, Opus, and AAC container metadata is strippe through ffmpeg with stream copying. The operation fails if ffmpeg is absent or cannot parse the input. +### Visible video removal supports Sora and Veo and is still experimental + +The experimental `video metadata` command and high level video API inspect and +strip supported AI provenance metadata without transcoding streams. + +`video visible` and `remove_video_visible` additionally support the moving +Sora 2 mascot and wordmark, the current Veo four-point diamond, and the legacy +`Veo` text. Detection requires a recurring visual candidate across adjacent +frames. Provider provenance can recover low-contrast runs only after visual +evidence exists, so metadata alone does not erase a clean API export. +Historical Sora Turbo exports use a small OpenAI swirl in the corner rather +than the moving mascot-and-wordmark design; that earlier variant is not +detected by the `sora` video mark. Other provider video labels and proprietary +invisible video watermarks are not supported yet. + +Visible removal transcodes the video stream and copies audio. Its frame-local +fill is not a motion-aware video inpainting model. OpenCV can leave a visible +smear where the mark overlaps a hard edge or structured texture, and the smear +can vary over time. MI-GAN and LaMa improve individual frames but do not +guarantee temporal coherence. The Veo diamond uses a shape mask to limit damage +outside the symbol, but OpenCV may still soften texture inside it. The current +encoder also emits a constant-frame-rate output at the decoded stream rate, so +variable-frame-rate preservation is not yet guaranteed. + +Native TC260 metadata in MP4/MOV is supported at its normative +`moov.udta.meta.keys/ilst` placement, including non-faststart files whose +`moov` follows a large media payload. MKV/WebM is supported at the normative +`Segment.Tags.Tag.SimpleTag` placement and uses ffmpeg for stream-copy removal. +The corresponding FLV and AVI native tags are not parsed yet. + +The current ISOBMFF stripper reads an MP4, MOV, or M4V container into memory +before rewriting its metadata boxes. A streaming box copier is required before +the experimental command is appropriate for very large video files. + ### Metadata transformation is fail safe `remove_ai_metadata` may copy an undecodable file through unchanged instead of diff --git a/docs/module-internals.md b/docs/module-internals.md index 8432062..cb4dba6 100644 --- a/docs/module-internals.md +++ b/docs/module-internals.md @@ -86,6 +86,70 @@ Regression coverage: - [`test_api.py`](../tests/test_api.py) - [`test_image_io.py`](../tests/test_image_io.py) +[`video.py`](../src/remove_ai_watermarks/video.py) provides the experimental +video entry point: + +- `inspect_video_metadata` +- `remove_video_metadata` +- `remove_video_visible` + +The video API validates both the supported extension and container signature, +then delegates all metadata detection and stripping to `metadata.py`. It +requires a separate same-container output, defaulting to `_clean`, so +the experimental path does not overwrite an original. The package root exposes +both functions lazily. + +Native MP4/MOV TC260 labels follow TC260-PG-20257A: +`moov.udta.meta.keys` maps an `AIGC` key to a raw JSON value in `ilst`. +[`noai/isobmff.py`](../src/remove_ai_watermarks/noai/isobmff.py) walks those +nested boxes by seeking, so detection reaches a tail `moov` without reading the +preceding `mdat`. Removal changes the four-byte key to `free` and blanks only +the validated JSON value with same-length spaces. This preserves every box +size, `stco`/`co64` offset, and encoded stream byte. A generic `AIGC` key whose +value has no TC260 field is ignored. + +[`noai/ebml.py`](../src/remove_ai_watermarks/noai/ebml.py) provides the +corresponding bounded Matroska/WebM reader. It seeks over clusters and accepts +only a `Segment.Tags.Tag.SimpleTag` pairing `TagName=AIGC` with a JSON +`TagString` carrying a TC260 field. The existing ffmpeg stream-copy path removes +those container tags without transcoding the encoded streams. + +[`video_visible.py`](../src/remove_ai_watermarks/video_visible.py) implements +the first pixel stages for Sora and Veo. The Sora detector searches a normalized +frame with a fully synthetic mascot-and-text silhouette at several scales. The +Veo detector uses separate synthetic silhouettes for the current four-point +diamond and legacy `Veo` text, with bounded bottom-right searches calibrated +independently from Sora. A strong relocated-diamond match may bypass the known +layout anchors, but weak free-corner matches never enter the temporal arbiter. +This prevents recurring scene details in clean API exports from being promoted +to a watermark. + +Every per-frame result is untrusted. The provider-specific stabilization +wrappers share one recurrence implementation, while retaining separate visual +floors and minimum-run policy. Provenance can relax a low-contrast run only +after recurring visual evidence exists. Sora transition frames follow the +nearest confirmed moving position only with Sora provenance. A confirmed Veo +run can cover low-contrast frames at its fixed position. This separation keeps +clean API exports from being modified merely because metadata names the same +generator. + +Removal runs in a second decode pass. Sora and legacy Veo text use padded box +masks. The square Veo diamond uses a synthetic shape mask so transparent box +corners do not erase unrelated pixels. Every mask goes through the shared +`watermark_registry.fill` backends. ffmpeg encodes the changed video stream and +copies optional audio. The default OpenCV fill is the speed floor; structured +backgrounds need MI-GAN or LaMa for better reconstruction. Invisible video +stages must continue to reuse the image and metadata implementations rather +than copying their logic. + +The inherited ISOBMFF metadata path currently reads the complete container into +memory; replacing that with a streaming box copier is a prerequisite for large +video inputs. + +Regression coverage: + +- [`test_video.py`](../tests/test_video.py) + ## Metadata and provenance ### C2PA @@ -110,6 +174,10 @@ Key contracts: scan. - ISOBMFF containers use [`noai/isobmff.py`](../src/remove_ai_watermarks/noai/isobmff.py). +- Native MP4/MOV TC260 `AIGC` entries are read from + `moov.udta.meta.keys/ilst` and blanked without changing box sizes. +- Native MKV/WebM TC260 `AIGC` entries are read from + `Segment.Tags.Tag.SimpleTag` and removed through the ffmpeg stream-copy path. - Supported non-ISOBMFF audio and video containers use ffmpeg stream copying. - The low-level remover is fail-safe and can copy an undecodable file through unchanged. diff --git a/docs/python-api.md b/docs/python-api.md index 4bc2af6..24a4fbb 100644 --- a/docs/python-api.md +++ b/docs/python-api.md @@ -128,6 +128,65 @@ path preserves the pixels but drops standard metadata. Treat a nonempty undecodable input through unchanged, so its return alone must not be presented as proof that metadata was removed. +## Inspect and strip video metadata + +The experimental high level video API supports MP4, MOV, M4V, WebM, and MKV: + +```python +import remove_ai_watermarks as raiw + +report = raiw.inspect_video_metadata("input.mp4") +if report.has_ai_metadata: + result = raiw.remove_video_metadata("input.mp4") + if result.remaining: + raise RuntimeError(f"AI metadata remains: {result.remaining}") +``` + +`remove_video_metadata` does not transcode video or audio streams. Its default +output is `input_clean.mp4`, leaving the source untouched. An explicit output +must use the same container extension as the source. + +The returned `VideoMetadataResult` records the source, output, metadata detected +before removal, and any markers remaining after the verified strip. MP4/MOV +inspection recognizes the native TC260 `AIGC` entry in +`moov.udta.meta.keys/ilst`; its removal preserves container size and encoded +stream bytes. MKV/WebM inspection recognizes the corresponding +`Segment.Tags.Tag.SimpleTag` representation; its removal requires ffmpeg for a +stream-copy remux. + +## Remove a visible Sora or Veo video mark + +```python +import remove_ai_watermarks as raiw + +result = raiw.remove_video_visible( + "sora.mp4", + "sora_clean.mp4", + backend="cv2", + strip_metadata=True, +) +if result.output is None: + print("No temporally stable Sora mark was found") + +veo_result = raiw.remove_video_visible( + "veo.mp4", + "veo_clean.mp4", + mark="veo", +) +``` + +`remove_video_visible` scans the complete video before writing output. It +combines synthetic multi-scale visual matching with temporal consistency, so an +isolated lookalike in one frame is not enough to authorize inpainting. The +supported `mark` values are `sora` and `veo`. The Veo detector recognizes the +current four-point diamond and the legacy `Veo` text with separate synthetic +silhouettes. + +The returned `VideoVisibleResult` records the total, detected, and removed frame +counts plus any AI metadata that survived the output encode. The function +returns `output=None` and writes no file when no stable mark is selected. Video +pixels are transcoded through ffmpeg while the source audio stream is copied. + ## Remove invisible watermarks ```python diff --git a/docs/supported-signals.md b/docs/supported-signals.md index d893e61..ba19ef5 100644 --- a/docs/supported-signals.md +++ b/docs/supported-signals.md @@ -29,6 +29,16 @@ the masked area. Marks from other vendors are not detected automatically. Use `erase --region` when you can select the affected area yourself. +### Visible video marks + +| Key | Mark | Motion | Important limit | +| --- | --- | --- | --- | +| `sora` | Sora 2 mascot and wordmark | Moves among frame positions | Requires a temporally recurring visual match; the older Sora Turbo corner swirl is a different unsupported mark. | +| `veo` | Current four-point diamond and legacy `Veo` text | Fixed bottom-right corner | Uses separate silhouettes and requires a recurring match; learned fill is preferable on structured backgrounds. | + +Use `video visible` for this registry. It is separate from the image `visible` +command because selection is made over a sequence rather than one raster. + ## Fill backends | Backend | Install | Behavior | @@ -48,7 +58,9 @@ The inspection and stripping code handles signals in these groups: - EXIF and XMP generator fields; - IPTC AI disclosure fields; - PNG text chunks and embedded generation parameters; -- China TC260 AIGC labels in supported metadata placements; +- China TC260 AIGC labels in supported image placements and the normative + MP4/MOV `moov.udta.meta.keys/ilst` and MKV/WebM + `Segment.Tags.Tag.SimpleTag` placements; - xAI and Grok EXIF signature fields; - Samsung AI editing markers; - Hugging Face job metadata; diff --git a/docs/watermarking-landscape.md b/docs/watermarking-landscape.md index 49f7729..d350f7c 100644 --- a/docs/watermarking-landscape.md +++ b/docs/watermarking-landscape.md @@ -38,7 +38,13 @@ pair. On the ISOBMFF path, `blank_ai_exif_tokens` provides the corresponding in-place scrub for supported EXIF values, TC260 AIGC blocks, and the xAI pair. - **China TC260 AIGC label (caught by `AIGC_MARKERS` / `metadata.aigc_label`, surfaced by `identify` as the `aigc` signal):** China-served generators embed an XMP `{"Label":"1","ContentProducer":...}` block — China's mandatory AI-content labeling (TC260 namespace `tc260.org.cn/ns/AIGC`). -**Doubao** (ByteDance) uses it (verified on a public issue sample; `ContentProducer` `001191110102MACQD9K64010000`, no C2PA/SynthID/imwatermark — the XMP block is the only signal; GitHub attachment upload did NOT strip it). The same standard is mandatory for Jimeng/Kling/Qwen/Ernie etc., so the one marker covers the whole China-AIGC-labeled ecosystem. `aigc_label` reads **four serializations** through a shared `_parse` helper: the HTML-entity-encoded XMP `TC260:AIGC` block in **either RDF form** — the nested element `{...}` (Doubao) or the attribute `TC260:AIGC="{...}"` (**PicWish**, `ContentProducer="picwish"`, verified on compatible samples) — via a container-agnostic raw-byte scan (any JSON object accepted), a raw-JSON PNG `AIGC` tEXt chunk (Doubao also writes the label this way, no namespaced marker at all — confirmed on compatible samples, `ContentProducer="doubao"`), a bare raw-JSON `{"AIGC":{...}}` object embedded in **JPEG EXIF (UserComment)** by some China-served generators, brace-matched from the scan head with `json.JSONDecoder().raw_decode` (no namespaced marker, no PNG chunk — confirmed on compatible samples, `ContentProducer="001191440300708461136T1308L"`), **and** a bare `AIGC{...}` blob (the label glued straight to its JSON, no `"AIGC":` key wrapper) embedded in a **JPEG APP segment near the JFIF header** — confirmed on compatible samples. The two raw-JSON forms are scanned in one loop (`'"AIGC"'` then `AIGC{`) that **falls through on a non-TC260 / undecodable hit instead of returning** — a quoted `"AIGC"` can appear later in an XMP packet while the real label is a bare `AIGC{...}` earlier in the file, so an unconditional early return on the quoted form would shadow the bare form (the exact bug behind the 06-10 misses). All three generic forms (the PNG chunk, the bare `{"AIGC":...}` object, and the bare `AIGC{...}` blob) are gated on at least one TC260 field (`_TC260_FIELDS`) so a generic `AIGC` key cannot false-positive; the namespaced XMP element is unambiguous and needs no gate. `_TC260_FIELDS` covers **two schemas**: the producer-side one (`Label` / `ContentProducer` / `ProduceID` / `ContentPropagator` / `PropagateID`, Doubao and most China gens) and the **service-provider** one (`ServiceProvider` / `ServiceUser`, plus generic `Time` / `ContentId` which are NOT gated on) — **Tencent Cloud's** AIGC variant (`ServiceProvider` = `腾讯云`), embedded in **EXIF `ImageDescription`**, verified on compatible samples. In `identify`, `aigc` fires on the parsed label **or** the `AIGC_MARKERS` byte scan (the latter preserves the laundering-tell case where the JSON payload is truncated). +**Doubao** (ByteDance) uses it (verified on a public issue sample; `ContentProducer` `001191110102MACQD9K64010000`, no C2PA/SynthID/imwatermark — the XMP block is the only signal; GitHub attachment upload did NOT strip it). The same standard is mandatory for Jimeng/Kling/Qwen/Ernie etc., so the one marker covers the whole China-AIGC-labeled ecosystem. `aigc_label` reads **four image serializations** through a shared `_parse` helper: the HTML-entity-encoded XMP `TC260:AIGC` block in **either RDF form** — the nested element `{...}` (Doubao) or the attribute `TC260:AIGC="{...}"` (**PicWish**, `ContentProducer="picwish"`, verified on compatible samples) — via a container-agnostic raw-byte scan (any JSON object accepted), a raw-JSON PNG `AIGC` tEXt chunk (Doubao also writes the label this way, no namespaced marker at all — confirmed on compatible samples, `ContentProducer="doubao"`), a bare raw-JSON `{"AIGC":{...}}` object embedded in **JPEG EXIF (UserComment)** by some China-served generators, brace-matched from the scan head with `json.JSONDecoder().raw_decode` (no namespaced marker, no PNG chunk — confirmed on compatible samples, `ContentProducer="001191440300708461136T1308L"`), **and** a bare `AIGC{...}` blob (the label glued straight to its JSON, no `"AIGC":` key wrapper) embedded in a **JPEG APP segment near the JFIF header** — confirmed on compatible samples. The two raw-JSON forms are scanned in one loop (`'"AIGC"'` then `AIGC{`) that **falls through on a non-TC260 / undecodable hit instead of returning** — a quoted `"AIGC"` can appear later in an XMP packet while the real label is a bare `AIGC{...}` earlier in the file, so an unconditional early return on the quoted form would shadow the bare form (the exact bug behind the 06-10 misses). Native MP4/MOV is a fifth serialization: TC260-PG-20257A stores an `AIGC` key in `moov.udta.meta.keys` and the raw JSON in the matching `ilst` item. The seeking parser reaches a tail `moov` without reading `mdat`; removal replaces the key with `free` and blanks the validated value at the same length so every box size and stream offset stays fixed. All generic forms are gated on at least one TC260 field (`TC260_AIGC_FIELDS`) so a generic `AIGC` key cannot false-positive; the namespaced XMP element is unambiguous and needs no gate. `TC260_AIGC_FIELDS` covers **two schemas**: the producer-side one (`Label` / `ContentProducer` / `ProduceID` / `ContentPropagator` / `PropagateID`, Doubao and most China gens) and the **service-provider** one (`ServiceProvider` / `ServiceUser`, plus generic `Time` / `ContentId` which are NOT gated on) — **Tencent Cloud's** AIGC variant (`ServiceProvider` = `腾讯云`), embedded in **EXIF `ImageDescription`**, verified on compatible samples. In `identify`, `aigc` fires on the parsed label **or** the `AIGC_MARKERS` byte scan (the latter preserves the laundering-tell case where the JSON payload is truncated). + +Native MKV/WebM is a sixth serialization. TC260-PG-20257A stores +`TagName=AIGC` and the raw JSON `TagString` in +`Segment.Tags.Tag.SimpleTag`. The bounded EBML reader skips cluster payloads; +the existing ffmpeg stream-copy path removes the tags without transcoding. + - **HuggingFace-hosted job (caught by `metadata.huggingface_job`, surfaced by `identify` as the `hf_job` signal, MEDIUM confidence):** HuggingFace Jobs / Spaces can stamp generated PNGs with an `hf-job-id` tEXt chunk holding the job UUID. It marks the *hosting job*, not a model, so it lifts an Unknown verdict to a tentative AI via `hf_only` but never overrides a hard metadata signal. `_HF_JOB_CAVEAT` states the limit. Removal drops the chunk through the PNG metadata whitelist. - **No detectable signal on some downloads:** Recraft exports and some hosted FLUX surfaces can arrive without a supported local signal. Midjourney samples @@ -49,13 +55,18 @@ in-place scrub for supported EXIF values, TC260 AIGC blocks, and the xAI pair. - **C2PA 2.4 "Durable Content Credentials" (April 2026; verified against the spec) raise the bar for metadata stripping.** 2.4 defines soft bindings (an invisible watermark or a content fingerprint) plus a server-side manifest repository and a new `c2pa.repository-receipt` assertion. Per the spec: "if a C2PA manifest is removed from an asset, but a copy of that manifest remains in a provenance store elsewhere, the manifest and asset may be matched using available soft bindings." So our local `metadata --remove` deletes the *embedded* manifest, but a fingerprint/watermark soft binding can still re-link the image to its manifest in a repository server-side. Stripping the file is becoming necessary-but-not-sufficient against durable provenance. (Our parsers target the stable embedded-manifest format documented in C2PA 2.1 §11; that format is unchanged in 2.4 -- the new pieces are repository/soft-binding infra, not the on-file box layout, so no parser change is implied.) Spec: https://spec.c2pa.org/specifications/specifications/2.4/specs/C2PA_Specification.html We now READ the soft-binding `alg` (`C2PA_SOFT_BINDINGS` / `soft_binding_vendors_in`) to name the forensic-watermark vendor, and locally DECODE the one open scheme, Adobe TrustMark (`trustmark_detector`); the rest (Digimarc/Imatag/Steg.AI/...) stay name-only (proprietary decoders). - **Built in the dated batch:** soft-binding vendor detection, IPTC Photo Metadata AI-disclosure fields, C2PA detection and stripping for supported - ISOBMFF video, and the optional Adobe TrustMark decoder. Visible video-logo - removal and proprietary audio-watermark detection remain outside the package. + ISOBMFF video, the optional Adobe TrustMark decoder, and temporally stabilized + visible Sora and Veo removal. Other visible video logos and proprietary + audio-watermark detection remain outside the package. Metadata stripping for supported audio containers is a separate implemented path. **Box detection window — now handled (v0.6.8):** detection no longer relies on a fixed first-MB read. `metadata.scan_head(path, size)` reads the first `size` bytes and, for ISOBMFF, appends the payloads of late provenance boxes found by `isobmff.scan_c2pa_region` (a file-seeking top-level box walker that skips past `mdat` by size without reading it), so a C2PA/AIGC/IPTC manifest placed AFTER a large `mdat` in a streaming/non-faststart MP4 is now caught. Every C2PA/marker byte scan (`has_ai_metadata`, `aigc_label`, `iptc_ai_system`, `synthid_source`, `exif_generator` XMP, `get_ai_metadata` soft-binding, and `identify`) goes through `scan_head`; it is behavior-neutral for non-ISOBMFF inputs (exactly `f.read(size)`). +Native TC260 MP4/MOV tags do not live in those top-level provenance boxes. +`tc260_aigc_payloads` separately seeks through `moov.udta.meta.keys/ilst`, so +the normative tag is also found when a large `mdat` precedes `moov`. + **Meta-box XMP and EXIF removal are handled in place:** an AI-label XMP packet stored as a meta-box `mime` item is blanked by `isobmff.blank_ai_xmp_packets`. Supported EXIF items are handled by @@ -69,6 +80,16 @@ against primary sources before adding jurisdiction-specific claims. **Google Gemini visible "sparkle" -- tier-dependent, and spec-undocumented by Google.** Google primary sources (the Nano Banana Pro blog and the gemini.google image-generation page, both WebFetch-verified) confirm Gemini images carry BOTH the invisible SynthID (on ALL Google-AI media) AND a visible sparkle, but the visible mark is **tier-gated**: applied for FREE and Google AI **Pro** users, and **REMOVED** for Google AI **Ultra** subscribers, inside **Google AI Studio**, and on **API / dev** output. So a Google-C2PA image with NO visible sparkle is expected (Ultra / API), not evidence it is clean -- this reinforces the `identify` "no visible mark != clean" rule. The ONLY official verifier is the SynthID flow (upload to the Gemini app, ask if it is AI-generated), which reads the INVISIBLE mark; there is **no official visible-sparkle detector**, and Google publishes **no** glyph geometry / size / opacity / color / locale / placement spec. So our capture-based sparkle template is the only source of truth and cannot be validated against a vendor spec -- keep reverse-engineering from real captures (do not expect a published spec). +**Google Veo video marks use two incompatible visible designs.** Public raw +clips verify the current four-point diamond and the legacy bottom-right `Veo` +text. The independent +[VeoWatermarkRemover](https://github.com/allenk/VeoWatermarkRemover) project +reports the same current-versus-legacy split, multiple output layouts, and the +need for cross-frame position agreement. Our implementation copies no logo +pixels or alpha maps from that project: it uses two synthetic silhouettes, +known-layout searches plus a strong relocated-diamond fallback, and a separate +temporal arbiter calibrated against raw watermarked clips and clean API exports. + **The faint-visible-mark precision/recall wall is fundamental, not a heuristic artifact.** The visible-watermark-detection literature has moved to LEARNED segmentation / object-detection (WDNet WACV'21 arXiv:2012.07616; SLBR ACM MM'21, open code+weights; the PRCV'18 large-scale detector; Su et al. survey 2025), but three verified findings bound what a learned detector actually buys: (1) a claim that a confidence threshold "cleanly separates" true from false matches even with a learned CNN front-end was **REFUTED** in verification (arXiv:1705.08593) -- the precision/recall wall persists even with learned features. (2) Learned detectors need a LARGE, pattern-diverse labeled dataset trained on synthetic composites (PRCV'18: 60k images / 80 watermark classes; CLWD: 60k / 160 marks), and off-distribution degradation is a documented real axis (models trained on limited-pattern LVW transfer worse; diversity of training patterns drives generalization). (3) Inference is cheap (WDNet ~8 ms at 256x256) -- the cost is the data pipeline, not runtime. Net: a learned detector shifts the frontier but does NOT remove the wall; for a SINGLE mark the cheapest next step is a small patch classifier (real-sparkle vs false-positive) on top of the existing NCC localizer, not a full segmentation model. SLBR is a ready baseline. The current NCC + false-positive gate (core-ring brightness margin + gradient-NCC crispness + white-core saturation) is a sound operating point, and the residual miss is the information-theoretic wall the literature confirms. **Visible-mark landscape beyond the registry.** Meta stamps a visible "Imagined with AI" mark (bottom-LEFT, a small symbol) on its OWN Meta AI / "Imagine" output; for third-party images it relies on C2PA / IPTC, not a visible mark. Samsung Galaxy AI additionally uses a **four-star icon** variant in a corner alongside the localized text wordmark `samsung_engine` calibrates (only the Italian text variant is covered) -- the icon is a distinct, uncovered variant. Every source agrees visible + metadata marks are trivially removable (crop / screenshot, ~2 s), which is the tool's premise. diff --git a/pyproject.toml b/pyproject.toml index dd846ca..2c96661 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,7 +1,7 @@ [project] name = "remove-ai-watermarks" version = "0.20.2" -description = "AI watermark remover: strip visible and invisible AI watermarks (Gemini / Nano Banana sparkle, SynthID) and provenance metadata (C2PA, EXIF) from images" +description = "Remove visible and invisible AI watermarks from images and provenance metadata from media containers" readme = "README.md" requires-python = ">=3.10.1" license = {text = "Apache-2.0"} diff --git a/src/remove_ai_watermarks/__init__.py b/src/remove_ai_watermarks/__init__.py index b7cceda..8ebbbcd 100644 --- a/src/remove_ai_watermarks/__init__.py +++ b/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}") diff --git a/src/remove_ai_watermarks/cli.py b/src/remove_ai_watermarks/cli.py index 3f8676c..02f0236 100644 --- a/src/remove_ai_watermarks/cli.py +++ b/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: _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: _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)) diff --git a/src/remove_ai_watermarks/metadata.py b/src/remove_ai_watermarks/metadata.py index 9de9459..016cbe7 100644 --- a/src/remove_ai_watermarks/metadata.py +++ b/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 ``{...}`` 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 {...} 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, diff --git a/src/remove_ai_watermarks/noai/ebml.py b/src/remove_ai_watermarks/noai/ebml.py new file mode 100644 index 0000000..6b87412 --- /dev/null +++ b/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) diff --git a/src/remove_ai_watermarks/noai/isobmff.py b/src/remove_ai_watermarks/noai/isobmff.py index 2086638..af135f7 100644 --- a/src/remove_ai_watermarks/noai/isobmff.py +++ b/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" diff --git a/src/remove_ai_watermarks/video.py b/src/remove_ai_watermarks/video.py new file mode 100644 index 0000000..1fcb364 --- /dev/null +++ b/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 ''}; 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 ''}" + ) + 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 ``_clean``. 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, + ) diff --git a/src/remove_ai_watermarks/video_visible.py b/src/remove_ai_watermarks/video_visible.py new file mode 100644 index 0000000..7b15ca0 --- /dev/null +++ b/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() diff --git a/tests/test_metadata.py b/tests/test_metadata.py index 85555ba..6cd268f 100644 --- a/tests/test_metadata.py +++ b/tests/test_metadata.py @@ -1103,7 +1103,7 @@ class TestAIGCLabel: """Tencent Cloud's AIGC variant uses a service-provider schema (``ServiceProvider`` / ``ServiceUser`` / ``Time`` / ``ContentId``) rather than the producer schema, embedded in EXIF ``ImageDescription`` -- none of - its fields overlap the producer-side ``_TC260_FIELDS``, so the generic + its fields overlap the producer-side ``TC260_AIGC_FIELDS``, so the generic ``"AIGC":{...}`` gate must recognize them too.""" import json diff --git a/tests/test_video.py b/tests/test_video.py new file mode 100644 index 0000000..ebde4be --- /dev/null +++ b/tests/test_video.py @@ -0,0 +1,701 @@ +"""Tests for the video processing API and CLI.""" + +from __future__ import annotations + +from typing import TYPE_CHECKING + +import cv2 +import numpy as np +import pytest +from click.testing import CliRunner +from PIL import Image, ImageDraw, ImageFont + +from remove_ai_watermarks.cli import main +from remove_ai_watermarks.metadata import C2PA_UUID + +if TYPE_CHECKING: + from pathlib import Path + + +_MP4_FTYP = b"\x00\x00\x00\x18ftypmp42\x00\x00\x00\x00mp42isom" +_VIDEO_PAYLOAD = b"synthetic-video-payload" +_TC260_AIGC = ( + b'{"Label":"1","ContentProducer":"00119144030008867405X210002",' + b'"ProduceID":"sample-001","ReservedCode1":"","ContentPropagator":"",' + b'"PropagateID":"","ReservedCode2":""}' +) + + +def _box(box_type: bytes, payload: bytes) -> bytes: + return (8 + len(payload)).to_bytes(4, "big") + box_type + payload + + +def _video_with_c2pa(path: Path) -> Path: + manifest = C2PA_UUID + b"OpenAI trainedAlgorithmicMedia" + path.write_bytes(_MP4_FTYP + _box(b"uuid", manifest) + _box(b"mdat", _VIDEO_PAYLOAD)) + return path + + +def _metadata_key(name: bytes) -> bytes: + return (8 + len(name)).to_bytes(4, "big") + b"mdta" + name + + +def _metadata_value(index: int, value: bytes) -> bytes: + data = _box(b"data", b"\x00\x00\x00\x01\x00\x00\x00\x00" + value) + return _box(index.to_bytes(4, "big"), data) + + +def _video_with_tc260(path: Path, *, media_payload: bytes = _VIDEO_PAYLOAD) -> Path: + keys = _box( + b"keys", + b"\x00\x00\x00\x00" + (2).to_bytes(4, "big") + _metadata_key(b"AIGC") + _metadata_key(b"title"), + ) + ilst = _box( + b"ilst", + _metadata_value(1, _TC260_AIGC) + _metadata_value(2, b"standard title"), + ) + meta = _box(b"meta", b"\x00\x00\x00\x00" + keys + ilst) + path.write_bytes(_MP4_FTYP + _box(b"mdat", media_payload) + _box(b"moov", _box(b"udta", meta))) + return path + + +def _ebml_size(value: int) -> bytes: + for length in range(1, 9): + if value < (1 << (7 * length)) - 1: + return ((1 << (7 * length)) | value).to_bytes(length, "big") + raise ValueError("EBML test value is too large") + + +def _ebml_element(element_id: bytes, payload: bytes) -> bytes: + return element_id + _ebml_size(len(payload)) + payload + + +def _video_with_tc260_ebml(path: Path, *, value: bytes = _TC260_AIGC) -> Path: + simple_tag = _ebml_element( + b"\x67\xc8", + _ebml_element(b"\x45\xa3", b"AIGC") + _ebml_element(b"\x44\x87", value), + ) + tags = _ebml_element(b"\x12\x54\xc3\x67", _ebml_element(b"\x73\x73", simple_tag)) + segment = _ebml_element(b"\x18\x53\x80\x67", tags) + path.write_bytes(_ebml_element(b"\x1a\x45\xdf\xa3", b"") + segment) + return path + + +class TestVideoMetadataApi: + def test_top_level_api_is_lazy_exported(self): + import remove_ai_watermarks as raiw + + assert raiw.inspect_video_metadata is not None + assert raiw.remove_video_metadata is not None + assert raiw.remove_video_visible is not None + + def test_inspects_video_metadata(self, tmp_path: Path): + from remove_ai_watermarks.video import inspect_video_metadata + + source = _video_with_c2pa(tmp_path / "source.mp4") + + report = inspect_video_metadata(source) + + assert report.source == source + assert report.has_ai_metadata is True + assert report.markers + + def test_removes_metadata_without_touching_video_payload(self, tmp_path: Path): + from remove_ai_watermarks.video import remove_video_metadata + + source = _video_with_c2pa(tmp_path / "source.mp4") + output = tmp_path / "clean.mp4" + + result = remove_video_metadata(source, output) + + assert result.output == output + assert result.detected + assert result.remaining == {} + assert _VIDEO_PAYLOAD in output.read_bytes() + assert C2PA_UUID not in output.read_bytes() + + def test_default_output_preserves_source(self, tmp_path: Path): + from remove_ai_watermarks.video import remove_video_metadata + + source = _video_with_c2pa(tmp_path / "source.mp4") + original = source.read_bytes() + + result = remove_video_metadata(source) + + assert result.output == tmp_path / "source_clean.mp4" + assert result.output.exists() + assert source.read_bytes() == original + + @pytest.mark.parametrize("suffix", [".mp4", ".mov"]) + def test_inspects_native_tc260_metadata(self, tmp_path: Path, suffix: str): + from remove_ai_watermarks.video import inspect_video_metadata + + source = _video_with_tc260(tmp_path / f"source{suffix}") + + report = inspect_video_metadata(source) + + assert report.has_ai_metadata is True + assert report.markers["aigc_label"].endswith("producer 00119144030008867405X210002") + + def test_inspects_native_tc260_metadata_after_large_media_payload(self, tmp_path: Path): + from remove_ai_watermarks.video import inspect_video_metadata + + source = _video_with_tc260( + tmp_path / "source.mp4", + media_payload=b"x" * (1024 * 1024), + ) + + report = inspect_video_metadata(source) + + assert report.has_ai_metadata is True + assert "aigc_label" in report.markers + + def test_removes_native_tc260_metadata_without_touching_media_or_standard_tag(self, tmp_path: Path): + from remove_ai_watermarks.video import remove_video_metadata + + source = _video_with_tc260(tmp_path / "source.mp4") + output = tmp_path / "clean.mp4" + + result = remove_video_metadata(source, output) + cleaned = output.read_bytes() + + assert result.detected["aigc_label"].startswith("China AIGC label") + assert result.remaining == {} + assert len(cleaned) == source.stat().st_size + assert _VIDEO_PAYLOAD in cleaned + assert b"standard title" in cleaned + assert b"AIGC" not in cleaned + assert _TC260_AIGC not in cleaned + + def test_ignores_generic_mp4_aigc_tag_without_tc260_fields(self, tmp_path: Path): + from remove_ai_watermarks.video import inspect_video_metadata + + source = _video_with_tc260(tmp_path / "source.mp4") + source.write_bytes(source.read_bytes().replace(_TC260_AIGC, b'{"description":"' + b"x" * 146 + b'"}')) + + report = inspect_video_metadata(source) + + assert report.has_ai_metadata is False + assert report.markers == {} + + @pytest.mark.parametrize("suffix", [".mkv", ".webm"]) + def test_inspects_native_tc260_ebml_metadata(self, tmp_path: Path, suffix: str): + from remove_ai_watermarks.video import inspect_video_metadata + + source = _video_with_tc260_ebml(tmp_path / f"source{suffix}") + + report = inspect_video_metadata(source) + + assert report.has_ai_metadata is True + assert report.markers["aigc_label"].endswith("producer 00119144030008867405X210002") + + def test_ignores_generic_ebml_aigc_tag_without_tc260_fields(self, tmp_path: Path): + from remove_ai_watermarks.video import inspect_video_metadata + + source = _video_with_tc260_ebml( + tmp_path / "source.mkv", + value=b'{"description":"ordinary application metadata"}', + ) + + report = inspect_video_metadata(source) + + assert report.has_ai_metadata is False + assert report.markers == {} + + def test_rejects_image_input(self, tmp_clean_png: Path): + from remove_ai_watermarks.video import inspect_video_metadata + + with pytest.raises(ValueError, match="Unsupported video format"): + inspect_video_metadata(tmp_clean_png) + + def test_rejects_image_with_video_extension(self, tmp_clean_png: Path, tmp_path: Path): + from remove_ai_watermarks.video import inspect_video_metadata + + disguised = tmp_path / "image.mp4" + disguised.write_bytes(tmp_clean_png.read_bytes()) + + with pytest.raises(ValueError, match="does not match"): + inspect_video_metadata(disguised) + + def test_rejects_output_container_change(self, tmp_path: Path): + from remove_ai_watermarks.video import remove_video_metadata + + source = _video_with_c2pa(tmp_path / "source.mp4") + + with pytest.raises(ValueError, match="must match"): + remove_video_metadata(source, tmp_path / "clean.mov") + + +class TestVideoMetadataCli: + def test_help(self): + runner = CliRunner() + + result = runner.invoke(main, ["video", "metadata", "--help"]) + + assert result.exit_code == 0, result.output + assert "AI metadata" in result.output + + def test_check_reports_metadata(self, tmp_path: Path): + runner = CliRunner() + source = _video_with_c2pa(tmp_path / "source.mp4") + + result = runner.invoke(main, ["video", "metadata", str(source), "--check"]) + + assert result.exit_code == 0, result.output + assert "AI metadata detected" in result.output + + def test_remove_reports_output(self, tmp_path: Path): + runner = CliRunner() + source = _video_with_c2pa(tmp_path / "source.mp4") + output = tmp_path / "clean.mp4" + + result = runner.invoke(main, ["video", "metadata", str(source), "--remove", "-o", str(output)]) + + assert result.exit_code == 0, result.output + assert "AI metadata stripped" in result.output + assert C2PA_UUID not in output.read_bytes() + + def test_rejects_image_input(self, tmp_clean_png: Path): + runner = CliRunner() + + result = runner.invoke(main, ["video", "metadata", str(tmp_clean_png), "--check"]) + + assert result.exit_code != 0 + assert "Unsupported video format" in result.output + + +class TestSoraFrameLocalization: + @staticmethod + def _sora_like_frame() -> tuple[np.ndarray, tuple[int, int, int, int]]: + frame = np.full((480, 840, 3), 36, dtype=np.uint8) + mark = Image.new("L", (180, 64), 0) + draw = ImageDraw.Draw(mark) + draw.ellipse((1, 14, 32, 54), fill=255) + draw.ellipse((25, 8, 62, 58), fill=255) + draw.ellipse((15, 20, 28, 44), fill=0) + draw.ellipse((37, 18, 50, 43), fill=0) + try: + font = ImageFont.load_default(size=49) + except TypeError: + font = ImageFont.load_default() + draw.text((68, 1), "Sora", font=font, fill=255, stroke_width=1) + mark_array = cv2.resize(np.asarray(mark), (124, 44), interpolation=cv2.INTER_AREA) + x, y = 620, 398 + alpha = mark_array.astype(np.float32)[:, :, None] / 255 * 0.78 + crop = frame[y : y + 44, x : x + 124].astype(np.float32) + frame[y : y + 44, x : x + 124] = np.clip(crop * (1 - alpha) + 255 * alpha, 0, 255).astype(np.uint8) + return frame, (x, y, 124, 44) + + def test_localizes_independently_rendered_sora_like_mark(self): + from remove_ai_watermarks.video_visible import _region_iou, detect_sora_frame + + frame, expected = self._sora_like_frame() + + detection = detect_sora_frame(frame) + + assert detection.region is not None + assert detection.confidence >= 0.58 + assert _region_iou(detection.region, expected) >= 0.45 + + def test_empty_frame_is_not_localized(self): + from remove_ai_watermarks.video_visible import detect_sora_frame + + detection = detect_sora_frame(np.empty((0, 0, 3), dtype=np.uint8)) + + assert detection.confidence == 0.0 + assert detection.region is None + + +class TestVeoFrameLocalization: + def test_localizes_independently_rendered_diamond_at_relocated_position(self): + from remove_ai_watermarks.video_visible import _region_iou, detect_veo_frame + + frame = np.full((720, 1280, 3), 28, dtype=np.uint8) + size = 48 + x, y = 1080, 570 + mark = Image.new("L", (size, size), 0) + points = ( + (size // 2, 1), + (round(size * 0.61), round(size * 0.38)), + (size - 2, size // 2), + (round(size * 0.61), round(size * 0.62)), + (size // 2, size - 2), + (round(size * 0.39), round(size * 0.62)), + (1, size // 2), + (round(size * 0.39), round(size * 0.38)), + ) + ImageDraw.Draw(mark).polygon(points, fill=255) + alpha = np.asarray(mark, dtype=np.float32)[:, :, None] / 255 * 0.72 + crop = frame[y : y + size, x : x + size].astype(np.float32) + frame[y : y + size, x : x + size] = np.clip( + crop * (1 - alpha) + 255 * alpha, + 0, + 255, + ).astype(np.uint8) + + detection = detect_veo_frame(frame) + + assert detection.region is not None + assert detection.confidence >= 0.70 + assert _region_iou(detection.region, (x, y, size, size)) >= 0.70 + + def test_localizes_independently_rendered_legacy_text(self): + from remove_ai_watermarks.video_visible import _region_iou, detect_veo_frame + + frame = np.full((720, 1280, 3), 42, dtype=np.uint8) + mark = Image.new("L", (60, 24), 0) + try: + font = ImageFont.load_default(size=19) + except TypeError: + font = ImageFont.load_default() + ImageDraw.Draw(mark).text((1, 0), "Veo", font=font, fill=255) + mark_array = np.asarray(mark) + ys, xs = np.where(mark_array > 0) + mark_array = mark_array[ys.min() : ys.max() + 1, xs.min() : xs.max() + 1] + mark_height, mark_width = mark_array.shape + x = frame.shape[1] - mark_width - 20 + y = frame.shape[0] - mark_height - 18 + alpha = mark_array.astype(np.float32)[:, :, None] / 255 * 0.66 + crop = frame[y : y + mark_height, x : x + mark_width].astype(np.float32) + frame[y : y + mark_height, x : x + mark_width] = np.clip( + crop * (1 - alpha) + 255 * alpha, + 0, + 255, + ).astype(np.uint8) + + detection = detect_veo_frame(frame) + + assert detection.region is not None + assert detection.confidence >= 0.55 + assert _region_iou(detection.region, (x, y, mark_width, mark_height)) >= 0.65 + + def test_empty_frame_is_not_localized(self): + from remove_ai_watermarks.video_visible import detect_veo_frame + + detection = detect_veo_frame(np.empty((0, 0, 3), dtype=np.uint8)) + + assert detection.confidence == 0.0 + assert detection.region is None + + def test_diamond_mask_preserves_transparent_box_corners(self): + from remove_ai_watermarks.video_visible import _mask_for_region + + mask = _mask_for_region( + np.zeros((100, 100, 3), dtype=np.uint8), + (20, 20, 48, 48), + padding_fraction=0.18, + mask_style="veo", + ) + + assert mask[44, 44] == 255 + assert mask[20, 20] == 0 + assert mask[67, 67] == 0 + + +class TestSoraTemporalArbiter: + _BOX = (40, 60, 150, 54) + + def test_four_frame_lookalike_run_is_too_short(self): + from remove_ai_watermarks.video_visible import FrameLocalization, stabilize_sora_localizations + + detections = [FrameLocalization(index, 0.70, self._BOX) for index in range(4)] + + assert stabilize_sora_localizations(detections, provenance=False) == [None] * 4 + + def test_provenance_accepts_recurring_low_contrast_visual_match(self): + from remove_ai_watermarks.video_visible import FrameLocalization, stabilize_sora_localizations + + detections = [ + FrameLocalization(0, 0.59, self._BOX), + FrameLocalization(1, 0.61, self._BOX), + FrameLocalization(2, 0.62, self._BOX), + FrameLocalization(3, 0.60, self._BOX), + FrameLocalization(4, 0.61, self._BOX), + ] + + assert stabilize_sora_localizations(detections, provenance=True) == [self._BOX] * 5 + + def test_confirmed_provenance_run_covers_transition_frames(self): + from remove_ai_watermarks.video_visible import FrameLocalization, stabilize_sora_localizations + + detections = [ + FrameLocalization(0, 0.30, (500, 300, 54, 54)), + FrameLocalization(1, 0.59, self._BOX), + FrameLocalization(2, 0.61, self._BOX), + FrameLocalization(3, 0.62, self._BOX), + FrameLocalization(4, 0.60, self._BOX), + FrameLocalization(5, 0.61, self._BOX), + FrameLocalization(6, 0.30, (300, 100, 54, 54)), + ] + + assert stabilize_sora_localizations(detections, provenance=True) == [self._BOX] * 7 + + def test_transition_prefers_low_score_match_at_a_confirmed_position(self): + from remove_ai_watermarks.video_visible import FrameLocalization, stabilize_sora_localizations + + other_box = (500, 300, 150, 54) + detections = [ + FrameLocalization(0, 0.61, self._BOX), + FrameLocalization(1, 0.62, self._BOX), + FrameLocalization(2, 0.63, self._BOX), + FrameLocalization(3, 0.61, self._BOX), + FrameLocalization(4, 0.62, self._BOX), + FrameLocalization(5, 0.20, (250, 180, 54, 54)), + FrameLocalization(6, 0.52, self._BOX), + FrameLocalization(7, 0.61, other_box), + FrameLocalization(8, 0.62, other_box), + FrameLocalization(9, 0.63, other_box), + FrameLocalization(10, 0.61, other_box), + FrameLocalization(11, 0.62, other_box), + ] + + stabilized = stabilize_sora_localizations(detections, provenance=True) + + assert stabilized[6] == self._BOX + assert stabilized[7:] == [other_box] * 5 + + def test_transition_without_a_match_keeps_previous_stable_position(self): + from remove_ai_watermarks.video_visible import FrameLocalization, stabilize_sora_localizations + + other_box = (500, 300, 150, 54) + detections = [ + FrameLocalization(0, 0.61, self._BOX), + FrameLocalization(1, 0.62, self._BOX), + FrameLocalization(2, 0.63, self._BOX), + FrameLocalization(3, 0.61, self._BOX), + FrameLocalization(4, 0.62, self._BOX), + FrameLocalization(5, 0.20, (250, 180, 54, 54)), + FrameLocalization(6, 0.20, (300, 200, 54, 54)), + FrameLocalization(7, 0.61, other_box), + FrameLocalization(8, 0.62, other_box), + FrameLocalization(9, 0.63, other_box), + FrameLocalization(10, 0.61, other_box), + FrameLocalization(11, 0.62, other_box), + ] + + stabilized = stabilize_sora_localizations(detections, provenance=True) + + assert stabilized[5:7] == [self._BOX, self._BOX] + + def test_unproven_weak_run_is_rejected(self): + from remove_ai_watermarks.video_visible import FrameLocalization, stabilize_sora_localizations + + detections = [ + FrameLocalization(0, 0.61, self._BOX), + FrameLocalization(1, 0.62, self._BOX), + FrameLocalization(2, 0.63, self._BOX), + FrameLocalization(3, 0.62, self._BOX), + FrameLocalization(4, 0.61, self._BOX), + ] + + assert stabilize_sora_localizations(detections, provenance=False) == [None] * 5 + + def test_strong_recurring_visual_run_needs_no_metadata(self): + from remove_ai_watermarks.video_visible import FrameLocalization, stabilize_sora_localizations + + detections = [ + FrameLocalization(0, 0.61, self._BOX), + FrameLocalization(1, 0.66, self._BOX), + FrameLocalization(2, 0.62, self._BOX), + FrameLocalization(3, 0.61, self._BOX), + FrameLocalization(4, 0.62, self._BOX), + ] + + assert stabilize_sora_localizations(detections, provenance=False) == [self._BOX] * 5 + + def test_isolated_lookalikes_at_different_positions_are_rejected(self): + from remove_ai_watermarks.video_visible import FrameLocalization, stabilize_sora_localizations + + detections = [ + FrameLocalization(0, 0.70, (10, 10, 150, 54)), + FrameLocalization(1, 0.70, (400, 200, 150, 54)), + FrameLocalization(2, 0.70, (650, 400, 150, 54)), + ] + + assert stabilize_sora_localizations(detections, provenance=True) == [None, None, None] + + def test_short_dropout_between_matching_boxes_is_filled(self): + from remove_ai_watermarks.video_visible import FrameLocalization, stabilize_sora_localizations + + detections = [ + FrameLocalization(0, 0.66, self._BOX), + FrameLocalization(1, 0.20, (500, 300, 54, 54)), + FrameLocalization(2, 0.67, self._BOX), + FrameLocalization(3, 0.66, self._BOX), + FrameLocalization(4, 0.66, self._BOX), + FrameLocalization(5, 0.66, self._BOX), + ] + + assert stabilize_sora_localizations(detections, provenance=False) == [self._BOX] * 6 + + +class TestVeoTemporalArbiter: + _BOX = (1132, 572, 56, 56) + + def test_eleven_frame_lookalike_run_is_too_short(self): + from remove_ai_watermarks.video_visible import FrameLocalization, stabilize_veo_localizations + + detections = [FrameLocalization(index, 0.70, self._BOX) for index in range(11)] + + assert stabilize_veo_localizations(detections, provenance=False) == [None] * 11 + + def test_strong_fixed_run_covers_video_without_metadata(self): + from remove_ai_watermarks.video_visible import FrameLocalization, stabilize_veo_localizations + + detections = [FrameLocalization(index, 0.60, self._BOX) for index in range(12)] + detections.extend(FrameLocalization(index, 0.20, (300, 200, 48, 48)) for index in range(12, 15)) + + assert stabilize_veo_localizations(detections, provenance=False) == [self._BOX] * 15 + + def test_google_provenance_accepts_recurring_low_contrast_diamond(self): + from remove_ai_watermarks.video_visible import FrameLocalization, stabilize_veo_localizations + + detections = [FrameLocalization(index, 0.47, self._BOX) for index in range(12)] + + assert stabilize_veo_localizations(detections, provenance=True) == [self._BOX] * 12 + + def test_unproven_weak_run_is_rejected(self): + from remove_ai_watermarks.video_visible import FrameLocalization, stabilize_veo_localizations + + detections = [FrameLocalization(index, 0.52, self._BOX) for index in range(12)] + + assert stabilize_veo_localizations(detections, provenance=False) == [None] * 12 + + +class TestVideoVisibleApi: + def test_removes_stable_sora_run_and_writes_output(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch): + from remove_ai_watermarks import video_visible + from remove_ai_watermarks.video import remove_video_visible + from remove_ai_watermarks.video_visible import FrameLocalization, VideoScan + + source = _video_with_c2pa(tmp_path / "source.mp4") + output = tmp_path / "clean.mp4" + box = (4, 4, 20, 8) + scan = VideoScan( + width=64, + height=64, + fps=24.0, + detections=tuple(FrameLocalization(index, 0.66, box) for index in range(5)), + ) + monkeypatch.setattr(video_visible, "scan_sora_video", lambda _source: scan) + + def fake_encode( + _source: Path, + target: Path, + _scan: VideoScan, + regions: list[tuple[int, int, int, int] | None], + **_kwargs: object, + ) -> int: + assert regions == [box] * 5 + target.write_bytes(_MP4_FTYP + _box(b"mdat", _VIDEO_PAYLOAD)) + return 5 + + monkeypatch.setattr(video_visible, "encode_clean_video", fake_encode) + + result = remove_video_visible(source, output) + + assert result.output == output + assert result.detected_frames == 5 + assert result.removed_frames == 5 + assert result.remaining_metadata == {} + + def test_no_stable_mark_writes_no_output(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch): + from remove_ai_watermarks import video_visible + from remove_ai_watermarks.video import remove_video_visible + from remove_ai_watermarks.video_visible import FrameLocalization, VideoScan + + source = _video_with_c2pa(tmp_path / "source.mp4") + output = tmp_path / "clean.mp4" + scan = VideoScan( + width=64, + height=64, + fps=24.0, + detections=( + FrameLocalization(0, 0.70, (1, 1, 20, 8)), + FrameLocalization(1, 0.70, (30, 30, 20, 8)), + FrameLocalization(2, 0.70, (1, 30, 20, 8)), + ), + ) + monkeypatch.setattr(video_visible, "scan_sora_video", lambda _source: scan) + + result = remove_video_visible(source, output) + + assert result.output is None + assert result.removed_frames == 0 + assert not output.exists() + + def test_dispatches_veo_detector_and_uses_tighter_mask( + self, + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + ): + from remove_ai_watermarks import video_visible + from remove_ai_watermarks.video import remove_video_visible + from remove_ai_watermarks.video_visible import FrameLocalization, VideoScan + + source = _video_with_c2pa(tmp_path / "source.mp4") + output = tmp_path / "clean.mp4" + box = (4, 4, 20, 20) + scan = VideoScan( + width=64, + height=64, + fps=24.0, + detections=tuple(FrameLocalization(index, 0.60, box) for index in range(12)), + ) + monkeypatch.setattr(video_visible, "scan_veo_video", lambda _source: scan) + + def fake_encode( + _source: Path, + target: Path, + _scan: VideoScan, + regions: list[tuple[int, int, int, int] | None], + **kwargs: object, + ) -> int: + assert regions == [box] * 12 + assert kwargs["padding_fraction"] == 0.18 + assert kwargs["mask_style"] == "veo" + target.write_bytes(_MP4_FTYP + _box(b"mdat", _VIDEO_PAYLOAD)) + return 12 + + monkeypatch.setattr(video_visible, "encode_clean_video", fake_encode) + + result = remove_video_visible(source, output, mark="veo") + + assert result.output == output + assert result.mark == "veo" + assert result.detected_frames == 12 + assert result.removed_frames == 12 + + +class TestVideoVisibleCli: + def test_help(self): + result = CliRunner().invoke(main, ["video", "visible", "--help"]) + + assert result.exit_code == 0, result.output + assert "temporally stable" in result.output + assert "sora|veo" in result.output + + def test_reports_removed_frames(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch): + from remove_ai_watermarks import video + from remove_ai_watermarks.video import VideoVisibleResult + + source = _video_with_c2pa(tmp_path / "source.mp4") + output = tmp_path / "clean.mp4" + monkeypatch.setattr( + video, + "remove_video_visible", + lambda *_args, **_kwargs: VideoVisibleResult( + source=source, + output=output, + mark="sora", + total_frames=12, + detected_frames=10, + removed_frames=10, + remaining_metadata={}, + ), + ) + + result = CliRunner().invoke(main, ["video", "visible", str(source), "-o", str(output)]) + + assert result.exit_code == 0, result.output + assert "10/12 frames" in result.output