app.py

import os
import subprocess
import importlib.util
import gradio as gr
import logging
from moviepy.editor import VideoFileClip
import json
import spaces
import torch

torch.set_num_threads(1)
torch.set_num_interop_threads(1)
torch.use_deterministic_algorithms(True)
torch.backends.cudnn.deterministic = True
torch.backends.cudnn.benchmark = False
torch.backends.cuda.matmul.allow_tf32 = False
torch.backends.cudnn.allow_tf32 = False

def truncate_video(video_file):
    """Truncates video to 15 seconds and saves it as a temporary file."""
    clip = VideoFileClip(video_file)
    truncated_clip = clip.subclip(0, min(15, clip.duration))
    truncated_video_file = "temp_truncated_video.mp4"
    truncated_clip.write_videofile(truncated_video_file, codec="libx264", audio_codec="aac")
    return truncated_video_file

def clone_repo():
    """Clone the GitHub repository containing the backend."""
    repo_url = "https://github.com/NeeravSood/AllMark-MVP.git"  # Update when changing
    repo_path = "./repository"

    github_pat = os.getenv("GITHUB_PAT")
    if not github_pat:
        raise RuntimeError("GitHub Personal Access Token (GITHUB_PAT) not found in environment variables.")
    authenticated_repo_url = f"https://{github_pat}@github.com/NeeravSood/AllMark-MVP.git"

    if os.path.exists(repo_path):
        print("Repository already cloned.")
    else:
        try:
            subprocess.run(
                ["git", "clone", authenticated_repo_url, repo_path],
                check=True,
                text=True,
                capture_output=True
            )
            print("Repository cloned successfully.")
        except subprocess.CalledProcessError as e:
            print("Output:", e.stdout)
            print("Error:", e.stderr)
            raise RuntimeError(f"Failed to clone repository: {e.stderr}")

def import_backend_script(script_name):
    """Dynamically import the backend script."""
    try:
        script_path = os.path.join("./repository", script_name)
        if not os.path.exists(script_path):
            raise FileNotFoundError(f"Script {script_name} not found in the repository.")
        
        spec = importlib.util.spec_from_file_location("backend_module", script_path)
        backend_module = importlib.util.module_from_spec(spec)
        spec.loader.exec_module(backend_module)
        return backend_module
    except Exception as e:
        logging.error(f"Error importing backend script: {str(e)}")
        raise RuntimeError(f"Failed to import backend script: {str(e)}")

clone_repo()
backend = import_backend_script("app.py") 
analyzer = backend.DeepfakeAnalyzer()

@spaces.GPU(duration=1000)
def analyze_video(video_file):
    try:
        truncated_video = truncate_video(video_file)
        results = analyzer.analyze_media(truncated_video)
        combined_assessment = int(results.get('combined_assessment', 0))
        analysis_result = "genuine/original" if combined_assessment < 50 else "a deepfake"
        output = {
            "message": f"According to our analysis, the video you uploaded appears to be {analysis_result}. "
                       f"{len(results['video_analysis']['frame_results'])} frames were analyzed in total."
        }
        
        return output

    except Exception as e:
        logging.error(f"Error during analysis: {e}")
        return {"error": "An error occurred during video analysis. Please check your input and try again."}

interface = gr.Interface(
    fn=analyze_video,
    inputs=gr.Video(label="Upload Video"),
    outputs="json",
    title="AllMark - Deepfake Analyzer",
    description="Upload a video to analyze. N.B. - Only mp4 files. Processing time 1-10 minutes. For any false negatives, please contact the publisher for verification. Incognito Mode Recommended"
)

if __name__ == "__main__":
    interface.launch()