feat: add watermark template matching

app/services/watermark_template_matcher.py

@@ -0,0 +1,110 @@
+from __future__ import annotations
+
+from pathlib import Path
+from typing import Any
+
+import cv2
+import numpy as np
+from PIL import Image, ImageOps
+
+from app.config import settings
+
+
+TEMPLATE_EXTENSIONS = {".png", ".jpg", ".jpeg", ".webp"}
+
+
+def match_watermark_templates(path: Path, verification_id: str, output_dir: Path | None = None) -> dict[str, Any]:
+    templates = [p for p in settings.watermark_templates_dir.iterdir() if p.suffix.lower() in TEMPLATE_EXTENSIONS]
+    if not templates:
+        return {
+            "checked": True,
+            "template_count": 0,
+            "found": False,
+            "detected_template": None,
+            "confidence": 0.0,
+            "location": None,
+            "annotated_image_url": None,
+            "risk_score": 0.0,
+            "flags": [],
+        }
+
+    try:
+        with Image.open(path) as img:
+            rgb = np.array(ImageOps.exif_transpose(img).convert("RGB"))
+        gray = cv2.cvtColor(rgb, cv2.COLOR_RGB2GRAY)
+        regions = _corner_regions(gray)
+        best: dict[str, Any] = {"confidence": 0.0, "template": None, "box": None, "location": None}
+
+        for template_path in templates:
+            tmpl = cv2.imread(str(template_path), cv2.IMREAD_GRAYSCALE)
+            if tmpl is None or tmpl.size == 0:
+                continue
+            for location, roi, offset in regions:
+                score, box = _match_one(roi, tmpl, offset)
+                if score > best["confidence"]:
+                    best = {"confidence": score, "template": template_path.name, "box": box, "location": location}
+
+        found = best["confidence"] >= 0.72
+        annotated_url = None
+        if found and best["box"] and output_dir:
+            annotated_url = _save_annotation(rgb, best["box"], verification_id, output_dir)
+
+        flags = [f"Visible watermark template matched: {best['template']}."] if found else []
+        return {
+            "checked": True,
+            "template_count": len(templates),
+            "found": found,
+            "detected_template": best["template"] if found else None,
+            "confidence": round(float(best["confidence"]), 3),
+            "location": best["location"] if found else None,
+            "annotated_image_url": annotated_url,
+            "risk_score": 0.85 if found else 0.0,
+            "flags": flags,
+        }
+    except Exception as exc:
+        return {
+            "checked": True,
+            "template_count": len(templates),
+            "found": False,
+            "detected_template": None,
+            "confidence": 0.0,
+            "location": None,
+            "annotated_image_url": None,
+            "risk_score": 0.0,
+            "flags": ["Watermark template matching failed."],
+            "error": str(exc),
+        }
+
+
+def _corner_regions(gray: np.ndarray) -> list[tuple[str, np.ndarray, tuple[int, int]]]:
+    h, w = gray.shape
+    rw = max(80, int(w * 0.4))
+    rh = max(60, int(h * 0.28))
+    return [
+        ("bottom_right", gray[h - rh : h, w - rw : w], (w - rw, h - rh)),
+        ("bottom_left", gray[h - rh : h, 0:rw], (0, h - rh)),
+        ("top_right", gray[0:rh, w - rw : w], (w - rw, 0)),
+        ("top_left", gray[0:rh, 0:rw], (0, 0)),
+    ]
+
+
+def _match_one(roi: np.ndarray, template: np.ndarray, offset: tuple[int, int]) -> tuple[float, tuple[int, int, int, int] | None]:
+    if template.shape[0] > roi.shape[0] or template.shape[1] > roi.shape[1]:
+        scale = min(roi.shape[0] / template.shape[0], roi.shape[1] / template.shape[1], 1.0)
+        if scale <= 0:
+            return 0.0, None
+        template = cv2.resize(template, (max(1, int(template.shape[1] * scale)), max(1, int(template.shape[0] * scale))))
+    result = cv2.matchTemplate(roi, template, cv2.TM_CCOEFF_NORMED)
+    _, max_val, _, max_loc = cv2.minMaxLoc(result)
+    x = offset[0] + max_loc[0]
+    y = offset[1] + max_loc[1]
+    return float(max_val), (x, y, template.shape[1], template.shape[0])
+
+
+def _save_annotation(rgb: np.ndarray, box: tuple[int, int, int, int], verification_id: str, output_dir: Path) -> str:
+    x, y, w, h = box
+    annotated = rgb.copy()
+    cv2.rectangle(annotated, (x, y), (x + w, y + h), (255, 80, 40), max(2, annotated.shape[1] // 300))
+    path = output_dir / f"{verification_id}_watermark_template.jpg"
+    Image.fromarray(annotated).save(path, quality=92)
+    return f"/outputs/{path.name}"