# apps/aroflo_connector_app/agent/cheap/zone_resolvers/users.py from __future__ import annotations from typing import Any, Dict, Set, List, Optional from ..plan import Plan, MissingParam from ..extractors import ( extract_page_and_pagesize, extract_where_tail, extract_join, extract_kv_params, normalize_join_keyword, ) from ..context_bridge import get_user_ref_by_index JOINABLE = { "customfields", "documentsandphotos", "featureaccess", "notes", "permissiongroups", "trackingcentredefaults", } WRITE_OPS = { "create_user", "update_user", "update_user_customfields", "update_user_permissiongroups", "update_user_featureaccess", } def _choose_join_op(join: str, available_ops: Set[str]) -> str: j = normalize_join_keyword(join) if j in JOINABLE: candidate = f"get_users_with_{j}" if candidate in available_ops: return candidate return "list_users" def _extract_user_index_local(q: str) -> Optional[int]: """ Parser ultra-barato para: - "user #2" / "usuario #3" - "user 2" / "usuario 4" """ import re s = (q or "").strip().lower() m = re.search(r"\b(user|usuario)\s*#?\s*(\d{1,3})\b", s) if not m: return None try: idx = int(m.group(2)) except Exception: return None return idx if idx > 0 else None def _coerce_obj_from_kv(params: Dict[str, Any], *, key: str, child_key: str) -> None: """ Permite alias barato: orgid=123 => org={orgid:123} sin romper si el usuario ya mandó org como dict/JSON. """ if key in params and isinstance(params.get(key), dict): return # orgid -> org alias = params.get(child_key) if alias and key not in params: params[key] = {child_key: alias} def build_plan(*, question: str, intent: str, available_ops: Set[str]) -> Plan: q = question or "" missing: List[MissingParam] = [] # Base op selection op_code = intent if intent in available_ops else "list_users" side_effect = "write" if op_code in WRITE_OPS else "read" params: Dict[str, Any] = {} # key=value parsing barato kv = extract_kv_params(q) # ---- Allowed keys por operación (NO inventar; filtrar duro) ---- READ_KEYS = { "page", "pageSize", "pagesize", "where", "order", "join", "userid", "UserID", "user_id", "raw", } CREATE_KEYS = { "givennames", "surname", "username", "password", "org", "orgid", "email", "email2", "phone", "fax", "mobile", "position", "accesstype", "address", "permissiongroups", "customfields", "featureaccess", "dry_run", "raw", } UPDATE_KEYS = { "userid", "UserID", "user_id", "givennames", "surname", "username", "email", "email2", "phone", "fax", "mobile", "position", "accesstype", "archived", "org", "orgid", "address", "permissiongroups", "customfields", "featureaccess", "dry_run", "raw", } UPDATE_CF_KEYS = {"userid", "UserID", "user_id", "customfields", "dry_run", "raw"} UPDATE_PG_KEYS = {"userid", "UserID", "user_id", "permissiongroups", "dry_run", "raw"} UPDATE_FA_KEYS = {"userid", "UserID", "user_id", "featureaccess", "dry_run", "raw"} # Choose allowed keys based on op_code allowed = set(READ_KEYS) if op_code == "create_user": allowed |= CREATE_KEYS elif op_code == "update_user": allowed |= UPDATE_KEYS elif op_code == "update_user_customfields": allowed |= UPDATE_CF_KEYS elif op_code == "update_user_permissiongroups": allowed |= UPDATE_PG_KEYS elif op_code == "update_user_featureaccess": allowed |= UPDATE_FA_KEYS # Copy only allowed kv pairs for k, v in kv.items(): if k in allowed: params[k] = v # Normalize pageSize/pagesize if "pagesize" in params and "pageSize" not in params: params["pageSize"] = params["pagesize"] # Join-based read selection join = extract_join(q) or (params.get("join") if isinstance(params.get("join"), str) else None) if op_code == "list_users" and join: op_code = _choose_join_op(join, available_ops) # where tail if provided where = extract_where_tail(q) if where: params["where"] = where # paging hints from natural language page, pagesize = extract_page_and_pagesize(q) if page is not None: params["page"] = page if pagesize is not None: params["pageSize"] = pagesize # get_user: resolve userid explicit or from context ("user #N") if op_code == "get_user": userid = params.get("userid") or params.get("UserID") or params.get("user_id") if not userid: idx = _extract_user_index_local(q) if idx: ref = get_user_ref_by_index(idx) if ref: userid = ref.get("userid") or ref.get("UserID") or ref.get("user_id") if userid: params["userid"] = userid else: missing.append(MissingParam("userid", "Provide userid=XXXX or refer to a recent list as 'user #2'.")) # ---- Mutations: normalizaciones + required ---- if op_code == "create_user": # soportar alias orgid=... -> org={orgid:...} _coerce_obj_from_kv(params, key="org", child_key="orgid") for req in ("givennames", "surname", "username", "password", "org"): if not params.get(req): if req == "org": missing.append(MissingParam("org.orgid", "Provide org={orgid:...} or orgid=...")) else: missing.append(MissingParam(req, f"Provide {req}=...")) if op_code in {"update_user", "update_user_customfields", "update_user_permissiongroups", "update_user_featureaccess"}: userid = params.get("userid") or params.get("UserID") or params.get("user_id") if userid: params["userid"] = userid else: idx = _extract_user_index_local(q) if idx: ref = get_user_ref_by_index(idx) if ref: params["userid"] = ref.get("userid") or ref.get("UserID") or ref.get("user_id") if not params.get("userid"): missing.append(MissingParam("userid", "Provide userid=XXXX or refer to a recent list as 'user #2'.")) if op_code == "update_user": # opcional: orgid -> org _coerce_obj_from_kv(params, key="org", child_key="orgid") # Asegurar que no sea update vacío (tu mutations.py lo valida, pero aquí lo hacemos explícito) allowed_update_fields = { "givennames", "surname", "username", "email", "email2", "phone", "fax", "mobile", "position", "accesstype", "archived", "address", "org", "permissiongroups", "customfields", "featureaccess", } has_any = any(k in params and params.get(k) is not None for k in allowed_update_fields) if not has_any: missing.append(MissingParam("fields", "Provide at least one field to update (e.g., email=..., archived=true).")) if op_code == "update_user_customfields" and not params.get("customfields"): missing.append(MissingParam("customfields", "Provide customfields=[{fieldid/name/type/value}, ...]")) if op_code == "update_user_permissiongroups" and not params.get("permissiongroups"): missing.append(MissingParam("permissiongroups", "Provide permissiongroups=[{groupid:...}, ...] (REPLACES)")) if op_code == "update_user_featureaccess" and not params.get("featureaccess"): missing.append(MissingParam("featureaccess", "Provide featureaccess=[{featureid:..., featurevalue:...}, ...]")) # Writes require confirmation needs_confirmation = (side_effect == "write") # Validate op exists; fallback if not if op_code not in available_ops: op_code = "list_users" side_effect = "read" needs_confirmation = False summary = f"users.{op_code} ({side_effect})" if join and side_effect == "read": summary += f" join={join}" return Plan( zone_code="users", op_code=op_code, params=params, side_effect=side_effect, # type: ignore[arg-type] needs_confirmation=needs_confirmation, summary=summary, missing=missing, )