run_api.py

#!/usr/bin/env python3
# ============================================================================
# agent verf - API with Real LLM Waterfall
# Version: 0.2.0
# Last Updated: 2026-01-24
#
# Real verification using Groq (triage) + Cerebras (synthesis)
# Run with: python3 run_api.py
# ============================================================================

import os
import sys
import json
import asyncio
import httpx
from uuid import uuid4, UUID
from datetime import datetime
from typing import Optional

# Load environment
from dotenv import load_dotenv
load_dotenv('.env.local')

from fastapi import FastAPI, HTTPException, BackgroundTasks
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import FileResponse
from pydantic import BaseModel
import uvicorn

# ============================================================================
# In-Memory Store
# ============================================================================

jobs = {}

# ============================================================================
# Models
# ============================================================================

class VerifyRequest(BaseModel):
    url: Optional[str] = None
    text: Optional[str] = None
    mode: str = "free"  # "free" or "venice"

class VerifyResponse(BaseModel):
    request_id: UUID
    status: str
    status_url: str
    estimated_seconds: int = 15

class StatusResponse(BaseModel):
    request_id: UUID
    status: str
    progress: Optional[float] = None
    current_step: Optional[str] = None
    receipt_url: Optional[str] = None
    error: Optional[str] = None

# ============================================================================
# LLM Calls
# ============================================================================

async def call_groq(prompt: str, max_tokens: int = 200) -> dict:
    """Call Groq for triage (8B model, fast)."""
    api_key = os.getenv("GROQ_API_KEY")
    if not api_key:
        raise Exception("GROQ_API_KEY not configured")

    async with httpx.AsyncClient() as client:
        response = await client.post(
            "https://api.groq.com/openai/v1/chat/completions",
            headers={
                "Authorization": f"Bearer {api_key}",
                "Content-Type": "application/json",
            },
            json={
                "model": "llama-3.1-8b-instant",
                "messages": [{"role": "user", "content": prompt}],
                "max_tokens": max_tokens,
                "temperature": 0.3,
            },
            timeout=30.0,
        )
        if response.status_code != 200:
            raise Exception(f"Groq error: {response.status_code} - {response.text[:200]}")
        return response.json()


async def call_cerebras(prompt: str, max_tokens: int = 500) -> dict:
    """Call Cerebras for synthesis (70B model, quality)."""
    api_key = os.getenv("CEREBRAS_API_KEY")
    if not api_key:
        raise Exception("CEREBRAS_API_KEY not configured")

    async with httpx.AsyncClient() as client:
        response = await client.post(
            "https://api.cerebras.ai/v1/chat/completions",
            headers={
                "Authorization": f"Bearer {api_key}",
                "Content-Type": "application/json",
            },
            json={
                "model": "llama-3.3-70b",
                "messages": [{"role": "user", "content": prompt}],
                "max_tokens": max_tokens,
                "temperature": 0.3,
            },
            timeout=60.0,
        )
        if response.status_code != 200:
            raise Exception(f"Cerebras error: {response.status_code} - {response.text[:200]}")
        return response.json()


async def call_venice(prompt: str, max_tokens: int = 500) -> dict:
    """Call Venice for premium users."""
    api_key = os.getenv("VENICE_API_KEY")
    if not api_key:
        raise Exception("VENICE_API_KEY not configured")

    async with httpx.AsyncClient() as client:
        response = await client.post(
            "https://api.venice.ai/api/v1/chat/completions",
            headers={
                "Authorization": f"Bearer {api_key}",
                "Content-Type": "application/json",
            },
            json={
                "model": "llama-3.3-70b",
                "messages": [{"role": "user", "content": prompt}],
                "max_tokens": max_tokens,
                "temperature": 0.3,
            },
            timeout=60.0,
        )
        if response.status_code != 200:
            raise Exception(f"Venice error: {response.status_code} - {response.text[:200]}")
        return response.json()


