# /apps/aroflo_connector_app/zones/trackingcenter/base.py from __future__ import annotations from typing import Any, Dict, List, Tuple, Optional from ..base import ZoneOperation, ParamSpec from ._join_utils import request, raw_wrap, coerce_page_size, coerce_order, build_list_params # ------------------------- # CONFIG: campo ID (ajustar si tu doc usa otro nombre) # ------------------------- ID_FIELD = "trackingcenterid" # si en tu doc aparece distinto (ej: trackingcentreid), cámbialo aquí. # ------------------------- # Operation codes (READ base) # ------------------------- OP_LIST = "list_trackingcentres" OP_GET = "get_trackingcentre" # agrega arriba con los otros OP_* OP_RESOLVE_ID = "resolve_trackingcentre_id_by_name" API_ZONE = "trackingcentres" #TrackingCentres, trackingcentres def get_operations() -> List[ZoneOperation]: return [ ZoneOperation( code=OP_LIST, label="List Tracking Centres", description=( "Devuelve una lista paginada de tracking centres (zone=trackingcenter). " "Soporta WHERE opcional, ORDER opcional y paginación." ), http_method="GET", side_effect="read", idempotent=True, default_params={ "where": None, "order": None, "page": 1, "pageSize": None, "raw": False, }, params=[ ParamSpec("where", "string", False, "Cláusula WHERE estilo AroFlo (opcional)."), ParamSpec("order", "string", False, "Orden: campo|asc o campo|desc (AroFlo 'order')."), ParamSpec("page", "integer", False, "Número de página (1..N)."), ParamSpec("pageSize", "integer", False, "Cantidad de registros por página (AroFlo pageSize)."), ParamSpec("raw", "boolean", False, "Si true, devuelve respuesta cruda + meta debug."), ], category="trackingcenter", use_cases=["Listar tracking centres", "Filtrar tracking centres usando WHERE", "Ordenar tracking centres usando ORDER"], risk_level="low", requires_confirmation=False, ), ZoneOperation( code=OP_GET, label="Get Tracking Centre", description=( "Obtiene un tracking centre específico por ID (zone=trackingcenter where=and|trackingcenterid|=|...). " f"Campo ID actual configurado como: {ID_FIELD}." ), http_method="GET", side_effect="read", idempotent=True, default_params={"raw": False}, params=[ ParamSpec("trackingcenterid", "string", True, "TrackingCenterID codificado en AroFlo (ver doc)."), ParamSpec("raw", "boolean", False, "Si true, devuelve respuesta cruda + meta debug."), ], category="trackingcenter", use_cases=["Consultar detalle de un tracking centre por ID"], risk_level="low", requires_confirmation=False, ), ZoneOperation( code=OP_RESOLVE_ID, label="Resolve TrackingCentreID by Name", description=( "Resuelve trackingcentreid a partir de un nombre EXACTO (where=and|name|=|...). " "Opcionalmente filtra por businessunit.orgid para desambiguar." ), http_method="GET", side_effect="read", idempotent=True, default_params={"raw": False}, params=[ ParamSpec("name", "string", True, "Nombre exacto del tracking centre (case-sensitive según AroFlo)."), ParamSpec("orgid", "string", False, "OrgID (business unit) para desambiguar si hay duplicados."), ParamSpec("raw", "boolean", False, "Si true, incluye meta/diagnóstico."), ], category="trackingcentres", use_cases=["Mapear nombre → trackingcentreid para luego filtrar Timesheets"], risk_level="low", requires_confirmation=False, ), ] def supports(operation_code: str) -> bool: return operation_code in {OP_LIST, OP_GET, OP_RESOLVE_ID} def execute(operation_code: str, client: Any, params: Dict[str, Any]) -> Any: raw = bool(params.get("raw", False)) pageSize = coerce_page_size(params) order = coerce_order(params) if operation_code == OP_LIST: where = params.get("where") # opcional page = int(params.get("page", 1)) if page < 1: raise ValueError("page debe ser >= 1.") params_list = build_list_params( zone=API_ZONE, where=where, join=None, order=order, page=page, pageSize=pageSize, ) resp = request(client, "GET", params_list) return raw_wrap(resp, params_list) if raw else resp if operation_code == OP_GET: # Por coherencia con ParamSpec, el parámetro se llama trackingcenterid. tc_id = params["trackingcenterid"] where = f"and|{ID_FIELD}|=|{tc_id}" params_list = [("zone", API_ZONE), ("where", where), ("page", "1")] resp = request(client, "GET", params_list) return raw_wrap(resp, params_list) if raw else resp if operation_code == OP_RESOLVE_ID: name = str(params["name"]).strip() if not name: raise ValueError("name no puede estar vacío.") orgid = params.get("orgid") raw = bool(params.get("raw", False)) # 1) traer por nombre exacto (AroFlo filtra server-side) where = f"and|name|=|{name}" params_list: List[Tuple[str, str]] = [("zone", API_ZONE), ("where", where), ("page", "1")] resp = request(client, "GET", params_list) data = raw_wrap(resp, params_list) if raw else resp # 2) extraer lista payload = data["data"] if raw else data zr = payload.get("zoneresponse", {}) if isinstance(payload, dict) else {} items = zr.get("trackingcentres", []) if isinstance(zr, dict) else [] # 3) filtrar por orgid si se pasó if orgid: items = [ x for x in items if isinstance(x, dict) and isinstance(x.get("businessunit"), dict) and x["businessunit"].get("orgid") == orgid ] if not items: raise ValueError(f"No se encontró tracking centre con name='{name}'" + (f" y orgid='{orgid}'" if orgid else "")) if len(items) > 1: # devolver error útil para UI/agente: candidatos candidates = [ { "trackingcentreid": x.get("trackingcentreid"), "name": x.get("name"), "businessunit": x.get("businessunit"), } for x in items ] raise ValueError( "Nombre ambiguo: hay múltiples tracking centres con ese name. " "Pasa --orgid para desambiguar. " f"Candidatos={candidates}" ) x = items[0] resolved = { "trackingcentreid": x.get("trackingcentreid"), "name": x.get("name"), "businessunit": x.get("businessunit"), } if raw: # preserva data/meta + resolved data["resolved"] = resolved return data return resolved raise ValueError(f"[TrackingCenter.base] Operación no soportada: {operation_code}")