Spaces:
Sleeping
Sleeping
| """ | |
| Batch Investigation Mode. | |
| Processes multiple images in a single request, returning a unified | |
| forensic report with cross-image analysis including: | |
| - Per-image full forensic report | |
| - Batch-level statistics (mean AI probability, class distribution) | |
| - Clustering: which images are most similar (hash-based) | |
| - Duplicate detection via perceptual hash comparison | |
| - Highest-risk images ranked by AI probability | |
| - Cross-image provenance consistency check | |
| Limits: | |
| MAX_BATCH_SIZE = 10 images per request | |
| MAX_IMAGE_SIZE = 5MB per image in batch mode | |
| Processing is sequential to avoid OOM on GPU | |
| """ | |
| import logging | |
| from typing import Dict, Any, List | |
| logger = logging.getLogger(__name__) | |
| MAX_BATCH_SIZE = 10 | |
| MAX_IMAGE_BYTES = 5 * 1024 * 1024 # 5MB per image in batch | |
| def _phash_distance(h1: str, h2: str) -> int: | |
| """Hamming distance between two perceptual hash hex strings.""" | |
| try: | |
| i1 = int(h1, 16) | |
| i2 = int(h2, 16) | |
| xor = i1 ^ i2 | |
| return bin(xor).count("1") | |
| except Exception: | |
| return 64 # Max distance on error | |
| def _find_duplicates(reports: List[Dict[str, Any]]) -> List[Dict[str, Any]]: | |
| """ | |
| Identify duplicate or near-duplicate images using perceptual hash. | |
| Threshold: Hamming distance <= 8 (out of 64 bits). | |
| """ | |
| pairs = [] | |
| for i in range(len(reports)): | |
| for j in range(i + 1, len(reports)): | |
| h1 = reports[i].get("hashes", {}).get("perceptual_hash", "") | |
| h2 = reports[j].get("hashes", {}).get("perceptual_hash", "") | |
| if h1 and h2: | |
| dist = _phash_distance(h1, h2) | |
| if dist <= 8: | |
| pairs.append({ | |
| "image_a": reports[i]["file_info"]["filename"], | |
| "image_b": reports[j]["file_info"]["filename"], | |
| "phash_distance": dist, | |
| "similarity": "identical" if dist == 0 else "near_duplicate", | |
| }) | |
| return pairs | |
| def _rank_by_risk(reports: List[Dict[str, Any]]) -> List[Dict[str, Any]]: | |
| """Return images ranked by AI probability descending.""" | |
| ranked = [] | |
| for r in reports: | |
| ranked.append({ | |
| "filename": r["file_info"]["filename"], | |
| "ai_probability": r["summary"]["ai_probability"], | |
| "classification": r["summary"]["ai_classification"], | |
| "evidence_id": r["evidence_id"], | |
| "predicted_generator": r.get("generator_attribution", {}).get("predicted_generator", "unknown"), | |
| "c2pa_status": r.get("c2pa_provenance", {}).get("provenance_status", "none"), | |
| }) | |
| return sorted(ranked, key=lambda x: x["ai_probability"], reverse=True) | |
| def _batch_statistics(reports: List[Dict[str, Any]]) -> Dict[str, Any]: | |
| """Compute aggregate statistics across all images in the batch.""" | |
| probs = [r["summary"]["ai_probability"] for r in reports] | |
| classes = [r["summary"]["ai_classification"] for r in reports] | |
| generators = [ | |
| r.get("generator_attribution", {}).get("predicted_generator", "unknown") | |
| for r in reports | |
| ] | |
| c2pa_statuses = [ | |
| r.get("c2pa_provenance", {}).get("provenance_status", "none") | |
| for r in reports | |
| ] | |
| class_counts: Dict[str, int] = {} | |
| for c in classes: | |
| class_counts[c] = class_counts.get(c, 0) + 1 | |
| generator_counts: Dict[str, int] = {} | |
| for g in generators: | |
| generator_counts[g] = generator_counts.get(g, 0) + 1 | |
| # Classification strings: "likely_ai_generated", "possibly_ai_generated", | |
| # "likely_authentic", "possibly_authentic", "uncertain" | |
| # Use exclusive categories to prevent double-counting. | |
| ai_count = sum(1 for cls in classes if "ai_generated" in cls) | |
| real_count = sum(1 for cls in classes if "authentic" in cls and "ai_generated" not in cls) | |
| uncertain_count = len(classes) - ai_count - real_count | |
| return { | |
| "total_images": len(reports), | |
| "ai_detected_count": ai_count, | |
| "authentic_count": real_count, | |
| "uncertain_count": uncertain_count, | |
| "mean_ai_probability": round(sum(probs) / len(probs), 4) if probs else 0.0, | |
| "max_ai_probability": round(max(probs), 4) if probs else 0.0, | |
| "min_ai_probability": round(min(probs), 4) if probs else 0.0, | |
| "classification_breakdown": class_counts, | |
| "generator_breakdown": generator_counts, | |
| "c2pa_has_credentials": sum(1 for s in c2pa_statuses if s != "none"), | |
| "batch_verdict": ( | |
| "high_risk" if ai_count / max(len(reports), 1) >= 0.6 else | |
| "mixed" if ai_count > 0 else | |
| "likely_authentic" | |
| ), | |
| } | |
| def _provenance_consistency(reports: List[Dict[str, Any]]) -> Dict[str, Any]: | |
| """ | |
| Check if C2PA provenance is consistent across images in the batch. | |
| A collection where some images have credentials and others do not | |
| may indicate selective credential stripping. | |
| """ | |
| statuses = [ | |
| r.get("c2pa_provenance", {}).get("provenance_status", "none") | |
| for r in reports | |
| ] | |
| has_credentials = sum(1 for s in statuses if s != "none") | |
| lacks_credentials = sum(1 for s in statuses if s == "none") | |
| if has_credentials == 0: | |
| consistency = "consistent_no_credentials" | |
| note = "No images in batch have C2PA credentials. This is normal for most images." | |
| elif lacks_credentials == 0: | |
| consistency = "consistent_all_credentialed" | |
| note = "All images have C2PA credentials. Strong provenance signal." | |
| else: | |
| consistency = "inconsistent" | |
| note = ( | |
| f"{has_credentials} of {len(reports)} images have C2PA credentials. " | |
| "Mixed provenance may indicate selective credential stripping." | |
| ) | |
| return { | |
| "consistency": consistency, | |
| "images_with_credentials": has_credentials, | |
| "images_without_credentials": lacks_credentials, | |
| "note": note, | |
| } | |
| def process_batch( | |
| images: List[Dict[str, Any]], | |
| ) -> Dict[str, Any]: | |
| """ | |
| Process a batch of images through the full forensic pipeline. | |
| Args: | |
| images: List of dicts with keys: | |
| filename - str | |
| data - bytes (image bytes) | |
| Returns: | |
| Batch forensic report with per-image results and aggregate analysis | |
| """ | |
| from backend.services.image_forensics import ImageForensics | |
| if len(images) > MAX_BATCH_SIZE: | |
| return { | |
| "error": f"Batch too large. Max {MAX_BATCH_SIZE} images per request.", | |
| "maximum": MAX_BATCH_SIZE, | |
| "received": len(images), | |
| } | |
| results = [] | |
| errors = [] | |
| reports = [] | |
| for item in images: | |
| filename = item.get("filename", "unknown") | |
| data = item.get("data", b"") | |
| if len(data) == 0: | |
| errors.append({"filename": filename, "error": "Empty file"}) | |
| continue | |
| if len(data) > MAX_IMAGE_BYTES: | |
| errors.append({ | |
| "filename": filename, | |
| "error": f"File too large for batch mode (max {MAX_IMAGE_BYTES // (1024*1024)}MB)" | |
| }) | |
| continue | |
| try: | |
| forensics = ImageForensics(data, filename) | |
| report = forensics.generate_forensic_report() | |
| reports.append(report) | |
| results.append({ | |
| "filename": filename, | |
| "status": "success", | |
| "evidence_id": report["evidence_id"], | |
| "report": report, | |
| }) | |
| logger.info( | |
| f"Batch: processed {filename} " | |
| f"(ai={report['summary']['ai_probability']:.3f})" | |
| ) | |
| except Exception as e: | |
| logger.warning(f"Batch: failed {filename}: {e}") | |
| errors.append({"filename": filename, "error": str(e)}) | |
| if not reports: | |
| return { | |
| "status": "failed", | |
| "processed": 0, | |
| "errors": errors, | |
| "error": "No images could be processed", | |
| } | |
| return { | |
| "status": "complete", | |
| "processed": len(reports), | |
| "failed": len(errors), | |
| "errors": errors, | |
| "statistics": _batch_statistics(reports), | |
| "risk_ranking": _rank_by_risk(reports), | |
| "duplicate_pairs": _find_duplicates(reports), | |
| "provenance_consistency": _provenance_consistency(reports), | |
| "individual_reports": results, | |
| } | |