""" Post-processing — applies final cleanup to provider results before they enter normalization. Examples: - de-duplicate scraped image URLs - clamp bounding boxes to image bounds - strip PII from raw responses (placeholder for future policy hooks) """ from __future__ import annotations from typing import Any, List from loguru import logger class ResultPostprocessor: """Cleans provider outputs before normalization.""" @staticmethod def dedupe_images(images: List[dict]) -> List[dict]: """Remove duplicate image URLs.""" seen: set[str] = set() out: List[dict] = [] for img in images: url = img.get("url") or img.get("image_url") if not url or url in seen: continue seen.add(url) out.append(img) return out @staticmethod def clamp_boxes(boxes: List[dict], width: int, height: int) -> List[dict]: """Clamp bounding boxes to image bounds.""" out: List[dict] = [] for b in boxes: x = max(0, min(b["x"], width - 1)) y = max(0, min(b["y"], height - 1)) x2 = max(0, min(b["x"] + b["w"], width)) y2 = max(0, min(b["y"] + b["h"], height)) out.append({"x": x, "y": y, "w": max(0, x2 - x), "h": max(0, y2 - y)}) return out @staticmethod def filter_low_confidence(boxes: List[dict], confs: List[float], threshold: float = 0.5) -> tuple[List[dict], List[float]]: """Drop detections below confidence threshold.""" out_boxes, out_confs = [], [] for b, c in zip(boxes, confs): if c >= threshold: out_boxes.append(b) out_confs.append(c) return out_boxes, out_confs