async def call_groq_vision(image_url: str, prompt: str = None) -> str:
    """Call Groq's Llama 3.2 Vision to extract text/content from an image."""
    api_key = os.getenv("GROQ_API_KEY")
    if not api_key:
        raise Exception("GROQ_API_KEY not configured")

    if prompt is None:
        prompt = """Analyze this image from a social media post. Extract and describe:
1. Any text visible in the image (quotes, headlines, captions, memes, screenshots)
2. Key visual content that makes factual claims
3. Any data, statistics, or numbers shown

Be concise and focus on factual claims that can be verified. If there's no text or verifiable claims, say "No verifiable content in image."
"""

    async with httpx.AsyncClient() as client:
        try:
            response = await client.post(
                "https://api.groq.com/openai/v1/chat/completions",
                headers={
                    "Authorization": f"Bearer {api_key}",
                    "Content-Type": "application/json",
                },
                json={
                    "model": "llama-3.2-90b-vision-preview",
                    "messages": [
                        {
                            "role": "user",
                            "content": [
                                {"type": "text", "text": prompt},
                                {"type": "image_url", "image_url": {"url": image_url}},
                            ],
                        }
                    ],
                    "max_tokens": 500,
                    "temperature": 0.3,
                },
                timeout=30.0,
            )
            if response.status_code != 200:
                print(f"Groq Vision error: {response.status_code} - {response.text[:200]}")
                return ""
            data = response.json()
            return data["choices"][0]["message"]["content"]
        except Exception as e:
            print(f"Groq Vision failed: {e}")
            return ""


# ============================================================================
# URL Content Extraction
# ============================================================================

def is_twitter_url(url: str) -> bool:
    """Check if URL is a Twitter/X post."""
    import re
    return bool(re.search(r'(twitter\.com|x\.com)/\w+/status/\d+', url))


def is_instagram_url(url: str) -> bool:
    """Check if URL is an Instagram post."""
    import re
    return bool(re.search(r'instagram\.com/(p|reel)/[\w-]+', url))


def get_twitter_token(tweet_id: str) -> str:
    """Generate token for syndication API (same algorithm as react-tweet)."""
    import math
    num = int(tweet_id)
    result = (num / 1e15) * math.pi
    # Convert to base 36
    chars = '0123456789abcdefghijklmnopqrstuvwxyz'
    token = ''
    n = abs(result)
    while n >= 1:
        token = chars[int(n % 36)] + token
        n = n // 36
    # Remove leading zeros and dots
    return token.lstrip('0').replace('.', '') or '0'


async def extract_twitter_content(url: str) -> Optional[dict]:
    """Extract tweet text and images from Twitter/X URL using syndication API (FREE, no auth)."""
    import re

    # Extract tweet ID from URL
    match = re.search(r'/status/(\d+)', url)
    if not match:
        return None

    tweet_id = match.group(1)
    token = get_twitter_token(tweet_id)

    # Build URL with all required parameters (matching react-tweet)
    params = {
        "id": tweet_id,
        "lang": "en",
        "token": token,
    }
    api_url = f"https://cdn.syndication.twimg.com/tweet-result"

    async with httpx.AsyncClient() as client:
        try:
            response = await client.get(api_url, params=params, timeout=10.0)
            if response.status_code == 200:
                data = response.json()

                # Extract text
                text = data.get("text", "")

                # Extract media URLs (photos)
                media_urls = []
                photos = data.get("photos", [])
                for photo in photos:
                    if photo.get("url"):
                        media_urls.append(photo["url"])

                # Also check mediaDetails (alternative structure)
                media_details = data.get("mediaDetails", [])
                for media in media_details:
                    if media.get("type") == "photo" and media.get("media_url_https"):
                        if media["media_url_https"] not in media_urls:
                            media_urls.append(media["media_url_https"])

                # Process images with vision model if present
                image_text = ""
                if media_urls:
                    print(f"Found {len(media_urls)} image(s) in tweet, processing with vision...")
                    for i, img_url in enumerate(media_urls[:3]):  # Limit to 3 images
                        try:
                            extracted = await call_groq_vision(img_url)
                            if extracted and "No verifiable content" not in extracted:
                                image_text += f"\n[IMAGE {i+1} CONTENT]: {extracted}"
                                print(f"  Image {i+1}: Extracted {len(extracted)} chars")
                        except Exception as e:
                            print(f"  Image {i+1}: Vision extraction failed - {e}")

                return {
                    "platform": "twitter",
                    "text": text,
                    "image_text": image_text.strip(),
                    "media_urls": media_urls,
                    "author": data.get("user", {}).get("screen_name", ""),
                    "created_at": data.get("created_at"),
                    "url": url,
                }
            else:
                print(f"Twitter API returned {response.status_code}")
        except Exception as e:
            print(f"Twitter extraction failed: {e}")
    return None


