from __future__ import annotations import json from fastapi import Depends, FastAPI, HTTPException from sqlalchemy import select, text from services.routing_service import engine as routing_engine from services.shared.core import Role, new_id, utc_now_iso from services.shared.db import engine, get_session from services.shared.models import ( AgentCreate, AgentPoolOut, AgentStatusUpdateIn, EscalationOut, HealthResponse, QueueCreate, QueueOut, RoutingAgentReserveIn, RoutingAgentReserveOut, ) from services.shared.security import require_roles from services.shared.sql_init import init_sql_schema from services.shared.sql_models import AgentRow, EscalationRow, IvrSessionRow, Queue, RoutingCounter app = FastAPI(title="routing-service", version="1.1.0") init_sql_schema() _AGENTS = { "voice": ["operator_a", "operator_b", "operator_c"], "telegram": ["operator_t1", "operator_t2"], "webchat": ["operator_w1"], "email": ["operator_e1"], } def _db_backend_name() -> str: return engine.url.get_backend_name() def _next_agent_index(session, *, channel: str, agent_count: int) -> int: if _db_backend_name() != "postgresql": counter = session.execute( select(RoutingCounter).where(RoutingCounter.channel == channel) ).scalar_one_or_none() if not counter: counter = RoutingCounter(channel=channel, counter=0) session.add(counter) session.flush() idx = counter.counter % agent_count counter.counter += 1 return idx counter_value = session.execute( text( """ INSERT INTO routing_counters(channel, counter) VALUES (:channel, 1) ON CONFLICT (channel) DO UPDATE SET counter = routing_counters.counter + 1 RETURNING counter """ ), {"channel": channel}, ).scalar_one() return (int(counter_value) - 1) % agent_count def _parse_rules_json(raw_rules_json: str | None) -> list[dict]: if not raw_rules_json: return [] try: data = json.loads(raw_rules_json) except (TypeError, json.JSONDecodeError): return [] if isinstance(data, list): return [item for item in data if isinstance(item, dict)] if isinstance(data, dict): nested_rules = data.get("rules") if isinstance(nested_rules, list): return [item for item in nested_rules if isinstance(item, dict)] if data: return [data] return [] return [] def _to_out(row: Queue) -> QueueOut: return QueueOut( queue_id=row.queue_id, name=row.name, description=row.description, rules=_parse_rules_json(row.rules_json), created_at=row.created_at, ) @app.get("/health", response_model=HealthResponse) def health() -> HealthResponse: return HealthResponse(status="ok", service="routing-service", version="v1.1") @app.post("/queues", response_model=QueueOut) def create_queue(payload: QueueCreate, _: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR))) -> QueueOut: session = get_session() try: row = Queue( queue_id=new_id("que"), name=payload.name, description=payload.description, rules_json=json.dumps([r.model_dump() for r in payload.rules], ensure_ascii=False), created_at=utc_now_iso(), ) session.add(row) session.commit() session.refresh(row) return _to_out(row) finally: session.close() @app.get("/queues", response_model=list[QueueOut]) def list_queues( _: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR)), ) -> list[QueueOut]: session = get_session() try: rows = session.execute(select(Queue).order_by(Queue.id.desc())).scalars().all() return [_to_out(r) for r in rows] finally: session.close() @app.patch("/queues/{queue_id}/rules", response_model=QueueOut) def update_rules( queue_id: str, payload: QueueCreate, _: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR)), ) -> QueueOut: session = get_session() try: row = session.execute(select(Queue).where(Queue.queue_id == queue_id)).scalar_one_or_none() if not row: raise HTTPException(status_code=404, detail="Queue not found") row.name = payload.name row.description = payload.description row.rules_json = json.dumps([r.model_dump() for r in payload.rules], ensure_ascii=False) session.commit() return _to_out(row) finally: session.close() @app.post("/queues/{queue_id}/route") def route_interaction( queue_id: str, channel: str, priority: int = 3, ivr_session_id: str | None = None, _: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR)), ) -> dict: session = get_session() try: original_queue = session.execute(select(Queue).where(Queue.queue_id == queue_id)).scalar_one_or_none() if not original_queue: raise HTTPException(status_code=404, detail="Queue not found") resolved_queue_id = queue_id ivr_flow_id = None ivr_outcome_code = None if ivr_session_id: ivr_session = session.execute( select(IvrSessionRow).where(IvrSessionRow.session_id == ivr_session_id) ).scalar_one_or_none() if not ivr_session: raise HTTPException(status_code=404, detail="IVR session not found") if ivr_session.status != "completed": raise HTTPException(status_code=400, detail="IVR session is not completed") if ivr_session.queue_id != queue_id: raise HTTPException(status_code=400, detail="IVR session does not belong to the requested queue") resolved_queue_id = ivr_session.resolved_queue_id or queue_id ivr_flow_id = ivr_session.flow_id ivr_outcome_code = ivr_session.outcome_code queue = session.execute(select(Queue).where(Queue.queue_id == resolved_queue_id)).scalar_one_or_none() if not queue: raise HTTPException(status_code=404, detail="Queue not found") agents = _AGENTS.get(channel) if not agents: raise HTTPException(status_code=400, detail="No agents for channel") idx = _next_agent_index(session, channel=channel, agent_count=len(agents)) sla_seconds = 30 for rule in _parse_rules_json(queue.rules_json): if rule.get("channel") == channel and int(rule.get("priority", 3)) == priority: sla_seconds = int(rule.get("sla_seconds", 30)) break session.commit() response = { "queue_id": resolved_queue_id, "channel": channel, "priority": priority, "assignee": agents[idx], "sla_seconds": sla_seconds, } if ivr_session_id: response.update( { "original_queue_id": queue_id, "resolved_queue_id": resolved_queue_id, "ivr_flow_id": ivr_flow_id, "ivr_outcome_code": ivr_outcome_code, "ivr_session_id": ivr_session_id, } ) return response finally: session.close() @app.delete("/queues/{queue_id}") def delete_queue( queue_id: str, _: dict = Depends(require_roles(Role.ADMIN)), ) -> dict: session = get_session() try: row = session.execute(select(Queue).where(Queue.queue_id == queue_id)).scalar_one_or_none() if not row: raise HTTPException(status_code=404, detail="Queue not found") session.delete(row) session.commit() return {"queue_id": queue_id, "deleted": True} finally: session.close() def _agent_to_out(row: AgentRow) -> AgentPoolOut: return AgentPoolOut( agent_id=row.agent_id, tenant_ids=routing_engine.parse_list_json(row.tenant_ids_json), extension=row.extension, endpoint=row.endpoint, display_name=row.display_name, level=row.level, skills=routing_engine.parse_list_json(row.skills_json), status=row.status, current_call_id=row.current_call_id, max_concurrent_calls=row.max_concurrent_calls, enabled=row.enabled, calls_handled_count=row.calls_handled_count, created_at=row.created_at, updated_at=row.updated_at, ) @app.post("/agents", response_model=AgentPoolOut) def create_agent( payload: AgentCreate, _: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR)), ) -> AgentPoolOut: session = get_session() try: now = utc_now_iso() row = AgentRow( agent_id=new_id("agt"), tenant_ids_json=json.dumps(payload.tenant_ids, ensure_ascii=False), extension=payload.extension, endpoint=payload.endpoint, display_name=payload.display_name, level=payload.level, skills_json=json.dumps(payload.skills, ensure_ascii=False), status="OFFLINE", max_concurrent_calls=payload.max_concurrent_calls, enabled=payload.enabled, created_at=now, updated_at=now, ) session.add(row) session.commit() session.refresh(row) return _agent_to_out(row) finally: session.close() @app.get("/agents", response_model=list[AgentPoolOut]) def list_agents( level: str | None = None, _: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR)), ) -> list[AgentPoolOut]: session = get_session() try: stmt = select(AgentRow).order_by(AgentRow.id.desc()) if level: stmt = stmt.where(AgentRow.level == level) rows = session.execute(stmt).scalars().all() return [_agent_to_out(r) for r in rows] finally: session.close() @app.patch("/agents/{agent_id}/status", response_model=AgentPoolOut) def update_agent_status( agent_id: str, payload: AgentStatusUpdateIn, _: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR)), ) -> AgentPoolOut: session = get_session() try: row = session.execute(select(AgentRow).where(AgentRow.agent_id == agent_id)).scalar_one_or_none() if not row: raise HTTPException(status_code=404, detail="Agent not found") row.status = payload.status if payload.status == "AVAILABLE": row.current_call_id = None row.updated_at = utc_now_iso() session.commit() return _agent_to_out(row) finally: session.close() def _escalation_to_out(row: EscalationRow) -> EscalationOut: return EscalationOut( escalation_id=row.escalation_id, call_id=row.call_id, tenant_id=row.tenant_id, from_level=row.from_level, to_level=row.to_level, reason_code=row.reason_code, required_skills=routing_engine.parse_list_json(row.required_skills_json), priority=row.priority, topic=row.topic, summary=row.summary, status=row.status, assigned_agent_id=row.assigned_agent_id, requested_at=row.requested_at, connected_at=row.connected_at, completed_at=row.completed_at, ) @app.get("/escalations", response_model=list[EscalationOut]) def list_escalations( status: str | None = None, limit: int = 50, _: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR)), ) -> list[EscalationOut]: session = get_session() try: stmt = select(EscalationRow).order_by(EscalationRow.id.desc()).limit(max(1, min(limit, 200))) if status: stmt = stmt.where(EscalationRow.status == status) rows = session.execute(stmt).scalars().all() return [_escalation_to_out(r) for r in rows] finally: session.close() @app.post("/internal/routing/reserve-agent", response_model=RoutingAgentReserveOut) def reserve_agent_endpoint( payload: RoutingAgentReserveIn, _: dict = Depends(require_roles(Role.ADMIN)), ) -> RoutingAgentReserveOut: session = get_session() try: agent = routing_engine.reserve_agent( session, call_id=payload.call_id, level=payload.level, tenant_id=payload.tenant_id, required_skills=payload.required_skills, exclude_agent_ids=payload.exclude_agent_ids, ) if agent is None: raise HTTPException(status_code=409, detail=f"No available {payload.level} agent") return RoutingAgentReserveOut( agent_id=agent.agent_id, extension=agent.extension, endpoint=agent.endpoint, display_name=agent.display_name, ) finally: session.close() @app.post("/internal/routing/release-agent") def release_agent_endpoint( payload: dict, _: dict = Depends(require_roles(Role.ADMIN)), ) -> dict: call_id = str(payload.get("call_id") or "").strip() if not call_id: raise HTTPException(status_code=400, detail="call_id is required") session = get_session() try: agent = routing_engine.release_agent_by_call_id(session, call_id=call_id) return {"call_id": call_id, "released_agent_id": agent.agent_id if agent else None} finally: session.close()