mirror of
https://github.com/guillaumemeyer/watermarks-remover.git
synced 2026-08-22 13:11:57 +02:00
feat: add pre-commit hook integration for staged-file checking/cleaning (#138)
CI gating for AI provenance marks already exists (audit_dir.py's -j concurrency + SARIF export from #101), but that only runs after a marked file has already been committed and pushed. Catch it at commit time instead, using git's own hook point. Adds two hooks via .pre-commit-hooks.yaml: - watermarks-remover-check: fails the commit and lists findings when staged files carry AI/C2PA marks. Wraps audit_lib.scan_file() / is_actionable() -- the exact per-file logic audit_dir.py already uses for CI, so the pre-commit gate and the CI gate agree on what counts as actionable. - watermarks-remover-clean (opt-in): rewrites staged files in place by shelling out to clean_file.py --in-place per file (no duplicated cleaning logic), then exits 1 so the developer reviews the diff and re-stages -- the same convention as auto-fixing hooks like ruff --fix. .pre-commit-hooks.yaml needed an explicit allow-rule in the deny-by- default .gitignore, same as every other root-level config file already listed there. Closes #135 Co-authored-by: Guillaume Meyer (The Opinionated Man) <1385518+guillaumemeyer@users.noreply.github.com>
This commit is contained in:
co-authored by
Guillaume Meyer
parent
7e5b4c1a14
commit
a3b414654f
@@ -0,0 +1,71 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Fail if any given file carries AI/C2PA provenance marks.
|
||||
|
||||
Designed to run as a pre-commit hook: the pre-commit framework invokes this
|
||||
with the staged files as positional arguments, so it never touches git state
|
||||
itself — it just scans whatever paths it's given. Reuses scan_file() /
|
||||
is_actionable() from audit_lib.py (audit_dir.py's own per-file logic), so
|
||||
the pre-commit gate and the CI gate (#101's SARIF export) agree on exactly
|
||||
what counts as actionable.
|
||||
|
||||
Exit codes match audit_dir.py: 0 = clean, 1 = actionable findings found,
|
||||
2 = usage error (a given path does not exist / is not a file).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent))
|
||||
|
||||
from audit_lib import is_actionable, scan_file
|
||||
from common import MAX_INPUT_BYTES, eprint
|
||||
|
||||
|
||||
def main() -> int:
|
||||
p = argparse.ArgumentParser(description=__doc__)
|
||||
p.add_argument("paths", nargs="+", type=Path, help="Files to check (e.g. staged files)")
|
||||
p.add_argument(
|
||||
"--check-stylometry",
|
||||
action="store_true",
|
||||
help="Also evaluate text files for AI statistical & stylometric signals",
|
||||
)
|
||||
args = p.parse_args()
|
||||
|
||||
actionable: list[dict] = []
|
||||
for path in args.paths:
|
||||
if not path.is_file():
|
||||
eprint(f"not a file: {path}")
|
||||
return 2
|
||||
if path.stat().st_size > MAX_INPUT_BYTES:
|
||||
eprint(f"skipping {path}: larger than {MAX_INPUT_BYTES} bytes")
|
||||
continue
|
||||
item = scan_file(path, check_stylometry=args.check_stylometry)
|
||||
if item.get("kind") == "unknown":
|
||||
continue
|
||||
if is_actionable(item):
|
||||
actionable.append(item)
|
||||
|
||||
if not actionable:
|
||||
return 0
|
||||
|
||||
eprint(f"watermarks-remover: {len(actionable)} file(s) carry AI/C2PA provenance marks:")
|
||||
for item in actionable:
|
||||
eprint(f" {item['path']}")
|
||||
for finding in item.get("findings", []):
|
||||
eprint(f" - {finding}")
|
||||
if item.get("has_c2pa"):
|
||||
eprint(" - C2PA manifest present")
|
||||
if item.get("has_ai_metadata"):
|
||||
eprint(" - AI-generator metadata present")
|
||||
eprint(
|
||||
"Run `python3 service/scripts/clean_file.py <path> --in-place` "
|
||||
"(or the watermarks-remover-clean pre-commit hook) to strip these before committing."
|
||||
)
|
||||
return 1
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
@@ -0,0 +1,88 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Strip AI/C2PA provenance marks from given files in place.
|
||||
|
||||
Designed to run as a pre-commit hook: the pre-commit framework invokes this
|
||||
with the staged files as positional arguments. It wraps clean_file.py
|
||||
--in-place as a subprocess per file — no cleaning logic is duplicated here,
|
||||
this is purely a batch-over-staged-files wrapper.
|
||||
|
||||
Rewriting a file a developer just staged is a real behavior change, not a
|
||||
lint-and-continue: following the standard pre-commit auto-fixer convention
|
||||
(same as e.g. black / ruff --fix hooks), this exits 1 whenever it modified
|
||||
at least one file, so the hook framework stops the commit and the developer
|
||||
reviews the diff and re-stages. Exit 0 means every file was already clean.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import subprocess
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent))
|
||||
|
||||
from common import eprint
|
||||
|
||||
CLEAN_FILE_PY = Path(__file__).resolve().parent / "clean_file.py"
|
||||
|
||||
|
||||
def _changed(result: dict) -> bool:
|
||||
stats = result.get("stats")
|
||||
if stats is not None:
|
||||
return bool(stats.get("removed_count") or stats.get("replaced_count"))
|
||||
return bool(result.get("actions"))
|
||||
|
||||
|
||||
def _clean_one(path: Path) -> str:
|
||||
"""Returns 'changed', 'unchanged', or 'skipped'."""
|
||||
proc = subprocess.run(
|
||||
[sys.executable, str(CLEAN_FILE_PY), str(path), "--in-place", "--json"],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
check=False,
|
||||
)
|
||||
if proc.returncode == 2:
|
||||
# Unrecognized format or oversized input; clean_file.py already
|
||||
# explained why on stderr, this is a non-fatal skip for the hook.
|
||||
return "skipped"
|
||||
if not proc.stdout.strip():
|
||||
eprint(f"{path}: {proc.stderr.strip() or 'clean_file.py produced no output'}")
|
||||
return "skipped"
|
||||
try:
|
||||
result = json.loads(proc.stdout)
|
||||
except json.JSONDecodeError:
|
||||
eprint(f"{path}: could not parse clean_file.py output")
|
||||
return "skipped"
|
||||
return "changed" if _changed(result) else "unchanged"
|
||||
|
||||
|
||||
def main() -> int:
|
||||
p = argparse.ArgumentParser(description=__doc__)
|
||||
p.add_argument(
|
||||
"paths", nargs="+", type=Path, help="Files to clean in place (e.g. staged files)"
|
||||
)
|
||||
args = p.parse_args()
|
||||
|
||||
changed_paths: list[Path] = []
|
||||
for path in args.paths:
|
||||
if not path.is_file():
|
||||
eprint(f"not a file: {path}")
|
||||
continue
|
||||
status = _clean_one(path)
|
||||
if status == "changed":
|
||||
changed_paths.append(path)
|
||||
|
||||
if not changed_paths:
|
||||
return 0
|
||||
|
||||
eprint(f"watermarks-remover: cleaned {len(changed_paths)} file(s) in place:")
|
||||
for path in changed_paths:
|
||||
eprint(f" {path}")
|
||||
eprint("Review the changes and re-stage before committing.")
|
||||
return 1
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
Reference in New Issue
Block a user