async def extract_url_content(url: str) -> Optional[dict]:
    """Extract content from social media URL."""
    if is_twitter_url(url):
        return await extract_twitter_content(url)
    elif is_instagram_url(url):
        # Instagram requires auth - return None, caller will use URL as-is
        return None
    return None


# ============================================================================
# Evidence Search (Brave)
# ============================================================================

async def search_brave(query: str, count: int = 5) -> list:
    """Search Brave for evidence. Returns list of {title, url, description}."""
    api_key = os.getenv("BRAVE_SEARCH_API_KEY")
    if not api_key:
        print("BRAVE_SEARCH_API_KEY not configured")
        return []

    async with httpx.AsyncClient() as client:
        try:
            response = await client.get(
                "https://api.search.brave.com/res/v1/web/search",
                headers={"X-Subscription-Token": api_key},
                params={"q": query, "count": count},
                timeout=15.0,
            )
            if response.status_code == 200:
                data = response.json()
                results = []
                for item in data.get("web", {}).get("results", [])[:count]:
                    results.append({
                        "title": item.get("title", ""),
                        "url": item.get("url", ""),
                        "description": item.get("description", ""),
                    })
                return results
            else:
                print(f"Brave search error: {response.status_code}")
        except Exception as e:
            print(f"Brave search failed: {e}")
    return []


def parse_json_response(content: str) -> dict:
    """Parse JSON from LLM response, handling markdown code blocks."""
    text = content.strip()
    if "```json" in text:
        text = text.split("```json")[1].split("```")[0]
    elif "```" in text:
        text = text.split("```")[1].split("```")[0]
    try:
        return json.loads(text.strip())
    except:
        return {"raw": content}


# ============================================================================
# Real Verification Pipeline
# ============================================================================

async def real_verification(job_id: UUID, content: str, mode: str):
    """Run real verification using LLM waterfall."""

    try:
        # Step 1: Triage - classify the content
        jobs[job_id] = {"status": "processing", "progress": 0.2, "current_step": "triage"}

        triage_prompt = f"""Analyze this social media post for fact-checking.

POST: "{content}"

Extract and classify any factual claims. Respond with JSON:
{{
    "claims": ["list of specific factual claims made"],
    "category": "factual|opinion|satire|unverifiable",
    "priority": "high|medium|low",
    "checkworthy": true/false,
    "reasoning": "brief explanation"
}}"""

        triage_response = await call_groq(triage_prompt, max_tokens=300)
        triage_content = triage_response["choices"][0]["message"]["content"]
        triage_result = parse_json_response(triage_content)

        # Step 2: Evidence gathering with Brave Search
        jobs[job_id] = {"status": "processing", "progress": 0.5, "current_step": "evidence"}

        # Build search query from claims
        claims = triage_result.get("claims", [])
        search_query = claims[0] if claims else content[:100]
        evidence = await search_brave(search_query, count=5)

        # Format evidence for the synthesis prompt
        evidence_text = ""
        if evidence:
            evidence_text = "\n\nEVIDENCE FROM WEB SEARCH:\n"
            for i, e in enumerate(evidence, 1):
                evidence_text += f"{i}. {e['title']}\n   URL: {e['url']}\n   {e['description'][:200]}\n\n"

        # Step 3: Synthesis - generate verdict
        jobs[job_id] = {"status": "processing", "progress": 0.8, "current_step": "synthesis"}

        claims_text = ", ".join(triage_result.get("claims", [content]))

        synthesis_prompt = f"""You are a fact-checker. Generate a verdict for these claims.

CLAIMS: {claims_text}

ORIGINAL POST: "{content}"
{evidence_text}
Analyze the claims against the evidence. Respond with JSON:
{{
    "verdict": "TRUE|MOSTLY_TRUE|MIXED|MOSTLY_FALSE|FALSE|UNVERIFIABLE",
    "confidence": 0.0-1.0,
    "summary": "2-3 sentence plain-language summary",
    "detailed_reasoning": "Thorough explanation of your analysis",
    "key_findings": ["finding 1", "finding 2", "finding 3"],
    "what_is_true": "what parts are accurate (if any)",
    "what_is_false": "what parts are inaccurate (if any)",
    "missing_context": "important context that was omitted"
}}"""

        # Use Venice for premium, Cerebras for free
        if mode == "venice" and os.getenv("VENICE_API_KEY"):
            try:
                synthesis_response = await call_venice(synthesis_prompt, max_tokens=600)
            except:
                synthesis_response = await call_cerebras(synthesis_prompt, max_tokens=600)
        else:
            synthesis_response = await call_cerebras(synthesis_prompt, max_tokens=600)

        synthesis_content = synthesis_response["choices"][0]["message"]["content"]
        synthesis_result = parse_json_response(synthesis_content)

        # Complete
        jobs[job_id] = {
            "status": "completed",
            "progress": 1.0,
            "current_step": "done",
            "receipt_url": f"http://localhost:8080/api/v1/receipt/{job_id}",
            "triage": triage_result,
            "verdict": synthesis_result.get("verdict", "UNVERIFIABLE"),
            "confidence": synthesis_result.get("confidence", 0.5),
            "summary": synthesis_result.get("summary", "Analysis complete."),
            "detailed_reasoning": synthesis_result.get("detailed_reasoning", ""),
            "key_findings": synthesis_result.get("key_findings", []),
            "what_is_true": synthesis_result.get("what_is_true", ""),
            "what_is_false": synthesis_result.get("what_is_false", ""),
            "missing_context": synthesis_result.get("missing_context", ""),
            "sources": evidence,  # Include search results as sources
            "original_content": content,
            "mode": mode,
        }

    except Exception as e:
        jobs[job_id] = {
            "status": "failed",
            "progress": 0.0,
            "current_step": "error",
            "error": str(e),
        }


