Update app.py

app.py

@@ -9,9 +9,11 @@ import spaces
 import torch
 import random
 import string
+from datetime import datetime, timedelta
+from collections import defaultdict
 
 # Set up logging
-logging.basicConfig(level=logging.ERROR)
+logging.basicConfig(level=logging.INFO, filename="api_access.log", format="%(asctime)s - %(message)s")
 
 torch.set_num_threads(1)
 torch.set_num_interop_threads(1)
@@ -21,6 +23,58 @@ torch.backends.cudnn.benchmark = False
 torch.backends.cuda.matmul.allow_tf32 = False
 torch.backends.cudnn.allow_tf32 = False
 
+# Data structures to track IP access
+ip_access_records = defaultdict(lambda: {
+    'total_access_time': timedelta(),
+    'blocked_until': None,
+    'session_start': None
+})
+
+SESSION_TIMEOUT = timedelta(hours=1)
+
+# Function to log API access attempts
+def log_api_access(successful_captcha, video_file=None, client_ip="unknown", blocked=False):
+    status = "Blocked" if blocked else "Allowed"
+    logging.info(f"Access Attempt - Status: {status}, Successful CAPTCHA: {successful_captcha}, "
+                 f"Video File: {video_file if video_file else 'N/A'}, Client IP: {client_ip}")
+
+# Function to get client IP address
+def get_client_ip(request: gr.Request):
+    return request.client.host
+
+# Function to check if IP is blocked
+def is_ip_blocked(client_ip):
+    record = ip_access_records[client_ip]
+    blocked_until = record.get('blocked_until')
+    if blocked_until and datetime.now() < blocked_until:
+        return True
+    return False
+
+# Function to update IP access time
+def update_ip_access_time(client_ip):
+    record = ip_access_records[client_ip]
+    now = datetime.now()
+    session_start = record.get('session_start')
+
+    if session_start:
+        elapsed = now - session_start
+        record['total_access_time'] += elapsed
+        record['session_start'] = now
+    else:
+        record['session_start'] = now
+
+    # Remove access times older than 24 hours
+    if 'last_access_time' in record and now - record['last_access_time'] > timedelta(hours=24):
+        record['total_access_time'] = timedelta()
+
+    record['last_access_time'] = now
+
+    # Check if total access time in last 24 hours exceeds 1 hour
+    if record['total_access_time'] >= timedelta(hours=1):
+        # Block IP for 48 hours
+        record['blocked_until'] = now + timedelta(hours=48)
+        record['total_access_time'] = timedelta()  # Reset total access time
+
 # Function to truncate video
 def truncate_video(video_file):
     """Truncates video to 15 seconds and saves it as a temporary file."""
@@ -66,7 +120,7 @@ def import_backend_script(script_name):
         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)
@@ -90,13 +144,32 @@ captcha_solution = generate_captcha()
 def verify_captcha(user_input):
     global captcha_solution
     if user_input.strip() == captcha_solution:
-        return "Captcha verified. You can now upload a video.", True, captcha_solution
+        return "Captcha verified. Video analysis will proceed.", True, captcha_solution
     else:
         captcha_solution = generate_captcha()  # Reset CAPTCHA on failure
         return f"Incorrect CAPTCHA. Please try again.", False, captcha_solution
 
 @spaces.GPU(duration=1000)
-def analyze_video(video_file):
+def analyze_video(video_file, captcha_input, request: gr.Request):
+    client_ip = get_client_ip(request)
+
+    # Check if IP is blocked
+    if is_ip_blocked(client_ip):
+        log_api_access(successful_captcha=False, video_file=video_file.name if video_file else "N/A", client_ip=client_ip, blocked=True)
+        return {"error": "Your IP has been blocked due to excessive usage. Please try again later."}
+
+    # Verify CAPTCHA
+    message, success, _ = verify_captcha(captcha_input)
+    if not success:
+        log_api_access(successful_captcha=False, video_file=video_file.name if video_file else "N/A", client_ip=client_ip)
+        return {"error": message}
+
+    # Update IP access time
+    update_ip_access_time(client_ip)
+
+    # Log API access attempt with successful CAPTCHA
+    log_api_access(successful_captcha=True, video_file=video_file.name if video_file else "N/A", client_ip=client_ip)
+
     try:
         # Video truncation and analysis
         truncated_video = truncate_video(video_file)
@@ -120,38 +193,35 @@ def analyze_video(video_file):
         logging.error(f"Error during analysis: {e}")
         return {"error": "An error occurred during video analysis. Please check your input and try again."}
 
-# Interface with CAPTCHA verification
+# Interface with CAPTCHA after video upload
 def main_interface():
     with gr.Blocks() as interface:
         gr.Markdown("# AllMark - Deepfake Analyzer")
-        gr.Markdown("Please solve the CAPTCHA to proceed with video analysis.")
-
-        # Display CAPTCHA and input box for user response
-        captcha_text = gr.Textbox(label="CAPTCHA", value=captcha_solution, interactive=False)
-        captcha_input = gr.Textbox(label="Enter CAPTCHA Here")
-        captcha_output = gr.Textbox(label="CAPTCHA Status", interactive=False)
-
-        # Verify CAPTCHA button
-        verify_button = gr.Button("Verify CAPTCHA")
-
-        # Video analysis components (hidden initially)
-        with gr.Column(visible=False) as analysis_components:
-            video_input = gr.Video(label="Upload Video")
-            analysis_output = gr.JSON(label="Analysis Result")
-
-            # Submit button for video analysis
-            analyze_button = gr.Button("Analyze Video")
-            analyze_button.click(analyze_video, inputs=video_input, outputs=analysis_output)
-
-        # Handle CAPTCHA verification
-        def verify_captcha_click(input_text):
-            message, success, new_captcha = verify_captcha(input_text)
-            if success:
-                analysis_components.visible = True
-            return message, new_captcha
-
-        # Update both CAPTCHA message and CAPTCHA text dynamically
-        verify_button.click(verify_captcha_click, inputs=captcha_input, outputs=[captcha_output, captcha_text])
+        gr.Markdown("Upload a video to proceed with analysis.")
+
+        # Video input
+        video_input = gr.Video(label="Upload Video")
+        captcha_text = gr.Textbox(label="CAPTCHA", value=captcha_solution, interactive=False, visible=False)
+        captcha_input = gr.Textbox(label="Enter CAPTCHA Here", visible=False)
+        captcha_output = gr.Textbox(label="CAPTCHA Status", interactive=False, visible=False)
+        analyze_button = gr.Button("Analyze Video", visible=False)
+        analysis_output = gr.JSON(label="Analysis Result")
+
+        # Function to show CAPTCHA after video upload
+        def show_captcha(video):
+            if video is not None:
+                return gr.update(visible=True), gr.update(visible=True), gr.update(visible=True), gr.update(visible=True)
+            else:
+                return gr.update(visible=False), gr.update(visible=False), gr.update(visible=False), gr.update(visible=False)
+
+        video_input.change(show_captcha, inputs=video_input, outputs=[captcha_text, captcha_input, captcha_output, analyze_button])
+
+        # Handle analysis
+        analyze_button.click(
+            analyze_video,
+            inputs=[video_input, captcha_input],
+            outputs=analysis_output
+        )
 
     return interface