from __future__ import annotations import json from fastapi import Depends, FastAPI, HTTPException from sqlalchemy import select, text from services.shared.core import Role, new_id, utc_now_iso from services.shared.db import engine, get_session from services.shared.models import HealthResponse, QueueCreate, QueueOut from services.shared.security import require_roles from services.shared.sql_init import init_sql_schema from services.shared.sql_models import 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()