# ============================================================================
# App
# ============================================================================

app = FastAPI(
    title="agent verf API",
    version="0.2.0",
    description="Social media fact-checking with LLM waterfall"
)

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)


@app.get("/")
async def root():
    """Serve the frontend UI."""
    import os
    index_path = os.path.join(os.path.dirname(__file__), "index.html")
    if os.path.exists(index_path):
        return FileResponse(index_path, media_type="text/html")
    # Fallback to API info if no frontend
    return {
        "name": "agent verf API",
        "version": "0.2.0",
        "status": "running",
        "docs": "/docs",
    }


@app.get("/health")
async def health():
    groq_ok = bool(os.getenv("GROQ_API_KEY"))
    cerebras_ok = bool(os.getenv("CEREBRAS_API_KEY"))
    venice_ok = bool(os.getenv("VENICE_API_KEY"))

    return {
        "status": "healthy" if (groq_ok and cerebras_ok) else "degraded",
        "timestamp": datetime.utcnow().isoformat(),
        "providers": {
            "groq": "ok" if groq_ok else "missing",
            "cerebras": "ok" if cerebras_ok else "missing",
            "venice": "ok" if venice_ok else "missing",
        },
        "modes": {
            "free": groq_ok and cerebras_ok,
            "venice": venice_ok,
        }
    }


@app.post("/api/v1/verify", response_model=VerifyResponse)
async def verify(request: VerifyRequest, background_tasks: BackgroundTasks):
    """Submit content for verification."""
    content = request.text
    source_url = request.url
    extracted = None

    # If URL provided, try to extract content
    if request.url and not request.text:
        extracted = await extract_url_content(request.url)
        if extracted:
            content = extracted.get("text", "")
            # Include image content if present
            if extracted.get("image_text"):
                content += "\n" + extracted["image_text"]
            print(f"Extracted from {extracted.get('platform')}: {content[:100]}...")
        else:
            # Fallback: use URL as content (user will need to provide text)
            content = request.url

    if not content:
        raise HTTPException(status_code=400, detail="'text' or 'url' required")

    job_id = uuid4()
    jobs[job_id] = {"status": "queued", "progress": 0.0, "source_url": source_url, "extracted": extracted}

    # Start real verification in background
    background_tasks.add_task(real_verification, job_id, content, request.mode)

    return VerifyResponse(
        request_id=job_id,
        status="queued",
        status_url=f"http://localhost:8080/api/v1/status/{job_id}",
        estimated_seconds=15,
    )


