Sync from GitHub via hub-sync

.env.example

@@ -34,6 +34,13 @@ LLM_MODEL=gemini-1.5-flash
 
 # News lookup (Phase 13)
 NEWS_API_KEY=
+NEWS_API_BASE_URL=https://newsdata.io/api/1/latest
+NEWS_API_ARCHIVE_BASE_URL=https://newsdata.io/api/1/archive
+NEWS_API_LANGUAGES=en,hi
+NEWS_API_RECENT_TIMEFRAME=1
+NEWS_API_OLDER_DAYS=7
+NEWS_API_PAGE_SIZE=10
+NEWS_API_PRIMARY_COUNTRY=in
 
 # Auth (REQUIRED in production — generate with python -c "import secrets; print(secrets.token_urlsafe(48))")
 JWT_SECRET_KEY=change-me-in-production

api/v1/analyze.py

@@ -71,7 +71,13 @@ from services.storage import (
 )
 from services.job_queue import registry as job_registry, run_job
 from utils.file_handler import read_upload_bytes, save_upload_to_tempfile
-from utils.scoring import compute_authenticity_score, compute_video_authenticity_score, get_verdict_label, maybe_clamp_to_uncertain
+from utils.scoring import (
+    apply_unverified_news_gate,
+    compute_authenticity_score,
+    compute_video_authenticity_score,
+    get_verdict_label,
+    maybe_clamp_to_uncertain,
+)
 
 router = APIRouter(prefix="/analyze", tags=["analyze"])
 
