chore:reverse

linguaphoto/api/image.py

@@ -9,6 +9,7 @@
 from linguaphoto.crud.image import ImageCrud
 from linguaphoto.models import Image
 from linguaphoto.schemas.image import ImageTranslateFragment
+from linguaphoto.socket_manager import notify_user
 from linguaphoto.utils.auth import get_current_user_id, subscription_validate
 
 router = APIRouter()
@@ -20,6 +21,8 @@ async def translate_background(image_id: str, image_crud: ImageCrud, user_id: st
         translating_images.append(image_id)
         image = await image_crud.translate(image_id, user_id)
         translating_images.remove(image_id)
+        if image:
+            await notify_user(user_id, image.model_dump())  # Use await here
 
 
 @router.post("/upload", response_model=Image)

linguaphoto/main.py

@@ -1,10 +1,14 @@
 """Defines the main entrypoint for the FastAPI app."""
 
+import socketio
 import uvicorn
 from fastapi import FastAPI
 from fastapi.middleware.cors import CORSMiddleware
 
 from linguaphoto.api.api import router
+from linguaphoto.socket_manager import (
+    sio,  # Import the `sio` and `notify_user` from socket.py
+)
 
 app = FastAPI()
 
@@ -19,6 +23,7 @@
 
 app.include_router(router, prefix="")
 
+app.mount("/", socketio.ASGIApp(sio, other_asgi_app=app))
 if __name__ == "__main__":
     print("Starting webserver...")
     uvicorn.run(app, port=8080, host="0.0.0.0")

linguaphoto/socket_manager.py

@@ -0,0 +1,47 @@
+"""Socket part."""
+
+import socketio
+
+from linguaphoto.settings import settings
+
+# Create a new Socket.IO server with CORS enabled
+sio = socketio.AsyncServer(
+    async_mode="asgi",
+    cors_allowed_origins=[settings.homepage_url],  # Update this to match your frontend URL
+)
+
+# Dictionary to store connected users by their socket ID
+connected_users: dict[str, str] = {}
+
+
+# Handle client connection
+@sio.event
+async def connect(sid: str) -> None:
+    print(f"User connected: {sid}")
+
+
+# Handle client disconnection
+@sio.event
+async def disconnect(sid: str) -> None:
+    if sid in connected_users:
+        print(f"User {connected_users[sid]} disconnected")
+        del connected_users[sid]
+
+
+# Event for registering a specific user (e.g., after authentication)
+@sio.event
+async def register_user(sid: str, user_id: str) -> None:
+    connected_users[user_id] = sid
+    print(f"User {user_id} registered with session ID {sid}")
+
+
+# Event to notify a specific user
+async def notify_user(user_id: str, message: dict) -> None:
+    sid = connected_users.get(user_id)
+    if sid:
+        await sio.emit("notification", message, room=sid)
+    else:
+        print(f"User {user_id} is not connected")
+
+
+# Export the `sio` instance so it can be used in other files