Retrocop86/Deepfake-Shield-AI
1
1from flask import Flask, request, jsonify, send_from_directory
2from flask_cors import CORS
3import torch
4import torch.nn as nn
5import numpy as np
6import cv2
7import os
8import json
9from datetime import datetime
10from torchvision import transforms
11from torchvision.models.video import r3d_18
12import requests
13
14app = Flask(__name__)
15CORS(app)
16
17UPLOAD_FOLDER = "uploads"
18os.makedirs(UPLOAD_FOLDER, exist_ok=True)
19
20# ✅ Use /data for persistence (HF Spaces)
21HISTORY_FILE = "history.json"
22
23# ✅ Ensure file exists
24if not os.path.exists(HISTORY_FILE):
25 with open(HISTORY_FILE, "w") as f:
26 json.dump([], f, indent=4)
27
28# ------------------ LOAD MODEL ------------------
29
30device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
31
32model = r3d_18(weights=None)
33model.fc = nn.Linear(model.fc.in_features, 2)
34
35try:
36 model.load_state_dict(torch.load("deepfake_model_550.pth", map_location=device))
37 print("MODEL LOADED SUCCESSFULLY!")
38except Exception as e:
39 print("❌ MODEL FAILED TO LOAD:", e)
40
41model.to(device)
42model.eval()
43
44# ------------------ LLM CONFIG ------------------
45
46OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
47
48def load_history():
49 try:
50 with open(HISTORY_FILE, "r") as f:
51 return json.load(f)
52 except:
53 return []
54
55def save_history(entry):
56 history = load_history()
57 history.append(entry)
58
59 with open(HISTORY_FILE, "w") as f:
60 json.dump(history, f, indent=4)
61
62def call_llm(label, confidence, artifacts):
63 history = load_history()
64 # Filter only useful learning cases
65 filtered_history = []
66
67 for h in history:
68 # Skip feedback entries
69 if "feedback" in h:
70 continue
71
72 model_label = h.get("model_label")
73 final_label = h.get("final_label")
74 artifacts = h.get("artifacts", [])
75
76 # Keep only correct or strong-signal cases
77 if model_label == final_label:
78 filtered_history.append(h)
79 elif "possible face manipulation artifacts" in artifacts:
80 filtered_history.append(h)
81
82 # Take last 5 meaningful cases
83 recent_history = filtered_history[-5:]
84
85 history_text = ""
86 for h in recent_history:
87 history_text += f"""
88 Past Case:
89 Model: {h.get("model_label")}
90 Final: {h.get("final_label")}
91 Artifacts: {h.get("artifacts")}
92 """
93
94 prompt = f"""
95 You are an expert in detecting deepfake videos.
96
97 Reference past correct decisions:
98 {history_text}
99
100 Current Case:
101 Model Prediction: {label}
102 Confidence: {confidence:.2f}
103 Observations: {", ".join(artifacts)}
104
105 Instructions:
106 - Think independently
107 - You are assisting a deepfake detection system.
108 IMPORTANT RULES:
109 - If model confidence > 0.75 → TRUST the model
110 - If confidence is between 0.5–0.75 → you may override IF strong reasoning
111 - If confidence < 0.5 → you can disagree
112
113 - Deepfake artifacts (motion, lighting, edges) are strong indicators of FAKE
114 - Do NOT assume video is real unless evidence is strong
115
116 Be strict. Avoid optimistic bias.
117 - Use simple language
118 - Focus on visible issues like face, motion, lighting
119 - Limit response to maximum 120 words
120
121 Output STRICTLY:
122
123 Final Decision: REAL or FAKE
124
125 Key Observations:
126 - Point 1
127 - Point 2
128
129 Explanation:
130 (2-3 simple sentences)
131 """
132
133 try:
134 response = requests.post(
135 "https://api.groq.com/openai/v1/chat/completions",
136 headers={
137 "Authorization": f"Bearer {OPENAI_API_KEY}",
138 "Content-Type": "application/json"
139 },
140 json={
141 "model": "llama-3.1-8b-instant",
142 "messages": [{"role": "user", "content": prompt}],
143 "max_tokens": 350
144 },
145 timeout=10
146 )
147
148 data = response.json()
149
150 if "choices" in data:
151 return data["choices"][0]["message"]["content"]
152 else:
153 return "Final Decision: UNKNOWN\nExplanation: LLM failed."
154
155 except Exception as e:
156 return f"Final Decision: UNKNOWN\nExplanation: {str(e)}"
157
158# ------------------ HELPERS ------------------
159
160def get_llm_label(llm_output):
161 text = llm_output.upper()
162
163 if "FINAL DECISION: FAKE" in text:
164 return "FAKE"
165 elif "FINAL DECISION: REAL" in text:
166 return "REAL"
167
168 return "UNKNOWN"
169
170def extract_artifacts(video_tensor):
171 artifacts = []
172
173 frames = video_tensor.squeeze(0).permute(1, 0, 2, 3)
174
175 diffs = []
176 brightness = []
177 edges = []
178
179 for i in range(len(frames) - 1):
180 f1 = frames[i].cpu().numpy()
181 f2 = frames[i+1].cpu().numpy()
182
183 diff = np.mean(np.abs(f1 - f2))
184 diffs.append(diff)
185
186 brightness.append(np.mean(f1))
187
188 gray = np.mean(f1, axis=0)
189 edge = np.mean(cv2.Canny((gray * 255).astype(np.uint8), 100, 200))
190 edges.append(edge)
191
192 avg_diff = np.mean(diffs)
193 brightness_var = np.var(brightness)
194 edge_var = np.var(edges)
195
196 if avg_diff > 0.18:
197 artifacts.append("motion looks unnatural or inconsistent")
198
199 if brightness_var > 0.02:
200 artifacts.append("lighting changes across frames")
201
202 if edge_var > 5:
203 artifacts.append("edges appear unstable")
204
205 if avg_diff > 0.12 and edge_var > 5:
206 artifacts.append("possible face manipulation artifacts")
207
208 if len(artifacts) == 0:
209 artifacts.append("no strong manipulation signs detected")
210
211 return artifacts
212
213# ------------------ PREPROCESS ------------------
214
215normalize = transforms.Normalize(mean=[0.5, 0.5, 0.5],
216 std=[0.5, 0.5, 0.5])
217
218def preprocess_video(video_path, frames_per_clip=4): # ✅ Reduced frames
219 cap = cv2.VideoCapture(video_path)
220 frames = []
221
222 frame_count = 0
223
224 while True:
225 ret, frame = cap.read()
226 if not ret or frame_count > 32: # ✅ limit frames
227 break
228
229 frame_count += 1
230
231 frame = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
232 frame = torch.from_numpy(frame).permute(2, 0, 1).float() / 255.0
233
234 frame = torch.nn.functional.interpolate(
235 frame.unsqueeze(0),
236 size=(112, 112),
237 mode='bilinear',
238 align_corners=False
239 ).squeeze(0)
240
241 frame = normalize(frame)
242 frames.append(frame)
243
244 cap.release()
245 os.remove(video_path)
246
247 if len(frames) == 0:
248 raise ValueError("No frames extracted")
249
250 if len(frames) >= frames_per_clip:
251 idx = torch.linspace(0, len(frames)-1, frames_per_clip).long()
252 frames = [frames[i] for i in idx]
253 else:
254 while len(frames) < frames_per_clip:
255 frames.append(frames[-1])
256
257 video = torch.stack(frames)
258 video = video.permute(1, 0, 2, 3)
259 return video.unsqueeze(0)
260
261# ------------------ ROUTES ------------------
262
263@app.route("/")
264def home():
265 return send_from_directory(".", "index.html")
266
267@app.route("/debug-history-file")
268def debug_history_file():
269 return jsonify({
270 "exists": os.path.exists(HISTORY_FILE),
271 "path": os.path.abspath(HISTORY_FILE)
272 })
273
274@app.route("/download-history")
275def download_history():
276 return send_from_directory("/data", "history.json", as_attachment=True)
277
278@app.route("/language")
279def language():
280 return send_from_directory(".", "language_selector.html")
281
282@app.route("/detector")
283def detector():
284 return send_from_directory(".", "detector.html")
285
286@app.route("/logo_light.jpg")
287def serve_logo():
288 return send_from_directory(".", "logo_light.jpg")
289
290# ------------------ DETECTION ------------------
291
292@app.route("/detect", methods=["POST"])
293def detect():
294 if "video" not in request.files:
295 return jsonify({"error": "No video uploaded"}), 400
296
297 file = request.files["video"]
298
299 if file.filename == "":
300 return jsonify({"error": "No file selected"}), 400
301
302 save_path = os.path.join(UPLOAD_FOLDER, file.filename)
303 file.save(save_path)
304
305 try:
306 # ------------------ MODEL INFERENCE ------------------
307 tensor = preprocess_video(save_path).to(device)
308 artifacts = extract_artifacts(tensor)
309
310 with torch.no_grad():
311 output = model(tensor)
312 probs = torch.softmax(output, dim=1)
313
314 fake_prob = probs[0][0].item()
315 real_prob = probs[0][1].item()
316
317 if fake_prob > real_prob:
318 label = "FAKE"
319 confidence = fake_prob
320 else:
321 label = "REAL"
322 confidence = real_prob
323
324 # ------------------ LLM CALL ------------------
325 llm_output = call_llm(label, confidence, artifacts)
326 llm_label = get_llm_label(llm_output)
327
328 # ------------------ FEEDBACK LEARNING ------------------
329
330 recent_feedback = [h for h in load_history() if h.get("feedback") == "incorrect"][-10:]
331
332 fake_bias = 0
333 real_bias = 0
334
335 for f in recent_feedback:
336 if f.get("correct_label") == "FAKE":
337 fake_bias += 0.02
338 elif f.get("correct_label") == "REAL":
339 real_bias += 0.02
340
341 # Adjust confidence
342 if label == "FAKE":
343 confidence += fake_bias
344 else:
345 confidence += real_bias
346
347
348 # ------------------ FINAL DECISION LOGIC ------------------
349
350 # Rule 1: Strong model confidence → TRUST MODEL
351 if confidence > 0.75:
352 final_label = label
353
354 # Rule 2: Artifact-based bias
355 elif "possible face manipulation artifacts" in str(artifacts).lower():
356 final_label = "FAKE"
357
358 # Rule 3: Medium confidence → allow LLM
359 elif 0.5 < confidence <= 0.75:
360 if llm_label != "UNKNOWN":
361 final_label = llm_label
362 else:
363 final_label = label
364
365 # Rule 4: Low confidence → trust LLM
366 else:
367 final_label = llm_label if llm_label != "UNKNOWN" else label
368
369 # ------------------ FINAL RESPONSE ------------------
370
371 return jsonify({
372 "result": final_label,
373 "model_label": label, # ✅ needed for feedback
374 "llm_output": llm_output, # ✅ needed for feedback
375 "final_label": final_label # optional but useful
376 })
377
378 except Exception as e:
379 print("DETECT ERROR:", str(e))
380 return jsonify({"error": str(e)}), 500
381
382# ------------------ FEEDBACK ------------------
383
384@app.route("/feedback", methods=["POST"])
385def feedback():
386 data = request.json
387
388 # ✅ (1) VALIDATION
389 if data.get("correct_label") not in ["REAL", "FAKE"]:
390 return jsonify({"error": "Invalid label"}), 400
391
392 # ✅ (3) DEBUG PRINT
393 print("FEEDBACK RECEIVED:", data)
394
395 # ✅ (2) SAVE WITH correct_label
396 save_history({
397 "feedback": "incorrect",
398 "model_label": data.get("model"),
399 "llm_output": data.get("llm"),
400 "correct_label": data.get("correct_label"),
401 "timestamp": str(datetime.now())
402 })
403
404 return jsonify({"status": "saved"})
405
406# ------------------ RUN ------------------
407
408if __name__ == "__main__":
409 app.run(host="0.0.0.0", port=7860)