Upload 11 files

README.md

@@ -1,13 +1,14 @@
 ---
-title: SachAI
-emoji: 🦀
-colorFrom: green
-colorTo: red
+title: SachAI API
+emoji: 🔍
+colorFrom: blue
+colorTo: indigo
 sdk: gradio
-sdk_version: 6.30.0
-python_version: '3.12'
+sdk_version: 4.44.1
+python_version: "3.10"
 app_file: app.py
 pinned: false
 ---
 
-Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference
+Backend API for SachAI (deepfake image/video detection and news-claim checking).
+Endpoints: `POST /predict`, `POST /predict-video`, `POST /predict-text`.

ai_explainer.py

@@ -0,0 +1,97 @@
+def generate_ai_explanation(model_prediction, model_confidence, fact_check):
+    """
+    Creates human-readable explanation using classifier + Google Fact Check data
+    """
+
+    confidence = model_confidence
+
+    # If Google Fact Check verified the claim
+    if fact_check.get("verified"):
+        publisher = fact_check.get("publisher", "a fact-checking source")
+        rating = fact_check.get("rating", "unverified")
+        url = fact_check.get("url", "")
+
+        rating_lower = rating.lower()
+
+        if rating_lower == "false":
+            final_verdict = "FAKE"
+            verdict_color = "red"
+
+        elif "mostly false" in rating_lower:
+            final_verdict = "MISLEADING"
+            verdict_color = "orange"
+
+        elif "misleading" in rating_lower:
+            final_verdict = "MISLEADING"
+            verdict_color = "orange"
+
+        elif rating_lower == "true":
+            final_verdict = "REAL"
+            verdict_color = "green"
+
+        elif "mostly true" in rating_lower:
+            final_verdict = "REAL"
+            verdict_color = "green"
+
+        else:
+            final_verdict = "NEEDS CONTEXT"
+            verdict_color = "yellow"
+
+        if final_verdict == "FAKE":
+            explanation = (
+                f"{publisher} reviewed this claim and rated it '{rating}'. "
+                "The claim contains misinformation and contradicts verified fact-checking sources."
+            )
+        
+        elif final_verdict == "MISLEADING":
+            explanation = (
+                f"{publisher} reviewed this claim and rated it '{rating}'. "
+                "The claim may contain partial truths but is presented in a misleading or incomplete way."
+            )
+        
+        elif final_verdict == "REAL":
+            explanation = (
+                f"{publisher} reviewed this claim and rated it '{rating}'. "
+                "The claim aligns with verified reporting and appears credible."
+            )
+        
+        else:
+            explanation = (
+                f"{publisher} reviewed this claim and rated it '{rating}'. "
+                "Additional context is needed to fully assess this claim."
+            )
+
+        return {
+            "final_verdict": final_verdict,
+            "verdict_color": verdict_color,
+            "explanation": explanation,
+            "source_url": url,
+            "source_publisher": publisher
+        }
+
+    # Fallback if no fact-check data
+    if model_prediction == "FAKE":
+        final_verdict = "FAKE"
+        verdict_color = "red"
+
+        explanation = (
+            f"The classifier flagged this content as potentially fake with {confidence}% confidence. "
+            "No direct fact-check source was found, so manual verification is recommended."
+        )
+
+    else:
+        final_verdict = "REAL"
+        verdict_color = "green"
+
+        explanation = (
+            f"The classifier found this content likely credible with {confidence}% confidence. "
+            "However, no direct fact-check source was found."
+        )
+
+    return {
+        "final_verdict": final_verdict,
+        "verdict_color": verdict_color,
+        "explanation": explanation,
+        "source_url": None,
+        "source_publisher": None
+    }

app.py

