Deploy FN Detector backend

app.py

@@ -2,7 +2,7 @@ from pathlib import Path
 import os
 import sys
 
-from fastapi import FastAPI, HTTPException
+from fastapi import FastAPI, File, HTTPException, UploadFile
 from fastapi.middleware.cors import CORSMiddleware
 from pydantic import BaseModel, Field
 
@@ -12,6 +12,10 @@ SRC_DIR = ROOT / "src"
 if str(SRC_DIR) not in sys.path:
     sys.path.insert(0, str(SRC_DIR))
 
+from fake_news_detector.deepfake_detection import (  # noqa: E402
+    get_deepfake_model_snapshot,
+    predict_deepfake_image,
+)
 from fake_news_detector.prediction import (  # noqa: E402
     combine_news_text,
     get_model_snapshot,
@@ -29,6 +33,7 @@ def _parse_cors_origins() -> list[str]:
 
 
 MODEL_DIR = ROOT / os.environ.get("MODEL_DIR", "deployment/model")
+DEEPFAKE_MODEL_DIR = ROOT / os.environ.get("DEEPFAKE_MODEL_DIR", "deployment/deepfake_model")
 
 app = FastAPI(
     title="FN Detector API",
@@ -56,12 +61,25 @@ def root() -> dict:
         "service": "fn-detector-api",
         "status": "online",
         "model_dir": str(MODEL_DIR),
-        "routes": ["/health", "/metrics", "/predict"],
+        "deepfake_model_dir": str(DEEPFAKE_MODEL_DIR),
+        "routes": [
+            "/health",
+            "/metrics",
+            "/predict",
+            "/deepfake/health",
+            "/deepfake/metrics",
+            "/deepfake/predict",
+            "/api/health",
+            "/api/metrics",
+            "/api/predict",
+            "/api/deepfake/health",
+            "/api/deepfake/metrics",
+            "/api/deepfake/predict",
+        ],
     }
 
 
-@app.get("/health")
-def health() -> dict:
+def _news_health() -> dict:
     try:
         snapshot = get_model_snapshot(MODEL_DIR)
     except FileNotFoundError as exc:
@@ -78,8 +96,13 @@ def health() -> dict:
     }
 
 
-@app.get("/metrics")
-def metrics() -> dict:
+@app.get("/health")
+@app.get("/api/health")
+def health() -> dict:
+    return _news_health()
+
+
+def _news_metrics() -> dict:
     try:
         return {
             "status": "ok",
@@ -93,8 +116,13 @@ def metrics() -> dict:
         raise HTTPException(status_code=500, detail=f"Unable to read metrics: {exc}") from exc
 
 
-@app.post("/predict")
-def predict(request: PredictionRequest) -> dict:
+@app.get("/metrics")
+@app.get("/api/metrics")
+def metrics() -> dict:
+    return _news_metrics()
+
+
+def _news_predict(request: PredictionRequest) -> dict:
     combined_text = combine_news_text(request.title, request.text)
     if not combined_text:
         raise HTTPException(status_code=400, detail="Please provide a title or article body.")
@@ -117,3 +145,68 @@ def predict(request: PredictionRequest) -> dict:
         },
         "prediction": prediction,
     }
