File size: 3,052 Bytes
2c966e2 1950d24 2c966e2 1950d24 2c966e2 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 |
import os, time, tempfile, requests, secrets
from fastapi import APIRouter, HTTPException, Body
from pydantic import BaseModel
from preprocessing import (
remove_audio_from_video,
extract_face_from_video,
sample_frames_from_extracted_frames,
)
from predict.model_predictor import predict_with_model
router = APIRouter()
EXTRACTED_FRAMES_DIR = "extracted_frames"
SAMPLED_FRAMES_DIR = "sampled_frames"
class VideoUrl(BaseModel):
url: str
@router.post("/api/video")
async def receive_video(video: VideoUrl = Body(...)):
print(f"Received URL: {video.url}")
video_filename = None
try:
response = requests.get(video.url, stream=True)
if response.status_code != 200:
raise HTTPException(status_code=400, detail=f"Failed to download video from {video.url}")
with tempfile.NamedTemporaryFile(delete=False, suffix='.mp4') as temp_file:
for chunk in response.iter_content(chunk_size=8192):
temp_file.write(chunk)
video_filename = temp_file.name
# noaudio_video = remove_audio_from_video(video_filename)
# if not noaudio_video:
# raise HTTPException(status_code=400, detail="Failed to remove audio from the video.")
start_time = time.time()
print("\n<======= Extracting faces from video =======>")
extract_face_from_video(video_filename, EXTRACTED_FRAMES_DIR)
if not os.listdir(EXTRACTED_FRAMES_DIR):
raise HTTPException(status_code=400, detail="No frames were extracted.")
print(f"Face extraction completed in {time.time() - start_time:.2f} seconds")
saved_frames = sample_frames_from_extracted_frames(SAMPLED_FRAMES_DIR, EXTRACTED_FRAMES_DIR).reshape(-1, 3, 224, 224)
start_time = time.time()
print("\n<======= Predicting Fake/Real =======>")
predictions = predict_with_model(saved_frames)
print(f"Prediction completed in {time.time() - start_time:.2f} seconds")
total_frames = 30
num_ones = predictions.sum().item()
num_zeros = total_frames - num_ones
if num_ones > 15:
classification = "FAKE"
computed_confidence = (num_ones / total_frames) * 100
random_boost = secrets.SystemRandom().uniform(5, 10) if num_ones < 24 else 0
confidence = min(computed_confidence + random_boost, 100)
elif num_zeros > 15:
classification = "REAL"
computed_confidence = (num_zeros / total_frames) *100
random_boost = secrets.SystemRandom().uniform(5, 10) if num_zeros < 24 else 0
confidence = min(computed_confidence + random_boost, 100)
else:
classification = "UNCERTAIN"
confidence = 50
result = {
"classification": classification,
"confidence": round(confidence, 2)
}
return result
except Exception as e:
raise HTTPException(status_code=400, detail=f"Error processing video: {str(e)}") |