@@ -0,0 +1,211 @@
+# ZeroGPU Spaces require a @spaces.GPU function to exist at startup, and `spaces`
+# must be imported before torch/tensorflow. The models here run on CPU, so this
+# probe is never actually called. Harmless when `spaces` isn't installed.
+try:
+    import spaces
+
+    @spaces.GPU
+    def _gpu_probe():
+        return True
+except ImportError:
+    pass
+
+from flask import Flask, request, jsonify
+from flask_cors import CORS
+import os
+import cv2
+import numpy as np
+from tensorflow.keras.models import load_model
+from video_utils import extract_frames
+from predict_utils import predict_frame, aggregate_results
+import threading
+from fake_news_model import predict_fake_news, warm_up
+from ai_explainer import generate_ai_explanation
+from source_verifier import verify_claim_with_google
+
+app = Flask(__name__)
+CORS(app)
+UPLOAD_FOLDER = "uploads"
+os.makedirs(UPLOAD_FOLDER, exist_ok=True)
+
+# ✅ Load your custom trained CNN model
+model = load_model("custom_cnn_model.h5")
+
+
+def predict_text_internal(text):
+    try:
+        model_result = predict_fake_news(text)
+        model_ok = bool(
+            model_result and "prediction" in model_result and "confidence" in model_result
+        )
+
+        fact_check = verify_claim_with_google(text)
+
+        # Need at least one signal: the classifier or a published fact-check.
+        if not model_ok and not fact_check.get("verified"):
+            return jsonify({
+                "error": "Fake news model failed",
+                "details": model_result
+            }), 500
+
+        if not model_ok:
+            model_result = {"prediction": "UNKNOWN", "confidence": 0}
+
+        ai_result = generate_ai_explanation(
+            model_result["prediction"],
+            model_result["confidence"],
+            fact_check
+        )
+
+        # Safety check
+        if not ai_result:
+            return jsonify({
+                "error": "AI explanation failed"
+            }), 500
+
+        source_label = "Read Full Fact Check" if ai_result.get("source_url") else None
+
+        analysis_mode = (
+            "Google Fact Check + AI Classifier"
+            if fact_check.get("verified")
+            else "AI Classifier Only"
+        )
+
+        confidence = model_result["confidence"]
+
+        if confidence >= 90:
+            confidence_label = "Very High"
+        elif confidence >= 75:
+            confidence_label = "High"
+        elif confidence >= 50:
+            confidence_label = "Moderate"
+        else:
+            confidence_label = "Low"
+
+        return jsonify({
+            "model_prediction": model_result["prediction"],
+            "model_confidence": model_result["confidence"],
+            "confidence_label": confidence_label,
+
+            "fact_check_rating": fact_check.get("rating"),
+
+            "final_verdict": ai_result.get("final_verdict"),
+            "verdict_color": ai_result.get("verdict_color"),
+
+            "ai_explanation": ai_result.get("explanation"),
+
+            "source_publisher": ai_result.get("source_publisher"),
+            "source_url": ai_result.get("source_url"),
+            "source_label": source_label,
+
+            "analysis_mode": analysis_mode
+        })
+
+    except Exception as e:
+        return jsonify({
+            "error": "predict_text_internal failed",
+            "details": str(e)
+        }), 500
+
+@app.route("/")
+def home():
+    return "Flask server is running with custom CNN! 🚀"
+
+@app.route("/predict", methods=["POST"])
+def predict():
+    if "image" not in request.files:
+        return jsonify({"error": "No image file provided"}), 400
+
+    file = request.files["image"]
+    if file.filename == "":
+        return jsonify({"error": "Empty filename"}), 400
+
+    filepath = os.path.join(UPLOAD_FOLDER, file.filename)
+    file.save(filepath)
+
+    # ✅ Preprocess image to match CNN input
+    img = cv2.imread(filepath)
+    if img is None:
+        return jsonify({"error": "Invalid image"}), 400
+
+    img = cv2.resize(img, (128, 128))  # Match training size
+    img = img.astype(np.float32) / 255.0
+    img = np.expand_dims(img, axis=0)
+
+    # ✅ Predict using custom CNN model
+    prediction = model.predict(img)[0][0]
+    print(f"Raw prediction score: {prediction}")
+
+    label = "real" if prediction >= 0.5 else "deepfake"
+    confidence = float(prediction if label == "real" else 1 - prediction)
+
+
+    result = {
+        "filename": file.filename,
+        "prediction": label,
+        "confidence": round(confidence, 2)
+    }
+    return jsonify(result)
+
+
+@app.route("/predict-video", methods=["POST"])
+def predict_video():
+    if "video" not in request.files:
+        return jsonify({"error": "No video uploaded"}), 400
+
+    video = request.files["video"]
+    video_path = os.path.join("uploads", video.filename)
+    video.save(video_path)
+
+    # ✅ Extract frames (now returns arrays, not paths)
+    frames = extract_frames(video_path)
+
+    # ✅ Predict
+    results = []
+
+    for frame in frames:
+        result = predict_frame(frame)
+        results.append(result)
+
+    print("Total frames:", len(frames))
+    print("Results stored:", len(results))
+
+    if len(results) == 0:
+        return jsonify({
+            "error": "No faces detected in video"
+        }), 400
+
+    # ✅ Aggregate
+    final = aggregate_results(results)
+
+    # ✅ Free memory (optional but good practice)
+    del frames
+
+    return jsonify({
+        "prediction": final["result"],
+        "confidence": final["confidence"],
+        "frames_analyzed": len(results)
+    })
+
+@app.route('/predict-text', methods=['POST'])
+def predict_text():
+    data = request.get_json()
+    text = data.get("text", "")
+
+    if not text:
+        return jsonify({"error": "No text provided"}), 400
+
+    return predict_text_internal(text)
+
+
+# Load the local fake-news model in the background so the first request is fast.
+threading.Thread(target=warm_up, daemon=True).start()
+
+if __name__ == "__main__":
+    # On a Hugging Face Space (SPACE_ID is set) listen publicly on 7860.
+    on_space = bool(os.environ.get("SPACE_ID"))
+    app.run(
+        host="0.0.0.0" if on_space else "127.0.0.1",
+        port=int(os.environ.get("PORT", 7860 if on_space else 5001)),
+    )
+

