"""Background analysis tasks. MVP uses FastAPI BackgroundTasks; signatures stay queue-agnostic so they can move to Celery later unchanged.""" import logging import time from PIL import Image from app.core.config import settings from app.db.models import Result, Scan, SessionLocal from app.detectors import ( ai_image, audio_spoof, deepfake_face, explain, forensics, ocr, video_pipeline, ) from app.fusion import engine from app.provenance import factcheck, phash_cache, reverse_search logger = logging.getLogger(__name__) # Video fusion weights: frame ensemble carries most of it; audio and the # temporal-jitter heuristic corroborate. _VIDEO_WEIGHTS = {"ai_gen": 0.30, "deepfake": 0.30, "audio": 0.25, "temporal": 0.15} # Two AI-committee members at/above this level count as independent witnesses _COMMITTEE_CORROBORATION_LEVEL = 90 def _committee_corroboration(ai_result: dict, deepfake_score: float | None) -> float | None: """Second-witness score from the AI committee, for faceless images only. The two-witness rule normally needs the face specialist to confirm a high AI-gen score. Faceless AI art has no specialist to ask, so when at least two committee members (different architectures, different training data) independently score >= 90, the second-highest vote fills the witness seat. Images WITH a face keep the specialist veto: measured real portraits get unanimous committee false positives, and only the deepfake model separates them from actual fakes. """ if deepfake_score is not None: return None votes = sorted((m["score"] for m in ai_result.get("committee", [])), reverse=True) if len(votes) >= 2 and votes[1] >= _COMMITTEE_CORROBORATION_LEVEL: return votes[1] return None def _analyze_image(scan: Scan) -> dict: image = Image.open(scan.stored_path) ai_result = ai_image.detect(image) deepfake_result = deepfake_face.detect(image) forensic_result = forensics.analyze(image, scan.id) # second_witness only exists when the deepfake detector abstained, so the # specialist seat is either the deepfake score or the committee witness. second_witness = _committee_corroboration(ai_result, deepfake_result["score"]) specialist = deepfake_result["score"] if second_witness is None else second_witness fused = engine.fuse_signals( [ (settings.weight_ai_gen, ai_result["score"]), (settings.weight_deepfake, specialist), (settings.weight_forensics, forensic_result["score"]), ] ) # "Why?" heatmap — only when there is a suspicion to explain ai_focus = None if fused["verdict"] != "authentic": try: ai_focus = explain.occlusion_map(image, ai_result.get("committee", []), scan.id) except Exception: # noqa: BLE001 - explanation is optional, never fail the scan logger.exception("focus map failed for %s", scan.id) # Text printed inside the image (headlines, screenshots) for fact-checking ocr_result = ocr.extract(image) return { "fused": fused, "scores": { "ai_gen_score": ai_result["score"], "deepfake_score": deepfake_result["score"], "forensics_score": forensic_result["score"], }, "evidence": { "ai_gen": ai_result, "deepfake": deepfake_result, "forensics": forensic_result, "committee_corroboration": second_witness, "ai_focus": ai_focus, "ocr": ocr_result, }, } def _analyze_audio(scan: Scan) -> dict: result = audio_spoof.detect(scan.stored_path, scan.id) fused = engine.fuse_signals([(1.0, result["score"])]) return { "fused": fused, "scores": {"audio_spoof_score": result["score"]}, "evidence": {"audio": result}, } def _analyze_video(scan: Scan) -> dict: result = video_pipeline.analyze(scan.stored_path, scan.id) audio_score = result["audio"]["score"] if result["audio"] else None fused = engine.fuse_signals( [ (_VIDEO_WEIGHTS["ai_gen"], result["ai_gen_score"]), (_VIDEO_WEIGHTS["deepfake"], result["deepfake_score"]), (_VIDEO_WEIGHTS["audio"], audio_score), (_VIDEO_WEIGHTS["temporal"], result["temporal"]["score"]), ] ) return { "fused": fused, "scores": { "ai_gen_score": result["ai_gen_score"], "deepfake_score": result["deepfake_score"], "audio_spoof_score": audio_score, "temporal_score": result["temporal"]["score"], }, "evidence": {"video": result}, } _ANALYZERS = {"image": _analyze_image, "audio": _analyze_audio, "video": _analyze_video} def _build_provenance(db, scan: Scan, verdict: str, fallback_claim: str | None = None) -> dict: provenance: dict = {} if scan.media_type == "image" and scan.perceptual_hash: entry = phash_cache.record_scan(db, scan.perceptual_hash, scan.id, verdict) provenance["viral"] = { "perceptual_hash": entry.perceptual_hash, "scan_count": entry.scan_count, "first_seen": entry.first_seen.isoformat() if entry.first_seen else None, } provenance["reverse_search"] = reverse_search.search(scan.id) # user-pasted caption wins; text OCR'd from inside the image is the fallback claim = scan.claim_text or fallback_claim if claim: provenance["factcheck"] = factcheck.search(claim) return provenance def run_analysis(scan_id: str) -> None: db = SessionLocal() try: scan = db.get(Scan, scan_id) if scan is None: return scan.status = "processing" db.commit() start = time.monotonic() outcome = _ANALYZERS[scan.media_type](scan) elapsed_ms = int((time.monotonic() - start) * 1000) fused = outcome["fused"] ocr_text = (outcome["evidence"].get("ocr") or {}).get("text") provenance = _build_provenance(db, scan, fused["verdict"], fallback_claim=ocr_text) db.add( Result( scan_id=scan_id, final_score=fused["final_score"], verdict=fused["verdict"], provenance_json=provenance, evidence_json={ **outcome["evidence"], "verdict_label": fused["verdict_label"], }, processing_time_ms=elapsed_ms, **outcome["scores"], ) ) scan.status = "done" db.commit() except Exception as exc: # noqa: BLE001 - status must reflect any failure logger.exception("analysis failed for scan %s", scan_id) db.rollback() scan = db.get(Scan, scan_id) if scan is not None: scan.status = "failed" scan.error = str(exc) db.commit() finally: db.close() # Backwards-compatible alias (pre-Phase-4 name) run_image_analysis = run_analysis