@@ -261,9 +267,9 @@ def generate_llm_endpoint(
             return {"llm_summary": existing_summary}
         raise HTTPException(status_code=500, detail="LLM generation failed")
 
-def _persist_response_payload(db: Session, record: AnalysisRecord, resp) -> None:
+def _persist_response_payload(db: Session, record: AnalysisRecord, resp, exclude: dict | None = None) -> None:
     """Keep reloaded/history responses aligned with the fresh API response."""
-    record.result_json = json.dumps(resp.model_dump())
+    record.result_json = json.dumps(resp.model_dump(exclude=exclude))
     db.add(record)
     db.commit()
 
@@ -479,7 +485,7 @@ async def analyze_image(
         media_type="image",
         verdict=label,
         authenticity_score=float(score),
-        result_json=json.dumps(resp.model_dump()),
+        result_json=json.dumps(resp.model_dump(exclude=_IMAGE_EXCLUDE)),
         media_hash=media_hash,
         media_path=media_path,
         thumbnail_url=thumbnail_url,
@@ -497,7 +503,7 @@ async def analyze_image(
         stages.append("llm_explanation")
 
     resp.processing_summary.stages_completed = stages
-    _persist_response_payload(db, record, resp)
+    _persist_response_payload(db, record, resp, exclude=_IMAGE_EXCLUDE)
 
     # ── Phase 14: VLM breakdown runs after response is returned ──
     if user is not None and vlm_bd is None:
@@ -795,7 +801,14 @@ async def analyze_text_endpoint(
         weighted = raw_score
 
     score = int(round(max(0.0, min(100.0, weighted))))
-    label, severity = get_verdict_label(score)
+    score, label, severity, news_gate = apply_unverified_news_gate(
+        score,
+        has_trusted_sources=bool(news.trusted_sources),
+        has_contradicting_evidence=bool(news.contradicting_evidence),
+        truth_override_applied=bool(news.truth_override and news.truth_override.applied),
+    )
+    if news_gate:
+        stages.append(news_gate)
     duration_ms = int((time.perf_counter() - start) * 1000)
 
     model_used = (
@@ -850,6 +863,7 @@ async def analyze_text_endpoint(
             total_duration_ms=duration_ms,
             model_used=model_used,
             calibrator_applied=False,
+            gating_applied=news_gate,
         ),
     )
 
@@ -972,7 +986,14 @@ async def analyze_screenshot_endpoint(
     if not full_text.strip():
         weighted = 50
     score = int(round(max(0.0, min(100.0, weighted))))
-    label, severity = get_verdict_label(score)
+    score, label, severity, news_gate = apply_unverified_news_gate(
+        score,
+        has_trusted_sources=bool(news.trusted_sources),
+        has_contradicting_evidence=bool(news.contradicting_evidence),
+        truth_override_applied=bool(news.truth_override and news.truth_override.applied),
+    )
+    if news_gate:
+        stages.append(news_gate)
     duration_ms = int((time.perf_counter() - start) * 1000)
 
     model_used_str = (
@@ -1025,6 +1046,7 @@ async def analyze_screenshot_endpoint(
             total_duration_ms=duration_ms,
             model_used=model_used_str,
             calibrator_applied=False,
+            gating_applied=news_gate,
         ),
     )
 

config.py

@@ -156,7 +156,13 @@ class Settings(BaseSettings):
 
     # News API
     NEWS_API_KEY: str = ""
-    NEWS_API_BASE_URL: str = "https://newsdata.io/api/1/news"
+    NEWS_API_BASE_URL: str = "https://newsdata.io/api/1/latest"
+    NEWS_API_ARCHIVE_BASE_URL: str = "https://newsdata.io/api/1/archive"
+    NEWS_API_LANGUAGES: str = "en,hi"
+    NEWS_API_RECENT_TIMEFRAME: str = "1"
+    NEWS_API_OLDER_DAYS: int = 7
+    NEWS_API_PAGE_SIZE: int = 10
+    NEWS_API_PRIMARY_COUNTRY: str = "in"
 
     # Reports
     REPORT_DIR: str = "/data/reports"
@@ -229,6 +235,13 @@ class Settings(BaseSettings):
     GENERAL_FAKE_GATING_THRESHOLD: float = 0.80
     GAN_ARTIFACT_GATING_THRESHOLD: float = 0.70
     GATING_FAKE_FLOOR: float = 0.50
+    # Synthetic still-image overrides. FaceForensics/DFDC models are trained for
+    # manipulated video faces, so they should not veto a strong still-image AI
+    # detector on generated portraits.
+    SYNTHETIC_STILL_HIGH_THRESHOLD: float = 0.80
+    SYNTHETIC_STILL_HIGH_FLOOR: float = 0.80
+    SYNTHETIC_STILL_VERY_HIGH_THRESHOLD: float = 0.90
+    SYNTHETIC_STILL_VERY_HIGH_FLOOR: float = 0.90
 
     # Video-frame weight overrides. When an image is detected as a low-res
     # video frame (face-swap deepfakes are extracted from video), the general
@@ -239,6 +252,10 @@ class Settings(BaseSettings):
     VIDEO_FRAME_GENERAL_WEIGHT: float = 0.15
     VIDEO_FRAME_FORENSICS_WEIGHT: float = 0.10
     VIDEO_FRAME_EXIF_WEIGHT: float = 0.05
+    # Per-frame video detector blend. FFPP ViT is trained on FaceForensics++
+    # face forgery frames, so it is the dominant signal for video analysis.
+    VIDEO_FFPP_WEIGHT: float = 0.70
+    VIDEO_EFFNET_WEIGHT: float = 0.30
     VIDEO_SAMPLE_FRAMES: int = 16  # frames to sample per video for inference
     EXIFTOOL_PATH: str = ""  # full path to ExifTool binary; empty = metadata write disabled
 

models/heatmap_generator.py

@@ -10,7 +10,6 @@ import torch
 from loguru import logger
 from PIL import Image
 from pytorch_grad_cam import GradCAMPlusPlus
-from pytorch_grad_cam.utils.image import show_cam_on_image
 from pytorch_grad_cam.utils.model_targets import ClassifierOutputTarget
 
 from config import settings
@@ -41,6 +40,16 @@ def _vit_reshape_transform(tensor: torch.Tensor, height: int = 14, width: int =
     return result
 
 
+def _find_class_index(model: torch.nn.Module, label_tokens: tuple[str, ...]) -> Optional[int]:
+    """Find the first class index whose label contains one of `label_tokens`."""
+    id2label: dict[int, str] = getattr(getattr(model, "config", None), "id2label", {}) or {}
+    for idx, label in id2label.items():
+        lowered = str(label).lower()
+        if any(token in lowered for token in label_tokens):
+            return int(idx)
+    return None
+
+
 def _preprocess_for_cam(pil_img: Image.Image, processor) -> tuple[torch.Tensor, np.ndarray]:
     """Return (input_tensor, rgb_float_224) where rgb_float_224 is a (H,W,3) float
     array in [0,1] matching the model input geometry — needed for overlaying.
@@ -58,7 +67,7 @@ def _preprocess_for_cam(pil_img: Image.Image, processor) -> tuple[torch.Tensor,
 
 
 def _encode_overlay_to_base64(overlay: np.ndarray) -> str:
-    """Encode a uint8 (H,W,3) RGB overlay to a base64 data-URL PNG."""
+    """Encode a uint8 RGB/RGBA overlay to a base64 data-URL PNG."""
     buf = io.BytesIO()
     Image.fromarray(overlay).save(buf, format="PNG")
     b64 = base64.b64encode(buf.getvalue()).decode("ascii")
@@ -93,9 +102,13 @@ def _compute_gradcam_pp(
 
     wrapped = _HFLogitsWrapper(model)
 
-    targets = None
-    if target_class_idx is not None:
-        targets = [ClassifierOutputTarget(int(target_class_idx))]
+    if target_class_idx is None:
+        target_class_idx = _find_class_index(
+            model,
+            ("fake", "deepfake", "manipulated", "ai", "generated", "synthetic"),
+        )
+
+    targets = [ClassifierOutputTarget(int(target_class_idx))] if target_class_idx is not None else None
 
     with GradCAMPlusPlus(
         model=wrapped,

schemas/common.py

@@ -4,7 +4,7 @@ from typing import List, Optional
 
 from pydantic import BaseModel, ConfigDict, Field
 
-ANALYSIS_CACHE_VERSION = "2026-05-06-phase-a-unified-fusion"
+ANALYSIS_CACHE_VERSION = "2026-05-07-archive-7d-news-fallback"
 
 
 class Verdict(BaseModel):

services/image_service.py

@@ -200,11 +200,18 @@ def _classify_no_face(
     models_used = [general.model_used if general else "no-face-forensic-fusion"]
 
     # Apply hard gating (Phase A4) on the no-face path too.
+    is_video_frame = _looks_like_video_frame(pil_img)
     gated_prob, gating_reason = _apply_hard_gating(
         fake_prob=fused.fake_probability,
         general_fake_prob=general.fake_probability if general else None,
         artifacts=artifact_indicators or [],
     )
+    gated_prob, synthetic_reason = _apply_synthetic_still_overrides(
+        fake_prob=gated_prob,
+        general_fake_prob=general.fake_probability if general else None,
+        is_video_frame=is_video_frame,
+    )
+    final_gating_reason = synthetic_reason or gating_reason
     final_label = "Fake" if gated_prob >= 0.5 else fused.label
 
     return ImageClassification(
@@ -219,8 +226,9 @@ def _classify_no_face(
             "weights": fused.weights,
             "method": fused.method,
             "pre_gating": fused.fake_probability,
+            "is_video_frame": is_video_frame,
         },
-        gating_applied=gating_reason,
+        gating_applied=final_gating_reason,
     )
 
 
@@ -265,6 +273,34 @@ def _apply_hard_gating(
     return fake_prob, None
 
 
+def _apply_synthetic_still_overrides(
+    *,
+    fake_prob: float,
+    general_fake_prob: Optional[float],
+    is_video_frame: bool,
+) -> Tuple[float, Optional[str]]:
+    """Keep still-image AI-generation evidence from being diluted by FFPP/DFDC.
+
+    FaceForensics/DFDC models are trained for manipulated real/video faces. They
+    are useful evidence, but they should not veto a high-confidence whole-image
+    AI detector on generated still portraits.
+    """
+    if is_video_frame or general_fake_prob is None:
+        return fake_prob, None
+
+    general = max(0.0, min(1.0, float(general_fake_prob)))
+    if general >= settings.SYNTHETIC_STILL_VERY_HIGH_THRESHOLD:
+        adjusted = max(fake_prob, settings.SYNTHETIC_STILL_VERY_HIGH_FLOOR)
+        if adjusted != fake_prob:
+            return adjusted, f"general_detector_very_high({general:.2f})"
+    elif general >= settings.SYNTHETIC_STILL_HIGH_THRESHOLD:
+        adjusted = max(fake_prob, settings.SYNTHETIC_STILL_HIGH_FLOOR)
+        if adjusted != fake_prob:
+            return adjusted, f"general_detector_high({general:.2f})"
+
+    return fake_prob, None
+
+
 def classify_image(
     pil_img: Image.Image,
     *,
@@ -409,6 +445,12 @@ def classify_image(
         general_fake_prob=general_fake_prob,
         artifacts=artifacts_list,
     )
+    ensemble_prob, synthetic_reason = _apply_synthetic_still_overrides(
+        fake_prob=ensemble_prob,
+        general_fake_prob=general_fake_prob,
+        is_video_frame=is_video_frame,
+    )
+    final_gating_reason = synthetic_reason or gating_reason
 
     method = f"unified_evidence_{face_stack_method}"
     label = "Fake" if ensemble_prob >= 0.5 else "Real"
@@ -417,7 +459,7 @@ def classify_image(
         f"face_stack={face_stack_prob:.3f} general={general_fake_prob if general_fake_prob is not None else 'n/a'} "
         f"forensics={components.get('forensics', 'n/a')} exif={components.get('exif', 'n/a')} "
         f"vlm={components.get('vlm', 'n/a')} -> {pre_gating_prob:.3f} "
-        f"(gated:{gating_reason or 'none'} -> {ensemble_prob:.3f})"
+        f"(gated:{final_gating_reason or 'none'} -> {ensemble_prob:.3f})"
     )
     return ImageClassification(
         label=label,
@@ -434,7 +476,7 @@ def classify_image(
             "pre_gating": pre_gating_prob,
             "is_video_frame": is_video_frame,
         },
-        gating_applied=gating_reason,
+        gating_applied=final_gating_reason,
     )
 
 

services/news_lookup.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 from dataclasses import dataclass
+from datetime import datetime, timedelta, timezone
 from typing import List, Optional
 from urllib.parse import urlparse
 
@@ -17,6 +18,9 @@ TRUSTED_DOMAINS = {
     "cnn.com": 0.9, "npr.org": 0.95, "aljazeera.com": 0.9,
     "thehindu.com": 0.9, "indianexpress.com": 0.9, "ndtv.com": 0.85,
     "hindustantimes.com": 0.85, "pti.news": 0.95,
+    "timesofindia.indiatimes.com": 0.85, "livemint.com": 0.85,
+    "deccanherald.com": 0.85, "scroll.in": 0.8, "theprint.in": 0.8,
+    "news18.com": 0.8, "business-standard.com": 0.85, "thewire.in": 0.8,
 }
 
 # Fact-check / contradiction sources
@@ -46,6 +50,78 @@ class NewsLookupResult:
     no_source_penalty: float = 0.0
 
 
+def _clean_param(value: object) -> str:
+    return str(value).strip()
+
+
+def _configured_languages() -> str:
+    return ",".join(
+        part.strip()
+        for part in _clean_param(settings.NEWS_API_LANGUAGES).split(",")
+        if part.strip()
+    ) or "en"
+
+
+def _page_size() -> int:
+    return max(1, min(int(settings.NEWS_API_PAGE_SIZE or 10), 50))
+
+
+def _archive_window() -> tuple[str, str]:
+    days = max(1, int(settings.NEWS_API_OLDER_DAYS or 7))
+    today = datetime.now(timezone.utc).date()
+    from_day = today - timedelta(days=days)
+    return from_day.isoformat(), today.isoformat()
+
+
+def _query_attempts(q: str, country: Optional[str]) -> list[dict]:
+    """Build a recency/country fallback ladder for NewsData lookups."""
+    primary_country = _clean_param(country or settings.NEWS_API_PRIMARY_COUNTRY or "in").lower()
+    recent_window = _clean_param(settings.NEWS_API_RECENT_TIMEFRAME or "1")
+    archive_from, archive_to = _archive_window()
+    base = {
+        "apikey": settings.NEWS_API_KEY,
+        "q": q,
+        "language": _configured_languages(),
+        "size": _page_size(),
+    }
+
+    attempts: list[dict] = []
+    countries: list[str | None] = [primary_country]
+    if country is None:
+        countries.append(None)
+    elif primary_country != _clean_param(settings.NEWS_API_PRIMARY_COUNTRY or "in").lower():
+        countries.append(_clean_param(settings.NEWS_API_PRIMARY_COUNTRY or "in").lower())
+        countries.append(None)
+
+    seen: set[tuple[str | None, str]] = set()
+    for country_code in countries:
+        latest_key = (country_code, "latest")
+        if latest_key not in seen:
+            seen.add(latest_key)
+            latest_params = dict(base)
+            latest_params["_endpoint"] = "latest"
+            latest_params["_url"] = settings.NEWS_API_BASE_URL
+            if country_code:
+                latest_params["country"] = country_code
+            if recent_window:
+                latest_params["timeframe"] = recent_window
+            attempts.append(latest_params)
+
+        archive_key = (country_code, "archive")
+        if archive_key not in seen:
+            seen.add(archive_key)
+            archive_params = dict(base)
+            archive_params["_endpoint"] = "archive"
+            archive_params["_url"] = settings.NEWS_API_ARCHIVE_BASE_URL
+            archive_params["from_date"] = archive_from
+            archive_params["to_date"] = archive_to
+            if country_code:
+                archive_params["country"] = country_code
+            attempts.append(archive_params)
+
+    return attempts
+
+
 def _domain_of(url: str) -> str:
     try:
         return urlparse(url).netloc.lower().replace("www.", "")
@@ -173,22 +249,56 @@ def _compute_truth_override(
         return None
 
 
-async def _fetch(q: str, country: Optional[str]) -> list[dict]:
-    params = {"apikey": settings.NEWS_API_KEY, "q": q, "language": "en", "size": 10, "country": country or "in"}
-    logger.info(f"News lookup query: {q!r} country={country or 'in'}")
-
+async def _fetch(params: dict) -> list[dict]:
+    url = params.get("_url") or settings.NEWS_API_BASE_URL
+    request_params = {k: v for k, v in params.items() if not k.startswith("_")}
+    redacted = {k: v for k, v in request_params.items() if k != "apikey"}
+    logger.info(f"News lookup query params: {redacted}")
     try:
         async with httpx.AsyncClient(timeout=httpx.Timeout(8.0, connect=3.0)) as c:
-            r = await c.get(settings.NEWS_API_BASE_URL, params=params)
+            r = await c.get(url, params=request_params)
             r.raise_for_status()
             results = (r.json() or {}).get("results") or []
-            logger.info(f"News lookup returned {len(results)} articles for query: {q!r}")
+            logger.info(f"News lookup returned {len(results)} articles for query: {request_params.get('q')!r}")
             return results
     except Exception as e:
-        logger.warning(f"News lookup failed (query={q!r}): {e}")
+        logger.warning(f"News lookup failed (query={params.get('q')!r}): {e}")
         return []
 
 
+def _collect_news_evidence(
+    articles: list[dict],
+    *,
+    seen: set[str],
+    trusted: List[TrustedSource],
+    contradictions: List[ContradictingEvidence],
+) -> None:
+    for art in articles:
+        url = art.get("link") or ""
+        if not url or url in seen:
+            continue
+        seen.add(url)
+
+        title = art.get("title") or ""
+        dom = _domain_of(url)
+        src_name = art.get("source_id") or dom or "news"
+
+        if _is_factcheck(url, title):
+            contradictions.append(ContradictingEvidence(
+                source_name=src_name, title=title, url=url, type="fact_check",
+            ))
+            continue
+
+        trusted.append(TrustedSource(
+            source_name=src_name,
+            title=title,
+            url=url,
+            description=art.get("description") or art.get("content"),
+            published_at=art.get("pubDate"),
+            relevance_score=_relevance(url),
+        ))
+
+
 async def search_news(
     keywords: List[str],
     limit: int = 6,
@@ -219,36 +329,17 @@ async def search_news_full(
         return NewsLookupResult([], [], 0)
 
     q = " ".join(keywords[:4])
-    articles = await _fetch(q, country)
-
+    total_articles = 0
     seen: set[str] = set()
     trusted: List[TrustedSource] = []
     contradictions: List[ContradictingEvidence] = []
 
-    for art in articles:
-        url = art.get("link") or ""
-        if not url or url in seen:
-            continue
-        seen.add(url)
-
-        title = art.get("title") or ""
-        dom = _domain_of(url)
-        src_name = art.get("source_id") or dom or "news"
-
-        if _is_factcheck(url, title):
-            contradictions.append(ContradictingEvidence(
-                source_name=src_name, title=title, url=url, type="fact_check",
-            ))
-            continue
-
-        trusted.append(TrustedSource(
-            source_name=src_name,
-            title=title,
-            url=url,
-            description=art.get("description") or art.get("content"),
-            published_at=art.get("pubDate"),
-            relevance_score=_relevance(url),
-        ))
+    for params in _query_attempts(q, country):
+        articles = await _fetch(params)
+        total_articles += len(articles)
+        _collect_news_evidence(articles, seen=seen, trusted=trusted, contradictions=contradictions)
+        if trusted or contradictions:
+            break
 
     trusted.sort(key=lambda s: -s.relevance_score)
     trusted = trusted[:limit]
@@ -270,7 +361,7 @@ async def search_news_full(
     return NewsLookupResult(
         trusted_sources=trusted,
         contradicting_evidence=contradictions[:limit],
-        total_articles=len(articles),
+        total_articles=total_articles,
         truth_override=truth_override,
         no_source_penalty=no_source_penalty,
     )

services/video_service.py

@@ -10,7 +10,7 @@ from PIL import Image
 
 from config import settings
 from models.model_loader import get_model_loader
-from services.image_service import _classify_vit
+from services.image_service import _classify_ffpp, _classify_vit
 from services.video_temporal import TemporalAnalysis, compute_temporal_score
 
 
@@ -107,6 +107,26 @@ def _score_efficientnet_face(eff, face) -> float:
     return float(eff._calibrate(raw_prob))
 
 
+def _blend_video_frame_scores(
+    *,
+    efficientnet_prob: Optional[float],
+    ffpp_prob: Optional[float],
+) -> float:
+    if ffpp_prob is not None and efficientnet_prob is not None:
+        total = settings.VIDEO_FFPP_WEIGHT + settings.VIDEO_EFFNET_WEIGHT
+        if total <= 0:
+            return float(ffpp_prob)
+        return float(
+            (settings.VIDEO_FFPP_WEIGHT * ffpp_prob + settings.VIDEO_EFFNET_WEIGHT * efficientnet_prob)
+            / total
+        )
+    if ffpp_prob is not None:
+        return float(ffpp_prob)
+    if efficientnet_prob is not None:
+        return float(efficientnet_prob)
+    return 0.0
+
+
 def _analyze_with_efficientnet(
     frames: List[Tuple[int, float, np.ndarray, Image.Image]],
 ) -> Tuple[List[FrameAnalysis], str, List[str], bool]:
@@ -137,13 +157,28 @@ def _analyze_with_efficientnet(
                 has_face = True
                 face_detector_used = "blazeface+crop_fallback"
 
-        fake_prob = 0.0
+        eff_prob: Optional[float] = None
+        ffpp_prob: Optional[float] = None
         label = "unknown"
         if has_face and faces:
             # Run EfficientNet on the best face/crop and apply the same calibration as image inference.
-            fake_prob = _score_efficientnet_face(eff, faces[0])
+            eff_prob = _score_efficientnet_face(eff, faces[0])
+            if settings.FFPP_ENABLED:
+                try:
+                    ffpp_res = _classify_ffpp(pil)
+                    if ffpp_res is not None:
+                        ffpp_prob = float(ffpp_res[0])
+                        if "ffpp-vit-local" not in models_used:
+                            models_used.append("ffpp-vit-local")
+                except Exception as exc:  # noqa: BLE001
+                    logger.debug(f"FFPP video frame scoring failed, using EfficientNet only: {exc}")
+            fake_prob = _blend_video_frame_scores(
+                efficientnet_prob=eff_prob,
+                ffpp_prob=ffpp_prob,
+            )
             label = "Fake" if fake_prob > 0.5 else "Real"
         elif not has_face:
+            fake_prob = 0.0
             label = "no_face"
 
         results.append(
@@ -155,7 +190,7 @@ def _analyze_with_efficientnet(
                 suspicious_prob=fake_prob,
                 is_suspicious=(fake_prob >= 0.5) and has_face,
                 has_face=has_face,
-                scored=has_face and faces,
+                scored=bool(has_face and faces),
             )
         )
 

tests/test_accuracy_regressions.py

@@ -1,14 +1,15 @@
 from __future__ import annotations
 
 import os
+import asyncio
 
 os.environ["DEBUG"] = "false"
 
 from schemas.common import TrustedSource
-from services.news_lookup import _compute_truth_override
+from services.news_lookup import _compute_truth_override, search_news_full
 from services.screenshot_service import OCRBox, extract_full_text
 from services.text_service import _scores_to_classification
-from utils.scoring import compute_video_authenticity_score
+from utils.scoring import apply_unverified_news_gate, compute_video_authenticity_score
 from schemas.common import ArtifactIndicator, ExifSummary, VLMComponentScore, VLMBreakdown
 from services.general_image_service import GeneralImageDetection, fuse_no_face_evidence
 
@@ -79,6 +80,105 @@ def test_truth_override_does_not_apply_from_headline_only_match(monkeypatch):
     assert override is None or not override.applied
 
 
+def test_unverified_news_gate_caps_real_scores_as_suspicious():
+    score, label, severity, reason = apply_unverified_news_gate(
+        92,
+        has_trusted_sources=False,
+        has_contradicting_evidence=False,
+        truth_override_applied=False,
+    )
+
+    assert score == 55
+    assert label == "Suspicious"
+    assert severity == "warning"
+    assert reason == "no_trusted_source"
+
+
+def test_unverified_news_gate_keeps_fake_scores_fake():
+    score, label, severity, reason = apply_unverified_news_gate(
+        18,
+        has_trusted_sources=False,
+        has_contradicting_evidence=False,
+        truth_override_applied=False,
+    )
+
+    assert score == 18
+    assert label == "Very Likely Fake"
+    assert severity == "critical"
+    assert reason == "no_trusted_source"
+
+
+def test_news_lookup_falls_back_from_recent_india_to_older_india(monkeypatch):
+    calls = []
+
+    async def fake_fetch(params):
+        calls.append(dict(params))
+        if params.get("country") == "in" and params.get("_endpoint") == "archive":
+            return [
+                {
+                    "link": "https://indianexpress.com/article/cities/kolkata/example",
+                    "title": "BJP leader aide shot dead in Bengal",
+                    "source_id": "indianexpress",
+                    "pubDate": "2026-05-07 00:43:00",
+                    "description": "Police launched an investigation.",
+                }
+            ]
+        return []
+
+    monkeypatch.setattr("services.news_lookup.settings.NEWS_API_KEY", "test-key")
+    monkeypatch.setattr("services.news_lookup._fetch", fake_fetch)
+    monkeypatch.setattr("services.news_lookup._compute_truth_override", lambda *args, **kwargs: None)
+
+    result = asyncio.run(
+        search_news_full(
+            ["BJP", "Suvendu", "Adhikari", "Madhyamgram"],
+            original_text="BJP leader Suvendu Adhikari's PA shot dead in West Bengal's Madhyamgram",
+        )
+    )
+
+    assert result.trusted_sources[0].source_name == "indianexpress"
+    assert calls[0]["country"] == "in"
+    assert calls[0]["timeframe"] == "1"
+    archive_call = next(call for call in calls if call.get("country") == "in" and call.get("_endpoint") == "archive")
+    assert archive_call["_url"].endswith("/archive")
+    assert "timeframe" not in archive_call
+    assert "from_date" in archive_call
+    assert "to_date" in archive_call
+
+
+def test_news_lookup_falls_back_to_global_when_india_has_no_results(monkeypatch):
+    calls = []
+
+    async def fake_fetch(params):
+        calls.append(dict(params))
+        if "country" not in params and params.get("timeframe") == "1":
+            return [
+                {
+                    "link": "https://www.reuters.com/world/example",
+                    "title": "US and EU announce new trade framework",
+                    "source_id": "reuters",
+                    "pubDate": "2026-05-07 01:05:00",
+                    "description": "Officials announced a new framework.",
+                }
+            ]
+        return []
+
+    monkeypatch.setattr("services.news_lookup.settings.NEWS_API_KEY", "test-key")
+    monkeypatch.setattr("services.news_lookup._fetch", fake_fetch)
+    monkeypatch.setattr("services.news_lookup._compute_truth_override", lambda *args, **kwargs: None)
+
+    result = asyncio.run(
+        search_news_full(
+            ["US", "EU", "trade", "framework"],
+            original_text="US and EU announce new trade framework",
+        )
+    )
+
+    assert result.trusted_sources[0].source_name == "reuters"
+    assert any(call.get("country") == "in" for call in calls)
+    assert any("country" not in call for call in calls)
+
+
 def test_no_face_fusion_uses_general_detector_forensic_and_exif_evidence():
     fused = fuse_no_face_evidence(
         general_fake_prob=0.72,
@@ -156,3 +256,109 @@ def test_no_face_image_route_skips_face_trained_classifiers(monkeypatch):
     assert result.ensemble_method == "no_face_general_forensic_fusion"
     assert result.models_used == ["test-general-detector"]
     assert result.no_face_analysis is not None
+
+
+def test_synthetic_still_override_keeps_strong_ai_detector_authoritative():
+    import services.image_service as image_service
+
+    adjusted, reason = image_service._apply_synthetic_still_overrides(
+        fake_prob=0.13,
+        general_fake_prob=0.93,
+        is_video_frame=False,
+    )
+
+    assert adjusted >= 0.90
+    assert reason == "general_detector_very_high(0.93)"
+
+
+def test_synthetic_still_override_does_not_affect_video_frame_route():
+    import services.image_service as image_service
+
+    adjusted, reason = image_service._apply_synthetic_still_overrides(
+        fake_prob=0.13,
+        general_fake_prob=0.93,
+        is_video_frame=True,
+    )
+
+    assert adjusted == 0.13
+    assert reason is None
+
+
+def test_heatmap_target_index_prefers_fake_label_tokens():
+    from types import SimpleNamespace
+
+    from models.heatmap_generator import _find_class_index
+
+    model = SimpleNamespace(config=SimpleNamespace(id2label={0: "real", 1: "fake"}))
+
+    assert _find_class_index(model, ("fake", "generated", "synthetic")) == 1
+
+
+def test_video_efficientnet_frame_scored_is_boolean(monkeypatch):
+    import numpy as np
+    from PIL import Image
+
+    import services.video_service as video_service
+
+    class FakeEfficientNet:
+        calibrator_applied = False
+
+        class FaceExtractor:
+            def process_image(self, img):
+                return {"faces": [np.zeros((16, 16, 3), dtype=np.uint8)]}
+
+        face_extractor = FaceExtractor()
+
+        def _fallback_face_crop(self, img):
+            return None
+
+    class FakeLoader:
+        def load_efficientnet(self):
+            return FakeEfficientNet()
+
+    monkeypatch.setattr(video_service, "get_model_loader", lambda: FakeLoader())
+    monkeypatch.setattr(video_service, "_score_efficientnet_face", lambda _eff, _face: 0.7)
+
+    frame = np.zeros((32, 32, 3), dtype=np.uint8)
+    results, *_ = video_service._analyze_with_efficientnet(
+        [(0, 0.0, frame, Image.fromarray(frame))]
+    )
+
+    assert results[0].scored is True
+    assert isinstance(results[0].scored, bool)
+
+
+def test_video_primary_path_weights_ffpp_vit_above_efficientnet(monkeypatch):
+    import numpy as np
+    from PIL import Image
+
+    import services.video_service as video_service
+
+    class FakeEfficientNet:
+        calibrator_applied = False
+
+        class FaceExtractor:
+            def process_image(self, img):
+                return {"faces": [np.zeros((16, 16, 3), dtype=np.uint8)]}
+
+        face_extractor = FaceExtractor()
+
+        def _fallback_face_crop(self, img):
+            return None
+
+    class FakeLoader:
+        def load_efficientnet(self):
+            return FakeEfficientNet()
+
+    monkeypatch.setattr(video_service, "get_model_loader", lambda: FakeLoader())
+    monkeypatch.setattr(video_service, "_score_efficientnet_face", lambda _eff, _face: 0.10)
+    monkeypatch.setattr(video_service, "_classify_ffpp", lambda _pil: (0.90, {"fake": 0.90, "real": 0.10}))
+
+    frame = np.zeros((32, 32, 3), dtype=np.uint8)
+    results, _detector, models_used, _calibrated = video_service._analyze_with_efficientnet(
+        [(0, 0.0, frame, Image.fromarray(frame))]
+    )
+
+    assert results[0].suspicious_prob > 0.60
+    assert results[0].label == "Fake"
+    assert "ffpp-vit-local" in models_used

utils/scoring.py

@@ -15,6 +15,7 @@ TRUST_SCALE = [
 # Score range for forced disagreement clamp
 UNCERTAIN_SCORE_LO = 56
 UNCERTAIN_SCORE_HI = 69
+UNVERIFIED_NEWS_SCORE_CAP = 55
 
 
 def _validate_weight_total(weights: list[float], context: str) -> None:
@@ -41,6 +42,31 @@ def get_verdict_label(score: int) -> Tuple[str, str]:
     return "Unknown", "warning"
 
 
+def apply_unverified_news_gate(
+    score: int,
+    *,
+    has_trusted_sources: bool,
+    has_contradicting_evidence: bool,
+    truth_override_applied: bool,
+) -> Tuple[int, str, str, str | None]:
+    """Prevent unverifiable news claims from receiving a real verdict.
+
+    The text classifier can judge writing style, but a news claim with no
+    corroborating trusted source should stay in the suspicious/verification band.
+    Already-fake scores remain fake; the gate only caps overly-real scores.
+    """
+    if has_trusted_sources or has_contradicting_evidence or truth_override_applied:
+        label, severity = get_verdict_label(score)
+        return score, label, severity, None
+
+    gated_score = min(score, UNVERIFIED_NEWS_SCORE_CAP)
+    if gated_score > 40:
+        return gated_score, "Suspicious", "warning", "no_trusted_source"
+
+    label, severity = get_verdict_label(gated_score)
+    return gated_score, label, severity, "no_trusted_source"
+
+
 def compute_video_authenticity_score(
     *,
     mean_suspicious_prob: float,