custom_cnn_model.h5

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:746d87faccdd53c0313483fa1b2639ba456db3f8fdc042a4ed74f0a7d933d63e
+size 39708648

fake_news_model.py

@@ -0,0 +1,61 @@
+import threading
+
+import torch
+from transformers import AutoModelForSequenceClassification, AutoTokenizer
+
+MODEL_ID = "hamzab/roberta-fake-news-classification"
+MAX_TOKENS = 512
+
+# Runs fully locally (free, offline after the first download).
+# Weights are cached by huggingface in ~/.cache/huggingface after first use.
+_lock = threading.Lock()
+_tokenizer = None
+_model = None
+_label_map = None
+
+
+def _load():
+    global _tokenizer, _model, _label_map
+    if _model is not None:
+        return
+    with _lock:
+        if _model is not None:
+            return
+        tokenizer = AutoTokenizer.from_pretrained(MODEL_ID)
+        model = AutoModelForSequenceClassification.from_pretrained(MODEL_ID).eval()
+        # config.id2label is {0: 'FAKE', 1: 'TRUE'}
+        _label_map = {
+            int(i): "FAKE" if str(name).upper() == "FAKE" else "REAL"
+            for i, name in model.config.id2label.items()
+        }
+        _tokenizer, _model = tokenizer, model
+
+
+def warm_up():
+    """Load the model ahead of the first request (call from a background thread)."""
+    try:
+        _load()
+    except Exception as e:
+        print(f"Fake-news model warm-up failed: {e}")
+
+
+def predict_fake_news(text):
+    try:
+        _load()
+        inputs = _tokenizer(
+            text, return_tensors="pt", truncation=True, max_length=MAX_TOKENS
+        )
+        with torch.no_grad():
+            probs = torch.softmax(_model(**inputs).logits, dim=-1)[0]
+
+        best = int(torch.argmax(probs))
+        return {
+            "prediction": _label_map[best],
+            "confidence": round(float(probs[best]) * 100, 2),
+        }
+
+    except Exception as e:
+        return {
+            "error": "Local model failed",
+            "details": str(e),
+        }

packages.txt

@@ -0,0 +1 @@
+libglib2.0-0

predict_utils.py

