# app/ws/agent.py from sqlalchemy import select from app.core.database import get_db from app.models.visitor import Visitor from app.models.agent import AgentAssignmentHistory from app.services.chat import ChatService from app.services.agent import get_agent_by_id from app.services.visitor import get_visitor_by_id from sqlalchemy.ext.asyncio import AsyncSession from app.ws.connection_manager import agent_manager, visitor_manager from fastapi import APIRouter, WebSocket, WebSocketDisconnect, Depends from app.services.assignment import ( assign_agent_to_visitor, release_assignment, is_agent_allowed_to_send) router = APIRouter() # @router.websocket("/ws/agent/{agent_id}") # async def agent_ws( # websocket: WebSocket, # agent_id: int, # db: AsyncSession = Depends(get_db), # ): # await websocket.accept() # # verify agent exists # agent = await get_agent_by_id(db, agent_id) # if not agent: # await websocket.close(code=1008) # return # agent_manager.connect(agent_id, websocket) # print(f"Agent {agent_id} connected") # try: # while True: # data = await websocket.receive_json() # msg_type = data.get("type") # # ----------------------------------- # # ASSIGN VISITOR TO THIS AGENT # # ----------------------------------- # if msg_type == "assign": # visitor_uid = data.get("visitor_uid") # if not visitor_uid: # continue # # get visitor row # v_res = await db.execute( # select(Visitor).where(Visitor.visitor_id == visitor_uid) # ) # visitor = v_res.scalar_one_or_none() # if not visitor: # await websocket.send_json( # {"type": "error", "message": "Visitor not found"} # ) # continue # try: # assignment = await assign_agent_to_visitor( # db, agent_id=agent_id, visitor_id=visitor.id # ) # except ValueError as e: # await websocket.send_json( # {"type": "error", "message": str(e)} # ) # continue # # notify all agents # await agent_manager.broadcast( # { # "type": "assignment_update", # "visitor_uid": visitor_uid, # "assigned_agent_id": agent_id, # "assigned_agent_name": agent.name, # } # ) # # ----------------------------------- # # RELEASE VISITOR # # ----------------------------------- # elif msg_type == "release": # visitor_uid = data.get("visitor_uid") # if not visitor_uid: # continue # v_res = await db.execute( # select(Visitor).where(Visitor.visitor_id == visitor_uid) # ) # visitor = v_res.scalar_one_or_none() # if not visitor: # continue # try: # await release_assignment(db, agent_id, visitor.id) # except ValueError as e: # await websocket.send_json( # {"type": "error", "message": str(e)} # ) # continue # await agent_manager.broadcast( # { # "type": "assignment_update", # "visitor_uid": visitor_uid, # "assigned_agent_id": None, # "assigned_agent_name": None, # } # ) # # ----------------------------------- # # AGENT TYPING # # ----------------------------------- # elif msg_type == "typing": # visitor_uid = data.get("to_visitor") # 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": "agent", # "text": text, # "isTyping": is_typing, # } # ) # # ----------------------------------- # # AGENT SEND MESSAGE # # ----------------------------------- # elif msg_type == "agent_message": # visitor_uid = data.get("to_visitor") # text = (data.get("text") or "").strip() # if not visitor_uid or not text: # continue # v_res = await db.execute( # select(Visitor).where(Visitor.visitor_id == visitor_uid) # ) # visitor = v_res.scalar_one_or_none() # if not visitor: # await websocket.send_json( # {"type": "error", "message": "Visitor not found"} # ) # continue # # check assignment rule # allowed = await is_agent_allowed_to_send( db, agent_id=agent_id, visitor_id=visitor.id ) # if not allowed: # await websocket.send_json( # { # "type": "error", # "message": "You must be assigned to this visitor to send messages", # } # ) # continue # # save message # chat_service = ChatService(db) # await chat_service.save_message( # client_id=visitor.client_id, # visitor_id=visitor.id, # sender="agent", # text=text, # ) # # send to visitor # target_ws = visitor_manager.get(visitor_uid) # if target_ws: # await target_ws.send_json( # { # "type": "agent_message", # "text": text, # } # ) # # echo to all agents # await agent_manager.broadcast( # { # "type": "agent_message", # "from_agent_id": agent_id, # "agent_name": agent.name, # "visitor_uid": visitor_uid, # "text": text, # } # ) # except WebSocketDisconnect: # print(f"Agent {agent_id} disconnected") # except Exception as e: # print("WS agent exception:", e) # finally: # agent_manager.disconnect(agent_id) # --------------------------------------------------------------------------------------------------------------------------------------- # app/ws/agent.py from sqlalchemy import select from app.core.database import get_db from app.models.visitor import Visitor from app.models.agent import * from datetime import datetime ,timezone from app.services.chat import ChatService from app.services.agent import * from sqlalchemy.ext.asyncio import AsyncSession from fastapi import APIRouter, WebSocket, WebSocketDisconnect, Depends from app.ws.connection_manager import agent_manager, visitor_manager ,admin_manager from app.services.assignment import ( assign_agent_to_visitor, release_assignment, is_agent_allowed_to_send ) router = APIRouter() @router.websocket("/ws/agent/{agent_id}") async def agent_ws( websocket: WebSocket, agent_id: int, db: AsyncSession = Depends(get_db)): await websocket.accept() agent = await get_agent_by_id(db, agent_id) if not agent: await websocket.close(code=1008) return # ----------------------------- # MARK AGENT ONLINE # ----------------------------- await update_agent_status(db, agent_id, True) agent_manager.connect(agent_id, websocket) print(f"Agent {agent_id} connected") # notify admins that this agent is online await admin_manager.broadcast( { "type": "agent_connected", "agent": { "id": agent.id, "name": agent.name, "email": agent.email, "is_online": True }, } ) # ---------------------------------------- await admin_manager.broadcast({ "type": "agent_status", "agent_id": agent_id, "is_online": True, }) try: while True: data = await websocket.receive_json() msg_type = data.get("type") # ----------------------------- # ASSIGN VISITOR TO AGENT # ----------------------------- if msg_type == "assign": visitor_uid = data.get("visitor_uid") if not visitor_uid: continue v_res = await db.execute( select(Visitor).where(Visitor.visitor_id == visitor_uid) ) visitor = v_res.scalar_one_or_none() if not visitor: await websocket.send_json( {"type": "error", "message": "Visitor not found"} ) continue try: await assign_agent_to_visitor( db, agent_id=agent_id, visitor_id=visitor.id ) except ValueError as e: await websocket.send_json( {"type": "error", "message": str(e)} ) continue payload = { "type": "assignment_update", "visitor_uid": visitor_uid, "assigned_agent_id": agent_id, "assigned_agent_name": agent.name, } await agent_manager.broadcast(payload) await admin_manager.broadcast(payload) # ----------------------------- # RELEASE VISITOR # ----------------------------- elif msg_type == "release": visitor_uid = data.get("visitor_uid") if not visitor_uid: continue print(">>>>>>>>>>>>>>>>data",data) v_res = await db.execute( select(Visitor).where(Visitor.visitor_id == visitor_uid) ) print(">>>>>>>>>>>>>>>>v_res",v_res) visitor = v_res.scalar_one_or_none() print(">>>>>>>>>>>>>>>>visitor",visitor) if not visitor: continue try: await release_assignment(db, agent_id, visitor.id) except ValueError as e: await websocket.send_json( {"type": "error", "message": str(e)} ) continue payload = { "type": "assignment_update", "visitor_uid": visitor_uid, "assigned_agent_id": None, "assigned_agent_name": None, } await agent_manager.broadcast(payload) await admin_manager.broadcast(payload) # ----------------------------- # AGENT TYPING # ----------------------------- elif msg_type == "typing": visitor_uid = data.get("to_visitor") 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": "agent", "agent_id": agent_id, "agent_name": agent.name, "text": text, "isTyping": is_typing, } ) # also show in admin panel await admin_manager.broadcast( { "type": "typing", "from": "agent", "agent_id": agent_id, "agent_name": agent.name, "visitor_uid": visitor_uid, "text": text, "isTyping": is_typing, } ) # ----------------------------- # AGENT SEND MESSAGE # ----------------------------- elif msg_type == "agent_message": visitor_uid = data.get("to_visitor") text = (data.get("text") or "").strip() if not visitor_uid or not text: continue v_res = await db.execute( select(Visitor).where(Visitor.visitor_id == visitor_uid) ) visitor = v_res.scalar_one_or_none() if not visitor: await websocket.send_json( {"type": "error", "message": "Visitor not found"} ) continue allowed = await is_agent_allowed_to_send( db, agent_id=agent_id, visitor_id=visitor.id ) if not allowed: await websocket.send_json( { "type": "error", "message": "You must be assigned to this visitor to send messages", } ) continue chat_service = ChatService(db) await chat_service.save_message( client_id=visitor.client_id, visitor_id=visitor.id, sender="agent", text=text, ) target_ws = visitor_manager.get(visitor_uid) if target_ws: await target_ws.send_json( { "type": "agent_message", "text": text, } ) payload = { "type": "agent_message", "from_agent_id": agent_id, "agent_name": agent.name, "visitor_uid": visitor_uid, "text": text, } await agent_manager.broadcast(payload) await admin_manager.broadcast(payload) except WebSocketDisconnect: print(f"Agent {agent_id} disconnected---------------") try: # ============================================ # FIND ALL ACTIVE ASSIGNMENTS FOR THIS AGENT # ============================================ res = await db.execute( select(AgentAssignmentHistory) .where( AgentAssignmentHistory.agent_id == agent_id, AgentAssignmentHistory.ended_at.is_(None) ) ) active_assignments = res.scalars().all() for assignment in active_assignments: visitor_id = assignment.visitor_id # END THE ASSIGNMENT assignment.ended_at = datetime.now(timezone.utc) await db.commit() # Get visitor_uid from Visitor table v_res = await db.execute( select(Visitor).where(Visitor.id == visitor_id) ) visitor = v_res.scalar_one_or_none() if not visitor: continue payload = { "type": "assignment_update", "visitor_uid": visitor.visitor_id, "assigned_agent_id": None, "assigned_agent_name": None, } await agent_manager.broadcast(payload) await admin_manager.broadcast(payload) # ----------------------------- # MARK AGENT OFFLINE # ----------------------------- await update_agent_status(db, agent_id, False) # remove agent from active ws agent_manager.disconnect(agent_id) # notify admin await admin_manager.broadcast( { "type": "agent_disconnected", "agent_id": agent_id, "is_online": False } ) # -------------------------------------- await admin_manager.broadcast({ "type": "agent_status", "agent_id": agent_id, "is_online": False, }) except Exception as e: print("Auto release error:", e) except Exception as e: print("WS agent exception:", e) finally: agent_manager.disconnect(agent_id) await admin_manager.broadcast( { "type": "agent_disconnected", "agent_id": agent_id, } ) # --------------------------------------- await admin_manager.broadcast({ "type": "agent_status", "agent_id": agent_id, "is_online": False, }) # --------------------------------------- # finally: # print(f"Agent {agent_id} disconnected") # try: # # ============================================ # # FIND ALL ACTIVE ASSIGNMENTS FOR THIS AGENT # # ============================================ # res = await db.execute( # select(AgentAssignmentHistory) # .where( # AgentAssignmentHistory.agent_id == agent_id, # AgentAssignmentHistory.ended_at.is_(None) # ) # ) # active_assignments = res.scalars().all() # for assignment in active_assignments: # visitor_id = assignment.visitor_id # # END THE ASSIGNMENT # assignment.ended_at = datetime.now(timezone.utc) # await db.commit() # # Get visitor_uid from Visitor table # v_res = await db.execute( # select(Visitor).where(Visitor.id == visitor_id) # ) # visitor = v_res.scalar_one_or_none() # if not visitor: # continue # payload = { # "type": "assignment_update", # "visitor_uid": visitor.visitor_id, # "assigned_agent_id": None, # "assigned_agent_name": None, # } # await agent_manager.broadcast(payload) # await admin_manager.broadcast(payload) # # ----------------------------- # # MARK AGENT OFFLINE # # ----------------------------- # await update_agent_status(db, agent_id, False) # # remove agent from active ws # agent_manager.disconnect(agent_id) # # notify admin # await admin_manager.broadcast( # { # "type": "agent_disconnected", # "agent_id": agent_id, # "is_online": False # } # ) # except Exception as e: # print("Auto release error:", e)