| """ |
| 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 |
|
|