@@ -0,0 +1,74 @@
+import numpy as np
+import cv2
+from tensorflow.keras.models import load_model
+
+# ✅ LOAD MODEL
+model = load_model("custom_cnn_model.h5")
+
+# ✅ LOAD FACE DETECTOR
+face_cascade = cv2.CascadeClassifier("haarcascade_frontalface_default.xml")
+
+
+def detect_face(frame):
+    gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
+
+    faces = face_cascade.detectMultiScale(
+        gray,
+        scaleFactor=1.1,
+        minNeighbors=3,
+        minSize=(30, 30)
+    )
+
+    print("Faces detected:", len(faces))
+
+    # ✅ If no face → use full frame
+    if len(faces) == 0:
+        return frame
+
+    x, y, w, h = faces[0]
+    face = frame[y:y+h, x:x+w]
+
+    # ✅ Safety check
+    if face is None or face.size == 0:
+        return frame
+
+    return face
+
+
+def preprocess_frame(frame):
+    img = cv2.resize(frame, (128, 128))
+    img = img / 255.0
+    img = np.reshape(img, (1, 128, 128, 3))
+    return img
+
+
+def predict_frame(frame):
+    # 🔥 Step 1: detect face or fallback
+    face_or_frame = detect_face(frame)
+
+    # 🔥 Step 2: preprocess
+    img = preprocess_frame(face_or_frame)
+
+    # 🔥 Step 3: predict
+    prediction = model.predict(img, verbose=0)[0][0]
+
+    if prediction > 0.5:
+        return "REAL", float(1 - prediction)
+    else:
+        return "FAKE", float(prediction)
+    
+
+def aggregate_results(results):
+    fake_scores = [s for l, s in results if l == "FAKE"]
+    real_scores = [s for l, s in results if l == "REAL"]
+
+    if len(fake_scores) > len(real_scores):
+        return {
+            "result": "FAKE",
+            "confidence": sum(fake_scores) / len(results)
+        }
+    else:
+        return {
+            "result": "REAL",
+            "confidence": sum(real_scores) / len(results)
+        }

requirements.txt

@@ -0,0 +1,10 @@
+flask==3.0.3
+flask-cors==5.0.0
+python-dotenv==1.0.1
+requests==2.31.0
+numpy==1.24.3
+pillow==10.3.0
+opencv-python-headless==4.10.0.84
+tensorflow==2.15.0
+torch==2.4.0
+transformers==4.46.3

source_verifier.py

@@ -0,0 +1,72 @@
+import os
+
+import requests
+from dotenv import load_dotenv
+
+load_dotenv()
+
+GOOGLE_FACT_CHECK_API_KEY = os.getenv("GOOGLE_FACT_CHECK_API_KEY")
+
+
+def verify_claim_with_google(claim):
+    if not GOOGLE_FACT_CHECK_API_KEY:
+        return {
+            "verified": False,
+            "error": "GOOGLE_FACT_CHECK_API_KEY is not set",
+        }
+
+    url = "https://factchecktools.googleapis.com/v1alpha1/claims:search"
+
+    params = {
+        "query": claim,
+        "key": GOOGLE_FACT_CHECK_API_KEY,
+        "languageCode": "en"
+    }
+
+    try:
+        response = requests.get(url, params=params, timeout=10)
+    except requests.RequestException as e:
+        return {
+            "verified": False,
+            "error": str(e)
+        }
+
+    if response.status_code != 200:
+        return {
+            "verified": False,
+            "error": response.text
+        }
+
+    data = response.json()
+
+    claims = data.get("claims", [])
+
+    if not claims:
+        return {
+            "verified": False,
+            "message": "No fact-check data found."
+        }
+
+    first_claim = claims[0]
+
+    claimant = first_claim.get("claimant", "Unknown")
+    text = first_claim.get("text", claim)
+
+    reviews = first_claim.get("claimReview", [])
+
+    if reviews:
+        review = reviews[0]
+
+        return {
+            "verified": True,
+            "claim": text,
+            "claimant": claimant,
+            "publisher": review.get("publisher", {}).get("name", "Unknown"),
+            "rating": review.get("textualRating", "Unknown"),
+            "url": review.get("url", "")
+        }
+
+    return {
+        "verified": False,
+        "message": "Claim found but no review available."
+    }

video_utils.py

@@ -0,0 +1,19 @@
+import cv2
+
+def extract_frames(video_path, frame_skip=7):
+    cap = cv2.VideoCapture(video_path)
+    frames = []
+    count = 0
+
+    while cap.isOpened():
+        ret, frame = cap.read()
+        if not ret:
+            break
+
+        if count % frame_skip == 0:
+            frames.append(frame)
+
+        count += 1
+
+    cap.release()
+    return frames