Muhammed Sayeedur Rahman
feat: Add v2 architecture β FastAPI + Gradio, core pipeline, CorefakeNet, training suite
46d358c ProofyX Architecture: FastAPI + Gradio
Version 2.0 | April 2026
Overview
ProofyX uses a FastAPI + Gradio mounted together architecture. FastAPI serves as the real backend (API-first), while Gradio provides the web dashboard UI. Both share the same core detection pipeline.
βββββββββββββββββββββββββββββββββββ
β FastAPI Server β
β (main.py) β
β β
Browser ββββββββββ€ /ui β Gradio Dashboard β
WhatsApp Bot βββββ€ /api/v1 β REST API Endpoints β
Mobile App βββββββ€ /docs β Swagger (auto-gen) β
Batch Script βββββ€ β
β ββββββββββββββββββββββββββββ β
β β core/pipeline.py β β
β β (shared detection logic) β β
β ββββββββββββββββββββββββββββ β
βββββββββββββββββββββββββββββββββββ
Directory Structure
authen_check/
βββ main.py # FastAPI app entry point + Gradio mount
β
βββ core/ # Framework-agnostic detection pipeline
β βββ __init__.py
β βββ pipeline.py # analyze_image(), analyze_video(), analyze_audio()
β βββ models.py # Model loading singleton, inference wrappers
β βββ reports.py # PDF/HTML forensic report generation
β βββ metadata.py # EXIF extraction, file analysis
β
βββ api/ # FastAPI REST API layer
β βββ __init__.py
β βββ routes.py # REST endpoints (/analyze/image, /analyze/video, etc.)
β βββ schemas.py # Pydantic request/response models
β βββ auth.py # API key authentication (future)
β
βββ ui/ # Gradio UI layer
β βββ __init__.py
β βββ gradio_app.py # Gradio Blocks UI (sidebar, pages, events)
β βββ components.py # HTML generators (gauge, bars, verdict, radar)
β βββ theme.py # CSS variables, theme config, JS injections
β
βββ db/ # Persistence layer
β βββ __init__.py
β βββ history.py # SQLite analysis history
β
βββ core_models/ # Model architecture definitions (unchanged)
β βββ corefakenet.py # CorefakeNet unified hybrid CNN
β βββ dinov2_auth_model.py # DINOv2 fine-tuned
β βββ efficientnet_auth_model.py
β βββ efficientnet_texture.py # EfficientNet-B4 texture
β βββ face_deepfake_model.py # ResNet50 face
β βββ frequency_cnn.py # Frequency CNN + FFT preprocessing
β βββ fusion_mlp.py # Learned fusion + temperature calibration
β
βββ pipeline/ # Analysis orchestration (unchanged)
β βββ video_analyzer.py # VideoAnalyzer, extract_frames, FrequencyAnalyzer
β βββ audio_analyzer.py # AudioAnalyzer
β βββ face_gate.py # Face presence detection
β
βββ training/ # Training scripts (unchanged)
β βββ train_all.py # Full training pipeline
β βββ train_corefakenet.py
β βββ dataset_portraits.py
β βββ ...
β
βββ utils/ # Utilities (unchanged)
β βββ explainability.py # Risk explanation generation
β βββ gradcam.py # GradCAM heatmap generation
β
βββ models/ # Trained model weights (.pth files)
βββ assets/ # Logo, static assets
βββ docs/ # Documentation
βββ RESEARCH_REPORT.md
βββ ARCHITECTURE.md # This file
βββ IMPLEMENTATION_PLAN.md
Core Pipeline Contract
All detection functions return plain Python dicts with no UI framework dependencies.
Image Analysis
# core/pipeline.py
def analyze_image(image_pil: PIL.Image, mode: str = "ensemble") -> dict:
"""
Analyze a single image for deepfake indicators.
Args:
image_pil: PIL Image object
mode: "ensemble" (7-model) or "fast" (CorefakeNet)
Returns:
{
"risk_score": float, # 0.0 to 1.0
"risk_percent": float, # 0.0 to 100.0
"verdict": str, # "LIKELY MANIPULATED" / "POSSIBLY MANIPULATED"
# / "UNCERTAIN" / "LIKELY AUTHENTIC"
"confidence": str, # "HIGH" / "MEDIUM" / "LOW"
"model_agreement": str, # "5/7 models detect manipulation"
"model_scores": {
"vit": float,
"texture": float,
"frequency": float,
"face": float,
"dino": float,
"efficientnet": float,
"forensic": float,
},
"fusion_mode": str, # "learned" / "weighted_avg" / "attention"
"face_detected": bool,
"face_aligned": bool,
"gradcam_image": PIL.Image | None, # PIL Image of heatmap overlay
"original_image": PIL.Image, # Original input for display
"models_used": int,
"model_versions": dict, # {"corefakenet": "epoch9", ...}
"processing_time_ms": float,
"metadata": {
"format": str,
"dimensions": [int, int],
"file_size_bytes": int,
"exif": dict | None,
"has_c2pa": bool,
},
}
"""
Video Analysis
def analyze_video(video_path: str, fps: float = 4.0,
aggregation: str = "weighted_avg") -> dict:
"""
Returns:
{
"risk_score": float,
"risk_percent": float,
"verdict": str,
"confidence": str,
"prediction": str, # "FAKE" / "REAL"
"total_frames_analyzed": int,
"fake_frames": int,
"real_frames": int,
"faces_detected_in_frames": int,
"frame_results": [ # Per-frame breakdown
{
"frame_index": int,
"timestamp": float,
"risk_score": float,
"has_face": bool,
"model_scores": dict,
},
],
"temporal_analysis": {
"score_variance": float,
"max_frame_jump": float,
"significant_jumps": int,
"risk_timeline": [float], # For chart rendering
},
"gradcam_video_path": str | None,
"video_info": {
"duration_sec": float,
"width": int,
"height": int,
"fps": float,
},
"processing_time_ms": float,
}
"""
Audio Analysis
def analyze_audio(audio_path: str) -> dict:
"""
Returns:
{
"risk_score": float, # P(fake)
"authenticity_score": float, # 0-100, inverted risk
"verdict": str,
"confidence": str,
"manipulation_type": str,
"evidence": [str],
"suspicious_timestamps": [float],
"segment_results": [
{
"start_time": float,
"end_time": float,
"fake_probability": float,
"real_probability": float,
},
],
"duration_sec": float,
"segments_analyzed": int,
"processing_time_ms": float,
}
"""
Multimodal Fusion
def analyze_multimodal(image: PIL.Image | None = None,
video_path: str | None = None,
audio_path: str | None = None) -> dict:
"""
Returns:
{
"risk_score": float,
"risk_percent": float,
"verdict": str,
"confidence": str,
"media_types": [str], # ["image", "video", "audio"]
"modality_scores": {
"image": float | None,
"video": float | None,
"audio": float | None,
},
"fusion_weights": dict,
"explanation": str,
"processing_time_ms": float,
}
"""
API Layer
REST Endpoints
| Method | Path | Description |
|---|---|---|
POST |
/api/v1/analyze/image |
Analyze uploaded image |
POST |
/api/v1/analyze/video |
Analyze uploaded video |
POST |
/api/v1/analyze/audio |
Analyze uploaded audio |
POST |
/api/v1/analyze/multimodal |
Analyze multiple media types |
POST |
/api/v1/analyze/url |
Analyze media from URL |
GET |
/api/v1/history |
List past analyses |
GET |
/api/v1/history/{id} |
Get specific analysis |
GET |
/api/v1/history/{id}/report |
Download PDF report |
GET |
/api/v1/models/status |
List loaded models and status |
GET |
/api/v1/health |
Health check |
Response Schema (Pydantic)
# api/schemas.py
class ModelScore(BaseModel):
name: str
score: float
confidence: str
class AnalysisResult(BaseModel):
id: str # UUID for this analysis
timestamp: datetime
risk_score: float
risk_percent: float
verdict: str
confidence: str
model_agreement: str
model_scores: dict[str, float]
face_detected: bool
models_used: int
processing_time_ms: float
media_type: str
metadata: dict
class AnalysisResponse(BaseModel):
success: bool
data: AnalysisResult | None
error: str | None
UI Layer (Gradio)
Page Structure
The Gradio UI simulates multi-page navigation using a sidebar with visibility toggling:
ββββββββββ¬βββββββββββββββββββββββββββββββββββββββββββββββ
β SIDEBARβ HEADER: "Ingestion Terminal" | SYSTEM ACTIVE β
β ββββββββββββββββββββββββββββββ¬ββββββββββββββββββ€
β [Scan] β β Detection β
β Anlysisβ Upload Zone β Modules Active β
β Histor β (drag & drop) β β ViT β
β Settngsβ β β EfficientNet β
β β β β Face CNN β
β β [Initialize Scan] β β Frequency β
β ββββββββββββββββββββββββββββββ€ β CorefakeNet β
β β βββββββββββββββββββ€
β β DETECTION RESULTS β RESULTS β
β β ββββββββ¬ββββββββ¬βββββββ β [Gauge] β
β β βGauge βVerdictβScoresβ β [Verdict] β
β β ββββββββ΄ββββββββ΄βββββββ β [Score Bars] β
β β β [Radar Chart] β
ββββββββββ΄βββββββββββββββββββββββββββββ΄ββββββββββββββββββ
Key Components
| Component | Implementation | Purpose |
|---|---|---|
| Stars background | gr.HTML + js_on_load |
Animated starfield |
| Sidebar navigation | gr.Sidebar or gr.Column + CSS |
Page switching |
| System status | gr.HTML + html_template |
Model status display |
| Upload zone | gr.HTML + @children wrapping gr.File |
Custom drag-drop |
| Detection modules | gr.HTML + html_template |
Module checklist |
| Risk gauge | SVG via gr.HTML |
Animated circular gauge |
| Score bars | HTML via gr.HTML |
Per-model breakdown |
| Radar chart | Chart.js via gr.HTML + head |
Visual score comparison |
| Verdict card | HTML via gr.HTML |
Color-coded verdict |
| Timeline chart | Chart.js via gr.HTML |
Video temporal analysis |
Data Flow
Single Image Analysis
User uploads image
β
βΌ
ui/gradio_app.py: analyze_btn.click()
β
βΌ
core/pipeline.py: analyze_image(pil_image, mode)
β
ββββΊ core/models.py: get_model_scores(image)
β ββββΊ ViT inference
β ββββΊ EfficientNet-B4 texture inference
β ββββΊ Frequency CNN inference
β ββββΊ DINOv2 inference
β ββββΊ EfficientNet auth inference
β ββββΊ Face model inference
β ββββΊ Forensic analysis (heuristic)
β
ββββΊ core/models.py: fuse_scores(scores, mode)
β ββββΊ FusionMLP (learned) OR
β ββββΊ CorefakeNet (fast)
β
ββββΊ utils/gradcam.py: generate_gradcam_image()
β
ββββΊ core/metadata.py: extract_metadata()
β
ββββΊ Returns plain dict
β
βΌ
ui/components.py: render results as HTML
db/history.py: save to SQLite (async)
β
βΌ
Gradio updates UI components
API Request
Client sends POST /api/v1/analyze/image
β
βΌ
api/routes.py: validate request (Pydantic)
β
βΌ
core/pipeline.py: analyze_image(pil_image, mode)
β (same pipeline as Gradio)
βΌ
api/routes.py: format as JSON response
β
βΌ
Client receives AnalysisResponse JSON
Model Loading
Models are loaded once at startup and shared across all requests:
# core/models.py
class ModelRegistry:
"""Singleton that loads all models once and provides inference methods."""
def __init__(self):
self.device = torch.device("cpu")
self.models = {}
self.loaded = []
self.missing = []
self._load_all()
def _load_all(self):
self._try_load("dino", DINOv2AuthModel, "dinov2_auth_model.pth")
self._try_load("efficientnet", EfficientNetAuthModel, "efficientnet_auth_model.pth")
self._try_load("face", FaceDeepfakeModel, "image_face_model.pth")
self._try_load("texture", EfficientNetTexture, "efficient.pth")
self._try_load("frequency", FrequencyCNN, "frequency.pth")
self._try_load("fusion", FusionMLP, "fusion_mlp.pth", n_inputs=4)
self._try_load("corefakenet", CorefakeNet, "corefakenet.pth")
self._load_vit()
self._load_audio()
def get_status(self) -> dict:
return {
"loaded": self.loaded,
"missing": self.missing,
"total": len(self.loaded),
"corefakenet_ready": "corefakenet" in self.models,
}
# Global singleton
registry = ModelRegistry()
Database Schema (SQLite)
CREATE TABLE analyses (
id TEXT PRIMARY KEY, -- UUID
timestamp DATETIME NOT NULL,
media_type TEXT NOT NULL, -- "image" / "video" / "audio" / "multimodal"
risk_score REAL NOT NULL,
verdict TEXT NOT NULL,
confidence TEXT NOT NULL,
model_scores TEXT NOT NULL, -- JSON blob
face_detected BOOLEAN,
models_used INTEGER,
processing_time_ms REAL,
file_name TEXT,
file_size_bytes INTEGER,
metadata TEXT, -- JSON blob (EXIF, etc.)
gradcam_path TEXT, -- Path to saved heatmap image
report_path TEXT -- Path to generated PDF
);
CREATE INDEX idx_analyses_timestamp ON analyses(timestamp DESC);
CREATE INDEX idx_analyses_media_type ON analyses(media_type);
CREATE INDEX idx_analyses_verdict ON analyses(verdict);
Future Integrations
WhatsApp Bot (Phase 4)
authen_check/
βββ bots/
βββ whatsapp.py # Twilio webhook
# POST /webhook β download media β pipeline.analyze_image() β reply
ONNX Export (Phase 3)
authen_check/
βββ core/
βββ onnx_inference.py # ONNX Runtime wrapper
# Same interface as models.py but uses .onnx files
Batch Processing (Phase 3)
# api/routes.py
@app.post("/api/v1/analyze/batch")
async def batch_analyze(files: list[UploadFile]):
results = [pipeline.analyze_image(open_image(f)) for f in files]
return {"results": results}