fyp

app/routes/websocket_listings.py

@@ -0,0 +1,160 @@
+# app/routes/websocket_listings.py
+from fastapi import APIRouter, WebSocket, Query, WebSocketDisconnect
+from typing import Set, Optional
+import json
+import logging
+from datetime import datetime
+from app.database import get_db
+from app.guards.jwt_guard import verify_token
+from app.models.listing import Listing
+
+logger = logging.getLogger(__name__)
+
+router = APIRouter(tags=["WebSocket"])
+
+
+class ConnectionManager:
+    """Manage WebSocket connections grouped by filter type"""
+    def __init__(self):
+        self.active_connections: dict[str, Set[WebSocket]] = {
+            "all": set(),
+            "rent": set(),
+            "roommate": set(),
+            "short-stay": set(),
+            "sale": set(),
+        }
+
+    async def connect(self, websocket: WebSocket, filter_type: str = "all"):
+        await websocket.accept()
+        if filter_type not in self.active_connections:
+            filter_type = "all"
+        self.active_connections[filter_type].add(websocket)
+        logger.info(f"[WS] Client connected: {filter_type} | Total: {len(self.active_connections[filter_type])}")
+
+    def disconnect(self, websocket: WebSocket, filter_type: str = "all"):
+        if filter_type in self.active_connections:
+            self.active_connections[filter_type].discard(websocket)
+        logger.info(f"[WS] Client disconnected: {filter_type}")
+
+    async def broadcast_to_filter(self, filter_type: str, message: dict):
+        """Send message to all clients subscribed to this filter"""
+        if filter_type not in self.active_connections:
+            return
+        
+        disconnected = set()
+        for connection in self.active_connections[filter_type]:
+            try:
+                await connection.send_json(message)
+            except Exception as e:
+                logger.warning(f"[WS] Error sending message: {e}")
+                disconnected.add(connection)
+        
+        # Clean up disconnected clients
+        for conn in disconnected:
+            self.active_connections[filter_type].discard(conn)
+
+    async def broadcast_to_all(self, message: dict):
+        """Send message to all connected clients"""
+        for filter_type in self.active_connections:
+            await self.broadcast_to_filter(filter_type, message)
+
+
+manager = ConnectionManager()
+
+
+@router.websocket("/ws/listings")
+async def websocket_listings_endpoint(websocket: WebSocket, token: str = Query(...)):
+    """
+    WebSocket endpoint for real-time listings updates
+    Connect with: ws://localhost:8000/ws/listings?token=YOUR_JWT_TOKEN
+    """
+    
+    # Verify JWT token
+    try:
+        user = verify_token(token)
+        if not user:
+            await websocket.close(code=4001, reason="Unauthorized")
+            return
+        logger.info(f"[WS] User authenticated: {user.get('email', 'unknown')}")
+    except Exception as e:
+        logger.error(f"[WS] Token verification failed: {e}")
+        await websocket.close(code=4001, reason="Unauthorized")
+        return
+
+    filter_type = "all"
+    await manager.connect(websocket, filter_type)
+
+    try:
+        # Send initial listings on connect
+        db = await get_db()
+        query = {"status": "active"}
+        cursor = db.listings.find(query).sort("created_at", -1)
+        listings = []
+        
+        async for doc in cursor:
+            if "_id" in doc:
+                doc["_id"] = str(doc["_id"])
+            try:
+                listing = Listing(**doc)
+                listings.append(listing.dict(by_alias=True))
+            except Exception as e:
+                logger.warning(f"[WS] Error parsing listing: {e}")
+                continue
+        
+        # Send initial data
+        await websocket.send_json({
+            "type": "initial_data",
+            "data": listings,
+            "count": len(listings),
+            "timestamp": datetime.utcnow().isoformat()
+        })
+
+        while True:
+            # Receive messages from client (filter changes, ping)
+            data = await websocket.receive_text()
+            message = json.loads(data)
+            
+            if message.get("type") == "filter_change":
+                new_filter = message.get("listing_type", "all")
+                if new_filter in manager.active_connections:
+                    # Move connection to new filter group
+                    manager.disconnect(websocket, filter_type)
+                    filter_type = new_filter
+                    await manager.connect(websocket, filter_type)
+                    
+                    # Send listings for new filter
+                    db = await get_db()
+                    query = {"status": "active"}
+                    if new_filter != "all":
+                        query["listing_type"] = new_filter
+                    
+                    cursor = db.listings.find(query).sort("created_at", -1)
+                    listings = []
+                    
+                    async for doc in cursor:
+                        if "_id" in doc:
+                            doc["_id"] = str(doc["_id"])
+                        try:
+                            listing = Listing(**doc)
+                            listings.append(listing.dict(by_alias=True))
+                        except Exception as e:
+                            logger.warning(f"[WS] Error parsing listing: {e}")
+                            continue
+                    
+                    await websocket.send_json({
+                        "type": "listings_update",
+                        "data": listings,
+                        "count": len(listings),
+                        "timestamp": datetime.utcnow().isoformat()
+                    })
+                    logger.info(f"[WS] Filter changed to: {filter_type}")
+            
+            elif message.get("type") == "ping":
+                # Respond to keep-alive ping
+                await websocket.send_json({"type": "pong"})
+
+    except WebSocketDisconnect:
+        manager.disconnect(websocket, filter_type)
+    except Exception as e:
+        logger.error(f"[WS] Error: {e}")
+        manager.disconnect(websocket, filter_type)

main.py

@@ -204,12 +204,14 @@ logger.info("Registering routers...")
 app.include_router(auth.router, prefix="/api/auth", tags=["Authentication"])
 app.include_router(ai_chat_router, prefix="/ai", tags=["Aida AI Chat"])
 
-# ---------- NEW ROUTERS ----------
+# ---------- LISTING ROUTERS ----------
 from app.routes.listing import router as listing_router
 from app.routes.user_public import router as user_public_router
+from app.routes.websocket_listings import router as ws_router
 
 app.include_router(listing_router, prefix="/api/listings", tags=["Listings"])
 app.include_router(user_public_router, prefix="/api/users", tags=["Users"])
+app.include_router(ws_router, tags=["WebSocket"])
 
 logger.info("All routers registered")