modularize api

app.py

@@ -1,11 +1,14 @@
-# semantix-api/app.py
-from fastapi import FastAPI, HTTPException
+
+
+from fastapi import FastAPI
 from fastapi.middleware.cors import CORSMiddleware
-from pydantic import BaseModel
 from loguru import logger
 import sys
-import os
+import socket
+import platform
 
+from config import settings
+from routes import game, health
 from services.word_service import WordEmbeddingService
 from services.game_service import GameService
 from services.visualization_service import VisualizationService
@@ -14,126 +17,39 @@ from services.visualization_service import VisualizationService
 logger.remove()
 logger.add(sys.stdout, level="INFO")
 
-app = FastAPI()
-
-# Configure CORS
-app.add_middleware(
-    CORSMiddleware,
-    allow_origins=["*"],  # Adjust this in production
-    allow_credentials=True,
-    allow_methods=["*"],
-    allow_headers=["*"],
-)
-
-# At the top of app.py, add:
-import socket
-import platform
-
-@app.on_event("startup")
-async def startup_event():
-    """Log information about the environment when the application starts"""
-    logger.info("=" * 40)
-    logger.info("Starting Semantix API")
-    logger.info(f"Python version: {platform.python_version()}")
-    logger.info(f"Host: {socket.gethostname()}")
-    logger.info(f"Model URL: {os.getenv('MODEL_URL')}")
-    logger.info("=" * 40)
-
-
-# Initialize services
-try:
+def create_app():
+    app = FastAPI(title=settings.app_name, version=settings.version)
+
+    # Configure CORS
+    app.add_middleware(
+        CORSMiddleware,
+        allow_origins=["*"],
+        allow_credentials=True,
+        allow_methods=["*"],
+        allow_headers=["*"],
+    )
+
+    @app.on_event("startup")
+    async def startup_event():
+        logger.info("=" * 40)
+        logger.info(f"Starting {settings.app_name}")
+        logger.info(f"Python version: {platform.python_version()}")
+        logger.info(f"Host: {socket.gethostname()}")
+        logger.info("=" * 40)
+
+    # Initialize services
     word_service = WordEmbeddingService()
     game_service = GameService(word_service)
     visualization_service = VisualizationService(word_service)
-    logger.info("Services initialized successfully")
-except Exception as e:
-    logger.error(f"Failed to initialize services: {str(e)}")
-    raise e
-
-# Pydantic models for request validation
-class WordCheck(BaseModel):
-    word: str
-
-class JokerUse(BaseModel):
-    joker_type: str
-
-
-@app.get("/api/game-state")
-async def get_game_state():
-    try:
-        return game_service.get_state()
-    except Exception as e:
-        logger.error(f"Error getting game state: {str(e)}")
-        raise HTTPException(status_code=500, detail="Internal server error")
-
-@app.post("/api/check-word")
-async def check_word(word_check: WordCheck):
-    try:
-        response = game_service.check_word(word_check.word)
-        print(f"Response sent: {response}")  
-        return response
-    except Exception as e:
-        logger.error(f"Error checking word: {str(e)}")
-        raise HTTPException(status_code=500, detail="Internal server error")
-
-@app.post("/api/use-joker")
-async def use_joker(joker: JokerUse):
-    try:
-        return game_service.use_joker(joker.joker_type)
-    except Exception as e:
-        logger.error(f"Error using joker: {str(e)}")
-        raise HTTPException(status_code=500, detail="Internal server error")
-
-@app.post("/api/reset-game")
-async def reset_game():
-    try:
-        return game_service.reset_game()
-    except Exception as e:
-        logger.error(f"Error resetting game: {str(e)}")
-        raise HTTPException(status_code=500, detail="Internal server error")
-
-@app.get("/api/visualization")
-async def get_visualization():
-    try:
-        state = game_service.get_state()
-        guessed_words = [attempt['word'] for attempt in state["attempts"]]
-        return visualization_service.prepare_3d_visualization(
-            state["target_word"],
-            guessed_words
-        )
-    except Exception as e:
-        logger.error(f"Error getting visualization: {str(e)}")
-        raise HTTPException(status_code=500, detail="Internal server error")
-
-
-# Add new health models
-class HealthResponse(BaseModel):
-    status: str
-    version: str = "1.0.0"
-    model_loaded: bool
 
-@app.get("/api/health")
-async def health_check():
-    """Health check endpoint"""
-    try:
-        # Check if model is loaded by accessing word service
-        model_loaded = word_service._model is not None
-        return {
-            "status": "healthy" if model_loaded else "degraded",
-            "version": "1.0.0",
-            "model_loaded": model_loaded
-        }
-    except Exception as e:
-        logger.exception("Health check failed")
-        raise HTTPException(status_code=500, detail=str(e))
+    # Include routers
+    app.include_router(game.init_router(game_service, visualization_service))
+    app.include_router(health.init_router(word_service))
 
+    return app
 
-@app.get("/")
-async def root():
-    """Health check endpoint"""
-    return {"status": "ok", "message": "Semantix API is running"}
+app = create_app()
 
 if __name__ == "__main__":
     import uvicorn
