CalvinBackup/remove-ai-watermarks
1
0
Fork 0
mirror of https://github.com/wiltodelta/remove-ai-watermarks.git synced 2026-09-02 02:30:48 +02:00
Code Issues Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'origin/main' into docs/arxiv-paper-review

Browse Source 
# Conflicts:
#	.claude/settings.json
#	docs/installation.md
#	docs/supported-signals.md
#	docs/synthid.md
#	docs/verification-plan.md
#	docs/watermarking-landscape.md
#	pyproject.toml
#	src/remove_ai_watermarks/identify.py
#	uv.lock
This commit is contained in:
Victor Kuznetsov
2026-08-26 12:39:05 -07:00
parent a1c64a0b1d 29f17105de
commit aa74f0eff7
84 changed files with 6888 additions and 1666 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/remove_ai_watermarks/__init__.py
+1 -1
View File
@@ -33,7 +33,7 @@ _os.environ.setdefault("TRANSFORMERS_VERBOSITY", "error")
_warnings.filterwarnings("ignore", message=r".*ImageProcessorFast.*")
__version__ = "0.26.1"
__version__ = "0.31.1"
__all__ = [
"BatchSummary",
src/remove_ai_watermarks/_internal/c2pa.py
+364 -2
View File
@@ -8,6 +8,7 @@ import json
import logging
import re
import struct
from collections import deque
from dataclasses import dataclass
from pathlib import Path
from typing import TYPE_CHECKING, Any, cast
@@ -37,6 +38,41 @@ with contextlib.suppress(Exception):
_C2PA_READER_AVAILABLE = _C2paReader is not None
_PNG_HEADER = struct.Struct(">I4s")
_CONTENT_BINDING_MATCHES = (
"assertion.dataHash.match",
"assertion.bmffHash.match",
"assertion.boxesHash.match",
"assertion.collectionHash.match",
"assertion.multiAssetHash.match",
)
_CONTENT_BINDING_FAILURE_MARKERS = (
"Hash.incorrectFileCount",
"Hash.malformed",
"Hash.mismatch",
"Hash.missingPart",
"assertion.hashedURI.mismatch",
"claim.hardBindings.missing",
"hashedUri.mismatch",
"ingredient.manifest.mismatch",
)
_SIGNATURE_FAILURES = frozenset(
{
"claimSignature.mismatch",
"claimSignature.missing",
"claimSignature.outsideValidity",
}
)
# A credential the issuer itself has disowned, disqualifying like a hash mismatch.
# ``signingCredential.expired`` is deliberately absent: an expired certificate does not
# imply the signed bytes changed, and a signature actually made outside validity already
# arrives as ``claimSignature.outsideValidity`` above.
_CREDENTIAL_FAILURES = frozenset(
{
"signingCredential.invalid",
"signingCredential.ocsp.revoked",
}
)
@dataclass(frozen=True)
class _PngChunk:
@@ -185,6 +221,184 @@ def _active_manifest(store: dict[str, Any]) -> dict[str, Any]:
return cast("dict[str, Any]", active) if isinstance(active, dict) else {}
def _manifest_chain(store: dict[str, Any]) -> list[dict[str, Any]]:
"""Return the active manifest followed by credential-usable reachable ingredients."""
manifests = store.get("manifests")
active_label = store.get("active_manifest")
if not isinstance(manifests, dict) or not isinstance(active_label, str):
return []
typed_manifests = cast("dict[object, object]", manifests)
pending = deque([active_label])
seen: set[str] = set()
chain: list[dict[str, Any]] = []
while pending:
label = pending.popleft()
if label in seen:
continue
seen.add(label)
manifest_value = typed_manifests.get(label)
if not isinstance(manifest_value, dict):
continue
manifest = cast("dict[str, Any]", manifest_value)
chain.append(manifest)
ingredients = manifest.get("ingredients")
if not isinstance(ingredients, list):
continue
for ingredient_value in cast("list[object]", ingredients):
if not isinstance(ingredient_value, dict):
continue
ingredient = cast("dict[object, object]", ingredient_value)
child = ingredient.get("active_manifest")
if (
isinstance(child, str)
and child not in seen
and not _store_has_invalid_credential(cast("dict[str, Any]", ingredient))
):
pending.append(child)
return chain
def _status_codes(store: dict[str, Any]) -> tuple[list[str], list[str], list[str]]:
"""Return ordered status codes for this store's active credential."""
successes: list[str] = []
informationals: list[str] = []
failures: list[str] = []
def extend(status_set: object) -> None:
if not isinstance(status_set, dict):
return
mapping = cast("dict[object, object]", status_set)
for key, target in (
("success", successes),
("informational", informationals),
("failure", failures),
):
values = mapping.get(key)
if not isinstance(values, list):
continue
for value in cast("list[object]", values):
if not isinstance(value, dict):
continue
code = cast("dict[object, object]", value).get("code")
if isinstance(code, str) and code:
target.append(code)
validation_results = store.get("validation_results")
if isinstance(validation_results, dict):
results = cast("dict[object, object]", validation_results)
extend(results.get("activeManifest"))
if not (successes or informationals or failures):
# Older readers expose only the flat non-success list. It still carries the
# exact codes needed to reject a mismatched binding or invalid credential.
values = store.get("validation_status")
if isinstance(values, list):
for value in cast("list[object]", values):
if not isinstance(value, dict):
continue
code = cast("dict[object, object]", value).get("code")
if isinstance(code, str) and code:
failures.append(code)
return successes, informationals, failures
def _store_has_invalid_credential(store: dict[str, Any]) -> bool:
"""Return whether this store's own active credential failed validation.
Deliberately routed through the same codes -> dimensions -> disqualified path the
report uses, rather than re-testing the raw codes here. When this walk classified
failures on its own, adding one rule meant editing it and
:func:`c2pa_info_has_invalid_credential` in lockstep, and the ingredient
reachability walk was free to drift away from the verdict it feeds.
"""
return c2pa_info_has_invalid_credential(_validation_fields(store))
def c2pa_info_has_invalid_credential(info: dict[str, Any]) -> bool:
"""Return whether parsed C2PA info reports a failed binding, signature, or credential."""
return (
info.get("c2pa_integrity") == "invalid"
or info.get("c2pa_signature") == "invalid"
or info.get("c2pa_signer_validity") == "invalid"
)
def c2pa_info_has_invismark(info: dict[str, Any]) -> bool:
"""Return whether parsed C2PA info declares Microsoft InvisMark."""
soft_bindings = info.get("soft_binding_vendors")
return isinstance(soft_bindings, list) and "Microsoft InvisMark" in soft_bindings
def c2pa_info_has_removal_hint(info: dict[str, Any]) -> bool:
"""Return whether a C2PA AI or watermark claim should keep removal fail-safe."""
return bool(info.get("ai_source_kind") or info.get("synthid_watermark") or c2pa_info_has_invismark(info))
def _has_suffix(codes: list[str], suffixes: tuple[str, ...]) -> bool:
return any(code.endswith(suffix) for code in codes for suffix in suffixes)
def _validation_fields(store: dict[str, Any]) -> dict[str, Any]:
successes, informationals, failures = _status_codes(store)
all_codes = list(dict.fromkeys([*successes, *informationals, *failures]))
disqualifying = [
code
for code in failures
if any(marker in code for marker in _CONTENT_BINDING_FAILURE_MARKERS)
or code in _SIGNATURE_FAILURES
or code in _CREDENTIAL_FAILURES
]
binding_failed = any(marker in code for code in disqualifying for marker in _CONTENT_BINDING_FAILURE_MARKERS)
binding_matched = _has_suffix(successes, _CONTENT_BINDING_MATCHES)
signature_failed = any(code in _SIGNATURE_FAILURES for code in disqualifying)
signature_validated = "claimSignature.validated" in successes
if binding_failed:
integrity = "invalid"
elif binding_matched:
integrity = "valid"
else:
integrity = "unknown"
if signature_failed:
signature = "invalid"
elif signature_validated:
signature = "valid"
else:
signature = "unknown"
if "signingCredential.trusted" in successes:
signer_trust = "trusted"
elif "signingCredential.untrusted" in failures:
signer_trust = "untrusted"
else:
signer_trust = "unknown"
if any(code in _CREDENTIAL_FAILURES for code in disqualifying):
signer_validity = "invalid"
elif "signingCredential.expired" in failures:
signer_validity = "expired"
elif "claimSignature.insideValidity" in successes:
signer_validity = "valid"
else:
signer_validity = "unknown"
state = store.get("validation_state")
return {
"c2pa_validation_source": "reader",
"c2pa_validation_state": state if isinstance(state, str) and state else "unknown",
"c2pa_integrity": integrity,
"c2pa_signature": signature,
"c2pa_signer_trust": signer_trust,
"c2pa_signer_validity": signer_validity,
"c2pa_validation_codes": all_codes,
# The subset of failures that actually drove a dimension to invalid. Callers
# rendering "why" must use this rather than re-classifying the full code list,
# or the reason shown and the verdict reached stop being the same rule.
"c2pa_failed_codes": disqualifying,
}
def _claim_generator_from_store(store: dict[str, Any]) -> str | None:
active = _active_manifest(store)
direct = active.get("claim_generator")
@@ -199,6 +413,141 @@ def _claim_generator_from_store(store: dict[str, Any]) -> str | None:
return None
def _structured_manifest_fields(store: dict[str, Any]) -> dict[str, Any]:
"""Extract provenance only from the active manifest and its ingredient graph."""
chain = _manifest_chain(store)
if not chain:
return {}
info: dict[str, Any] = {}
issuers: list[str] = []
tools: list[str] = []
actions: list[str] = []
source_types: list[str] = []
soft_binding_algorithms: list[str] = []
soft_binding_values: list[str] = []
claim_generator_asserts_ai = False
def add_tool_matches(value: str, *, asserts_ai: bool = False) -> None:
nonlocal claim_generator_asserts_ai
matches = _ordered_matches(value.encode(), C2PA_AI_TOOLS)
tools.extend(matches)
if asserts_ai and matches:
claim_generator_asserts_ai = True
for manifest in chain:
signature_value = manifest.get("signature_info")
if isinstance(signature_value, dict):
signature = cast("dict[object, object]", signature_value)
for key in ("issuer", "common_name", "certificate_issuer"):
value = signature.get(key)
if isinstance(value, str):
issuers.extend(_ordered_matches(value.encode(), C2PA_ISSUERS))
direct_generator = manifest.get("claim_generator")
if isinstance(direct_generator, str):
add_tool_matches(direct_generator, asserts_ai=True)
candidates = manifest.get("claim_generator_info")
if isinstance(candidates, list):
for candidate_value in cast("list[object]", candidates):
if not isinstance(candidate_value, dict):
continue
name = cast("dict[object, object]", candidate_value).get("name")
if isinstance(name, str):
add_tool_matches(name, asserts_ai=True)
assertions = manifest.get("assertions")
if not isinstance(assertions, list):
continue
for assertion_value in cast("list[object]", assertions):
if not isinstance(assertion_value, dict):
continue
assertion = cast("dict[object, object]", assertion_value)
label = assertion.get("label")
data = assertion.get("data")
if isinstance(label, str) and label.startswith("c2pa.soft-binding") and isinstance(data, dict):
soft_binding = cast("dict[object, object]", data)
algorithm = soft_binding.get("alg")
if isinstance(algorithm, str):
soft_binding_algorithms.append(algorithm)
blocks = soft_binding.get("blocks")
if isinstance(blocks, list):
for block_value in cast("list[object]", blocks):
if not isinstance(block_value, dict):
continue
value = cast("dict[object, object]", block_value).get("value")
if isinstance(value, str) and value.isprintable() and len(value) <= 256:
soft_binding_values.append(value)
if not (isinstance(label, str) and label.startswith("c2pa.actions") and isinstance(data, dict)):
continue
action_values = cast("dict[object, object]", data).get("actions")
if not isinstance(action_values, list):
continue
for action_value in cast("list[object]", action_values):
if not isinstance(action_value, dict):
continue
action = cast("dict[object, object]", action_value)
action_name = action.get("action")
if isinstance(action_name, str):
actions.append(C2PA_ACTIONS.get(action_name.encode(), action_name.removeprefix("c2pa.")))
source_type = action.get("digitalSourceType")
if isinstance(source_type, str):
source_types.append(source_type)
software_agent = action.get("softwareAgent")
if isinstance(software_agent, dict):
name = cast("dict[object, object]", software_agent).get("name")
if isinstance(name, str):
add_tool_matches(name)
issuers = list(dict.fromkeys(issuers))
tools = list(dict.fromkeys(tools))
actions = list(dict.fromkeys(actions))
if issuers:
info["issuer"] = ", ".join(issuers)
if tools:
info["ai_tool"] = ", ".join(tools)
if actions:
info["actions"] = ", ".join(actions)
if claim_generator_asserts_ai:
info["c2pa_identity_ai"] = True
generated = any(
"trainedAlgorithmicMedia" in value and "compositeWithTrainedAlgorithmicMedia" not in value
for value in source_types
)
enhanced = any("compositeWithTrainedAlgorithmicMedia" in value for value in source_types)
procedural = any("algorithmicMedia" in value for value in source_types)
if generated:
info.update(source_type="trainedAlgorithmicMedia (AI-generated)", ai_source_kind="generated")
elif enhanced:
info.update(source_type="compositeWithTrainedAlgorithmicMedia (AI-enhanced)", ai_source_kind="enhanced")
elif procedural:
info["source_type"] = "algorithmicMedia"
has_watermark_action = any(action.startswith("watermarked") for action in actions)
if has_watermark_action:
info["watermarked"] = True
if info.get("ai_source_kind"):
selected_bytes = json.dumps(chain, ensure_ascii=False).encode()
synthid = synthid_evidence_vendors_in(selected_bytes, has_watermark_action=has_watermark_action)
if synthid:
info["synthid_vendors"] = synthid
info["synthid_watermark"] = synthid_verdict(", ".join(synthid))
if soft_binding_algorithms:
soft_binding_algorithms = list(dict.fromkeys(soft_binding_algorithms))
algorithms = "\n".join(soft_binding_algorithms).encode()
soft_bindings = soft_binding_vendors_in(algorithms)
if soft_bindings:
info["soft_binding_vendors"] = soft_bindings
info["soft_binding"] = ", ".join(soft_bindings)
info["soft_binding_algorithm"] = ", ".join(soft_binding_algorithms)
if soft_binding_values:
info["soft_binding_value"] = ", ".join(dict.fromkeys(soft_binding_values))
return info
def synthid_verdict(vendors: str) -> str:
"""Describe why supported provenance establishes a SynthID watermark."""
return f"present according to {vendors} provenance"
@@ -276,16 +625,29 @@ def _populate_registry_fields(buffer: bytes, info: dict[str, Any]) -> bool:
def _base_info(byte_count: int, *, fallback: bool = False) -> dict[str, Any]:
container = "C2PA manifest" if fallback else "C2PA manifest store"
return {
info: dict[str, Any] = {
"has_c2pa": True,
"type": "C2PA (Coalition for Content Provenance and Authenticity)",
"c2pa_manifest": f"{container} ({byte_count} bytes)",
}
if fallback:
info.update(
c2pa_validation_source="fallback",
c2pa_validation_state="unknown",
c2pa_integrity="unknown",
c2pa_signature="unknown",
c2pa_signer_trust="unknown",
c2pa_signer_validity="unknown",
c2pa_validation_codes=[],
c2pa_failed_codes=[],
)
return info
def _info_from_store(store: dict[str, Any], encoded: bytes) -> dict[str, Any]:
info = _base_info(len(encoded))
_populate_registry_fields(encoded, info)
info.update(_structured_manifest_fields(store))
info.update(_validation_fields(store))
generator = _claim_generator_from_store(store)
if generator is not None:
info["claim_generator"] = generator
src/remove_ai_watermarks/_internal/constants.py
+21 -5
View File
@@ -65,7 +65,7 @@ def _vendor(
# Order is product priority when a manifest mentions more than one organization.
C2PA_AI_VENDORS: tuple[C2paAiVendor, ...] = (
_vendor(b"Microsoft", "Microsoft", "Microsoft (Bing Image Creator / Designer)", "Microsoft"),
_vendor(b"Microsoft", "Microsoft", "Microsoft (Copilot / Designer)", "Microsoft"),
_vendor(b"Adobe", "Adobe", "Adobe Firefly", "Adobe"),
_vendor(
b"OpenAI",
@@ -78,18 +78,25 @@ C2PA_AI_VENDORS: tuple[C2paAiVendor, ...] = (
_vendor(b"Google", "Google LLC", "Google (Gemini / Imagen)", "Google", synthid=True),
_vendor(b"Stability AI", "Stability AI", "Stability AI (Stable Image / DreamStudio)", "Stability AI"),
_vendor(b"Black Forest Labs", "Black Forest Labs", "Black Forest Labs (FLUX)", "Black Forest Labs"),
_vendor(b"volcengine", "ByteDance (Volcano Engine)", "ByteDance (Doubao / Jimeng / Volcano Engine)", "ByteDance"),
_vendor(
b"volcengine",
"ByteDance (Volcano Engine)",
"ByteDance (Doubao / Jimeng / Dreamina / Volcano Engine)",
"ByteDance",
),
_vendor(
"北京火山引擎科技有限公司",
"ByteDance (Volcano Engine)",
"ByteDance (Doubao / Jimeng / Volcano Engine)",
"ByteDance (Doubao / Jimeng / Dreamina / Volcano Engine)",
"ByteDance",
),
_vendor(b"Byteplus", "BytePlus (ByteDance)", "ByteDance (Doubao / Jimeng / Volcano Engine)", "ByteDance"),
_vendor(
b"Byteplus", "BytePlus (ByteDance)", "ByteDance (Doubao / Jimeng / Dreamina / Volcano Engine)", "ByteDance"
),
_vendor(
b"Dreamina",
"ByteDance (Dreamina)",
"ByteDance (Doubao / Jimeng / Volcano Engine)",
"ByteDance (Doubao / Jimeng / Dreamina / Volcano Engine)",
"ByteDance",
asserts_ai=True,
),
@@ -107,6 +114,9 @@ C2PA_IDENTITY_AI_ORGS = frozenset(vendor.org for vendor in C2PA_AI_VENDORS if ve
# Keep this attribution beside the issuer registry so every C2PA consumer has one
# canonical source rather than maintaining a derived product map in identify.py.
C2PA_CLAIM_GENERATOR_PLATFORMS: tuple[tuple[str, str], ...] = (
("adobe_firefly", "Adobe Firefly"),
("firefly", "Adobe Firefly"),
("dreamina", "ByteDance (Doubao / Jimeng / Dreamina / Volcano Engine)"),
("higgsfield ai", "Higgsfield AI"),
("topaz labs image api", "Topaz Labs"),
("tiktok ad creative toolbox", "TikTok Ad Creative Toolbox"),
@@ -122,6 +132,7 @@ C2PA_AI_TOOLS = {
("DALL", "DALL-E"),
("Imagen", "Imagen"),
("Firefly", "Firefly"),
("Dreamina", "Dreamina"),
)
}
@@ -167,6 +178,11 @@ AI_GENERATOR_TOKENS = frozenset(
"dreamstudio",
"novelai",
"reve.com",
# Luma AI stamps PNG tEXt Source="Luma AI" / Comment="Generated by
# Luma AI's Uni-1 model (https://lumalabs.ai)"; the space-bearing token
# avoids matching incidental "luma" runs (luma/chroma key names etc.).
"luma ai",
"lumalabs",
"aphrodite ai",
"apple photos clean up",
"fal-ai",
src/remove_ai_watermarks/_internal/isobmff.py
+110 -55
View File
@@ -12,7 +12,10 @@ so pixel, video, and audio 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.
place so fast-start media offsets remain valid. Two serialization variants are
covered: the ISO form (``meta`` as a FullBox under ``udta``) and the QuickTime
form Doubao's iOS export writes (a bare ``meta`` box as a direct ``moov`` child,
no FullBox header).
This file intentionally avoids dependencies on format-specific libraries
(pillow-heif, pillow-jxl, pymp4) so it works on systems where they aren't
@@ -184,68 +187,116 @@ def _tc260_key_indices(
return found
# The box types a TC260-bearing ``meta`` box opens with or contains: ISO files
# have ``hdlr`` then ``keys``/``ilst``; QuickTime metadata lists have ``hdlr``
# and ``ilst`` only. Which payload offset (0 vs 4) yields such children is what
# disambiguates the QuickTime form (no FullBox header) from the ISO one.
_META_CHILD_TYPES = frozenset({b"hdlr", b"keys", b"ilst"})
def _meta_child_boxes(
stream: BinaryIO,
meta_payload: int,
meta_end: int,
) -> Iterator[tuple[int, int, bytes, int]]:
"""Yield the child boxes of one ``meta`` box in either serialized form.
ISO serializes ``meta`` as a FullBox, so its children start 4 bytes into
the payload; QuickTime writes a bare box, so they start at 0. Both real
forms open with a recognized child (``hdlr``), so the form is picked by
which offset's first box type is one of ``_META_CHILD_TYPES``; a wrong
probe reads garbage header bytes that match no known type.
"""
for offset in (0, 4):
boxes = iter_file_boxes(stream, meta_payload + offset, meta_end)
first = next(boxes, None)
if first is not None and first[2] in _META_CHILD_TYPES:
yield first
yield from boxes
return
def _iter_tc260_meta_boxes(
stream: BinaryIO,
moov_payload: int,
moov_end: int,
) -> Iterator[tuple[int, int]]:
"""Yield ``(payload, end)`` of every ``meta`` box that may hold a TC260 label.
The normative ISO placement is ``moov.udta.meta``; Doubao's iOS MOV export
instead stores the label in a QuickTime-form ``meta`` box that hangs
directly off ``moov``. Both are yielded so one consumer covers them.
"""
for _start, end, box_type, payload in iter_file_boxes(stream, moov_payload, moov_end):
if box_type == b"udta":
for _udta_start, udta_end, udta_type, udta_payload in iter_file_boxes(stream, payload, end):
if udta_type == b"meta":
yield udta_payload, udta_end
elif box_type == b"meta":
yield payload, end
def _tc260_aigc_regions(
stream: BinaryIO,
file_size: int,
) -> list[tuple[int, int, int, int, bytes]]:
) -> list[tuple[tuple[int, int] | None, int, int, bytes]]:
"""Locate validated native TC260 entries without reading media payloads.
Each tuple is ``(key_start, key_end, value_start, value_end, value)``.
Each tuple is ``(key_span, value_start, value_end, value)``; ``key_span`` is
the byte span of the ``AIGC`` key when the normative ``keys`` box maps the
item, or None for the QuickTime metadata-list form (``hdlr=mdir``) Doubao's
iOS export writes, where the JSON sits in a bare ``ilst`` data item with no
key name to blank.
"""
regions: list[tuple[int, int, int, int, bytes]] = []
regions: list[tuple[tuple[int, int] | None, 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(
for meta_payload, meta_end in _iter_tc260_meta_boxes(stream, moov_payload, moov_end):
keys: dict[int, tuple[int, int]] = {}
ilst_boxes: list[tuple[int, int]] = []
keyed = False
for _child_start, child_end, child_type, child_payload in _meta_child_boxes(
stream,
udta_payload,
udta_end,
meta_payload,
meta_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(
if child_type == b"keys":
keyed = True
keys.update(_tc260_key_indices(stream, child_payload, child_end))
elif child_type == b"ilst":
ilst_boxes.append((child_payload, child_end))
if not ilst_boxes:
continue
for ilst_payload, ilst_end in ilst_boxes:
for _item_start, item_end, item_type, item_payload in iter_file_boxes(
stream,
meta_payload + 4,
meta_end,
ilst_payload,
ilst_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(
index = int.from_bytes(item_type, "big")
key_span = keys.get(index)
if keyed and key_span is None:
# A keyed (ISO) meta box maps items through ``keys``;
# an unmapped index is not an AIGC entry, and reading
# its value would pull arbitrary metadata (e.g. cover
# art) through the JSON parser on every scan. Only the
# keyless QuickTime list falls through to content
# validation below.
continue
for _data_start, data_end, data_type, data_payload in iter_file_boxes(
stream,
ilst_payload,
ilst_end,
item_payload,
item_end,
):
index = int.from_bytes(item_type, "big")
key_span = keys.get(index)
if key_span is None:
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
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 parse_tc260_aigc_json(value) is not None:
regions.append((*key_span, value_start, data_end, value))
stream.seek(value_start)
value = stream.read(value_size)
if len(value) == value_size and parse_tc260_aigc_json(value) is not None:
regions.append((key_span, value_start, data_end, value))
return regions
@@ -257,7 +308,7 @@ def tc260_aigc_payloads(path: str | Path) -> tuple[bytes, ...]:
return ()
stream.seek(0, 2)
file_size = stream.tell()
return tuple(region[4] for region in _tc260_aigc_regions(stream, file_size))
return tuple(region[3] for region in _tc260_aigc_regions(stream, file_size))
except OSError:
return ()
@@ -268,6 +319,8 @@ def blank_tc260_aigc_tags(data: bytes) -> tuple[bytes, int]:
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.
A keyless QuickTime metadata-list entry has no key name, so only its value
is blanked.
"""
if not is_isobmff(data):
return data, 0
@@ -276,9 +329,10 @@ def blank_tc260_aigc_tags(data: bytes) -> tuple[bytes, int]:
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"
for key_span, value_start, value_end, _value in regions:
if key_span is not None:
key_spans.add(key_span)
out[key_span[0] : key_span[1]] = b"free"
out[value_start:value_end] = b" " * (value_end - value_start)
return bytes(out), len(key_spans)
@@ -456,7 +510,7 @@ def strip_isobmff_media_file(
max_scan=max_box_scan,
)
tc260_regions = _tc260_aigc_regions(stream, file_size) if targets is not None else []
tc260_key_spans = {(region[0], region[1]) for region in tc260_regions}
tc260_key_spans = {region[0] for region in tc260_regions if region[0] is not None}
with atomic_video_output(output_path) as temporary_path:
with source_path.open("rb") as source_stream, temporary_path.open("r+b") as temporary:
@@ -466,9 +520,10 @@ def strip_isobmff_media_file(
temporary.seek(box_start + 4)
temporary.write(b"free")
_overwrite_range(temporary, payload_start, box_end, byte=b"\x00")
for key_start, _key_end, value_start, value_end, _value in tc260_regions:
temporary.seek(key_start)
temporary.write(b"free")
for key_span, value_start, value_end, _value in tc260_regions:
if key_span is not None:
temporary.seek(key_span[0])
temporary.write(b"free")
_overwrite_range(temporary, value_start, value_end, byte=b" ")
temporary.flush()
os.fsync(temporary.fileno())
src/remove_ai_watermarks/_internal/qwen_zimage_pipeline.py
+82 -10
View File
@@ -30,6 +30,8 @@ from remove_ai_watermarks._internal.watermark_profiles import resolve_seed
if TYPE_CHECKING:
from collections.abc import Callable
from remove_ai_watermarks._internal.text_restoration import VerifiedTextManifest
log = logging.getLogger(__name__)
QWEN_IMAGE_2512_MODEL_ID = "Qwen/Qwen-Image-2512"
@@ -973,6 +975,30 @@ class QwenZImagePipeline:
result = result.resize(image.size, Image.Resampling.LANCZOS)
return result.convert("RGB")
def _qwen_vae_roundtrip(self, image: Image.Image) -> Image.Image:
"""Reconstruct source pixels through the already loaded Qwen VAE."""
import torch
pipe, _controlnet_input_cls = self._load_qwen()
source_width, source_height = image.size
pad_width = (-source_width) % 8
pad_height = (-source_height) % 8
padded = image.convert("RGB")
if pad_width or pad_height:
padded = Image.fromarray(
np.pad(
np.asarray(padded),
((0, pad_height), (0, pad_width), (0, 0)),
mode="edge",
)
)
pipe.load_models_to_device(["vae"])
tensor = pipe.preprocess_image(padded).to(device=self.device, dtype=self.torch_dtype)
with torch.inference_mode():
latents = pipe.vae.encode(tensor)
decoded = pipe.vae.decode(latents)
return pipe.vae_output_to_image(decoded).crop((0, 0, source_width, source_height)).convert("RGB")
@staticmethod
def _detail_size(
crop_size: tuple[int, int],
@@ -1044,10 +1070,33 @@ class QwenZImagePipeline:
tile: bool = False,
tile_size: int = 1024,
tile_overlap: int = 128,
text_manifest: VerifiedTextManifest | None = None,
fidelity_anchor: bool = False,
) -> Image.Image:
"""Execute global regeneration and masked face repair."""
self._require_cuda()
seed = resolve_seed(seed)
donor = None
if text_manifest is not None:
self._progress("Reconstructing the verified text donor with the Qwen VAE...")
if tile and max(image.size) > tile_size:
from remove_ai_watermarks._internal.tiling import run_tiled
donor = run_tiled(
self._qwen_vae_roundtrip,
image,
tile_size,
tile_overlap,
lambda message: self._progress(
message.replace(
"Tiled diffusion",
"Reconstructing the verified text donor",
1,
)
),
)
else:
donor = self._qwen_vae_roundtrip(image)
global_strength = (
resolution_adaptive_denoise(image.width, image.height) if strength is None else float(strength)
)
@@ -1068,14 +1117,37 @@ class QwenZImagePipeline:
boxes = detect_faces(image)
if not boxes:
self._progress("No faces detected; keeping the Qwen global result.")
return global_result
masks = self._sam_masks(image, boxes)
face_strength = largest_face_denoise(boxes, image.size) * FACE_DENOISE_SCALE
return self._run_faces(
image,
global_result,
boxes,
masks,
strength=face_strength,
seed=seed,
result = global_result
else:
masks = self._sam_masks(image, boxes)
face_strength = largest_face_denoise(boxes, image.size) * FACE_DENOISE_SCALE
result = self._run_faces(
image,
global_result,
boxes,
masks,
strength=face_strength,
seed=seed,
)
if text_manifest is None:
return result
if donor is None:
raise RuntimeError("Verified text restoration requires a Qwen-VAE donor")
from remove_ai_watermarks._internal.text_restoration import (
blend_fidelity_anchor,
restore_verified_text,
)
# The anchor is OFF by default since 0.27.1: blending 15% of the Qwen-VAE
# donor ACROSS THE WHOLE FRAME returned detector-visible OpenAI SynthID on
# poster-scale manifests (official Content Provenance API, 2026-08-19:
# detected x6 with the anchor, clean x6 without it, base clean throughout;
# see docs/text-protection-research.md). ``fidelity_anchor=True`` keeps the
# 0.27.0 research behavior for reproduction.
if fidelity_anchor:
self._progress("Blending the Qwen-VAE fidelity anchor...")
anchor = blend_fidelity_anchor(result, donor)
else:
anchor = result
self._progress(f"Restoring {len(text_manifest.lines)} verified text lines...")
return restore_verified_text(image, anchor, donor, text_manifest.lines)
src/remove_ai_watermarks/_internal/text_restoration.py
+446
View File
@@ -0,0 +1,446 @@
"""Opt-in restoration of verified text from a Qwen VAE reconstruction."""
# 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
from __future__ import annotations
import hashlib
import json
import math
from dataclasses import dataclass
from typing import TYPE_CHECKING, Any
import cv2
import numpy as np
from PIL import Image
from remove_ai_watermarks._internal.schema import require_schema_version
if TYPE_CHECKING:
from collections.abc import Sequence
from pathlib import Path
from numpy.typing import NDArray
TEXT_MANIFEST_SCHEMA = 2
_SUPPORTED_TEXT_MANIFEST_SCHEMAS = frozenset({1, TEXT_MANIFEST_SCHEMA})
FIDELITY_BLEND_ALPHA = 0.15
GLYPH_FEATHER = 0.5
GLYPH_SIDE_PAD_RATIO = 0.12
GLYPH_SIDE_PAD_LIMIT = 1.0
@dataclass(frozen=True)
class VerifiedTextLine:
"""One operator-verified source region in source-pixel coordinates."""
box: tuple[int, int, int, int]
text: str | None = None
script: str | None = None
angle: float = 0.0
@dataclass(frozen=True)
class VerifiedTextManifest:
"""Verified text regions cryptographically bound to one decoded RGB source."""
source_pixel_sha256: str
width: int
height: int
lines: tuple[VerifiedTextLine, ...]
def source_pixel_sha256(image: Image.Image) -> str:
"""Hash decoded RGB geometry and bytes, independent of container metadata."""
rgb = image.convert("RGB")
digest = hashlib.sha256()
digest.update(rgb.width.to_bytes(8, "big"))
digest.update(rgb.height.to_bytes(8, "big"))
digest.update(rgb.tobytes())
return digest.hexdigest()
def load_verified_text_manifest(path: Path, source: Image.Image) -> VerifiedTextManifest:
"""Load and validate an operator-verified manifest for exactly ``source``."""
try:
payload = json.loads(path.read_text(encoding="utf-8"))
except (OSError, json.JSONDecodeError) as exc:
raise ValueError(f"Cannot read text manifest {path}: {exc}") from exc
if not isinstance(payload, dict):
raise ValueError("Text manifest must be a JSON object")
schema_version = require_schema_version(
payload.get("schema_version"),
contract="text manifest",
supported=_SUPPORTED_TEXT_MANIFEST_SCHEMAS,
)
if payload.get("verified") is not True:
raise ValueError("Text manifest must contain verified=true after operator verification")
rgb = source.convert("RGB")
width = _manifest_integer(payload, "width")
height = _manifest_integer(payload, "height")
if (width, height) != rgb.size:
raise ValueError(f"Text manifest dimensions {width}x{height} do not match source {rgb.width}x{rgb.height}")
expected_hash = payload.get("source_pixel_sha256")
if not isinstance(expected_hash, str) or len(expected_hash) != 64:
raise ValueError("Text manifest source_pixel_sha256 must be a 64-character SHA-256")
actual_hash = source_pixel_sha256(rgb)
if expected_hash.casefold() != actual_hash:
raise ValueError("Text manifest source_pixel_sha256 does not match the decoded source pixels")
raw_lines = payload.get("lines")
if not isinstance(raw_lines, list) or not raw_lines:
raise ValueError("Text manifest lines must be a non-empty list")
lines = tuple(_load_line(item, width, height, index, schema_version) for index, item in enumerate(raw_lines))
if list(lines) != sorted(lines, key=lambda line: (line.box[1], line.box[0])):
raise ValueError("Text manifest lines must be in top-to-bottom, left-to-right reading order")
return VerifiedTextManifest(actual_hash, width, height, lines)
def _manifest_integer(payload: dict[str, Any], key: str) -> int:
value = payload.get(key)
if isinstance(value, bool) or not isinstance(value, int) or value <= 0:
raise ValueError(f"Text manifest {key} must be a positive integer")
return value
def _load_line(item: Any, width: int, height: int, index: int, schema_version: int) -> VerifiedTextLine:
if not isinstance(item, dict):
raise ValueError(f"Text manifest line {index} must be an object")
raw_box = item.get("box")
if (
not isinstance(raw_box, list)
or len(raw_box) != 4
or any(isinstance(value, bool) or not isinstance(value, int) for value in raw_box)
):
raise ValueError(f"Text manifest line {index} box must contain four integers")
box = tuple(raw_box)
x1, y1, x2, y2 = box
if not (0 <= x1 < x2 <= width and 0 <= y1 < y2 <= height):
raise ValueError(f"Text manifest line {index} box is outside the source dimensions")
text: str | None = None
script: str | None = None
if schema_version == 1:
text = item.get("text")
script = item.get("script")
if not isinstance(text, str) or not text.strip():
raise ValueError(f"Text manifest line {index} text must be non-empty")
if not isinstance(script, str) or not script.strip():
raise ValueError(f"Text manifest line {index} script must be non-empty")
angle_value = item.get("angle", 0.0)
if isinstance(angle_value, bool) or not isinstance(angle_value, int | float):
raise ValueError(f"Text manifest line {index} angle must be numeric")
angle = float(angle_value)
if not math.isfinite(angle) or abs(angle) > 30.0:
raise ValueError(f"Text manifest line {index} angle must be between -30 and 30 degrees")
return VerifiedTextLine(box, text, script, angle)
def blend_fidelity_anchor(clean: Image.Image, donor: Image.Image) -> Image.Image:
"""Blend 15% Qwen-VAE reconstruction into the oracle-clean pipeline output."""
clean_rgb = np.asarray(clean.convert("RGB"), dtype=np.float32)
donor_rgb = np.asarray(donor.convert("RGB"), dtype=np.float32)
if clean_rgb.shape != donor_rgb.shape:
raise ValueError("Clean result and Qwen-VAE donor dimensions must match")
blended = np.rint(clean_rgb * (1.0 - FIDELITY_BLEND_ALPHA) + donor_rgb * FIDELITY_BLEND_ALPHA)
return Image.fromarray(np.clip(blended, 0, 255).astype(np.uint8))
def restore_verified_text(
source: Image.Image,
candidate: Image.Image,
donor: Image.Image,
lines: tuple[VerifiedTextLine, ...],
) -> Image.Image:
"""Erase candidate glyphs, then composite verified Qwen-VAE glyph cores."""
from remove_ai_watermarks import region_eraser
if not region_eraser.lama_available():
raise RuntimeError(
"Verified text restoration requires LaMa. Install: pip install 'remove-ai-watermarks[text-restoration]'"
)
source_rgb = np.asarray(source.convert("RGB"))
candidate_rgb = np.asarray(candidate.convert("RGB"))
donor_rgb = np.asarray(donor.convert("RGB"))
if source_rgb.shape != candidate_rgb.shape or source_rgb.shape != donor_rgb.shape:
raise ValueError("Source, candidate, and Qwen-VAE donor dimensions must match")
source_masks = [source_silhouette_mask(source_rgb, line.box, line.angle) for line in lines]
for index, mask in enumerate(source_masks):
if not np.any(mask):
raise ValueError(f"Verified text line {index} produced no source glyph pixels")
candidate_masks = [source_silhouette_mask(candidate_rgb, line.box, line.angle) for line in lines]
erase_masks = []
for line, source_mask, candidate_mask in zip(lines, source_masks, candidate_masks, strict=True):
radius = 5 if line.box[3] - line.box[1] >= 48 else 3
kernel = cv2.getStructuringElement(cv2.MORPH_ELLIPSE, (2 * radius + 1,) * 2)
erase_masks.append(cv2.dilate(np.maximum(source_mask, candidate_mask), kernel))
del candidate_masks
groups = group_text_lines(lines)
background = cv2.cvtColor(candidate_rgb, cv2.COLOR_RGB2BGR)
for group in groups:
background = region_eraser.erase_lama(background, np.maximum.reduce([erase_masks[index] for index in group]))
background_rgb = cv2.cvtColor(background, cv2.COLOR_BGR2RGB)
residual_masks = [
residual_glyph_mask(background_rgb, mask, line.box) for line, mask in zip(lines, erase_masks, strict=True)
]
for group in groups:
residual = np.maximum.reduce([residual_masks[index] for index in group])
if np.any(residual):
background = region_eraser.erase_lama(background, residual)
del erase_masks, residual_masks
restored = cv2.cvtColor(background, cv2.COLOR_BGR2RGB)
restored = composite_fresh_text_edges(source_rgb, restored, lines, source_masks)
source_glyph_mask = np.maximum.reduce(source_masks)
restored = composite_reconstructed_glyphs(donor_rgb, restored, source_glyph_mask)
return Image.fromarray(restored)
def _glyph_crop_box(
box: tuple[int, int, int, int],
width: int,
height: int,
left_pad: int,
right_pad: int,
) -> tuple[int, int, int, int]:
"""Widen the detector box so glyph edges stay inside the silhouette crop.
Detector boxes can stop inside a leading flourish or trailing punctuation,
and Paddle line boxes sit 2-5 px above the true ink bottom on the poster
fixtures. The crop is otherwise the box itself, so those pixels never reach
the donor composite. Start 12% sideways, expand each side independently
while foreground reaches its boundary, and pad 8% up and 25% down.
"""
x1, y1, x2, y2 = box
line_h = max(1, y2 - y1)
pad_top = max(1, round(line_h * 0.08))
pad_bot = max(2, round(line_h * 0.25))
return (
max(0, x1 - left_pad),
max(0, y1 - pad_top),
min(width, x2 + right_pad),
min(height, y2 + pad_bot),
)
def _silhouette_crop_mask(
source_rgb: NDArray[Any],
crop_box: tuple[int, int, int, int],
angle: float,
) -> NDArray[Any]:
"""Threshold one candidate crop so its horizontal boundaries can be tested."""
height, width = source_rgb.shape[:2]
x1, y1, x2, y2 = crop_box
gray = cv2.cvtColor(source_rgb[y1:y2, x1:x2], cv2.COLOR_RGB2GRAY)
support = np.ones(gray.shape, dtype=np.uint8)
if angle:
box_width, box_height = x2 - x1, y2 - y1
theta = math.radians(abs(angle))
cosine, sine = math.cos(theta), math.sin(theta)
denominator = cosine * cosine - sine * sine
rect_width = (box_width * cosine - box_height * sine) / denominator
rect_height = (box_height * cosine - box_width * sine) / denominator
rotated = cv2.boxPoints(
((box_width / 2, box_height / 2), (max(1.0, rect_width * 0.92), max(1.0, rect_height * 0.62)), -angle)
)
support.fill(0)
cv2.fillConvexPoly(support, np.rint(rotated).astype(np.int32), 1)
values = gray[support > 0]
background_luma = float(np.median(values))
else:
ring_pad = max(6, min(20, (y2 - y1) // 4))
rx1, ry1, rx2, ry2 = _clip_box(crop_box, width, height, pad=ring_pad)
context = cv2.cvtColor(source_rgb[ry1:ry2, rx1:rx2], cv2.COLOR_RGB2GRAY)
ring = np.ones(context.shape, dtype=bool)
ring[y1 - ry1 : y2 - ry1, x1 - rx1 : x2 - rx1] = False
background_luma = float(np.median(context[ring])) if ring.any() else float(np.median(gray))
values = gray.reshape(-1)
low, high = float(np.percentile(values, 2)), float(np.percentile(values, 98))
dark_contrast, light_contrast = background_luma - low, high - background_luma
threshold = max(16.0, min(56.0, max(light_contrast, dark_contrast) * 0.22))
if light_contrast > dark_contrast:
crop_mask = (gray.astype(np.float32) >= background_luma + threshold).astype(np.uint8) * 255
else:
crop_mask = (gray.astype(np.float32) <= background_luma - threshold).astype(np.uint8) * 255
crop_mask[support == 0] = 0
return crop_mask
def _anchored_components_reach_sides(
crop_mask: NDArray[Any],
anchor_box: tuple[int, int, int, int],
) -> tuple[bool, bool]:
"""Whether anchored foreground reaches the left and right crop sides."""
count, labels, stats, _centroids = cv2.connectedComponentsWithStats(crop_mask, 8)
x1, y1, x2, y2 = anchor_box
anchor_counts = np.bincount(labels[y1:y2, x1:x2].reshape(-1), minlength=count)
anchor_counts[0] = 0
def reaches(edge_labels: NDArray[Any]) -> bool:
return any(
label != 0 and anchor_counts[label] >= max(3, round(stats[label, cv2.CC_STAT_AREA] * 0.1))
for label in np.unique(edge_labels)
)
return reaches(labels[y1:y2, :2]), reaches(labels[y1:y2, -2:])
def source_silhouette_mask(
source_rgb: NDArray[Any],
box: tuple[int, int, int, int],
angle: float = 0.0,
) -> NDArray[Any]:
"""Recover a thresholded glyph shape without retaining source amplitudes."""
height, width = source_rgb.shape[:2]
box_x1, box_y1, box_x2, box_y2 = box
line_h = max(1, box_y2 - box_y1)
step = max(2, round(line_h * GLYPH_SIDE_PAD_RATIO))
limit = max(step, round(line_h * GLYPH_SIDE_PAD_LIMIT))
left_pad = right_pad = step
while True:
x1, y1, x2, y2 = _glyph_crop_box(box, width, height, left_pad, right_pad)
crop_mask = _silhouette_crop_mask(source_rgb, (x1, y1, x2, y2), angle)
core_y1 = max(0, box_y1 - y1)
core_y2 = min(y2 - y1, box_y2 - y1)
anchor_box = (max(0, box_x1 - x1), core_y1, min(x2 - x1, box_x2 - x1), core_y2)
can_expand_left = x1 > 0 and left_pad < limit
can_expand_right = x2 < width and right_pad < limit
left_anchored, right_anchored = (
_anchored_components_reach_sides(crop_mask, anchor_box)
if can_expand_left or can_expand_right
else (False, False)
)
left_touches = can_expand_left and left_anchored
right_touches = can_expand_right and right_anchored
if not left_touches and not right_touches:
break
if left_touches:
left_pad = min(limit, left_pad + step)
if right_touches:
right_pad = min(limit, right_pad + step)
result = np.zeros((height, width), dtype=np.uint8)
result[y1:y2, x1:x2] = crop_mask
return result
def residual_glyph_mask(
background_rgb: NDArray[Any],
original_mask: NDArray[Any],
box: tuple[int, int, int, int],
) -> NDArray[Any]:
"""Find glyph-like contrast left after the first inpaint pass."""
residual = _foreground_mask(background_rgb, box)
residual = cv2.bitwise_and(residual, original_mask)
return cv2.dilate(residual, np.ones((5, 5), np.uint8), iterations=1)
def composite_fresh_text_edges(
source_rgb: NDArray[Any],
background_rgb: NDArray[Any],
lines: tuple[VerifiedTextLine, ...],
masks: list[NDArray[Any]],
) -> NDArray[Any]:
"""Render fresh antialiased edges for source-derived glyph masks."""
restored = background_rgb
for line, mask in zip(lines, masks, strict=True):
color = _sample_text_color(source_rgb, mask, line.box)
restored = composite_fresh_silhouette(restored, mask, color)
return restored
def composite_reconstructed_glyphs(
donor_rgb: NDArray[Any],
background_rgb: NDArray[Any],
glyph_mask: NDArray[Any],
*,
feather: float = GLYPH_FEATHER,
) -> NDArray[Any]:
"""Composite an exact reconstructed core with a narrow donor edge."""
if donor_rgb.shape != background_rgb.shape or donor_rgb.shape[:2] != glyph_mask.shape:
raise ValueError("donor, background, and glyph mask dimensions must match")
blurred = cv2.GaussianBlur(glyph_mask, (0, 0), feather) if feather > 0 else glyph_mask
alpha = np.maximum(glyph_mask, blurred).astype(np.float32) / 255.0
combined = donor_rgb.astype(np.float32) * alpha[..., None] + background_rgb.astype(np.float32) * (
1.0 - alpha[..., None]
)
return np.clip(np.rint(combined), 0, 255).astype(np.uint8)
def composite_fresh_silhouette(
background_rgb: NDArray[Any],
glyph_mask: NDArray[Any],
color: tuple[int, int, int],
*,
feather: float = 0.35,
) -> NDArray[Any]:
"""Render a binary source shape with fresh color and antialiasing."""
if background_rgb.shape[:2] != glyph_mask.shape:
raise ValueError("background and glyph mask dimensions must match")
antialiased = cv2.GaussianBlur(glyph_mask, (0, 0), feather) if feather > 0 else glyph_mask
alpha = antialiased.astype(np.float32)[..., None] / 255.0
foreground = np.empty_like(background_rgb)
foreground[:, :] = color
combined = foreground.astype(np.float32) * alpha + background_rgb.astype(np.float32) * (1.0 - alpha)
return np.clip(combined, 0, 255).astype(np.uint8)
def _clip_box(box: tuple[int, int, int, int], width: int, height: int, pad: int = 0) -> tuple[int, int, int, int]:
x1, y1, x2, y2 = box
return max(0, x1 - pad), max(0, y1 - pad), min(width, x2 + pad), min(height, y2 + pad)
def _foreground_mask(source_rgb: NDArray[Any], box: tuple[int, int, int, int]) -> NDArray[Any]:
height, width = source_rgb.shape[:2]
line_height = box[3] - box[1]
x1, y1, x2, y2 = _clip_box(box, width, height, pad=max(6, int(line_height * 0.12)))
gray = cv2.cvtColor(source_rgb[y1:y2, x1:x2], cv2.COLOR_RGB2GRAY)
ring_pad = max(8, min(24, (y2 - y1) // 5))
rx1, ry1, rx2, ry2 = _clip_box((x1, y1, x2, y2), width, height, pad=ring_pad)
context = cv2.cvtColor(source_rgb[ry1:ry2, rx1:rx2], cv2.COLOR_RGB2GRAY)
ring = np.ones(context.shape, dtype=bool)
ring[y1 - ry1 : y2 - ry1, x1 - rx1 : x2 - rx1] = False
background_luma = float(np.median(context[ring])) if ring.any() else float(np.median(gray))
low, high = float(np.percentile(gray, 4)), float(np.percentile(gray, 96))
dark_contrast, light_contrast = background_luma - low, high - background_luma
threshold = max(24.0, min(72.0, max(light_contrast, dark_contrast) * 0.32))
if light_contrast > dark_contrast:
mask = (gray.astype(np.float32) >= background_luma + threshold).astype(np.uint8) * 255
else:
mask = (gray.astype(np.float32) <= background_luma - threshold).astype(np.uint8) * 255
mask = cv2.morphologyEx(mask, cv2.MORPH_OPEN, np.ones((2, 2), np.uint8))
dilation = 5 if line_height >= 48 else 3
mask = cv2.dilate(mask, cv2.getStructuringElement(cv2.MORPH_ELLIPSE, (2 * dilation + 1,) * 2))
result = np.zeros((height, width), dtype=np.uint8)
result[y1:y2, x1:x2] = mask
return result
def _sample_text_color(
source_rgb: NDArray[Any], mask: NDArray[Any], box: tuple[int, int, int, int]
) -> tuple[int, int, int]:
height, width = source_rgb.shape[:2]
x1, y1, x2, y2 = _clip_box(box, width, height, pad=2)
crop = source_rgb[y1:y2, x1:x2]
pixels = crop[mask[y1:y2, x1:x2] > 0]
luma = pixels.mean(axis=1)
background_luma = float(crop[[0, -1], :, :].reshape(-1, 3).mean(axis=1).mean())
selected = (
pixels[luma <= np.percentile(luma, 20)] if background_luma >= 128 else pixels[luma >= np.percentile(luma, 80)]
)
return tuple(int(value) for value in np.median(selected, axis=0))
def group_text_lines(lines: Sequence[VerifiedTextLine]) -> list[list[int]]:
"""Group nearby same-script lines for a shared LaMa erase pass."""
groups: list[list[int]] = []
for index, line in enumerate(lines):
if not groups:
groups.append([index])
continue
previous = lines[groups[-1][-1]]
gap = line.box[1] - previous.box[3]
if line.script != previous.script or gap > max(60, int((previous.box[3] - previous.box[1]) * 1.1)):
groups.append([index])
else:
groups[-1].append(index)
return groups
src/remove_ai_watermarks/_internal/watermark_profiles.py
+49 -8
View File
@@ -69,9 +69,33 @@ SDXL_ZIMAGE_OPENAI_STRENGTH = 0.15
SDXL_ZIMAGE_GEMINI_STRENGTH = 0.25
SDXL_ZIMAGE_UNKNOWN_STRENGTH = SDXL_ZIMAGE_GEMINI_STRENGTH
# qwen-zimage keeps its resolution curve for unknown content, but measured vendor
# cohorts bypass it. Google remained detectable through 0.24375; 0.25 cleared all
# three valid sources and 0.27 was separately repeated clean across them and three
# accounts, so the independently checked 0.27 candidate is the operating floor.
QWEN_ZIMAGE_GOOGLE_STRENGTH = 0.27
# sdxl-zimage picks its strength from the VENDOR (unlike qwen-zimage, which derives it
# from image area). An unlisted or unknown vendor falls back to the Gemini value.
# The two OpenAI sources first cleared at 0.06225 and 0.0695. Add one full observed
# cross-source spread (0.00725) to the worst clean boundary: 0.0695 + 0.00725.
QWEN_ZIMAGE_OPENAI_STRENGTH = 0.07675
# Microsoft's public detector returned Inconclusive rather than an API-level
# watermark-negative verdict. Three valid Paint sources first cleared at 0.04125,
# 0.055, and 0.095. Add one full observed cross-source spread to the worst clean
# boundary: 0.095 + (0.095 - 0.04125) = 0.14875, rounded up to 0.15. This is a
# measured corpus margin, not a universal InvisMark threshold.
QWEN_ZIMAGE_MICROSOFT_STRENGTH = 0.15
_QWEN_ZIMAGE_FLAT_STRENGTH_BY_VENDOR: dict[str, float] = {
"google": QWEN_ZIMAGE_GOOGLE_STRENGTH,
"openai": QWEN_ZIMAGE_OPENAI_STRENGTH,
"microsoft": QWEN_ZIMAGE_MICROSOFT_STRENGTH,
}
# sdxl-zimage always picks its strength from the vendor. qwen-zimage instead uses
# the vendor only for measured cohorts and image area for unknown content. An
# unlisted or unknown SDXL vendor falls back to the Gemini value.
_SDXL_ZIMAGE_STRENGTH_BY_VENDOR: dict[str, float] = {
"openai": SDXL_ZIMAGE_OPENAI_STRENGTH,
"google": SDXL_ZIMAGE_GEMINI_STRENGTH,
@@ -103,7 +127,9 @@ def resolve_adaptive_polish(adaptive_polish: bool | None, pipeline: str) -> bool
def strength_default_help() -> str:
"""Describe the live default policy without duplicating its values."""
return (
"profile-adaptive (qwen-zimage uses resolution-adaptive denoise; sdxl-zimage "
"profile-adaptive (qwen-zimage uses resolution-adaptive denoise, with a "
f"flat OpenAI {QWEN_ZIMAGE_OPENAI_STRENGTH} / Google {QWEN_ZIMAGE_GOOGLE_STRENGTH} / "
f"Microsoft InvisMark {QWEN_ZIMAGE_MICROSOFT_STRENGTH} floors; sdxl-zimage "
f"uses OpenAI {SDXL_ZIMAGE_OPENAI_STRENGTH} / Google {SDXL_ZIMAGE_GEMINI_STRENGTH} / "
f"unknown {SDXL_ZIMAGE_UNKNOWN_STRENGTH}, from the C2PA issuer)"
)
@@ -118,10 +144,14 @@ def resolve_strength(
) -> float:
"""Resolve a user override or the calibrated policy for a profile and vendor.
Total by design. qwen-zimage picks its strength from image area rather than from
the vendor, so it needs ``size``; returning ``None`` for it instead would push that
Total by design. qwen-zimage picks its strength from image area rather than
from the vendor, so it needs ``size``; returning ``None`` for it instead would push that
branch onto every caller and move one of the two strength policies outside this
module. ``size`` is required for qwen-zimage without an explicit strength.
Measured vendor exceptions bypass the area curve: OpenAI and Microsoft use one
additional observed cross-source spread over their worst clean boundaries, while
Google uses a separately repeated cross-source candidate. See the constants'
comments for the exact derivations.
"""
if strength is not None:
return strength
@@ -129,21 +159,32 @@ def resolve_strength(
return _SDXL_ZIMAGE_STRENGTH_BY_VENDOR.get((vendor or "").casefold(), SDXL_ZIMAGE_UNKNOWN_STRENGTH)
if size is None:
raise ValueError("qwen-zimage resolves strength from image area, so size is required")
vendor_strength = _QWEN_ZIMAGE_FLAT_STRENGTH_BY_VENDOR.get((vendor or "").casefold())
if vendor_strength is not None:
return vendor_strength
from remove_ai_watermarks._internal.qwen_zimage_pipeline import resolution_adaptive_denoise
return resolution_adaptive_denoise(*size)
def vendor_for_strength(image_path: Path) -> Literal["openai", "google"] | None:
"""Select the strength cohort using the input's SynthID provenance evidence."""
def vendor_for_strength(image_path: Path) -> Literal["openai", "google", "microsoft"] | None:
"""Select the strength cohort from non-invalid pixel-watermark provenance."""
try:
from remove_ai_watermarks._internal.c2pa import (
c2pa_info_has_invalid_credential,
c2pa_info_has_invismark,
extract_c2pa_info,
)
from remove_ai_watermarks.metadata import synthid_source
evidence = (synthid_source(image_path) or "").casefold()
info = extract_c2pa_info(image_path)
evidence = (synthid_source(image_path, c2pa_info=info) or "").casefold()
except Exception:
return None
if "google" in evidence:
return "google"
if "openai" in evidence:
return "openai"
if not c2pa_info_has_invalid_credential(info) and c2pa_info_has_invismark(info):
return "microsoft"
return None
src/remove_ai_watermarks/_internal/watermark_remover.py
+10
View File
@@ -26,6 +26,8 @@ if TYPE_CHECKING:
from collections.abc import Callable
from pathlib import Path
from remove_ai_watermarks._internal.text_restoration import VerifiedTextManifest
logger = logging.getLogger(__name__)
try:
@@ -187,6 +189,8 @@ class WatermarkRemover:
tile: bool = False,
tile_size: int = 1024,
tile_overlap: int = 128,
text_manifest: VerifiedTextManifest | None = None,
fidelity_anchor: bool = False,
) -> Path:
"""Regenerate image pixels and write the result without AI metadata.
@@ -203,6 +207,10 @@ class WatermarkRemover:
resolved_strength = resolve_strength(strength, vendor, self.model_profile, size=source.size)
if not 0.0 <= resolved_strength <= 1.0:
raise ValueError(f"Strength must be between 0.0 and 1.0, got {resolved_strength}")
if text_manifest is not None and self.model_profile == SDXL_ZIMAGE_PROFILE:
raise ValueError("Verified text restoration is supported only by the qwen-zimage profile")
if text_manifest is not None and tile:
raise ValueError("Verified text restoration is not calibrated with tiled diffusion")
result = self._load_qwen_zimage_pipeline().run(
source,
@@ -211,6 +219,8 @@ class WatermarkRemover:
tile=tile,
tile_size=tile_size,
tile_overlap=tile_overlap,
text_manifest=text_manifest,
fidelity_anchor=fidelity_anchor,
)
self._write_output(result, destination)
return destination
src/remove_ai_watermarks/api.py
+6 -1
View File
@@ -243,6 +243,8 @@ class InvisibleOptions:
tile: bool = False
tile_size: int = 1024
tile_overlap: int = 128
text_manifest: Path | None = None
fidelity_anchor: bool = False
# What the invisible stage did. "unavailable" is the one outcome the caller must
@@ -329,12 +331,13 @@ class _SourceEvidence:
if evidence is None:
return True
try:
from remove_ai_watermarks._internal.c2pa import c2pa_info_has_removal_hint
from remove_ai_watermarks.identify import identify_from_evidence
report = identify_from_evidence(evidence, image_path=self._path, check_invisible=True)
except Exception:
return True
return bool(report.ai_from_metadata)
return report.ai_from_metadata or c2pa_info_has_removal_hint(evidence.c2pa_info)
def _provenance_from_report(report: Any, path: Path) -> frozenset[str]:
@@ -522,6 +525,8 @@ def _run_invisible(
tile=opts.tile,
tile_size=opts.tile_size,
tile_overlap=opts.tile_overlap,
text_manifest=opts.text_manifest,
fidelity_anchor=opts.fidelity_anchor,
)
say("invisible", "removed")
return "removed"
src/remove_ai_watermarks/cli.py
+76 -14
View File
@@ -311,6 +311,26 @@ _cpu_offload_option = click.option(
),
)
_text_manifest_option = click.option(
"--text-manifest",
type=click.Path(exists=True, dir_okay=False, path_type=Path),
default=None,
help=(
"Experimental verified-text restoration manifest. Requires qwen-zimage, "
"the text-restoration extra, native untiled geometry, and no postprocessing."
),
)
_fidelity_anchor_option = click.option(
"--fidelity-anchor/--no-fidelity-anchor",
default=False,
help=(
"With --text-manifest: blend 15% of the Qwen-VAE donor across the whole "
"frame. Off by default - the global blend was measured to return "
"detector-visible OpenAI SynthID on poster-scale manifests."
),
)
_visible_backend_option = click.option(
"--backend",
@@ -787,6 +807,8 @@ def cmd_erase(
@_tile_options
@_force_option
@_cpu_offload_option
@_text_manifest_option
@_fidelity_anchor_option
@click.pass_context
def cmd_invisible(
ctx: click.Context,
@@ -806,6 +828,8 @@ def cmd_invisible(
tile_overlap: int,
force: bool,
cpu_offload: bool,
text_manifest: Path | None,
fidelity_anchor: bool,
) -> None:
"""Remove invisible AI watermarks (SynthID, StableSignature, TreeRing).
@@ -853,20 +877,26 @@ def cmd_invisible(
console.print(f" Strength: {_resolved_strength_for_display(source, strength, vendor, pipeline)}")
t0 = time.monotonic()
result_path = engine.remove_watermark(
image_path=source,
output_path=output,
strength=strength,
seed=seed,
humanize=humanize,
unsharp=unsharp,
adaptive_polish=adaptive_polish,
max_resolution=max_resolution,
vendor=vendor,
tile=tile,
tile_size=tile_size,
tile_overlap=tile_overlap,
)
try:
result_path = engine.remove_watermark(
image_path=source,
output_path=output,
strength=strength,
seed=seed,
humanize=humanize,
unsharp=unsharp,
adaptive_polish=adaptive_polish,
max_resolution=max_resolution,
vendor=vendor,
tile=tile,
tile_size=tile_size,
tile_overlap=tile_overlap,
text_manifest=text_manifest,
fidelity_anchor=fidelity_anchor,
)
except (OSError, RuntimeError, ValueError) as exc:
console.print(f" Error: {exc}")
raise SystemExit(1) from exc
elapsed = time.monotonic() - t0
size_kb = result_path.stat().st_size / 1024
@@ -874,10 +904,27 @@ def cmd_invisible(
# ── Metadata operations ──
def _print_metadata_not_a_clean_verdict() -> None:
"""Repeat the identify empty-scan limit on metadata check and strip success.
``metadata --check`` and ``metadata --remove`` answer a narrower question than
``identify``: they report embedded AI metadata only. A quiet result used to
stop at "No AI metadata found" / "AI metadata stripped", which readers treat
as a clean-image verdict. The pixel channel is unchanged, and this project has
no local SynthID decoder, so the command must say so in the same words
``identify`` already uses.
"""
console.print(
" This is not the same as 'clean': a pixel watermark such as SynthID cannot be\n"
" detected here once its metadata proxy is absent."
)
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}")
_print_metadata_not_a_clean_verdict()
return
console.print(f" Warning: AI metadata detected in {source.name}:")
@@ -944,6 +991,7 @@ def cmd_metadata(
console.print(" the file could not be decoded, so it was copied through unchanged")
raise SystemExit(1)
console.print(f" AI metadata stripped -> {out}")
_print_metadata_not_a_clean_verdict()
# ── Video pipeline ──
@@ -1085,6 +1133,7 @@ def cmd_video_metadata(
console.print(f" still present: {', '.join(sorted(result.remaining))}")
raise SystemExit(1)
console.print(f" AI metadata stripped -> {result.output}")
_print_metadata_not_a_clean_verdict()
@cmd_video.command("invisible")
@@ -1402,6 +1451,13 @@ def cmd_identify(ctx: click.Context, source: Path, no_visible: bool, as_json: bo
verdict = "AI-generated (fully synthetic)"
console.print(f"\n Verdict: {verdict} (confidence: {report.confidence})")
console.print(f" Platform: {report.platform or 'undetermined'}")
if report.c2pa_validation is not None:
validation = report.c2pa_validation
console.print(
" C2PA validation: "
f"integrity={validation['integrity']}, signature={validation['signature']}, "
f"signer trust={validation['signer_trust']}, signer validity={validation['signer_validity']}"
)
if report.is_ai_generated is None:
console.print(
@@ -1448,6 +1504,8 @@ def cmd_identify(ctx: click.Context, source: Path, no_visible: bool, as_json: bo
@_tile_options
@_force_option
@_cpu_offload_option
@_text_manifest_option
@_fidelity_anchor_option
@click.pass_context
def cmd_all(
ctx: click.Context,
@@ -1469,6 +1527,8 @@ def cmd_all(
tile_overlap: int,
force: bool,
cpu_offload: bool,
text_manifest: Path | None,
fidelity_anchor: bool,
) -> None:
"""Remove ALL watermarks: visible + invisible + metadata.
@@ -1546,6 +1606,8 @@ def cmd_all(
tile=tile,
tile_size=tile_size,
tile_overlap=tile_overlap,
text_manifest=text_manifest,
fidelity_anchor=fidelity_anchor,
),
force=force,
progress=progress,
src/remove_ai_watermarks/dwt_dct.py
+106 -39
View File
@@ -1,7 +1,11 @@
"""DWT-DCT decoder compatible with invisible-watermark's ``dwtDct`` path.
Derived from ShieldMnt/invisible-watermark ``imwatermark/maxDct.py`` (MIT),
trimmed to the matrix path used by Stable Diffusion, SDXL, and FLUX.
trimmed to the matrix path used by Stable Diffusion, SDXL, and FLUX. The block
scan is vectorized rather than transcribed, so the file no longer reads line by
line against upstream; what it preserves is the output, bit for bit. See
[`docs/module-internals.md`](../../docs/module-internals.md) for the
measurements and for why a faster hand-rolled transform is not available.
Copyright (c) 2021 ShieldMnt
@@ -23,7 +27,39 @@ if TYPE_CHECKING:
from numpy.typing import NDArray
_DEFAULT_SCALES = (0, 36, 36)
_DEFAULT_BLOCK = 4
# Fixed by the format being decoded, and the 4-way fold chains in `_frame_bits`
# are written for it. It was a constructor parameter while the block scan was a
# generic Python loop; nothing ever passed another value, and a knob that now
# silently returns wrong bits is worse than no knob.
_BLOCK = 4
# Block-rows of the approximation band handled per strip. Not a tuned value:
# every height measured landed inside the others' noise. What matters is strips
# at all rather than a full-plane intermediate, not this number.
_STRIP = 16
def _approximation(rows: NDArray[Any]) -> NDArray[Any]:
"""One Haar pass along the last axis, approximation band only.
``pywt.dwt(x, "haar", axis=1)[0]`` computes and allocates the detail band as
well, and dispatches per row. Flattening lets one ``downcoef`` call do the
whole plane, and it is the same numbers in the same order **only while the
last axis is even**: Haar's filter is length 2, so an even row length makes
every pair fall inside its own row with no boundary extension. On an odd
width the pairs walk across row boundaries and the reshape below still
succeeds whenever the total is even -- wrong bits, no exception, and the
upstream-parity test is `skipif`-gated. Hence the explicit check.
The transform itself stays inside pywt however slow that looks. Its C
convolution contracts into an FMA that no numpy expression reproduces, and
the caller's threshold is ``peak % 36 > 18.0`` against values that are exact
multiples of 0.5, so a 1-ulp difference flips real bits.
"""
height, width = rows.shape
if width % 2:
raise RuntimeError(f"row length {width} is odd; the raveled Haar pass requires an even last axis")
return pywt.downcoef("a", rows.ravel(), "haar").reshape(height, width // 2)
class _DecodeMaxDct:
@@ -33,54 +69,85 @@ class _DecodeMaxDct:
self,
wm_lengths: tuple[int, ...],
scales: tuple[int, int, int] = _DEFAULT_SCALES,
block: int = _DEFAULT_BLOCK,
) -> None:
self._wm_lengths = wm_lengths
self._scales = scales
self._block = block
def decode(self, bgr: NDArray[Any]) -> dict[int, NDArray[Any]]:
row, col, _channels = bgr.shape
yuv = cv2.cvtColor(bgr, cv2.COLOR_BGR2YUV)
trimmed = yuv[: row // 4 * 4, : col // 4 * 4]
scores_by_length = {wm_len: ([0] * wm_len, [0] * wm_len) for wm_len in self._wm_lengths}
for channel in range(2):
if self._scales[channel] <= 0:
continue
ca1, _detail = pywt.dwt2(yuv[: row // 4 * 4, : col // 4 * 4, channel], "haar")
self._decode_frame(ca1, self._scales[channel], scores_by_length)
per_channel = [
self._plane_bits(trimmed, channel, self._scales[channel])
for channel in range(2)
if self._scales[channel] > 0
]
# Each channel restarts the bit index at 0, so the buckets come from a
# per-channel arange rather than one running counter.
index = np.concatenate([np.arange(bits.size) for bits in per_channel] or [np.zeros(0, dtype=np.int64)])
weights = np.concatenate(per_channel or [np.zeros(0)])
return {
wm_len: np.asarray(sums) * 255 > np.asarray(counts) * 127
for wm_len, (sums, counts) in scores_by_length.items()
}
decoded: dict[int, NDArray[Any]] = {}
for wm_len in self._wm_lengths:
bucket = index % wm_len
sums = np.bincount(bucket, weights=weights, minlength=wm_len)
counts = np.bincount(bucket, minlength=wm_len)
decoded[wm_len] = sums * 255 > counts * 127
return decoded
def _decode_frame(
self,
frame: NDArray[Any],
scale: int,
scores_by_length: dict[int, tuple[list[int], list[int]]],
) -> None:
row, col = frame.shape
bit_index = 0
for i in range(row // self._block):
for j in range(col // self._block):
block = frame[
i * self._block : i * self._block + self._block,
j * self._block : j * self._block + self._block,
]
inferred = self._infer_bit(block, scale)
for wm_len, (sums, counts) in scores_by_length.items():
bucket = bit_index % wm_len
sums[bucket] += inferred
counts[bucket] += 1
bit_index += 1
def _plane_bits(self, trimmed: NDArray[Any], channel: int, scale: int) -> NDArray[Any]:
"""Block bits for one colour plane, a strip of block-rows at a time.
def _infer_bit(self, block: NDArray[Any], scale: int) -> int:
position = int(np.argmax(np.abs(block.flatten()[1:]))) + 1
i, j = position // self._block, position % self._block
value = abs(float(block[i][j]))
return int((value % scale) > 0.5 * scale)
``dwt2`` is ``dwtn``: it transforms along axis 0, then along axis 1 over
both halves, and three of the four bands it returns are discarded here.
Only the approximation is ever asked for, and transposing between the
two passes lets pywt walk a contiguous axis instead of a column -- that
access pattern, not the arithmetic saved, is where the time goes.
Strips mean no full-plane float64 intermediate is ever materialized, and
they are seam-free for the reason ``_approximation`` documents: output
row ``k`` reads input rows ``2k`` and ``2k + 1`` only. Strips start on
multiples of ``2 * _BLOCK``, so neither a pair nor a 4x4 block straddles
one.
"""
if trimmed.shape[0] == 0 or trimmed.shape[1] == 0:
# Reachable: a 1x65536 image clears the caller's area check and
# trims to an empty plane. Left to dwt2 so the exception stays the
# one this module has always raised -- returning empty bits here
# instead would silently turn a raise into an all-false verdict.
return pywt.dwt2(trimmed[:, :, channel], "haar")[0]
rows = trimmed.shape[0] // (2 * _BLOCK)
cols = trimmed.shape[1] // (2 * _BLOCK)
if rows == 0 or cols == 0:
return np.zeros(0, dtype=np.float64)
width = cols * 2 * _BLOCK
pieces: list[NDArray[Any]] = []
for start in range(0, rows, _STRIP):
stop = min(start + _STRIP, rows)
strip = trimmed[start * 2 * _BLOCK : stop * 2 * _BLOCK, :width]
columns = cv2.transpose(cv2.extractChannel(strip, channel))
band = _approximation(cv2.transpose(_approximation(columns)))
pieces.append(self._frame_bits(band, scale))
return np.concatenate(pieces)
def _frame_bits(self, frame: NDArray[Any], scale: int) -> NDArray[Any]:
"""One bit per 4x4 block, in row-major block order.
Upstream's per-block loop, said to numpy once instead of to the
interpreter ~135k times per image. Zeroing the DC term in the absolute
band says "ignore index 0" without materializing a ``(nblocks, 16)``
copy, and the 4-way ``np.maximum`` chains reduce over contiguous rows.
"""
rows = frame.shape[0] // _BLOCK
cols = frame.shape[1] // _BLOCK
band = np.abs(frame[: rows * _BLOCK, : cols * _BLOCK])
band[::_BLOCK, ::_BLOCK] = 0.0
folded = np.maximum(np.maximum(band[0::4], band[1::4]), np.maximum(band[2::4], band[3::4]))
folded = folded.reshape(rows * cols, _BLOCK)
peak = np.maximum(np.maximum(folded[:, 0], folded[:, 1]), np.maximum(folded[:, 2], folded[:, 3]))
return ((peak % scale) > 0.5 * scale).astype(np.float64)
def decode_dwt_dct(bgr: NDArray[Any], wm_len: int) -> NDArray[Any]:
src/remove_ai_watermarks/identify.py
+211 -44
View File
@@ -6,13 +6,15 @@ Aggregates every locally-readable signal into a single :class:`ProvenanceReport`
the signing platform (OpenAI, Google, Adobe, Microsoft).
- **IPTC ``digitalSourceType``** "Made with AI" marker (Meta, X, others).
- **PNG text / EXIF generation parameters** (Stable Diffusion, ComfyUI, InvokeAI).
- **SynthID evidence** -- supported C2PA provenance (Google AI, or current OpenAI
with an explicit watermark action). There is no local SynthID pixel decoder.
- **SynthID provenance evidence** -- Google AI C2PA follows Google's all-media
policy; current OpenAI C2PA explicitly declares a watermark action.
- **Registered visible marks** (optional; needs cv2/numpy, no GPU) through the
shared watermark registry.
Hard limit: Google does not publish its payload decoder, and this package does
not ship one. Absence of signals is reported as ``Unknown``, never as "clean".
Hard limit: a stripped image (re-encoded, screenshotted, social-media upload)
loses all metadata, and the SynthID *pixel* watermark is not locally decodable
(proprietary decoder). Absence of signals is therefore reported as ``Unknown``,
never as "clean". See CLAUDE.md "SynthID detection is metadata-only".
"""
from __future__ import annotations
@@ -26,6 +28,9 @@ from typing import TYPE_CHECKING, Any, cast
from remove_ai_watermarks._internal.c2pa import (
c2pa_info_from_manifest_store,
c2pa_info_has_invalid_credential,
c2pa_info_has_invismark,
c2pa_info_has_removal_hint,
cbor_text_after,
extract_c2pa_info,
soft_binding_vendors_in,
@@ -108,7 +113,27 @@ _STRIP_CAVEAT = (
"Absence of metadata is not proof the image is clean: C2PA, EXIF, and PNG "
"text chunks are stripped by re-encoding, screenshots, or social-media upload."
)
_SYNTHID_CAVEAT = "SynthID presence comes from supported provenance here. Confirm other cases with the provider oracle."
_SYNTHID_CAVEAT = (
"SynthID presence comes from supported provenance here; the pixel watermark is not locally "
"decoded (proprietary decoder). Confirm via the Gemini app or openai.com/verify."
)
_C2PA_UNTRUSTED_CAVEAT = (
"The C2PA claim signature and asset binding validate, but no trust anchor list is configured here, "
"so the signer identity was never checked against one; treat the named platform as a signed claim, "
"not a verified identity."
)
_C2PA_EXPIRED_CAVEAT = (
"The C2PA signing certificate has expired and no trusted timestamp establishes when the claim was "
"signed. The binding to these bytes still holds; only the signing time is unproven."
)
_C2PA_UNVALIDATED_CAVEAT = (
"The C2PA marker was parsed without cryptographic validation; treat its origin and watermark "
"assertions as unverified claims."
)
_C2PA_INVALID_CAVEAT = (
"The embedded C2PA claim no longer validates against this asset. Its origin and watermark assertions "
"are retained only as removal hints, not as verified provenance."
)
_IPTC_ONLY_CAVEAT = "The IPTC 'Made with AI' tag flags AI provenance but does not identify the specific platform."
_INVISIBLE_WM_CAVEAT = (
"The open invisible watermark is fragile: it does not survive JPEG re-encoding "
@@ -126,6 +151,10 @@ _C2PA_CLOUD_CAVEAT = (
"It marks Content Credentials, not AI origin: the cloud manifest may describe a "
"human edit, and reading it needs a network fetch this tool does not make."
)
_SOFT_BINDING_CAVEAT = (
"Removing the embedded C2PA manifest does not remove its soft binding: the named "
"watermark or fingerprint may remain in the pixels and re-link the asset to provenance."
)
_SAMSUNG_GENAI_CAVEAT = (
"Samsung's genAIType marker shows a Galaxy AI editing tool (Generative Edit, "
"Sketch to Image, ...) touched the image; it is an undocumented proprietary "
@@ -369,6 +398,8 @@ def evidence_from_metadata_record(
"actions",
"synthid_watermark",
"soft_binding",
"soft_binding_algorithm",
"soft_binding_value",
):
if key in c2pa_info:
ai_metadata.setdefault(key, str(c2pa_info[key]))
@@ -445,20 +476,29 @@ class ProvenanceReport:
ai_source_kind: str | None = None
# True when the AI verdict rests on a metadata or embedded-invisible signal
# (C2PA AI issuer / SynthID provenance, IPTC, AIGC, local gen params, EXIF/xAI, or
# an open DWT-DCT / TrustMark decode) -- as opposed to a visible mark or a
# weak medium-confidence hint (hf-job, Samsung genAIType). It is exactly the
# set of signals an invisible/diffusion scrub targets: a visible-only or
# no-signal image has it False. Equivalent to ``confidence == "high"``;
# surfaced as a field so callers gate on intent, not on the string.
# an open DWT-DCT decode) -- as opposed to a visible mark, provenance-only
# TrustMark, or a weak medium-confidence hint (hf-job, Samsung genAIType). This
# is the set of signals an invisible/diffusion scrub targets: a visible-only
# or no-signal image has it False. Callers should gate on intent, not on the
# confidence string.
ai_from_metadata: bool = False
watermarks: list[str] = field(default_factory=list[str])
signals: list[Signal] = field(default_factory=list["Signal"])
caveats: list[str] = field(default_factory=list[str])
# Contradictions between independent provenance signals (e.g. two different
# AI vendors both claiming the image, or camera-capture credentials next to
# AI-generation markers). Non-empty means the provenance is internally
# inconsistent -- a strong tell of spoofed, transplanted, or laundered metadata.
# AI-generation markers), and credentials that failed validation. Non-empty means
# the provenance is internally inconsistent -- a strong tell of spoofed,
# transplanted, or laundered metadata. A failed credential sends
# ``is_ai_generated`` to None and lands here instead, so a consumer that reads
# only the verdict field turns a broken AI manifest into silence; see the
# ``integrity_clashes`` note in docs/python-api.md.
integrity_clashes: list[str] = field(default_factory=list[str])
# Orthogonal C2PA checks. A valid content binding does not make an untrusted
# signer identity trusted, and an expired credential does not by itself mean the
# signed bytes changed. Fallback parsing reports each dimension as unknown;
# None means no C2PA result exists.
c2pa_validation: dict[str, Any] | None = None
def to_dict(
self,
@@ -494,9 +534,41 @@ class ProvenanceReport:
],
"caveats": list(self.caveats),
"integrity_clashes": list(self.integrity_clashes),
"c2pa_validation": self.c2pa_validation,
}
def _c2pa_validation(info: dict[str, Any]) -> dict[str, Any] | None:
source = info.get("c2pa_validation_source")
if not isinstance(source, str):
return None
codes = info.get("c2pa_validation_codes")
return {
"source": source,
"state": str(info.get("c2pa_validation_state", "unknown")),
"integrity": str(info.get("c2pa_integrity", "unknown")),
"signature": str(info.get("c2pa_signature", "unknown")),
"signer_trust": str(info.get("c2pa_signer_trust", "unknown")),
"signer_validity": str(info.get("c2pa_signer_validity", "unknown")),
"codes": list(cast("list[object]", codes)) if isinstance(codes, list) else [],
}
def _c2pa_credential_level(info: dict[str, Any]) -> str:
"""Return invalid, verified, or unverified for provenance attribution.
``verified`` means the reader tied this manifest to these bytes: the hard binding
matched and the claim signature validated. Signer trust is deliberately NOT a
condition -- no trust anchors ship, so gating on it made this branch unreachable.
Read the trust-anchor paragraph in docs/module-internals.md before changing this.
"""
if c2pa_info_has_invalid_credential(info):
return "invalid"
if info.get("c2pa_integrity") == "valid" and info.get("c2pa_signature") == "valid":
return "verified"
return "unverified"
def extract_provenance_evidence(image_path: Path) -> ProvenanceEvidence:
"""Read all file-backed metadata needed by provenance verdict logic once."""
return ProvenanceEvidence(
@@ -714,6 +786,7 @@ _AI_VENDOR_TOKENS: tuple[tuple[str, str], ...] = (
("google", "Google"),
("firefly", "Adobe"),
("adobe", "Adobe"),
("copilot", "Microsoft"),
("bing", "Microsoft"),
("designer", "Microsoft"),
("microsoft", "Microsoft"),
@@ -740,6 +813,8 @@ _AI_VENDOR_TOKENS: tuple[tuple[str, str], ...] = (
("fal-ai", "fal.ai"),
("bria", "Bria"),
("apple photos clean up", "Apple"),
("luma ai", "Luma AI"),
("lumalabs", "Luma AI"),
)
@@ -1079,10 +1154,16 @@ def _identify_from_evidence(
# ── C2PA Content Credentials ────────────────────────────────────
has_c2pa = bool(info) or c2pa_marker_in(head)
c2pa_validation = _c2pa_validation(info)
c2pa_level = _c2pa_credential_level(info)
c2pa_usable = c2pa_level != "invalid"
# The reader already named which failures moved a dimension; re-deriving that here
# by substring made the displayed reason a second, looser rule than the verdict.
failed_c2pa_codes = [str(code) for code in cast("list[object]", info.get("c2pa_failed_codes", []))]
issuers = [info["issuer"]] if info.get("issuer") else _issuers_in(region)
# Full AI generation (trainedAlgorithmicMedia) vs an AI-enhanced real photo
# (compositeWithTrainedAlgorithmicMedia). The structured kind is parsed once in
# _internal.c2pa._populate_registry_fields (covers PNG + any container the c2pa-python
# _internal.c2pa._structured_manifest_fields (covers PNG + any container the c2pa-python
# reader handles); fall back to a raw head scan for the non-PNG raw-blob path
# where extract_c2pa_info returns {}. Full generation wins when both appear.
source_kind = _metadata_source_kind(info, head)
@@ -1092,8 +1173,11 @@ def _identify_from_evidence(
# Restricted to the ``asserts_ai`` vendors (distinctive brand strings), so it
# does not reopen the incidental-mention problem the common-word issuers have.
issuer_blob = " ".join(issuers)
c2pa_identity_ai = has_c2pa and any(org in issuer_blob for org in C2PA_IDENTITY_AI_ORGS)
c2pa_is_ai = source_kind is not None or c2pa_identity_ai
c2pa_identity_ai = has_c2pa and (
bool(info.get("c2pa_identity_ai")) or any(org in issuer_blob for org in C2PA_IDENTITY_AI_ORGS)
)
c2pa_claims_ai = source_kind is not None or c2pa_identity_ai
c2pa_is_ai = c2pa_usable and c2pa_claims_ai
# Generator string (for the signal detail): structured for PNG, CBOR-scanned
# for other containers. Best-effort -- some manifests key it as
# `claim_generator_info` (Pixel), so this can be None even when a device is
@@ -1113,20 +1197,43 @@ def _identify_from_evidence(
camera_label
or signer_label
or (_claim_generator_platform(generator) if c2pa_is_ai else None)
or (_claim_generator_platform(str(info.get("ai_tool"))) if c2pa_is_ai and info.get("ai_tool") else None)
or _attribute_platform(issuers, is_ai=c2pa_is_ai)
)
if has_c2pa
if has_c2pa and c2pa_usable
else None
)
if has_c2pa:
detail = ", ".join(filter(None, [", ".join(issuers), generator, info.get("source_type")]))
signals.append(Signal("c2pa", detail or "C2PA manifest present", "high"))
watermarks.append(f"C2PA Content Credentials ({', '.join(issuers) or 'unknown signer'})")
if c2pa_level == "invalid":
suffix = f"; {', '.join(failed_c2pa_codes)}" if failed_c2pa_codes else ""
signals.append(Signal("c2pa", f"C2PA manifest present, credential integrity invalid{suffix}", "medium"))
watermarks.append("C2PA Content Credentials (invalid asset binding or signature)")
caveats.append(_C2PA_INVALID_CAVEAT)
else:
signals.append(
Signal("c2pa", detail or "C2PA manifest present", "high" if c2pa_level == "verified" else "medium")
)
watermarks.append(f"C2PA Content Credentials ({', '.join(issuers) or 'unknown signer'})")
if c2pa_level == "verified":
# A missing trust bundle is not a signer that failed against one.
if info.get("c2pa_signer_trust") != "trusted":
caveats.append(_C2PA_UNTRUSTED_CAVEAT)
if info.get("c2pa_signer_validity") == "expired":
caveats.append(_C2PA_EXPIRED_CAVEAT)
else:
caveats.append(_C2PA_UNVALIDATED_CAVEAT)
# Record the AI-origin vendor for clash detection only when the source is
# actually AI -- classify the issuer attribution / generator, NOT the
# resolved `platform` (which may be a camera device token whose label,
# e.g. "Google Pixel", would mis-normalize to an AI vendor).
if c2pa_is_ai and (v := (_vendor_of(_attribute_platform(issuers, is_ai=True)) or _vendor_of(generator))):
if c2pa_is_ai and (
v := (
_vendor_of(_attribute_platform(issuers, is_ai=True))
or _vendor_of(generator)
or _vendor_of(str(info.get("ai_tool", "")))
)
):
ai_vendor_claims["c2pa"] = v
# ── C2PA cloud-manifest reference (Durable Content Credentials) ─
@@ -1164,9 +1271,13 @@ def _identify_from_evidence(
if not synthid and trained_source and c2pa_marker_in(head) and (vendors := synthid_evidence_vendors_in(region)):
synthid = synthid_verdict(", ".join(vendors))
if synthid:
watermarks.append(f"SynthID watermark ({synthid})")
watermarks.append(
f"SynthID watermark ({synthid})"
if c2pa_usable
else f"SynthID watermark claimed by invalid C2PA credentials ({synthid})"
)
caveats.append(_SYNTHID_CAVEAT)
if v := _vendor_of(synthid):
if c2pa_usable and (v := _vendor_of(synthid)):
ai_vendor_claims["synthid"] = v
# ── C2PA soft-binding: a named forensic/third-party watermark vendor ─
@@ -1174,12 +1285,36 @@ def _identify_from_evidence(
# the watermark itself can't be decoded; names whose watermark stamped the pixels.
soft_binding = meta.get("soft_binding") or (", ".join(v) if (v := soft_binding_vendors_in(region)) else None)
if soft_binding:
signals.append(Signal("soft_binding", f"C2PA soft binding: {soft_binding}", "high"))
watermarks.append(f"Forensic watermark soft binding ({soft_binding})")
soft_binding_algorithm = meta.get("soft_binding_algorithm") or info.get("soft_binding_algorithm")
soft_binding_value = meta.get("soft_binding_value") or info.get("soft_binding_value")
soft_binding_details = "; ".join(
str(value) for value in (soft_binding, soft_binding_algorithm, soft_binding_value) if value
)
signals.append(
Signal(
"soft_binding",
f"C2PA soft binding: {soft_binding_details}",
"high" if c2pa_level == "verified" else "medium",
)
)
if c2pa_info_has_invismark(info):
# Keep the generic soft-binding row for schema-1 compatibility, and add
# the pixel-removal target as its own stable signal for clients that need
# to select diffusion without parsing human-readable detail.
signals.append(
Signal(
"invismark",
f"Microsoft InvisMark pixel watermark: {soft_binding_details}",
"high" if c2pa_level == "verified" else "medium",
)
)
watermarks.append(f"Forensic watermark soft binding ({soft_binding_details})")
caveats.append(_SOFT_BINDING_CAVEAT)
# ── IPTC "Made with AI" (Meta etc.), only meaningful without C2PA ─
iptc = any(m in head for m in IPTC_AI_MARKERS)
if iptc and not has_c2pa:
standalone_iptc = iptc and not has_c2pa
if standalone_iptc:
signals.append(Signal("iptc", "digitalSourceType (Made with AI)", "high"))
watermarks.append("IPTC digitalSourceType (Made with AI)")
caveats.append(_IPTC_ONLY_CAVEAT)
@@ -1305,8 +1440,18 @@ def _identify_from_evidence(
exif_gen = any(s.name == "exif_generator" for s in signals)
xai_sig = any(s.name == "xai_signature" for s in signals)
ai_from_metadata = bool(
(has_c2pa and (c2pa_is_ai or synthid))
or iptc
(has_c2pa and c2pa_usable and (c2pa_is_ai or synthid))
or standalone_iptc
or iptc_ai
or aigc
or local_keys
or invisible_wm
or exif_gen
or xai_sig
)
high_ai_from_metadata = bool(
(has_c2pa and c2pa_level == "verified" and (c2pa_is_ai or synthid))
or standalone_iptc
or iptc_ai
or aigc
or local_keys
@@ -1324,7 +1469,7 @@ def _identify_from_evidence(
if ai_from_metadata:
is_ai: bool | None = True
confidence = "high"
confidence = "high" if high_ai_from_metadata else "medium"
elif visible_only or hf_only or samsung_only:
is_ai = True
confidence = "medium"
@@ -1333,7 +1478,17 @@ def _identify_from_evidence(
confidence = "none"
# ── Integrity clashes: contradictions between independent signals ─
clashes = _integrity_clashes(ai_vendor_claims, camera_label, camera_has_ai_marker=bool(ai_vendor_claims))
clashes = _integrity_clashes(
ai_vendor_claims,
camera_label if c2pa_usable else None,
camera_has_ai_marker=bool(ai_vendor_claims),
)
if c2pa_level == "invalid":
clashes.insert(
0,
"C2PA credentials failed integrity validation"
+ (f": {', '.join(failed_c2pa_codes)}" if failed_c2pa_codes else "."),
)
caveats.append(_STRIP_CAVEAT)
# De-duplicate while preserving order.
@@ -1346,12 +1501,13 @@ def _identify_from_evidence(
confidence=confidence,
# Meaningful for the same digitalSourceType whether carried by C2PA or a
# standalone IPTC/XMP label. Other AI signals leave it None.
ai_source_kind=source_kind if (is_ai and (has_c2pa or iptc)) else None,
ai_source_kind=(source_kind if (is_ai and ((has_c2pa and c2pa_usable) or standalone_iptc)) else None),
ai_from_metadata=ai_from_metadata,
watermarks=watermarks,
signals=signals,
caveats=caveats,
integrity_clashes=clashes,
c2pa_validation=c2pa_validation,
)
@@ -1400,9 +1556,8 @@ def identify(
image_path: Path to the image (PNG, JPEG, WebP, or ISOBMFF container).
check_visible: Also run the registered visible-mark detectors through cv2.
Set False for a metadata-only, dependency-light scan.
check_invisible: Also run optional pixel detectors for the supported
SynthID carrier and open SD/SDXL/FLUX watermarks. No-op when their
numeric extras are not installed.
check_invisible: Also decode optional open invisible watermarks
(SD/SDXL/FLUX). No-op when the decoder extra is not installed.
File-backed metadata extraction runs first. The extracted evidence is then
evaluated independently, followed by the optional pixel-backed visible and
@@ -1427,25 +1582,37 @@ def has_invisible_target(image_path: Path) -> bool:
"""True when a locally-detectable invisible/metadata AI signal is present.
The decision gate for the diffusion scrub (``invisible`` / ``all`` / ``batch``):
regenerating pixels removes an invisible watermark (SynthID, open DWT-DCT,
TrustMark) but degrades a real photo, so it must not run when there is nothing
to remove. Runs :func:`identify` with ``check_visible=False`` -- a visible mark
is handled by the separate visible pass and is NOT a diffusion target -- and
``check_invisible=True`` so an open watermark counts. Returns
``report.ai_from_metadata`` (C2PA AI issuer / SynthID provenance, IPTC,
AIGC, local gen params, EXIF/xAI, open DWT-DCT / TrustMark).
regenerating pixels removes an AI-specific invisible watermark (SynthID,
open DWT-DCT) but degrades a real photo, so it must not run when there is
nothing to remove. It runs the same evidence pipeline as :func:`identify`
with visible checks disabled and invisible checks enabled, so a visible mark
is handled by the separate visible pass and is NOT a diffusion target. Returns
True for ``report.ai_from_metadata`` (C2PA AI issuer / SynthID provenance,
IPTC, AIGC, local gen params, EXIF/xAI, or open DWT-DCT), and also when an
invalid C2PA claim retains an AI or watermark removal hint. TrustMark alone
does not trigger the scrub because it also protects human-authored work and
therefore is not an AI signal by itself.
IMPORTANT -- this cannot prove a pixel SynthID is absent: Google does not
publish the payload decoder, and this package does not ship one. A False
therefore means "no supported locally-detectable invisible target", not
IMPORTANT -- this cannot prove a pixel SynthID is absent: SynthID is detectable
only through its C2PA proxy, so a metadata-stripped AI image reads as no signal
here. A False therefore means "no locally-detectable invisible target", not
"clean". Callers must NOT present a skip as a finished clean result.
Fail-safe: any error resolves to True so the removal still runs -- leaving a
watermark on a paid removal is worse than over-regenerating a clean image.
"""
try:
report = identify(image_path, check_visible=False, check_invisible=True)
evidence = extract_provenance_evidence(image_path)
report = _identify_from_evidence(
evidence,
image_path=image_path,
check_visible=False,
check_invisible=True,
)
except Exception: # unreadable / detector error -> do not skip the removal
logger.debug("has_invisible_target: identify failed, defaulting to run", exc_info=True)
return True
return report.ai_from_metadata
# An asset edit can invalidate the C2PA binding while leaving the declared pixel
# watermark intact. Keep the removal gate fail-safe without promoting that broken
# claim back into the provenance verdict.
return report.ai_from_metadata or c2pa_info_has_removal_hint(evidence.c2pa_info)
src/remove_ai_watermarks/invisible_engine.py
+41
View File
@@ -18,6 +18,7 @@ from typing import TYPE_CHECKING
from ._internal.watermark_profiles import (
DEFAULT_PROFILE,
QWEN_ZIMAGE_PROFILE,
REMOVAL_MODULES,
resolve_adaptive_polish,
resolve_seed,
@@ -148,6 +149,8 @@ class InvisibleEngine:
tile: bool = False,
tile_size: int = 1024,
tile_overlap: int = 128,
text_manifest: Path | None = None,
fidelity_anchor: bool = False,
) -> Path:
"""Remove invisible watermark from an image.
@@ -180,6 +183,20 @@ class InvisibleEngine:
Engages only when the long side exceeds ``tile_size``.
tile_size: Tile dimension in px (default 1024).
tile_overlap: Overlap between adjacent tiles in px (default 128).
text_manifest: Operator-verified text lines bound to the decoded source
pixels. Enables the experimental Qwen-VAE ``vae-glyphs`` post-pass.
Requires the ``text-restoration`` extra and the ``qwen-zimage``
profile. Incompatible with downscaling, humanize, unsharp, and
adaptive polish. Tiling is supported: the VAE donor uses the same
overlapping tiles as the global pass, then glyph restore runs on
the blended full frame.
fidelity_anchor: Blend 15% of the Qwen-VAE donor across the whole frame
before glyph restoration. OFF by default since 0.27.1: that global
blend was measured to return detector-visible OpenAI SynthID on
poster-scale manifests (detected x6 with the anchor vs clean x6
without it, base clean; official Content Provenance API,
2026-08-19 - docs/text-protection-research.md). ``True`` reproduces
the 0.27.0 research behavior. Requires ``text_manifest``.
Returns:
Path to the cleaned image.
@@ -189,6 +206,23 @@ class InvisibleEngine:
seed = resolve_seed(seed)
adaptive_polish = resolve_adaptive_polish(adaptive_polish, self._remover.model_profile)
if fidelity_anchor and text_manifest is None:
raise ValueError("fidelity_anchor requires a text manifest")
if text_manifest is not None:
if self._remover.model_profile != QWEN_ZIMAGE_PROFILE:
raise ValueError("--text-manifest is supported only by the qwen-zimage profile")
if max_resolution != 0:
raise ValueError("--text-manifest requires --max-resolution 0")
if humanize > 0.0 or unsharp > 0.0 or adaptive_polish:
raise ValueError("--text-manifest requires humanize=0, unsharp=0, and adaptive polish disabled")
from remove_ai_watermarks import region_eraser
if not region_eraser.lama_available():
raise RuntimeError(
"Verified text restoration requires LaMa. Install: "
"pip install 'remove-ai-watermarks[text-restoration]'"
)
from PIL import Image, ImageOps
# Resolution policy: a max_resolution cap (0 = none) bounds memory on huge
@@ -205,6 +239,11 @@ class InvisibleEngine:
# Full-res original, kept for the adaptive-polish detail target (image is
# reassigned to the resized copy below; PIL resize returns a new object).
reference_pil = image
verified_text = None
if text_manifest is not None:
from remove_ai_watermarks._internal.text_restoration import load_verified_text_manifest
verified_text = load_verified_text_manifest(text_manifest, reference_pil)
# Both profiles run at the input's native geometry, so only the explicit max
# cap can move it, and it can only ever scale down.
@@ -240,6 +279,8 @@ class InvisibleEngine:
tile=tile,
tile_size=tile_size,
tile_overlap=tile_overlap,
text_manifest=verified_text,
fidelity_anchor=fidelity_anchor,
)
# Post-processing chain: decode the diffusion output ONCE, apply the
src/remove_ai_watermarks/metadata.py
+28 -4
View File
@@ -792,7 +792,7 @@ def _iptc_ai_system_impl(image_path: Path) -> str | None:
return iptc_ai_system_in(scan_head(image_path))
def synthid_source(image_path: Path) -> str | None:
def synthid_source(image_path: Path, *, c2pa_info: dict[str, Any] | None = None) -> str | None:
"""Return the vendor name(s) when provenance establishes SynthID.
This is provenance-based, not a local pixel decode. Google states that all
@@ -809,14 +809,25 @@ def synthid_source(image_path: Path) -> str | None:
Args:
image_path: Path to the image (PNG, JPEG, WebP, or ISOBMFF container).
c2pa_info: Already extracted normalized C2PA info, for callers that also need
another claim from the same manifest.
Returns:
Comma-joined vendor name(s) (e.g. ``"OpenAI"``) or None.
"""
from remove_ai_watermarks._internal.c2pa import extract_c2pa_info, synthid_evidence_vendors_in
from remove_ai_watermarks._internal.c2pa import (
c2pa_info_has_invalid_credential,
extract_c2pa_info,
synthid_evidence_vendors_in,
)
# PNG: the caBX chunk parser gives a clean, structured issuer.
vendors = extract_c2pa_info(image_path).get("synthid_vendors")
# Prefer the official reader's structured result. A failed asset binding or
# signature cannot establish the claim, and the raw fallback below must not
# resurrect it from the same damaged manifest bytes.
c2pa = extract_c2pa_info(image_path) if c2pa_info is None else c2pa_info
if c2pa_info_has_invalid_credential(c2pa):
return None
vendors = c2pa.get("synthid_vendors")
if vendors:
return ", ".join(vendors)
@@ -1152,6 +1163,14 @@ def get_ai_metadata(image_path: Path) -> dict[str, str]:
"actions",
"synthid_watermark",
"soft_binding",
"soft_binding_algorithm",
"soft_binding_value",
"c2pa_validation_source",
"c2pa_validation_state",
"c2pa_integrity",
"c2pa_signature",
"c2pa_signer_trust",
"c2pa_signer_validity",
):
if key in c2pa:
result.setdefault(key, str(c2pa[key]))
@@ -1170,6 +1189,11 @@ def get_ai_metadata(image_path: Path) -> dict[str, str]:
if (aigc := aigc_label(image_path)) is not None:
producer = aigc.get("ContentProducer", "")
result["aigc_label"] = f"China AIGC label (TC260){f'; producer {producer}' if producer else ''}"
# The structural producer beside its display rendering: a machine
# consumer (video-visible provenance) must not parse the formatted
# sentence above, whose wording can change without notice.
if producer:
result["aigc_producer"] = producer
app_scan = scan_head(image_path)
app_provenance, app_generator = _app_metadata_evidence(app_scan)
src/remove_ai_watermarks/qwen_engine.py
+1 -1
View File
@@ -1,4 +1,4 @@
"""Qwen (Tongyi Qianwen, Alibaba) visible watermark detector/localizer.
"""Qwen (Alibaba) visible watermark detector/localizer.
Qwen stamps its generations with a visible "千问AI生成" text strip in the
bottom-right corner -- the explicit AIGC label mandated by China's GB 45438-2025
src/remove_ai_watermarks/text_draft.py
+299
View File
@@ -0,0 +1,299 @@
"""Draft operator-verifiable text lines for verified-text manifests.
Proposal-only OCR: PaddleOCR detection plus three script-chosen recognition
engines (Latin / Cyrillic / CJK), accepted only when three crop paddings
normalize identically and every confidence clears the floor. ``accepted``
means crop-stable, NEVER ground-truth-correct: on the reference posters the
draft's exact-text precision was 90.0% and 94.4% because high-confidence OCR
still lost punctuation (one English comma dropped, one ideographic comma
replaced with ASCII). Every accepted line needs a human yes/no before it may
enter a manifest with ``verified: true``.
Heavy imports (paddle, and the numpy/cv2/PIL pixel stack) stay inside the
call so importing this module costs nothing without the ``text-draft`` extra;
``draft_available()`` reports whether the extra is installed.
"""
# 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
from __future__ import annotations
import unicodedata
from dataclasses import dataclass
from typing import TYPE_CHECKING, Any
from remove_ai_watermarks._internal.text_restoration import source_pixel_sha256
if TYPE_CHECKING:
from pathlib import Path
__all__ = [
"DraftLine",
"RejectedLine",
"TextDraft",
"choose_language",
"draft_available",
"draft_text_lines",
"group_word_boxes",
"source_pixel_sha256",
"stable_recognition",
]
# Crop paddings probed per line: recognition must be invariant across them.
JITTER_RATIOS: tuple[float, ...] = (0.08, 0.12, 0.2)
DETECT_SCORE_FLOOR = 0.5
@dataclass(frozen=True)
class DraftLine:
"""One crop-stable proposal. ``accepted`` means stable, not correct."""
box: tuple[int, int, int, int]
text: str
script: str
language: str
min_score: float
@dataclass(frozen=True)
class RejectedLine:
"""A detected line whose recognition was unstable or low-confidence."""
box: tuple[int, int, int, int]
script: str
language: str
reads: tuple[tuple[str, float], ...]
@dataclass(frozen=True)
class TextDraft:
"""The full proposal set for one image."""
accepted: tuple[DraftLine, ...] = ()
rejected: tuple[RejectedLine, ...] = ()
def normalize_text(text: str) -> str:
"""Normalize text for layout-independent comparison (casefold + no spaces)."""
return "".join(unicodedata.normalize("NFC", text).casefold().split())
def _has_script(text: str, script: str) -> bool:
return any(script in unicodedata.name(character, "") for character in text)
def choose_language(probes: dict[str, tuple[str, float]]) -> str:
"""Pick ``ch``/``ru``/``en`` from what each probe engine actually read."""
if _has_script(probes["ch"][0], "CJK"):
return "ch"
if _has_script(probes["ru"][0], "CYRILLIC"):
return "ru"
return "en"
def stable_recognition(reads: list[tuple[str, float]], min_score: float = 0.85) -> str | None:
"""The proposal text when every read normalizes identically and clears the floor."""
normalized = {normalize_text(text) for text, _score in reads}
if len(normalized) != 1 or min(score for _text, score in reads) < min_score:
return None
return reads[0][0]
def _vertical_overlap_ratio(left: tuple[int, int, int, int], right: tuple[int, int, int, int]) -> float:
overlap = max(0, min(left[3], right[3]) - max(left[1], right[1]))
return overlap / max(1, min(left[3] - left[1], right[3] - right[1]))
def _row_center(box: tuple[int, int, int, int]) -> float:
return (box[1] + box[3]) / 2
def group_word_boxes(boxes: list[tuple[int, int, int, int]]) -> list[tuple[int, int, int, int]]:
"""Merge word detections into line boxes by vertical overlap and gap."""
groups: list[tuple[int, int, int, int]] = []
for box in sorted(boxes, key=lambda item: (_row_center(item), item[0])):
matches: list[int] = []
for index, group in enumerate(groups):
if _vertical_overlap_ratio(box, group) < 0.45:
continue
horizontal_gap = max(0, max(box[0], group[0]) - min(box[2], group[2]))
line_height = min(box[3] - box[1], group[3] - group[1])
if horizontal_gap <= max(24, line_height * 3):
matches.append(index)
if not matches:
groups.append(box)
continue
index = max(matches, key=lambda item: _vertical_overlap_ratio(box, groups[item]))
x1, y1, x2, y2 = groups[index]
groups[index] = min(x1, box[0]), min(y1, box[1]), max(x2, box[2]), max(y2, box[3])
return sorted(groups, key=_row_center)
def _recognition_box(
box: tuple[int, int, int, int],
script: str,
width: int,
height: int,
vertical_pad_ratio: float | None = None,
) -> tuple[int, int, int, int]:
x1, y1, x2, y2 = box
line_height = y2 - y1
if script == "cjk":
left_pad = max(16, round(line_height * 0.2))
right_pad = max(16, round(line_height * 0.6))
return max(0, x1 - left_pad), y1, min(width, x2 + right_pad), y2
pad_x = max(16, line_height)
pad_y = max(8, line_height // 3) if vertical_pad_ratio is None else max(8, round(line_height * vertical_pad_ratio))
return max(0, x1 - pad_x), max(0, y1 - pad_y), min(width, x2 + pad_x), min(height, y2 + pad_y)
def _recognize(
engine: Any,
image: Any,
box: tuple[int, int, int, int],
script: str,
vertical_pad_ratio: float | None = None,
) -> tuple[str, float]:
import cv2
height, width = image.shape[:2]
x1, y1, x2, y2 = _recognition_box(box, script, width, height, vertical_pad_ratio)
crop = image[y1:y2, x1:x2]
if crop.shape[0] < 64:
scale = 64 / crop.shape[0]
crop = cv2.resize(crop, None, fx=scale, fy=scale, interpolation=cv2.INTER_CUBIC)
result = next(iter(engine.predict(crop)))
return str(result.get("rec_text", "")), float(result.get("rec_score", 0.0))
def _detect_line_boxes(engine: Any, source_rgb: Any) -> list[tuple[int, int, int, int]]:
import numpy as np
boxes: list[tuple[int, int, int, int]] = []
for page in engine.predict(source_rgb):
detected = page.get("rec_boxes", None)
if detected is None or len(detected) == 0:
detected = page.get("rec_polys", [])
for score, raw_box in zip(page.get("rec_scores", []), detected, strict=False):
if float(score) < DETECT_SCORE_FLOOR:
continue
points = np.asarray(raw_box, dtype=np.float32).reshape(-1)
if points.size == 4:
x1, y1, x2, y2 = points
else:
points = points.reshape(-1, 2)
x1, y1 = points.min(axis=0)
x2, y2 = points.max(axis=0)
boxes.append((round(float(x1)), round(float(y1)), round(float(x2)), round(float(y2))))
return group_word_boxes(boxes)
def draft_available() -> bool:
"""True when the ``text-draft`` extra (paddleocr + paddle) can run."""
from remove_ai_watermarks.optional_deps import module_available
return module_available("paddleocr") and module_available("paddle")
def _build_engines() -> tuple[Any, dict[str, Any]]:
from paddleocr import PaddleOCR, TextRecognition
# enable_mkldnn=False is load-bearing on linux: paddle's oneDNN PIR executor
# raises NotImplementedError on the text-detection graph there (seen on the
# Modal draft container, 2026-08-19), while the reference CPU path runs. The
# eval scripts never hit it because macOS pads take a different path.
detector = PaddleOCR(
lang="ch",
use_doc_orientation_classify=False,
use_doc_unwarping=False,
use_textline_orientation=False,
enable_mkldnn=False,
)
engines = {
"en": TextRecognition(model_name="en_PP-OCRv5_mobile_rec", enable_mkldnn=False),
"ru": TextRecognition(model_name="eslav_PP-OCRv5_mobile_rec", enable_mkldnn=False),
"ch": TextRecognition(model_name="PP-OCRv5_server_rec", enable_mkldnn=False),
}
return detector, engines
def draft_text_lines(
image: Path,
*,
min_score: float = 0.85,
detector: Any | None = None,
engines: dict[str, Any] | None = None,
stable: bool = True,
) -> TextDraft:
"""Propose verified-text manifest lines for ``image``; never verified ones.
Args:
image: path of the source image (draft boxes are in ITS pixel space).
min_score: recognition confidence floor. With ``stable=True`` every
jittered read must clear it; with ``stable=False`` the single
probe for the chosen language must.
detector/engines: injectable Paddle objects (tests use fakes); when
None they are built from the ``text-draft`` extra, which must be
installed (``draft_available()`` reports it).
stable: when True (default), accept a line only if three crop paddings
normalize identically. When False, take one recognition trio and
accept on score alone. Automatic restoration uses box/script only,
so the jitter gate is optional there.
Returns:
``TextDraft`` with crop-stable ``accepted`` proposals and unstable
``rejected`` lines. Both lists need human review before any manifest
may claim ``verified: true``; accepted means crop-stable, NOT
ground-truth-correct.
"""
import os
import numpy as np
from PIL import Image
if detector is None or engines is None:
if not draft_available():
raise RuntimeError(
"Text drafting requires PaddleOCR. Install: pip install 'remove-ai-watermarks[text-draft]'"
)
os.environ.setdefault("PADDLE_PDX_DISABLE_MODEL_SOURCE_CHECK", "True")
detector, engines = _build_engines()
source_rgb = np.asarray(Image.open(image).convert("RGB"))
accepted: list[DraftLine] = []
rejected: list[RejectedLine] = []
for box in _detect_line_boxes(detector, source_rgb):
probes: dict[str, tuple[str, float]] = {}
for language, engine in engines.items():
script = "cjk" if language == "ch" else "alphabetic"
probes[language] = _recognize(engine, source_rgb, box, script, 0.1)
language = choose_language(probes)
script = "cjk" if language == "ch" else "alphabetic"
if stable:
reads = [_recognize(engines[language], source_rgb, box, script, ratio) for ratio in JITTER_RATIOS]
text = stable_recognition(reads, min_score)
else:
text, score = probes[language]
reads = [(text, score)]
if not text.strip() or score < min_score:
text = None
if text is None:
rejected.append(
RejectedLine(
box=box,
script=script,
language=language,
reads=tuple((value, score) for value, score in reads),
)
)
else:
accepted.append(
DraftLine(
box=box,
text=text,
script=script,
language=language,
min_score=min(score for _value, score in reads),
)
)
return TextDraft(accepted=tuple(accepted), rejected=tuple(rejected))
src/remove_ai_watermarks/trustmark_detector.py
+21 -12
View File
@@ -32,6 +32,9 @@ logger = logging.getLogger(__name__)
# Adobe ships Variant P in production (com.adobe.trustmark.P).
_MODEL_TYPE = "P"
# Schema 3 (BCH_3) is below the detector's measured precision threshold; the
# calibration history is canonical in docs/module-internals.md.
_SUPPORTED_SCHEMAS = frozenset({0, 1, 2})
# Lazily constructed singleton -- model load + first-use download is expensive.
# Guarded by a lock so concurrent callers don't double-construct/double-download.
_tm: Any = None
@@ -57,8 +60,9 @@ def _decoder() -> Any:
# JPEG quality for the false-positive durability gate (see detect_trustmark).
# Deliberately mild: a genuine TrustMark survives far harsher, while every
# observed false positive collapsed even at this quality.
# Deliberately mild: a genuine TrustMark survives far harsher. The round-trip
# still needs payload and schema validation because content-correlated false
# positives can survive this compression level.
_REENCODE_QUALITY = 95
@@ -79,8 +83,8 @@ def detect_trustmark(image_path: Path) -> str | None:
cannot carry Adobe's watermark, and decoded a random-bytes secret). A genuine
TrustMark is a *durable* soft binding engineered to survive re-encoding (that
is its entire purpose once C2PA is stripped), so we re-decode after a mild
JPEG round-trip and require the same schema both times. Every observed false
positive collapsed under this gate.
JPEG round-trip and require the same binary payload and schema both times.
Only the calibrated schemas 0-2 count as high-precision positives.
"""
if not is_available():
return None
@@ -90,10 +94,17 @@ def detect_trustmark(image_path: Path) -> str | None:
with Image.open(image_path) as img:
cover = img.convert("RGB")
decoder = _decoder()
_wm_secret, wm_present, wm_schema = decoder.decode(cover)
wm_secret, wm_present, wm_schema = decoder.decode(cover, "binary")
if not wm_present:
return None
if not _survives_reencode(decoder, cover, wm_schema):
if wm_schema not in _SUPPORTED_SCHEMAS:
logger.debug(
"TrustMark decode for %s used weak schema %s; treating as false positive",
image_path,
wm_schema,
)
return None
if not _survives_reencode(decoder, cover, wm_secret, wm_schema):
logger.debug("TrustMark decode for %s did not survive re-encode; treating as false positive", image_path)
return None
except Exception as exc: # model download / decode failure / unreadable image
@@ -102,10 +113,8 @@ def detect_trustmark(image_path: Path) -> str | None:
return f"Adobe TrustMark (variant {_MODEL_TYPE}, schema {wm_schema})"
def _survives_reencode(decoder: Any, cover: Any, schema: int) -> bool:
"""True if the watermark re-decodes with the same schema after a mild JPEG
round-trip -- the durability a genuine TrustMark guarantees, which a BCH
false positive (content noise) does not."""
def _survives_reencode(decoder: Any, cover: Any, payload: str, schema: int) -> bool:
"""True if the same watermark re-decodes after a mild JPEG round-trip."""
import io
from PIL import Image
@@ -114,5 +123,5 @@ def _survives_reencode(decoder: Any, cover: Any, schema: int) -> bool:
cover.save(buffer, "JPEG", quality=_REENCODE_QUALITY)
buffer.seek(0)
with Image.open(buffer) as reencoded:
_secret, present, reencoded_schema = decoder.decode(reencoded.convert("RGB"))
return bool(present) and reencoded_schema == schema
reencoded_payload, present, reencoded_schema = decoder.decode(reencoded.convert("RGB"), "binary")
return bool(present) and reencoded_schema == schema and reencoded_payload == payload
src/remove_ai_watermarks/video.py
+2
View File
@@ -239,6 +239,7 @@ def _visible_removal_plan(
from remove_ai_watermarks.video_visible import (
VISIBLE_MARK_POLICIES,
has_bytedance_video_provenance,
has_hailuo_video_provenance,
has_sora_provenance,
has_veo_provenance,
stabilize_localizations,
@@ -249,6 +250,7 @@ def _visible_removal_plan(
"veo": has_veo_provenance,
"seedance": has_bytedance_video_provenance,
"dola": has_bytedance_video_provenance,
"hailuo": has_hailuo_video_provenance,
}.get(selected_mark)
policy = VISIBLE_MARK_POLICIES[selected_mark]
regions = stabilize_localizations(
src/remove_ai_watermarks/video_visible.py
+23 -6
View File
@@ -890,11 +890,14 @@ class VisibleMarkPolicy:
Every value here is MEASURED per provider; the arbiter itself
(:func:`_stabilize_localizations`) is shared and knows nothing about providers.
``accepts_provenance`` is load-bearing rather than cosmetic. Hailuo and Kling have
no metadata that could confirm them, so their rows force ``provenance=False``; that
used to be guaranteed structurally by wrappers that took no ``provenance``
parameter at all, and this flag is what preserves the guarantee now that one entry
point serves every mark.
``accepts_provenance`` is load-bearing rather than cosmetic. A vendor with no
metadata that could confirm it forces ``provenance=False``; that used to be
guaranteed structurally by wrappers that took no ``provenance`` parameter at
all, and this flag is what preserves the guarantee now that one entry point
serves every mark. Kling keeps the flag off (its TC260 producer code is image
evidence; no kling video corpus row ties a producer to the moving mark), while
Hailuo accepts it through the MiniMax TC260 label
(:func:`has_hailuo_video_provenance`).
``padding_fraction`` and ``mask_style`` belong to the removal plan rather than the
arbiter, but they are per-provider constants like the rest, so they live on the same
@@ -969,7 +972,9 @@ VISIBLE_MARK_POLICIES: dict[str, VisibleMarkPolicy] = {
anchor_iou=0.80,
padding_fraction=0.12,
mask_style="box",
accepts_provenance=False,
# No provenance_weak_floor: the measured weak floor stays the entry bar;
# a MiniMax TC260 label only drops the strong-frame requirement for an
# already-stable run (see has_hailuo_video_provenance).
),
"kling": VisibleMarkPolicy(
weak_floor=_KLING_WEAK_CONFIDENCE,
@@ -1458,3 +1463,15 @@ def has_bytedance_video_provenance(markers: dict[str, str]) -> bool:
).lower()
source_type = markers.get("source_type", "").lower()
return ("bytedance" in identity or "byteplus" in identity) and "trainedalgorithmicmedia" in source_type
def has_hailuo_video_provenance(markers: dict[str, str]) -> bool:
"""Whether a TC260 label names MiniMax, Hailuo's maker, as the producer.
Hailuo video exports carry no C2PA; their TC260 ``ContentProducer`` is the
bare name ``MiniMax`` (verified on retained MiniMax-hailuo clips). The
producer travels as its own structural marker (``aigc_producer``, set in
``metadata.get_ai_metadata``), so this matches the field exactly rather
than parsing the human-readable ``aigc_label`` sentence.
"""
return markers.get("aigc_producer", "").strip().lower() == "minimax"
src/remove_ai_watermarks/watermark_registry.py
+2 -2
View File
@@ -20,7 +20,7 @@ Entries:
- ``gemini`` -- Google Gemini / Nano Banana sparkle, bottom-right.
- ``doubao`` -- ByteDance Doubao "豆包AI生成" text strip, bottom-right.
- ``jimeng`` -- ByteDance Jimeng / Dreamina "★ 即梦AI" wordmark, bottom-right.
- ``qwen`` -- Alibaba Tongyi Qianwen "千问AI生成" text strip, bottom-right.
- ``qwen`` -- Alibaba Qwen "千问AI生成" text strip, bottom-right.
- ``kling`` -- Kuaishou Kling "可灵AI 3.0" text strip, bottom-right.
- ``yuanbao`` -- Tencent Yuanbao "元宝 / AI生成" two-line mark, bottom-right.
- ``samsung`` -- Samsung Galaxy AI "Contenuti generati dall'AI" strip, bottom-left.
@@ -651,7 +651,7 @@ _REGISTRY: tuple[KnownMark, ...] = (
"qwen",
"Qwen 千问AI生成 text",
"bottom-right",
platform="Alibaba Tongyi Qianwen (visible 千问AI生成 mark detected)",
platform="Alibaba Qwen (visible 千问AI生成 mark detected)",
tc260_producer_codes=("91440101MA9Y9T4H7A",),
),
_text_mark(