+
+
+@app.post("/predict")
+@app.post("/api/predict")
+def predict(request: PredictionRequest) -> dict:
+    return _news_predict(request)
+
+
+def _deepfake_health() -> dict:
+    try:
+        snapshot = get_deepfake_model_snapshot(DEEPFAKE_MODEL_DIR)
+    except FileNotFoundError as exc:
+        raise HTTPException(status_code=503, detail=str(exc)) from exc
+    return {
+        "status": "ok",
+        "service": "fn-detector-api",
+        "deepfake_model": snapshot["summary"].get("model_name"),
+        "mode": snapshot["summary"].get("status"),
+        "feature_count": len(snapshot["feature_names"]),
+    }
+
+
+@app.get("/deepfake/health")
+@app.get("/api/deepfake/health")
+def deepfake_health() -> dict:
+    return _deepfake_health()
+
+
+def _deepfake_metrics() -> dict:
+    try:
+        return {
+            "status": "ok",
+            "snapshot": get_deepfake_model_snapshot(DEEPFAKE_MODEL_DIR),
+        }
+    except FileNotFoundError as exc:
+        raise HTTPException(status_code=503, detail=str(exc)) from exc
+
+
+@app.get("/deepfake/metrics")
+@app.get("/api/deepfake/metrics")
+def deepfake_metrics() -> dict:
+    return _deepfake_metrics()
+
+
+def _deepfake_predict(filename: str, payload: bytes) -> dict:
+    try:
+        prediction = predict_deepfake_image(
+            model_dir=DEEPFAKE_MODEL_DIR,
+            image_bytes=payload,
+            filename=filename,
+        )
+    except ValueError as exc:
+        raise HTTPException(status_code=400, detail=str(exc)) from exc
+    except FileNotFoundError as exc:
+        raise HTTPException(status_code=503, detail=str(exc)) from exc
+    except Exception as exc:  # pragma: no cover - defensive for deployment only
+        raise HTTPException(status_code=500, detail=f"Deepfake prediction failed: {exc}") from exc
+
+    return {"status": "ok", "prediction": prediction}
+
+
+@app.post("/deepfake/predict")
+@app.post("/api/deepfake/predict")
+async def deepfake_predict(file: UploadFile = File(...)) -> dict:
+    return _deepfake_predict(file.filename or "upload", await file.read())

deployment/deepfake_model/bundle.json

@@ -0,0 +1,105 @@
+{
+  "feature_names": [
+    "gray_mean",
+    "gray_std",
+    "laplacian_var",
+    "gradient_mean",
+    "gradient_std",
+    "high_frequency_ratio",
+    "blockiness",
+    "jpeg_residual_mean",
+    "jpeg_residual_std",
+    "mirror_difference",
+    "saturation_mean",
+    "saturation_std"
+  ],
+  "scaler": {
+    "mean": [
+      0.3894135057926178,
+      0.16643017530441284,
+      0.0016245938604697585,
+      0.01306572463363409,
+      0.0241470355540514,
+      0.30300813913345337,
+      1.007244348526001,
+      0.008029050193727016,
+      0.007909948006272316,
+      0.11933569610118866,
+      0.17327596247196198,
+      0.07436472177505493
+    ],
+    "scale": [
+      0.0831027701497078,
+      0.04393522068858147,
+      0.0009263058891519904,
+      0.004046801012009382,
+      0.006974863354116678,
+      0.03189605101943016,
+      0.08480057865381241,
+      0.001521392841823399,
+      0.0017849425785243511,
+      0.06415991485118866,
+      0.04083705693483353,
+      0.032066166400909424
+    ]
+  },
+  "model": {
+    "coefficients": [
+      -0.1552580323,
+      0.4495079971,
+      -0.1030570684,
+      0.6508357217,
+      -0.2056009272,
+      -0.4020637292,
+      -0.1068058709,
+      -0.3841619248,
+      -0.6691807026,
+      0.9157355483,
+      -1.0115699767,
+      0.3580950632
+    ],
+    "intercept": 0.0585737738
+  },
+  "training_summary": {
+    "status": "trained",
+    "model_name": "lightweight-deepfake-linear",
+    "dataset_dir": "C:\\Users\\harma\\OneDrive\\Desktop\\FN\\FNdetector\\external\\DeepFake-Detect\\prepared_dataset",
+    "dataset_rows": 22,
+    "train_rows": 16,
+    "test_rows": 6,
+    "max_images_per_label": 32,
+    "accuracy": 1.0,
+    "classification_report": {
+      "real": {
+        "precision": 1.0,
+        "recall": 1.0,
+        "f1-score": 1.0,
+        "support": 3.0
+      },
+      "fake": {
+        "precision": 1.0,
+        "recall": 1.0,
+        "f1-score": 1.0,
+        "support": 3.0
+      },
+      "accuracy": 1.0,
+      "macro avg": {
+        "precision": 1.0,
+        "recall": 1.0,
+        "f1-score": 1.0,
+        "support": 6.0
+      },
+      "weighted avg": {
+        "precision": 1.0,
+        "recall": 1.0,
+        "f1-score": 1.0,
+        "support": 6.0
+      }
+    },
+    "system_profile": {
+      "cpu": "Intel Core i7-8550U class",
+      "ram_gb": 16,
+      "recommended_reason": "Conservative cap to avoid memory spikes on laptop-class CPUs."
+    }
+  }
+}

