Merge pull request #14 from Tobkubos/backend-setup

backend/app/api/routes.py

@@ -1,7 +1,11 @@
+import asyncio
+from collections import defaultdict
 import logging
 from fastapi import APIRouter, HTTPException, Request, status
+from app.services.queue import get_queue_service
 from slowapi.errors import RateLimitExceeded
 from limits import parse
+from redis.exceptions import LockError
 
 from app.models.schemas import (
     AnalysisRequest,
@@ -23,6 +27,7 @@ from app.config_manager import _load_all_configs, save_guild_config
 logger = logging.getLogger(__name__)
 
 router = APIRouter()
+local_guild_locks = defaultdict(asyncio.Lock)
 
 @router.get(
     "/",
@@ -109,26 +114,8 @@ async def get_discord_guild_config(guild_id: str):
         "log_channel_id": guild_config.get("log_channel_id", None)
     }
 
-@router.post(
-    "/analyze",
-    response_model=AnalysisResponse,
-    responses={
-        400: {"model": ErrorResponse, "description": "Bad request"},
-        408: {"model": ErrorResponse, "description": "Request timeout"},
-        415: {"model": ErrorResponse, "description": "Unsupported media type"},
-        429: {"model": ErrorResponse, "description": "Too many requests"},
-        500: {"model": ErrorResponse, "description": "Internal server error"},
-    },
-    tags=["Analysis"],
-    summary="Analyze content for deepfake detection",
-)
-async def analyze(request: Request, payload: AnalysisRequest) -> AnalysisResponse:
-    guild_id = payload.guild_id
-    limit_item = parse("1/5seconds")
-    
-    if not limiter.limiter.hit(limit_item, f"analyze:{guild_id}"):
-        raise HTTPException(status_code=429, detail="Rate limit exceeded for this guild")
-    
+async def _execute_analysis(payload: AnalysisRequest, guild_id: str, settings) -> dict:
+    """Funkcja pomocnicza wykonująca właściwy proces pobierania i analizy."""
     if isinstance(payload, TextAnalysisRequest):
         content_type = "text"
     elif isinstance(payload, ImageAnalysisRequest):
@@ -136,28 +123,29 @@ async def analyze(request: Request, payload: AnalysisRequest) -> AnalysisRespons
     else:
         raise HTTPException(
             status_code=status.HTTP_415_UNSUPPORTED_MEDIA_TYPE, 
-            detail="Unsupported file/content type. Only text and image are currently supported."
+            detail="Unsupported file/content type."
         )
 
-    settings = get_settings()
-
     try:
         if content_type == "text":
             if len(payload.text) > settings.MAX_CONTENT_SIZES["text"]:
-                raise ValueError(f"Text content exceeds maximum length of {settings.MAX_CONTENT_SIZES['text']} characters")
+                raise ValueError(f"Text content exceeds maximum length.")
             if len(payload.text) < 50:
                 raise ValueError("Text content must be at least 50 characters")
             
-            analysis_result = await analyze_text(payload.text, guild_id)
+            result = await analyze_text(payload.text, guild_id)
 
         elif content_type == "image":
             image_bytes = await download_file(str(payload.image_url))
             if not image_bytes:
                 raise ValueError("Failed to download image")
             if len(image_bytes) > settings.MAX_CONTENT_SIZES["image"]:
-                raise ValueError(f"Image size exceeds maximum of {settings.MAX_CONTENT_SIZES['image']} bytes")
+                raise ValueError(f"Image size exceeds maximum.")
             
-            analysis_result = await analyze_image(image_bytes, guild_id)
+            result = await analyze_image(image_bytes, guild_id)
+            
+        result["content_type"] = content_type
+        return result
 
     except ValueError as e:
         raise HTTPException(status_code=400, detail=str(e))
@@ -166,10 +154,66 @@ async def analyze(request: Request, payload: AnalysisRequest) -> AnalysisRespons
     except DeepfakeDetectionError as e:
         raise HTTPException(status_code=e.status_code, detail=e.message)
     except Exception as e:
-        logger.error(f"{content_type.capitalize()} analysis error: {str(e)}", exc_info=True)
+        logger.error(f"Analysis error: {str(e)}", exc_info=True)
         raise HTTPException(status_code=500, detail=f"Failed to analyze {content_type}")
+    
+@router.post(
+    "/analyze",
+    response_model=AnalysisResponse,
+    responses={
+        400: {"model": ErrorResponse, "description": "Bad request"},
+        408: {"model": ErrorResponse, "description": "Request timeout"},
+        415: {"model": ErrorResponse, "description": "Unsupported media type"},
+        429: {"model": ErrorResponse, "description": "Too many requests"},
+        500: {"model": ErrorResponse, "description": "Internal server error"},
+    },
+    tags=["Analysis"],
+    summary="Analyze content for deepfake detection",
+)
+async def analyze(request: Request, payload: AnalysisRequest) -> AnalysisResponse:
+    guild_id = payload.guild_id
+    user_id = payload.user_id
+    
+    limit_item = parse("1/5seconds")
+    if not limiter.limiter.hit(limit_item, f"analyze:user:{user_id}"):
+        logger.warning(f"Użytkownik {user_id} przekroczył limit zapytań (1/5s).")
+        raise HTTPException(
+            status_code=429, 
+            detail="Przekroczyłeś limit zapytań. Możesz wykonać tylko 1 analizę na 5 sekund."
+        )
+    
+    settings = get_settings()
+    queue_service = get_queue_service()
+    
+    # 2. Sprawdzamy, czy Redis jest aktywny w pliku konfiguracyjnym oraz czy klient został zainicjalizowany
+    if settings.REDIS_ENABLED and queue_service.redis_client is not None:
+        logger.info(f"Używam rozproszonej blokady Redis dla użytkownika {user_id}")
+        
+        # Tworzymy blokadę przy użyciu klienta z QueueService [1.2.6]
+        redis_lock = queue_service.redis_client.lock(
+            "global_analysis_queue_lock", timeout=60, blocking_timeout=120
+        )
+        try:
+            async with redis_lock:
+                analysis_result = await _execute_analysis(payload, guild_id, settings)
+        except LockError:
+            logger.error(f"Użytkownik {user_id} odrzucony z kolejki Redis z powodu timeoutu.")
+            raise HTTPException(
+                status_code=503, 
+                detail="Serwer jest zbyt zajęty (kolejka Redis przepełniona). Spróbuj ponownie za chwilę."
+            )
+    else:
+        # FALLBACK: Jeśli Redis jest wyłączony, aplikacja automatycznie używa kolejki in-memory
+        logger.info(f"Redis jest wyłączony. Używam lokalnej blokady in-memory dla gildii {guild_id}")
+        
+        local_lock = local_guild_locks[guild_id]
+        async with local_lock:
+            analysis_result = await _execute_analysis(payload, guild_id, settings)
 
+    # 3. Zwrócenie wyniku analizy
+    content_type = analysis_result["content_type"]
     logger.info(f"{content_type.capitalize()} analysis completed. Result: {analysis_result}")
+    
     used_model = analysis_result.get("used_model", settings.AVAILABLE_MODELS.get(content_type)[0])
     
     return AnalysisResponse(

backend/app/models/schemas.py

@@ -6,6 +6,7 @@ class TextAnalysisRequest(BaseModel):
     content_type: Literal["text"]
     text: str = Field(..., description="Text content to analyze for deepfake detection")
     guild_id: str = Field(..., description="ID serwera Discord, z którego pochodzi żądanie")
+    user_id: str = Field(..., description="ID użytkownika Discord, który wywołał analizę")
     
     class Config:
         json_schema_extra = {
@@ -20,6 +21,7 @@ class ImageAnalysisRequest(BaseModel):
     content_type: Literal["image"]
     image_url: HttpUrl = Field(..., description="URL of the image to analyze")
     guild_id: str = Field(..., description="ID serwera Discord, z którego pochodzi żądanie")
+    user_id: str = Field(..., description="ID użytkownika Discord, który wywołał analizę")
     
     class Config:
         json_schema_extra = {

backend/app/services/queue.py

@@ -1,6 +1,7 @@
 import logging
 from typing import Optional, Any, Dict
 import json
+import redis.asyncio as redis
 
 from app.core.config import get_settings
 
@@ -17,17 +18,23 @@ class QueueService:
     def __init__(self):
         """Initialize the queue service."""
         self.settings = get_settings()
-        self.redis_client = None
+        self.redis_client = Optional[redis.Redis] = None
         
         if self.settings.REDIS_ENABLED:
             self._initialize_redis()
     
     def _initialize_redis(self):
-        """Initialize Redis connection (future implementation)."""
-        # This will be implemented when Redis support is added
-        logger.info(
-            f"Redis queue service initialized: {self.settings.REDIS_URL}"
-        )
+        
+        try:
+            self.redis_client = redis.from_url(
+                self.settings.REDIS_URL, decode_responses=True
+            )
+            logger.info(
+                f"Redis queue service initialized successfully: {self.settings.REDIS_URL}"
+            )
+        except Exception as e:
+            logger.error(f"Failed to initialize Redis client: {e}")
+            self.redis_client = None
     
     async def enqueue_analysis(
         self,
@@ -36,15 +43,7 @@ class QueueService:
         task_id: str,
     ) -> bool:
         """
-        Enqueue an analysis task.
-        
-        Args:
-            file_url: URL of the file to analyze
-            model: Detector model to use
-            task_id: Unique task identifier
-            
-        Returns:
-            True if successful, False otherwise
+        Enqueue an analysis task (future background worker implementation).
         """
         task_data = {
             "task_id": task_id,
@@ -54,33 +53,24 @@ class QueueService:
         
         logger.info(f"Enqueuing analysis task: {task_id}")
         
-        if self.settings.REDIS_ENABLED:
-            # Future: Push to Redis queue
-            # await self.redis_client.lpush(
-            #     self.settings.REDIS_QUEUE_NAME,
-            #     json.dumps(task_data)
-            # )
-            pass
+        if self.settings.REDIS_ENABLED and self.redis_client:
+            await self.redis_client.lpush(
+                self.settings.REDIS_QUEUE_NAME,
+                json.dumps(task_data)
+            )
+            return True
         
-        return True
+        return False
     
     async def get_task_result(self, task_id: str) -> Optional[Dict[str, Any]]:
         """
         Get analysis result for a task.
-        
-        Args:
-            task_id: Task identifier
-            
-        Returns:
-            Analysis result or None if not found
         """
         logger.info(f"Retrieving result for task: {task_id}")
         
-        if self.settings.REDIS_ENABLED:
-            # Future: Get from Redis
-            # result = await self.redis_client.get(f"result:{task_id}")
-            # return json.loads(result) if result else None
-            pass
+        if self.settings.REDIS_ENABLED and self.redis_client:
+            result = await self.redis_client.get(f"result:{task_id}")
+            return json.loads(result) if result else None
         
         return None
 

backend/requirements.txt

@@ -12,4 +12,5 @@ protobuf
 Pillow
 slowapi
 pytest==7.4.3
-pytest-asyncio==0.21.1
+pytest-asyncio==0.21.1
+redis

index.js

@@ -233,6 +233,7 @@ async function handleAnalysis(interaction, userContent, targetMessage = null, ex
 	try {
 		const { type, payload } = preparePayload(userContent, explicitContentType);
 		payload.guild_id = interaction.guildId;
+		payload.user_id = interaction.user.id;
 
 		console.log(`Wysyłanie zapytania typu: ${type} do API z modelem: ${payload.model || "domyślny"}...`);