-    logger.info("Starting FastAPI server in development mode...")
-    uvicorn.run(app, host="0.0.0.0", port=8000, log_level="info")
+    uvicorn.run("main:app", host="0.0.0.0", port=8000, reload=settings.debug)

config.py

@@ -0,0 +1,9 @@
+from pydantic_settings import BaseSettings
+
+class Settings(BaseSettings):
+    app_name: str = "Semantix API"
+    version: str = "1.0.0"
+    debug: bool = False
+    model_url: str = "https://huggingface.co/Miroir/cc.fr.300.reduced/resolve/main/cc.fr.300.reduced.vec"
+
+settings = Settings()

models/game.py

@@ -0,0 +1,10 @@
+from pydantic import BaseModel
+
+class WordCheck(BaseModel):
+    word: str
+
+class JokerUse(BaseModel):
+    joker_type: str
+
+class SetTargetWord(BaseModel):
+    word: str

models/health.py

@@ -0,0 +1,6 @@
+from pydantic import BaseModel
+
+class HealthResponse(BaseModel):
+    status: str
+    version: str = "1.0.0"
+    model_loaded: bool

requirements.txt

@@ -9,4 +9,5 @@ requests==2.31.0
 scikit-learn==1.3.2
 umap-learn==0.5.5
 requests==2.31.0
-pydantic==2.5.2
+pydantic==2.5.2
+pydantic-settings

routes/game.py

@@ -0,0 +1,64 @@
+from fastapi import APIRouter, HTTPException
+from loguru import logger
+from models.game import WordCheck, JokerUse, SetTargetWord
+from services.game_service import GameService
+from services.visualization_service import VisualizationService
+
+router = APIRouter(prefix="/api")
+
+def init_router(game_service: GameService, visualization_service: VisualizationService):
+    @router.get("/game-state")
+    async def get_game_state():
+        try:
+            return game_service.get_state()
+        except Exception as e:
+            logger.error(f"Error getting game state: {str(e)}")
+            raise HTTPException(status_code=500, detail="Internal server error")
+
+    @router.post("/check-word")
+    async def check_word(word_check: WordCheck):
+        try:
+            response = game_service.check_word(word_check.word)
+            return response
+        except Exception as e:
+            logger.error(f"Error checking word: {str(e)}")
+            raise HTTPException(status_code=500, detail="Internal server error")
+
+    @router.post("/use-joker")
+    async def use_joker(joker: JokerUse):
+        try:
+            return game_service.use_joker(joker.joker_type)
+        except Exception as e:
+            logger.error(f"Error using joker: {str(e)}")
+            raise HTTPException(status_code=500, detail="Internal server error")
+
+    @router.post("/reset-game")
+    async def reset_game():
+        try:
+            return game_service.reset_game()
+        except Exception as e:
+            logger.error(f"Error resetting game: {str(e)}")
+            raise HTTPException(status_code=500, detail="Internal server error")
+
+    @router.get("/visualization")
+    async def get_visualization():
+        try:
+            state = game_service.get_state()
+            guessed_words = [attempt['word'] for attempt in state["attempts"]]
+            return visualization_service.prepare_3d_visualization(
+                state["target_word"],
+                guessed_words
+            )
+        except Exception as e:
+            logger.error(f"Error getting visualization: {str(e)}")
+            raise HTTPException(status_code=500, detail="Internal server error")
+
+    @router.post("/set-target")
+    async def set_target_word(target: SetTargetWord):
+        try:
+            return game_service.set_target_word(target.word)
+        except Exception as e:
+            logger.error(f"Error setting target word: {str(e)}")
+            raise HTTPException(status_code=500, detail="Internal server error")
+
+    return router

routes/health.py

@@ -0,0 +1,22 @@
+from fastapi import APIRouter, HTTPException
+from loguru import logger
+from app.models.health import HealthResponse
+from services.word_service import WordEmbeddingService
+
+router = APIRouter(prefix="/api")
+
+def init_router(word_service: WordEmbeddingService):
+    @router.get("/health")
+    async def health_check():
+        try:
+            model_loaded = word_service._model is not None
+            return {
+                "status": "healthy" if model_loaded else "degraded",
+                "version": "1.0.0",
+                "model_loaded": model_loaded
+            }
+        except Exception as e:
+            logger.exception("Health check failed")
+            raise HTTPException(status_code=500, detail=str(e))
+
+    return router

services/game_service.py

@@ -194,6 +194,19 @@ class GameService:
             logger.exception("Error getting history")
             return []
 
