feat: Add fake news detection with OCR text extraction

app/agent.py

@@ -38,7 +38,32 @@ def invoke_with_fallback(messages: list) -> str:
     raise RuntimeError(f"All Gemini models exhausted. Last error: {last_err}")
 
 
+def _merge_factcheck(result: AnalysisResult, fc: dict) -> AnalysisResult:
+    """Merge a fact-check result dict into an existing AnalysisResult."""
+    verdict = fc.get("verdict", "UNVERIFIABLE")
+    content_type = fc.get("content_type", "unknown")
+    simplified = fc.get("simplified_explanation", "")
+    fc_score = float(fc.get("risk_score", 0.5))
+    fc_threats = fc.get("threat_types", [])
+
+    combined_score = round(max(result.risk_score, fc_score * 0.6), 3)
+    combined_threats = list(set(result.threat_types + fc_threats))
+
+    return AnalysisResult(
+        risk_score=combined_score,
+        risk_level=_risk_level(combined_score),
+        verdict=verdict,
+        content_type=content_type,
+        threat_types=combined_threats,
+        explanation=result.explanation,
+        simplified_explanation=simplified,
+        tool_outputs={**result.tool_outputs, "fact_check": fc},
+    )
+
+
 def run_text_agent(text: str, url_flags: dict) -> AnalysisResult:
+    from app.tools.fakenews_tools import classify_and_fact_check
+
     system = (
         "You are a cybersecurity expert specializing in phishing detection. "
         "Analyse the provided text for phishing indicators: urgency language, "
@@ -51,7 +76,8 @@ def run_text_agent(text: str, url_flags: dict) -> AnalysisResult:
     raw = raw.strip().strip("```json").strip("```").strip()
     data = json.loads(raw)
     score = float(data["risk_score"])
-    return AnalysisResult(
+
+    base = AnalysisResult(
         risk_score=score,
         risk_level=_risk_level(score),
         threat_types=data.get("threat_types", []),
@@ -59,8 +85,11 @@ def run_text_agent(text: str, url_flags: dict) -> AnalysisResult:
         tool_outputs={"gemini_text": data, "url_scan": url_flags},
     )
 
+    fc = classify_and_fact_check(text)
+    return _merge_factcheck(base, fc)
 
-def run_image_agent(hf_result: dict, gemini_result: dict) -> AnalysisResult:
+
+def run_image_agent(hf_result: dict, gemini_result: dict, fc_result: dict | None = None) -> AnalysisResult:
     hf_score = hf_result.get("deepfake_score", 0.0)
     gemini_score = gemini_result.get("risk_score", 0.0)
     combined = round((hf_score * 0.5) + (gemini_score * 0.5), 3)
@@ -68,17 +97,20 @@ def run_image_agent(hf_result: dict, gemini_result: dict) -> AnalysisResult:
         set(hf_result.get("threat_types", []) + gemini_result.get("threat_types", []))
     )
     explanation = (
-        f"HuggingFace deepfake model: {hf_result.get('label', 'N/A')} "
+        f"Deepfake model: {hf_result.get('label', 'N/A')} "
         f"(confidence {hf_score:.2f}). "
         f"Gemini vision analysis: {gemini_result.get('explanation', '')}"
     )
-    return AnalysisResult(
+    base = AnalysisResult(
         risk_score=combined,
         risk_level=_risk_level(combined),
         threat_types=threat_types,
         explanation=explanation,
         tool_outputs={"hf_deepfake": hf_result, "gemini_vision": gemini_result},
     )
+    if fc_result:
+        return _merge_factcheck(base, fc_result)
+    return base
 
 
 def run_video_agent(gemini_result: dict, frame_scores: list[float]) -> AnalysisResult:
@@ -93,8 +125,15 @@ def run_video_agent(gemini_result: dict, frame_scores: list[float]) -> AnalysisR
     return AnalysisResult(
         risk_score=combined,
         risk_level=_risk_level(combined),
+        verdict="FAKE" if combined > 0.5 else "UNVERIFIABLE",
+        content_type="unknown",
         threat_types=gemini_result.get("threat_types", ["deepfake_video"]),
         explanation=explanation,
+        simplified_explanation=(
+            "This video shows signs of AI manipulation or deepfake content."
+            if combined > 0.5 else
+            "No definitive deepfake signals detected, but proceed with caution."
+        ),
         tool_outputs={"gemini_video": gemini_result, "frame_scores": frame_scores},
     )
 
@@ -114,7 +153,14 @@ def run_audio_agent(hf_result: dict, gemini_result: dict) -> AnalysisResult:
     return AnalysisResult(
         risk_score=combined,
         risk_level=_risk_level(combined),
+        verdict="FAKE" if combined > 0.5 else "UNVERIFIABLE",
+        content_type="unknown",
         threat_types=threat_types,
         explanation=explanation,
+        simplified_explanation=(
+            "This audio appears to be AI-generated or synthetically cloned. Do not trust its authenticity."
+            if combined > 0.5 else
+            "No strong deepfake signals in audio, but remain cautious."
+        ),
         tool_outputs={"hf_audio": hf_result, "gemini_audio": gemini_result},
     )

app/models.py

@@ -9,6 +9,9 @@ class TextRequest(BaseModel):
 class AnalysisResult(BaseModel):
     risk_score: float = Field(..., ge=0.0, le=1.0, description="0.0 = safe, 1.0 = critical threat")
     risk_level: str = Field(..., description="LOW | MEDIUM | HIGH | CRITICAL")
+    verdict: str = Field(default="", description="REAL | FAKE | UNVERIFIABLE | PHISHING")
+    content_type: str = Field(default="", description="phishing | fake_news | real_news | unknown")
     threat_types: list[str] = Field(default_factory=list)
     explanation: str
+    simplified_explanation: str = Field(default="", description="Plain-language verdict summary")
     tool_outputs: dict[str, Any] = Field(default_factory=dict)

app/routers/image.py

@@ -1,6 +1,7 @@
 from fastapi import APIRouter, UploadFile, File, HTTPException
 from app.models import AnalysisResult
-from app.tools.image_tools import hf_detect_image_deepfake, gemini_analyze_image
+from app.tools.image_tools import hf_detect_image_deepfake, gemini_analyze_image, gemini_extract_image_text
+from app.tools.fakenews_tools import extract_text_from_image_for_factcheck
 from app.agent import run_image_agent
 
 router = APIRouter()
@@ -15,6 +16,10 @@ async def analyze_image(file: UploadFile = File(...)):
         image_bytes = await file.read()
         hf_result = hf_detect_image_deepfake(image_bytes, mime_type=file.content_type)
         gemini_result = gemini_analyze_image(image_bytes, mime_type=file.content_type)
-        return run_image_agent(hf_result, gemini_result)
+
+        extracted_text = gemini_extract_image_text(image_bytes, mime_type=file.content_type)
+        fc_result = extract_text_from_image_for_factcheck(extracted_text) if extracted_text else None
+
+        return run_image_agent(hf_result, gemini_result, fc_result)
     except Exception as e:
         raise HTTPException(status_code=500, detail=str(e))

app/routers/text.py

@@ -1,6 +1,6 @@
 from fastapi import APIRouter, HTTPException
 from app.models import TextRequest, AnalysisResult
-from app.tools.text_tools import analyze_urls_in_text, gemini_analyze_text
+from app.tools.text_tools import analyze_urls_in_text
 from app.agent import run_text_agent
 
 router = APIRouter()
@@ -10,7 +10,7 @@ router = APIRouter()
 async def analyze_text(request: TextRequest):
     try:
         url_flags = analyze_urls_in_text(request.text)
-        gemini_data = gemini_analyze_text(request.text)
         return run_text_agent(request.text, url_flags)
     except Exception as e:
         raise HTTPException(status_code=500, detail=str(e))
+

app/tools/fakenews_tools.py

@@ -0,0 +1,119 @@
+"""
+Fake-news / fact-check tools.
+Uses Gemini with Google Search grounding (already configured).
+No additional API keys needed.
+"""
+import json
+import re
+from langchain_google_genai import ChatGoogleGenerativeAI
+from langchain_core.messages import HumanMessage, SystemMessage
+from app.config import GEMINI_API_KEY, GEMINI_MODEL, GEMINI_MODEL_FALLBACKS
+from app.tools.retry_utils import execute_with_retry
+
+try:
+    from google.genai import types as genai_types
+    _SEARCH_TOOL = genai_types.Tool(google_search=genai_types.GoogleSearch())
+    _HAS_SEARCH = True
+except Exception:
+    _SEARCH_TOOL = None
+    _HAS_SEARCH = False
+
+
+def _invoke_grounded(messages: list) -> str:
+    for model in [GEMINI_MODEL] + GEMINI_MODEL_FALLBACKS:
+        try:
+            llm = ChatGoogleGenerativeAI(
+                model=model,
+                google_api_key=GEMINI_API_KEY,
+                temperature=0.1,
+            )
+            kwargs = {"tools": [_SEARCH_TOOL]} if _HAS_SEARCH and _SEARCH_TOOL else {}
+            return execute_with_retry(lambda m=llm: m.invoke(messages, **kwargs).content)
+        except Exception as e:
+            if "429" not in str(e) and "RESOURCE_EXHAUSTED" not in str(e):
+                raise
+    raise RuntimeError("All Gemini models quota exhausted")
+
+
+def _clean_json(raw: str) -> str:
+    raw = raw.strip()
+    raw = re.sub(r"^```(?:json)?", "", raw).rstrip("`").strip()
+    return raw
+
+
+def classify_and_fact_check(text: str) -> dict:
+    """
+    Determines if the content is a phishing attempt or news/claim,
+    fact-checks it using Google Search grounding, and returns a
+    skeptical verdict leaning towards FAKE when evidence is weak.
+
+    Returns:
+        {
+          "content_type": "phishing|fake_news|real_news|unknown",
+          "verdict": "FAKE|REAL|UNVERIFIABLE|PHISHING",
+          "risk_score": float,
+          "checked_claims": [{"claim": str, "verdict": str, "source": str}],
+          "explanation": str,
+          "simplified_explanation": str,
+          "threat_types": [str]
+        }
+    """
+    system = (
+        "You are an extremely skeptical AI fact-checker and cybersecurity analyst, "
+        "similar to Twitter/X Community Notes and Grok fact-checking. "
+        "Your default assumption is that unverified or sensational content is FAKE or PHISHING. "
+        "You MUST use the Google Search tool to look up every specific claim, company, event, or person mentioned. "
+        "\n\nYour analysis pipeline:"
+        "\n1. CLASSIFY: Is this (a) a phishing/scam attempt, (b) a news claim or viral content, or (c) unknown?"
+        "\n2. EXTRACT: Identify all specific claims, company names, events, people, and dates."
+        "\n3. VERIFY: Search for each claim. Trusted sources: established news outlets, official .gov/.edu sites, Wikipedia."
+        "\n4. VERDICT: Apply these strict rules:"
+        "\n   - If phishing indicators found → verdict=PHISHING"
+        "\n   - If claims are confirmed by 2+ credible sources → verdict=REAL"
+        "\n   - If claims are contradicted by credible sources → verdict=FAKE"
+        "\n   - If evidence is weak, missing, or only from unverified sources → verdict=FAKE (lean skeptical)"
+        "\n   - If completely unverifiable even after searching → verdict=UNVERIFIABLE"
+        "\n5. EXPLAIN: Write a simplified_explanation as if explaining to a non-technical person in 2-3 sentences. "
+        "Be direct: start with 'This appears to be FAKE/REAL/etc because...'"
+        "\n\nRespond ONLY with valid JSON:"
+        '{"content_type": "phishing|fake_news|real_news|unknown", '
+        '"verdict": "FAKE|REAL|UNVERIFIABLE|PHISHING", '
+        '"risk_score": <float 0.0-1.0>, '
+        '"checked_claims": [{"claim": <str>, "verdict": <str>, "source": <str>}], '
+        '"explanation": <str>, '
+        '"simplified_explanation": <str>, '
+        '"threat_types": [<str>]}'
+    )
+    try:
+        raw = _invoke_grounded([SystemMessage(content=system), HumanMessage(content=text)])
+        if not isinstance(raw, str):
+            raw = str(raw)
+        return json.loads(_clean_json(raw))
+    except Exception as e:
+        return {
+            "content_type": "unknown",
+            "verdict": "UNVERIFIABLE",
+            "risk_score": 0.5,
+            "checked_claims": [],
+            "explanation": f"Fact-check failed: {e}",
+            "simplified_explanation": "Could not verify this content. Treat with caution.",
+            "threat_types": [],
+        }
+
+
+def extract_text_from_image_for_factcheck(gemini_text_response: str) -> dict:
+    """
+    Given text already extracted from an image (e.g. from gemini_analyze_image),
+    runs classify_and_fact_check on it.
+    """
+    if not gemini_text_response or len(gemini_text_response.strip()) < 10:
+        return {
+            "content_type": "unknown",
+            "verdict": "UNVERIFIABLE",
+            "risk_score": 0.0,
+            "checked_claims": [],
+            "explanation": "No text extracted from image to fact-check.",
+            "simplified_explanation": "No readable text was found in the image.",
+            "threat_types": [],
+        }
+    return classify_and_fact_check(gemini_text_response)

app/tools/image_tools.py

@@ -55,6 +55,36 @@ def hf_detect_image_deepfake(image_bytes: bytes, mime_type: str = "image/jpeg")
 
 
 
+def gemini_extract_image_text(image_bytes: bytes, mime_type: str = "image/jpeg") -> str:
+    """Extract all visible text/claims from an image for fact-checking."""
+    b64 = base64.b64encode(image_bytes).decode()
+    system = (
+        "You are an OCR and content extraction system. "
+        "Extract ALL visible text from this image verbatim: headlines, captions, tweets, posts, labels, watermarks. "
+        "If it contains news or claims, summarize the key claim in one sentence at the end. "
+        "Return plain text only, no JSON."
+    )
+    message = HumanMessage(
+        content=[
+            {"type": "text", "text": system},
+            {"type": "image_url", "image_url": {"url": f"data:{mime_type};base64,{b64}"}},
+        ]
+    )
+    from app.tools.retry_utils import execute_with_retry
+    for model in [GEMINI_MODEL] + GEMINI_MODEL_FALLBACKS:
+        try:
+            resp = execute_with_retry(
+                lambda m=model: ChatGoogleGenerativeAI(
+                    model=m, google_api_key=GEMINI_API_KEY, temperature=0.0
+                ).invoke([message])
+            )
+            return resp.content if isinstance(resp.content, str) else str(resp.content)
+        except Exception as e:
+            if "429" not in str(e) and "RESOURCE_EXHAUSTED" not in str(e):
+                raise
+    return ""
+
+
 def gemini_analyze_image(image_bytes: bytes, mime_type: str = "image/jpeg") -> dict:
     b64 = base64.b64encode(image_bytes).decode()
     system = (