proofyx / docs /ARCHITECTURE.md
Muhammed Sayeedur Rahman
feat: Add v2 architecture β€” FastAPI + Gradio, core pipeline, CorefakeNet, training suite
46d358c
|
Raw
History Blame Contribute Delete
17.3 kB

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}