requirements.txt

@@ -3,3 +3,5 @@ transformers>=4.51,<5.0
 safetensors>=0.5,<1.0
 fastapi>=0.117,<1.0
 uvicorn>=0.35,<1.0
+python-multipart>=0.0.20,<1.0
+Pillow>=11.2,<12.0

src/fake_news_detector/deepfake_detection.py

@@ -0,0 +1,257 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from functools import lru_cache
+import io
+import json
+from pathlib import Path
+
+import numpy as np
+from PIL import Image
+
+
+IMAGE_SIZE = 128
+MAX_IMAGE_BYTES = 8 * 1024 * 1024
+FEATURE_NAMES = [
+    "gray_mean",
+    "gray_std",
+    "laplacian_var",
+    "gradient_mean",
+    "gradient_std",
+    "high_frequency_ratio",
+    "blockiness",
+    "jpeg_residual_mean",
+    "jpeg_residual_std",
+    "mirror_difference",
+    "saturation_mean",
+    "saturation_std",
+]
+
+
+@dataclass(frozen=True)
+class DeepfakeModelBundle:
+    feature_names: tuple[str, ...]
+    mean: np.ndarray
+    scale: np.ndarray
+    coefficients: np.ndarray
+    intercept: float
+    training_summary: dict
+
+
+def _sigmoid(value: float) -> float:
+    return float(1.0 / (1.0 + np.exp(-value)))
+
+
+def _resolve_model_dir(model_dir: Path) -> Path:
+    resolved = Path(model_dir).resolve()
+    if not resolved.exists():
+        raise FileNotFoundError(f"Deepfake model directory does not exist: {resolved}")
+    return resolved
+
+
+def _load_image(source: bytes) -> Image.Image:
+    if not source:
+        raise ValueError("No image content was provided.")
+    if len(source) > MAX_IMAGE_BYTES:
+        raise ValueError("Image is too large. Please upload an image under 8 MB.")
+    try:
+        image = Image.open(io.BytesIO(source))
+    except Exception as exc:  # pragma: no cover - defensive for malformed uploads
+        raise ValueError("Unable to decode the uploaded image.") from exc
+    return image.convert("RGB")
+
+
+def _prepare_rgb_array(image: Image.Image) -> np.ndarray:
+    resized = image.resize((IMAGE_SIZE, IMAGE_SIZE), Image.Resampling.BILINEAR)
+    return np.asarray(resized, dtype=np.float32) / 255.0
+
+
+def _compute_gray(rgb: np.ndarray) -> np.ndarray:
+    return np.dot(rgb[..., :3], np.array([0.2989, 0.5870, 0.1140], dtype=np.float32))
+
+
+def _compute_laplacian(gray: np.ndarray) -> np.ndarray:
+    center = gray[1:-1, 1:-1]
+    return (
+        gray[:-2, 1:-1]
+        + gray[2:, 1:-1]
+        + gray[1:-1, :-2]
+        + gray[1:-1, 2:]
+        - (4.0 * center)
+    )
+
+
+def _compute_frequency_ratio(gray: np.ndarray) -> float:
+    centered = gray - float(gray.mean())
+    spectrum = np.abs(np.fft.rfft2(centered))
+    if not np.any(spectrum):
+        return 0.0
+
+    height, width = gray.shape
+    y_coords = np.fft.fftfreq(height)[:, None]
+    x_coords = np.fft.rfftfreq(width)[None, :]
+    radius = np.sqrt((y_coords**2) + (x_coords**2))
+    high_mask = radius >= 0.18
+    high_energy = float(spectrum[high_mask].sum())
+    total_energy = float(spectrum.sum()) + 1e-8
+    return high_energy / total_energy
+
+
+def _compute_blockiness(gray: np.ndarray) -> float:
+    vertical_boundaries = gray[:, 8::8] - gray[:, 7:-1:8]
+    horizontal_boundaries = gray[8::8, :] - gray[7:-1:8, :]
+    all_vertical = np.diff(gray, axis=1)
+    all_horizontal = np.diff(gray, axis=0)
+    boundary_energy = float(np.mean(np.abs(vertical_boundaries))) + float(
+        np.mean(np.abs(horizontal_boundaries))
+    )
+    overall_energy = float(np.mean(np.abs(all_vertical))) + float(np.mean(np.abs(all_horizontal))) + 1e-8
+    return boundary_energy / overall_energy
+
+
+def _compute_jpeg_residual(rgb_uint8: np.ndarray) -> tuple[float, float]:
+    image = Image.fromarray(rgb_uint8, mode="RGB")
+    buffer = io.BytesIO()
+    image.save(buffer, format="JPEG", quality=72, optimize=True)
+    recompressed = Image.open(io.BytesIO(buffer.getvalue())).convert("RGB")
+    diff = np.abs(rgb_uint8.astype(np.float32) - np.asarray(recompressed, dtype=np.float32)) / 255.0
+    return float(diff.mean()), float(diff.std())
+
+
+def _compute_saturation(rgb: np.ndarray) -> np.ndarray:
+    channel_max = rgb.max(axis=2)
+    channel_min = rgb.min(axis=2)
+    return channel_max - channel_min
+
+
+def extract_feature_dict(image: Image.Image) -> dict[str, float]:
+    rgb = _prepare_rgb_array(image)
+    rgb_uint8 = np.clip(np.round(rgb * 255.0), 0, 255).astype(np.uint8)
+    gray = _compute_gray(rgb)
+    laplacian = _compute_laplacian(gray)
+    gradients = np.concatenate(
+        [
+            np.abs(np.diff(gray, axis=0)).ravel(),
+            np.abs(np.diff(gray, axis=1)).ravel(),
+        ]
+    )
+    saturation = _compute_saturation(rgb)
+    mirrored = np.flip(rgb, axis=1)
+    jpeg_mean, jpeg_std = _compute_jpeg_residual(rgb_uint8)
+
+    values = {
+        "gray_mean": float(gray.mean()),
+        "gray_std": float(gray.std()),
+        "laplacian_var": float(laplacian.var()),
+        "gradient_mean": float(gradients.mean()),
+        "gradient_std": float(gradients.std()),
+        "high_frequency_ratio": _compute_frequency_ratio(gray),
+        "blockiness": _compute_blockiness(gray),
+        "jpeg_residual_mean": jpeg_mean,
+        "jpeg_residual_std": jpeg_std,
+        "mirror_difference": float(np.mean(np.abs(rgb - mirrored))),
+        "saturation_mean": float(saturation.mean()),
+        "saturation_std": float(saturation.std()),
+    }
+    return values
+
+
+def extract_feature_vector(image: Image.Image) -> np.ndarray:
+    feature_dict = extract_feature_dict(image)
+    return np.array([feature_dict[name] for name in FEATURE_NAMES], dtype=np.float32)
+
+
+def _default_training_summary(model_dir: Path) -> dict:
+    return {
+        "status": "heuristic",
+        "model_name": "artifact-heuristic-baseline",
+        "model_dir": str(model_dir),
+        "notes": "No trained lightweight deepfake bundle was found, so the fallback heuristic is active.",
+    }
+
+
+@lru_cache(maxsize=2)
+def load_deepfake_bundle(model_dir_value: str) -> DeepfakeModelBundle | None:
+    model_dir = _resolve_model_dir(Path(model_dir_value))
+    bundle_path = model_dir / "bundle.json"
+    if not bundle_path.exists():
+        return None
+
+    payload = json.loads(bundle_path.read_text(encoding="utf-8"))
+    return DeepfakeModelBundle(
+        feature_names=tuple(payload.get("feature_names", FEATURE_NAMES)),
+        mean=np.array(payload["scaler"]["mean"], dtype=np.float32),
+        scale=np.array(payload["scaler"]["scale"], dtype=np.float32),
+        coefficients=np.array(payload["model"]["coefficients"], dtype=np.float32),
+        intercept=float(payload["model"]["intercept"]),
+        training_summary=payload.get("training_summary", {}),
+    )
+
+
+def get_deepfake_model_snapshot(model_dir: Path) -> dict:
+    resolved = _resolve_model_dir(model_dir)
+    bundle = load_deepfake_bundle(str(resolved))
+    summary = bundle.training_summary if bundle else _default_training_summary(resolved)
+
+    return {
+        "model_dir": str(resolved),
+        "available": bundle is not None,
+        "feature_names": FEATURE_NAMES,
+        "summary": summary,
+    }
+
+
+def _heuristic_prediction(features: dict[str, float]) -> float:
+    raw_score = (
+        2.8 * features["jpeg_residual_mean"]
+        + 1.6 * features["high_frequency_ratio"]
+        + 0.7 * max(features["blockiness"] - 1.0, 0.0)
+        + 1.4 * features["mirror_difference"]
+        + 8.0 * features["laplacian_var"]
+        - 0.6 * features["saturation_mean"]
+    )
+    centered = (raw_score - 0.22) * 4.0
+    return min(max(_sigmoid(centered), 0.02), 0.98)
+
+
+def predict_deepfake_image(model_dir: Path, image_bytes: bytes, filename: str | None = None) -> dict:
+    image = _load_image(image_bytes)
+    features = extract_feature_dict(image)
+    vector = np.array([features[name] for name in FEATURE_NAMES], dtype=np.float32)
+
+    bundle = load_deepfake_bundle(str(_resolve_model_dir(model_dir)))
+    if bundle is None:
+        fake_score = _heuristic_prediction(features)
+        model_name = "artifact-heuristic-baseline"
+        model_status = "heuristic"
+    else:
+        normalized = (vector - bundle.mean) / np.where(bundle.scale == 0, 1.0, bundle.scale)
+        logit = float(np.dot(normalized, bundle.coefficients) + bundle.intercept)
+        fake_score = _sigmoid(logit)
+        model_name = str(bundle.training_summary.get("model_name", "lightweight-deepfake-linear"))
+        model_status = "trained"
+
+    real_score = 1.0 - fake_score
+    top_feature_names = sorted(
+        FEATURE_NAMES,
+        key=lambda feature_name: abs(features[feature_name]),
+        reverse=True,
+    )[:4]
+
+    return {
+        "prediction": "fake" if fake_score >= 0.5 else "real",
+        "confidence": float(max(fake_score, real_score)),
+        "scores": {
+            "fake": float(fake_score),
+            "real": float(real_score),
+        },
+        "model_name": model_name,
+        "model_status": model_status,
+        "filename": filename or "upload",
+        "image_size": {"width": image.width, "height": image.height},
+        "features": features,
+        "top_signals": [
+            {"name": name, "value": float(features[name])}
+            for name in top_feature_names
+        ],
+    }

src/fake_news_detector/prediction.py

@@ -2,8 +2,12 @@ from dataclasses import dataclass
 from functools import lru_cache
 from pathlib import Path
 import json
+import os
 import re
 
+os.environ.setdefault("TRANSFORMERS_NO_TF", "1")
+os.environ.setdefault("USE_TF", "0")
+
 import torch
 from transformers import (
     AutoModelForSequenceClassification,