@app.post("/api/v1/verify/quick")
async def verify_quick(request: VerifyRequest):
    """Synchronous verification - waits for result."""
    content = request.text
    source_url = request.url
    extracted = None

    # If URL provided, try to extract content
    if request.url and not request.text:
        extracted = await extract_url_content(request.url)
        if extracted:
            content = extracted.get("text", "")
            # Include image content if present
            if extracted.get("image_text"):
                content += "\n" + extracted["image_text"]
            print(f"Extracted from {extracted.get('platform')}: {content[:100]}...")
        else:
            # Fallback: use URL as content
            content = request.url

    if not content:
        raise HTTPException(status_code=400, detail="'text' or 'url' required")

    job_id = uuid4()
    jobs[job_id] = {"status": "queued", "progress": 0.0, "source_url": source_url, "extracted": extracted}

    # Run verification and wait
    await real_verification(job_id, content, request.mode)

    job = jobs[job_id]
    if job.get("status") == "failed":
        raise HTTPException(status_code=500, detail=job.get("error", "Verification failed"))

    response = {
        "request_id": str(job_id),
        "verdict": job.get("verdict"),
        "confidence": job.get("confidence"),
        "summary": job.get("summary"),
        "key_findings": job.get("key_findings"),
        "sources": job.get("sources", []),  # Evidence links
        "mode": job.get("mode"),
    }

    # Include extraction info if URL was processed
    if extracted:
        response["source"] = {
            "platform": extracted.get("platform"),
            "author": extracted.get("author"),
            "url": source_url,
            "has_images": bool(extracted.get("media_urls")),
            "images_analyzed": len(extracted.get("media_urls", [])),
        }

    return response


@app.get("/api/v1/status/{job_id}", response_model=StatusResponse)
async def status(job_id: UUID):
    """Check verification status."""
    if job_id not in jobs:
        raise HTTPException(status_code=404, detail="Job not found")

    job = jobs[job_id]
    return StatusResponse(
        request_id=job_id,
        status=job.get("status", "unknown"),
        progress=job.get("progress"),
        current_step=job.get("current_step"),
        receipt_url=job.get("receipt_url"),
        error=job.get("error"),
    )


@app.get("/api/v1/extract")
async def extract(url: str):
    """Extract content from a social media URL (for testing)."""
    if not url:
        raise HTTPException(status_code=400, detail="'url' query param required")

    result = await extract_url_content(url)
    if result:
        return {"success": True, **result}

    # Check if it's a known platform we couldn't extract from
    if is_instagram_url(url):
        return {"success": False, "error": "Instagram requires authentication", "platform": "instagram"}

    return {"success": False, "error": "Unsupported or invalid URL"}


@app.get("/api/v1/receipt/{receipt_id}")
async def get_receipt(receipt_id: UUID):
    """Get full verification receipt."""
    job = jobs.get(receipt_id)
    if not job or job.get("status") != "completed":
        raise HTTPException(status_code=404, detail="Receipt not found or not ready")

    return {
        "id": str(receipt_id),
        "created_at": datetime.utcnow().isoformat(),
        "original_content": job.get("original_content"),
        "mode": job.get("mode"),
        "triage": job.get("triage"),
        "verdict": {
            "status": job.get("verdict"),
            "confidence": job.get("confidence"),
            "summary": job.get("summary"),
            "detailed_reasoning": job.get("detailed_reasoning"),
            "key_findings": job.get("key_findings"),
            "what_is_true": job.get("what_is_true"),
            "what_is_false": job.get("what_is_false"),
            "missing_context": job.get("missing_context"),
        },
        "sources": job.get("sources", []),  # Evidence links for user to verify
    }


# ============================================================================
# Run
# ============================================================================

if __name__ == "__main__":
    port = int(os.getenv("PORT", 8080))

    print("\n" + "=" * 60)
    print("  agent verf API - Real LLM Verification")
    print("=" * 60)
    print(f"  API:  http://localhost:{port}")
    print(f"  Docs: http://localhost:{port}/docs")
    print("=" * 60)

    # Check providers
    groq = "OK" if os.getenv("GROQ_API_KEY") else "MISSING"
    cerebras = "OK" if os.getenv("CEREBRAS_API_KEY") else "MISSING"
    venice = "OK" if os.getenv("VENICE_API_KEY") else "MISSING"
    brave = "OK" if os.getenv("BRAVE_SEARCH_API_KEY") else "MISSING"
    print(f"  Groq:     {groq}")
    print(f"  Cerebras: {cerebras}")
    print(f"  Venice:   {venice}")
    print(f"  Brave:    {brave}")
    print("=" * 60 + "\n")

    uvicorn.run(app, host="0.0.0.0", port=port)