+
+    # In game_service.py
+    def set_target_word(self, word: str):
+        """Set a specific target word for testing"""
+        if not self.word_service.get_vector(word):
+            raise ValueError(f"Word '{word}' not found in vocabulary")
+        self.target_word = word
+        self.attempts = []
+        self.jokers_used = 0
+        return self.get_state()
+
+
+
     def check_word(self, word: str) -> Dict:
         """
         Check a word against the target word and save the attempt.

services/model_downloader.py

@@ -1,29 +0,0 @@
-import os
-import requests
-from loguru import logger
-from pathlib import Path
-
-def download_model(url: str, model_path: str):
-    """Download the model file if it doesn't exist."""
-    if os.path.exists(model_path):
-        logger.info(f"Model file already exists at {model_path}")
-        return
-
-    logger.info(f"Downloading model from {url}")
-    os.makedirs(os.path.dirname(model_path), exist_ok=True)
-    
-    try:
-        response = requests.get(url, stream=True)
-        response.raise_for_status()
-        
-        total_size = int(response.headers.get('content-length', 0))
-        block_size = 1024  # 1 KB
-
-        with open(model_path, 'wb') as f:
-            for data in response.iter_content(block_size):
-                f.write(data)
-                
-        logger.info(f"Model downloaded successfully to {model_path}")
-    except Exception as e:
-        logger.error(f"Error downloading model: {str(e)}")
-        raise

services/study_service.py

@@ -0,0 +1,105 @@
+from loguru import logger
+from typing import List, Dict, Optional
+import numpy as np
+from .word_service import WordEmbeddingService
+
+class StudyService:
+    def __init__(self, word_service: WordEmbeddingService):
+        self.word_service = word_service
+
+    def analyze_word_neighborhood(self, word: str, n_neighbors: int = 20) -> Dict:
+        """
+        Get detailed analysis of a word's semantic neighborhood
+        """
+        try:
+            similar_words = self.word_service.get_most_similar_words(word, n=n_neighbors)
+            vector = self.word_service.get_vector(word)
+            
+            return {
+                "word": word,
+                "in_vocabulary": vector is not None,
+                "similar_words": similar_words,
+                "vector_norm": float(np.linalg.norm(vector)) if vector is not None else None
+            }
+        except Exception as e:
+            logger.exception(f"Error analyzing word neighborhood: {e}")
+            return {
+                "word": word,
+                "in_vocabulary": False,
+                "similar_words": [],
+                "vector_norm": None
+            }
+
+    def compare_words(self, words: List[str]) -> Dict:
+        """
+        Compare multiple words to understand their relationships
+        """
+        results = []
+        similarity_matrix = []
+        
+        for i, word1 in enumerate(words):
+            row = []
+            for j, word2 in enumerate(words):
+                sim = self.word_service.calculate_similarity(word1, word2)
+                row.append(sim)
+            similarity_matrix.append(row)
+            
+            # Get vector if available
+            vector = self.word_service.get_vector(word1)
+            results.append({
+                "word": word1,
+                "in_vocabulary": vector is not None,
+                "vector_norm": float(np.linalg.norm(vector)) if vector is not None else None
+            })
+
+        return {
+            "words": results,
+            "similarity_matrix": similarity_matrix
+        }
+
+    def analyze_similarity_distribution(self, target_word: str, test_words: List[str]) -> Dict:
+        """
+        Analyze similarity distribution across test words
+        """
+        similarities = []
+        for word in test_words:
+            sim = self.word_service.calculate_similarity(target_word, word)
+            similarities.append({
+                "word": word,
+                "similarity": sim
+            })
+        
+        # Sort by similarity
+        similarities.sort(key=lambda x: x["similarity"], reverse=True)
+        
+        return {
+            "target_word": target_word,
+            "similarities": similarities,
+            "statistics": {
+                "max_similarity": max(s["similarity"] for s in similarities),
+                "min_similarity": min(s["similarity"] for s in similarities),
+                "mean_similarity": np.mean([s["similarity"] for s in similarities]),
+                "median_similarity": np.median([s["similarity"] for s in similarities])
+            }
+        }
+
+    def get_similarity_ranges(self, word: str) -> Dict:
+        """
+        Get words in different similarity ranges to understand semantic distances
+        """
+        ranges = [
+            (0.9, 1.0, "very_high"),
+            (0.7, 0.9, "high"),
+            (0.5, 0.7, "medium"),
+            (0.3, 0.5, "low"),
+            (0.1, 0.3, "very_low")
+        ]
+        
+        results = {}
+        for min_sim, max_sim, range_name in ranges:
+            words = self.word_service.get_words_in_range(
+                word, min_sim, max_sim, n=5
+            )
+            results[range_name] = words
+            
+        return results

services/word_service.py

@@ -105,10 +105,10 @@ class WordEmbeddingService:
         self._ensure_model_loaded()
         try:
             w = target_word.lower()
-            if w not in WordEmbeddingService._model:  # Changed from self.model
+            if w not in WordEmbeddingService._model:
                 logger.warning(f"Target word not found in vocab: {target_word}")
                 return []
-            similar = WordEmbeddingService._model.most_similar(w, topn=n)  # Changed from self.model
+            similar = WordEmbeddingService._model.most_similar(w, topn=n) 
             return [{'word': word, 'similarity': float(sim)} for word, sim in similar]
         except Exception:
             logger.exception(f"Error finding similar words for: {target_word}")