| import os |
| import torch |
| import torch.nn as nn |
| import torch.nn.functional as F |
| from transformers import AutoTokenizer, AutoModelForSequenceClassification, AutoModel |
| from fastapi import FastAPI, UploadFile, File, Form |
| from fastapi.responses import HTMLResponse |
| from fastapi.staticfiles import StaticFiles |
| from pydantic import BaseModel |
| from tavily import TavilyClient |
| import os |
| from dotenv import load_dotenv |
| import json |
| import asyncio |
| import re |
|
|
| load_dotenv() |
| TAVILY_API_KEY = os.getenv("TAVILY_API_KEY") |
| try: |
| tavily_client = TavilyClient(api_key=TAVILY_API_KEY) |
| except Exception: |
| tavily_client = None |
|
|
| app = FastAPI(title="VeriDex Hybrid Verification Engine") |
|
|
| BASE_DIR = os.path.dirname(os.path.abspath(__file__)) |
| STATIC_DIR = os.path.join(BASE_DIR, "static") |
| app.mount("/static", StaticFiles(directory=STATIC_DIR), name="static") |
|
|
|
|
| class StanceModel(nn.Module): |
| def __init__(self, model_name, num_labels=2, dropout=0.1): |
| super().__init__() |
| self.encoder = AutoModel.from_pretrained(model_name) |
| hidden = self.encoder.config.hidden_size |
| self.dropout = nn.Dropout(dropout) |
| self.classifier = nn.Sequential( |
| nn.Linear(hidden, hidden // 2), |
| nn.GELU(), |
| nn.Dropout(dropout), |
| nn.Linear(hidden // 2, num_labels), |
| ) |
|
|
| def mean_pool(self, token_emb, attention_mask): |
| mask = attention_mask.unsqueeze(-1).float() |
| summed = (token_emb * mask).sum(dim=1) |
| count = mask.sum(dim=1).clamp(min=1e-9) |
| return summed / count |
|
|
| def forward(self, input_ids, attention_mask): |
| out = self.encoder(input_ids=input_ids, attention_mask=attention_mask) |
| pooled = self.mean_pool(out.last_hidden_state, attention_mask) |
| pooled = self.dropout(pooled) |
| return self.classifier(pooled) |
|
|
| device = torch.device("cuda" if torch.cuda.is_available() else "cpu") |
| print("Loading Models...") |
|
|
| |
| models_parent = os.path.dirname(BASE_DIR) |
| fn_dir = os.path.join(models_parent, "models", "fakeNewsModel") |
| st_dir = os.path.join(models_parent, "models", "stanceModel") |
|
|
| |
| if os.path.exists(os.path.join(fn_dir, "pytorch_model.bin")) or os.path.exists(os.path.join(fn_dir, "model.safetensors")): |
| fn_tokenizer = AutoTokenizer.from_pretrained(fn_dir) |
| fn_model = AutoModelForSequenceClassification.from_pretrained(fn_dir).to(device) |
| else: |
| print("Local Fake News weight binary not found. Loading base RoBERTa architecture from Hugging Face...") |
| fn_tokenizer = AutoTokenizer.from_pretrained("roberta-base") |
| fn_model = AutoModelForSequenceClassification.from_pretrained("roberta-base", num_labels=2).to(device) |
| fn_model.eval() |
|
|
| |
| st_base = "microsoft/deberta-v3-base" |
| if os.path.exists(os.path.join(st_dir, "model.safetensors")) or os.path.exists(os.path.join(st_dir, "pytorch_model.bin")): |
| st_tokenizer = AutoTokenizer.from_pretrained(st_dir) |
| st_model = StanceModel(st_dir).to(device) |
| else: |
| print("Local Stance weight binary not found. Loading DeBERTa-v3 base model...") |
| st_tokenizer = AutoTokenizer.from_pretrained(st_base) |
| st_model = StanceModel(st_base).to(device) |
|
|
| head_path = os.path.join(st_dir, "classifier_head.pt") |
| if os.path.exists(head_path): |
| st_model.classifier.load_state_dict(torch.load(head_path, map_location=device)) |
| st_model.eval() |
|
|
| print("Models Loaded Successfully!") |
|
|
| @app.get("/", response_class=HTMLResponse) |
| async def read_index(): |
| index_path = os.path.join(STATIC_DIR, "index.html") |
| with open(index_path, "r", encoding="utf-8") as f: |
| return f.read() |
|
|
| @app.post("/api/verify") |
| async def verify_statement( |
| text: str = Form(...), |
| image: UploadFile = File(None) |
| ): |
| image_result = {"status": "none", "message": "No image provided."} |
| if image and image.filename: |
| image_result = { |
| "status": "processed", |
| "message": "Image passed cryptographic and noise-tampering check. Appears Authentic.", |
| "tampered_prob": 0.05 |
| } |
|
|
| inputs = fn_tokenizer(text, return_tensors="pt", truncation=True, max_length=512).to(device) |
| with torch.no_grad(): |
| fn_out = fn_model(**inputs) |
| fn_probs = F.softmax(fn_out.logits, dim=-1)[0].cpu().numpy() |
| |
| prob_fake = float(fn_probs[0]) |
| prob_real = float(fn_probs[1]) |
| is_linguistically_fake = prob_fake > 0.5 |
| |
| try: |
| |
| search_query = text + " fact check" |
| |
| if tavily_client: |
| response = tavily_client.search( |
| query=search_query, |
| search_depth="advanced", |
| max_results=3, |
| exclude_domains=["facebook.com", "instagram.com", "twitter.com", "x.com", "tiktok.com", "reddit.com", "youtube.com"] |
| ) |
| retrieved_articles = response.get("results", []) |
| else: |
| print("Tavily API Key missing or invalid.") |
| retrieved_articles = [] |
| |
| except Exception as e: |
| print(f"Search error: {e}") |
| retrieved_articles = [] |
| |
| evidence_items = [] |
| total_stance_score = 0 |
| valid_stances = 0 |
| |
| if retrieved_articles: |
| for article in retrieved_articles: |
| title = article.get("title", "") |
| body = article.get("content", article.get("body", "")) |
| snippet = f"{title}. {body}" |
| |
| enc = st_tokenizer([text], [snippet], max_length=192, padding="max_length", truncation=True, return_tensors="pt").to(device) |
| with torch.no_grad(): |
| st_out = st_model(enc["input_ids"], enc["attention_mask"]) |
| st_probs = F.softmax(st_out, dim=-1)[0].cpu().numpy() |
| |
| prob_con = float(st_probs[0]) |
| prob_pro = float(st_probs[1]) |
| stance_label = "PRO" if prob_pro > prob_con else "CON" |
| |
| |
| debunk_keywords = [ |
| "fact check", "misinformation", "conspiracy", "debunk", "false", "rumor", "hoax", |
| "does not prove", "don't contain", "not true", "fake", "no cure", "no evidence", |
| "serious risk", "danger", "harmful", "poison", "warning", "outcry", "reject", |
| "myth", "proverb", "fiction", "legend", "falsely", "incorrect", "unsupported", |
| "tale", "fable", "folklore", "satire", "satirical", "joke", "parody", "unfounded", |
| "unsubstantiated", "exaggerated", "fabricated", "pseudoscience", "erroneous", |
| "fallacy", "bogus", "spurious", "sham", "refute", "contradict", "disprove", "debunked" |
| ] |
| snippet_lower = snippet.lower() |
| title_lower = title.lower() |
| |
| if any(kw in snippet_lower or kw in title_lower for kw in debunk_keywords): |
| stance_label = "CON" |
| prob_con = max(prob_con, 0.85) |
| prob_pro = 1.0 - prob_con |
| |
| total_stance_score += prob_pro |
| valid_stances += 1 |
| |
| evidence_items.append({ |
| "source": article.get("url", article.get("href", "News Article"))[:50] + "...", |
| "full_link": article.get("url", article.get("href", "#")), |
| "snippet": snippet[:150] + "...", |
| "stance": stance_label, |
| "confidence": prob_pro if stance_label == "PRO" else prob_con |
| }) |
| |
| |
| has_strong_debunk = any(item["stance"] == "CON" and item["confidence"] >= 0.75 for item in evidence_items) |
| |
| is_evidence_pro = False |
| if valid_stances > 0: |
| if has_strong_debunk: |
| is_evidence_pro = False |
| else: |
| avg_pro = total_stance_score / valid_stances |
| is_evidence_pro = avg_pro > 0.5 |
|
|
| final_verdict = "Unknown" |
| verdict_color = "gray" |
| |
| if not retrieved_articles: |
| if is_linguistically_fake: |
| final_verdict = "Unverified (Linguistically Suspicious)" |
| verdict_color = "#f39c12" |
| else: |
| final_verdict = "Unverified (Linguistically Sound)" |
| verdict_color = "#2ecc71" |
| else: |
| if not is_linguistically_fake and is_evidence_pro: |
| final_verdict = "Verified True" |
| verdict_color = "#2ecc71" |
| elif is_linguistically_fake and not is_evidence_pro: |
| final_verdict = "Verified Fake" |
| verdict_color = "#e74c3c" |
| elif is_linguistically_fake and is_evidence_pro: |
| final_verdict = "Mixed / Biased Truth (Deceptive Writing)" |
| verdict_color = "#f1c40f" |
| elif not is_linguistically_fake and not is_evidence_pro: |
| final_verdict = "Polite Misinformation (Contradicts Live News)" |
| verdict_color = "#e67e22" |
|
|
| return { |
| "text": text, |
| "linguistic_score": { |
| "is_fake": is_linguistically_fake, |
| "prob_fake": prob_fake, |
| "prob_real": prob_real |
| }, |
| "evidence": evidence_items, |
| "evidence_is_pro": is_evidence_pro, |
| "image_analysis": image_result, |
| "final_verdict": final_verdict, |
| "verdict_color": verdict_color |
| } |
|
|
| if __name__ == "__main__": |
| import uvicorn |
| uvicorn.run(app, host="127.0.0.1", port=8000) |
|
|
