Serhii milestone 2 fix (#55)

* chores_lint_errors_and_path_issues

* feature_db_py

* fix_audio_issue

* chores_lint_error

* fix_openAPI_switching

* chore_module_insert

* fix_edit_collection_fix

* fix_frontend_critical_update

* fix_feedback2_translation_async

* chore:lint

* chore:loading_issue

* chore:module

* chore:version-python-socketio[asgi]

* chore:main.py

* chore:path

linguaphoto/api/image.py

@@ -9,7 +9,7 @@
 from linguaphoto.crud.image import ImageCrud
 from linguaphoto.models import Image
 from linguaphoto.schemas.image import ImageTranslateFragment
-from linguaphoto.socket import notify_user
+from linguaphoto.socket_manager import notify_user
 from linguaphoto.utils.auth import get_current_user_id, subscription_validate
 
 router = APIRouter()

linguaphoto/main.py

@@ -6,7 +6,7 @@
 import socketio
 
 from linguaphoto.api.api import router
-from linguaphoto.socket import sio  # Import the `sio` and `notify_user` from socket.py
+from linguaphoto.socket_manager import sio  # Import the `sio` and `notify_user` from socket.py
 
 app = FastAPI()
 

linguaphoto/requirements.txt

@@ -41,3 +41,4 @@ openai
 requests
 stripe
 python-socketio
+

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