# app/ws/admin.py # from fastapi import APIRouter, WebSocket # from app.ws.connection_manager import visitor_manager, admin_manager # from app.services.chat import ChatService # from app.core.database import get_db # router = APIRouter() # @router.websocket("/ws/admin") # async def admin_ws(websocket: WebSocket): # await websocket.accept() # admin_manager.connect(websocket) # try: # while True: # data = await websocket.receive_json() # # typing # if data["type"] == "typing": # target = visitor_manager.get(data["to"]) # if target: # await target.send_json({ # "type": "typing", # "from": "admin", # "text": data["text"] # }) # # send admin message # if data["type"] == "admin_message": # visitor_uid = data["to"] # text = data["text"] # # Save to DB # async for db in get_db(): # service = ChatService(db) # visitor = await service.get_or_create_visitor( # widget_id=data.get("widget_id"), # visitor_uid=visitor_uid # ) # await service.save_message( # client_id=visitor.client_id, # visitor_id=visitor.id, # sender="admin", # text=text # ) # break # # Send to visitor # target = visitor_manager.get(visitor_uid) # if target: # await target.send_json({ # "type": "admin_message", # "text": text # }) # # Echo to admins # await admin_manager.broadcast({ # "type": "admin_message", # "to": visitor_uid, # "text": text # }) # except: # pass # finally: # admin_manager.disconnect(websocket) # ------------------------------------------------------------------------------------------------------------------------------------ # ------------------------------------------------------------------------------------------------------------------------------------ # app/ws/admin.py # from fastapi import APIRouter, WebSocket, WebSocketDisconnect # from sqlalchemy.ext.asyncio import AsyncSession # from app.core.database import get_db # from app.services.chat import ChatService # from app.ws.connection_manager import visitor_manager, admin_manager # router = APIRouter() # @router.websocket("/ws/admin") # async def admin_ws(websocket: WebSocket): # """ # Single central admin panel. # Frontend connects to: # ws://localhost:8000/ws/admin # """ # await websocket.accept() # admin_manager.connect(websocket) # try: # while True: # data = await websocket.receive_json() # msg_type = data.get("type") # # ---------------------------- # # Admin is typing to a visitor # # ---------------------------- # if msg_type == "typing": # visitor_uid = data.get("to") # if not visitor_uid: # continue # target_ws = visitor_manager.get(visitor_uid) # if target_ws: # text = data.get("text", "") # is_typing = data.get("isTyping", bool(text)) # await target_ws.send_json( # { # "type": "typing", # "from": "admin", # "text": text, # "isTyping": is_typing, # } # ) # # ------------------------- # # Admin sends a message # # ------------------------- # elif msg_type == "admin_message": # visitor_uid = data.get("to") # text = (data.get("text") or "").strip() # widget_id = data.get("widget_id") # should be sent from Admin.jsx # if not visitor_uid or not text: # continue # # Save message in DB # async for db in get_db(): # get AsyncSession # service = ChatService(db) # # Make sure visitor exists (and is linked to this client) # visitor = await service.get_or_create_visitor( # widget_id=widget_id, # visitor_uid=visitor_uid, # ) # await service.save_message( # client_id=visitor.client_id, # visitor_id=visitor.id, # sender="admin", # text=text, # ) # break # # Send to visitor (if online) # target_ws = visitor_manager.get(visitor_uid) # if target_ws: # await target_ws.send_json( # { # "type": "admin_message", # "text": text, # } # ) # # Echo to all admins (so your own UI updates) # # await admin_manager.broadcast( # # { # # "type": "admin_message", # # "to": visitor_uid, # # "text": text, # # } # # ) # except WebSocketDisconnect: # pass # except Exception: # # you can log this # pass # finally: # admin_manager.disconnect(websocket) # app/ws/admin.py from fastapi import APIRouter, WebSocket, WebSocketDisconnect from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy import select from app.core.database import get_db from app.services.chat import ChatService from app.services.agent import assign_agent_to_visitor, release_assignment, get_active_assignment from app.ws.connection_manager import visitor_manager, admin_manager from app.models.visitor import Visitor router = APIRouter() # @router.websocket("/ws/admin") # async def admin_ws(websocket: WebSocket): # """ # Multi-agent admin/agent websocket. # Frontend connects to: # ws://localhost:8000/ws/admin # Agents send messages like: # { "type": "admin_message", "to": "", "widget_id": "...", "text": "...", "agent_id": 1 } # { "type": "typing", "to": "", "from": "agent", "text": "...", "agent_id": 1 } # { "type": "claim_chat", "visitor_uid": "", "agent_id": 1 } # { "type": "release_chat", "visitor_uid": "", "agent_id": 1 } # """ # await websocket.accept() # admin_manager.connect(websocket) # try: # while True: # data = await websocket.receive_json() # msg_type = data.get("type") # # ------------------------- # # Agent typing to a visitor # # ------------------------- # if msg_type == "typing": # visitor_uid = data.get("to") # if not visitor_uid: # continue # target_ws = visitor_manager.get(visitor_uid) # if target_ws: # text = data.get("text", "") # is_typing = data.get("isTyping", bool(text)) # await target_ws.send_json( # { # "type": "typing", # "from": "admin", # "text": text, # "isTyping": is_typing, # } # ) # # ------------------------- # # Agent claims a chat # # ------------------------- # elif msg_type == "claim_chat": # agent_id = data.get("agent_id") # visitor_uid = data.get("visitor_uid") # if not agent_id or not visitor_uid: # continue # async for db in get_db(): # async_db: AsyncSession = db # # Find visitor by visitor_uid # result = await async_db.execute( # select(Visitor).where(Visitor.visitor_id == visitor_uid) # ) # visitor = result.scalar_one_or_none() # if not visitor: # # notify this agent only # await websocket.send_json( # { # "type": "error", # "message": "Visitor not found", # "visitor_uid": visitor_uid, # } # ) # break # try: # await assign_agent_to_visitor(async_db, agent_id=agent_id, visitor_id=visitor.id) # except ValueError as e: # # someone else already assigned # await websocket.send_json( # { # "type": "assignment_blocked", # "visitor_uid": visitor_uid, # "reason": str(e), # } # ) # break # # broadcast assignment to all agents # await admin_manager.broadcast( # { # "type": "assignment_update", # "visitor_uid": visitor_uid, # "agent_id": agent_id, # "status": "assigned", # } # ) # break # # ------------------------- # # Agent releases a chat # # ------------------------- # elif msg_type == "release_chat": # agent_id = data.get("agent_id") # visitor_uid = data.get("visitor_uid") # if not agent_id or not visitor_uid: # continue # async for db in get_db(): # async_db: AsyncSession = db # result = await async_db.execute( # select(Visitor).where(Visitor.visitor_id == visitor_uid) # ) # visitor = result.scalar_one_or_none() # if not visitor: # break # await release_assignment(async_db, agent_id=agent_id, visitor_id=visitor.id) # await admin_manager.broadcast( # { # "type": "assignment_update", # "visitor_uid": visitor_uid, # "agent_id": None, # "status": "released", # } # ) # break # # ------------------------- # # Agent sends a message # # ------------------------- # elif msg_type == "admin_message": # visitor_uid = data.get("to") # text = (data.get("text") or "").strip() # widget_id = data.get("widget_id") # agent_id = data.get("agent_id") # if not visitor_uid or not text or not widget_id or not agent_id: # continue # # Save message & enforce assignment # async for db in get_db(): # get AsyncSession # async_db: AsyncSession = db # service = ChatService(async_db) # # Make sure visitor exists (and is linked to this client) # visitor = await service.get_or_create_visitor( # widget_id=widget_id, # visitor_uid=visitor_uid, # ) # # Check assignment # active_assignment = await get_active_assignment(async_db, visitor_id=visitor.id) # if active_assignment and active_assignment.agent_id != agent_id: # # Not allowed # await websocket.send_json( # { # "type": "assignment_blocked", # "visitor_uid": visitor_uid, # "reason": "Chat is assigned to another agent.", # } # ) # break # # If no assignment yet, auto-assign to this agent # if not active_assignment: # from app.services.agent import assign_agent_to_visitor # await assign_agent_to_visitor(async_db, agent_id=agent_id, visitor_id=visitor.id) # await admin_manager.broadcast( # { # "type": "assignment_update", # "visitor_uid": visitor_uid, # "agent_id": agent_id, # "status": "assigned", # } # ) # # Save message to DB # await service.save_message( # client_id=visitor.client_id, # visitor_id=visitor.id, # sender="agent", # text=text, # ) # break # # Send to visitor (if online) # target_ws = visitor_manager.get(visitor_uid) # if target_ws: # await target_ws.send_json( # { # "type": "admin_message", # "text": text, # } # ) # # Echo to all agents # await admin_manager.broadcast( # { # "type": "admin_message", # "to": visitor_uid, # "text": text, # "agent_id": agent_id, # } # ) # except WebSocketDisconnect: # pass # except Exception: # # log if you want # pass # finally: # admin_manager.disconnect(websocket) # app/ws/admin.py from fastapi import APIRouter, WebSocket, WebSocketDisconnect, Depends from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy import select from app.core.database import get_db from app.ws.connection_manager import admin_manager, visitor_manager, agent_manager from app.models.visitor import Visitor from app.models.client import Client from app.models.agent import Agent, AgentAssignmentHistory from app.services.chat import ChatService router = APIRouter() @router.websocket("/ws/admin") async def admin_ws( websocket: WebSocket, db: AsyncSession = Depends(get_db), ): await websocket.accept() admin_manager.connect(websocket) print("Admin connected") try: # --------------------------- # INITIAL SNAPSHOT # --------------------------- # online visitors (from WS memory) visitor_uids = list(visitor_manager.active_connections.keys()) visitors_payload = [] if visitor_uids: res = await db.execute( select(Visitor, Client) .join(Client, Visitor.client_id == Client.id) .where(Visitor.visitor_id.in_(visitor_uids)) ) rows = res.all() visitor_ids = [v.id for (v, _) in rows] # active assignments ass_res = await db.execute( select(AgentAssignmentHistory).where( AgentAssignmentHistory.visitor_id.in_(visitor_ids), AgentAssignmentHistory.ended_at.is_(None), ) ) assignments = ass_res.scalars().all() ass_map = {a.visitor_id: a for a in assignments} for v, c in rows: a = ass_map.get(v.id) visitors_payload.append( { "visitor_uid": v.visitor_id, "name": v.name, "client_id": v.client_id, "city": v.city, "country": v.country, "region": v.region, "ip_address": v.ip_address, "client_name": c.client_name if c else None, "assigned_agent_id": a.agent_id if a else None, } ) # online agents agent_ids = list(agent_manager.active_connections.keys()) agents_payload = [] if agent_ids: res = await db.execute(select(Agent).where(Agent.id.in_(agent_ids))) for a in res.scalars().all(): agents_payload.append( { "id": a.id, "name": a.name, "email": a.email, } ) await websocket.send_json( { "type": "initial_state", "agents": agents_payload, "visitors": visitors_payload, } ) # --------------------------- # LISTEN ADMIN EVENTS # --------------------------- while True: data = await websocket.receive_json() msg_type = data.get("type") # ADMIN TYPING if msg_type == "typing": visitor_uid = data.get("to") text = data.get("text", "") is_typing = data.get("isTyping", bool(text)) ws = visitor_manager.get(visitor_uid) if ws: await ws.send_json( { "type": "typing", "from": "admin", "text": text, "isTyping": is_typing, } ) # also show to other admins await admin_manager.broadcast( { "type": "typing", "from": "admin", "visitor_uid": visitor_uid, "text": text, "isTyping": is_typing, } ) # ADMIN SEND MESSAGE elif msg_type == "admin_message": visitor_uid = data.get("to") text = (data.get("text") or "").strip() widget_id = data.get("widget_id") if not visitor_uid or not text: continue # resolve visitor via ChatService (using widget_id + visitor_uid) service = ChatService(db) visitor = await service.get_or_create_visitor( widget_id=widget_id, visitor_uid=visitor_uid, ) await service.save_message( client_id=visitor.client_id, visitor_id=visitor.id, sender="admin", text=text, ) # send to visitor (if online) v_ws = visitor_manager.get(visitor_uid) if v_ws: await v_ws.send_json( { "type": "admin_message", "text": text, } ) # echo to all admins await admin_manager.broadcast( { "type": "admin_message", "visitor_uid": visitor_uid, "text": text, } ) except WebSocketDisconnect: print("Admin disconnected") except Exception as e: print("Admin WS error:", e) finally: admin_manager.disconnect(websocket)