# ============================================================================== # FILE: apps/leave_form_app/api/v1/routes_users.py # PROJECT: leave_form_app (api/v1) # PURPOSE: # - Endpoints relacionados con usuarios para el frontend (wizard). # - Paso 3: exponer lista de managers para poblar el dropdown #lf-manager. # # DESIGN: # - Prioridad 1 (fallback explícito): LEAVE_APP_MANAGERS_JSON (lista manual en env). # - Prioridad 2 (auto): intenta listar usuarios desde AroFlo usando CLI (si disponible), # resolver rol con RolePositionMap, y filtrar roles tipo manager/admin. # # RESPONSE: # { "status": "ok", "managers": [ { "username": "...", "email": "...", "userid": "...", "positionname": "...", "orgname": "...", "role": "manager" }, ... ] } # # NOTES: # - El comando exacto para listar usuarios via CLI puede variar según tu CLI actual. # Si el comando falla, te deja el fallback por env. # ============================================================================== from __future__ import annotations import os import json import sys import subprocess from pathlib import Path from typing import Any, Dict, List from flask import Blueprint, request, jsonify, current_app from config.db import get_session from apps.leave_form_app.models import RolePositionMap users_bp = Blueprint("leave_api_users", __name__) def _project_root() -> str: return str(Path(__file__).resolve().parents[4]) def _resolve_role_from_position(positionid: str, positionname: str) -> str: session = get_session() try: q = session.query(RolePositionMap) row = None if positionid: row = q.filter(RolePositionMap.positionid == positionid).first() if not row and positionname: row = q.filter(RolePositionMap.positionname == positionname).first() if row and row.is_enabled and row.role: return row.role return "worker" finally: session.close() def _normalize_user_from_list_item(u: Dict[str, Any]) -> Dict[str, str]: # Estructura típica: puede variar según el output del CLI. # Se intenta ser tolerante a keys comunes. org = u.get("org") or {} userposition = u.get("userposition") or {} givennames = (u.get("givennames") or "").strip() surname = (u.get("surname") or "").strip() full_name = (f"{givennames} {surname}").strip() or (u.get("username") or "").strip() positionid = (userposition.get("positionid") or u.get("positionid") or "").strip() positionname = (userposition.get("positionname") or u.get("positionpositionname") or u.get("positionname") or u.get("position") or "").strip() return { "userid": (u.get("userid") or "").strip(), "email": (u.get("email") or "").strip(), "username": (u.get("username") or full_name or "").strip(), "full_name": full_name, "orgid": (org.get("orgid") or u.get("orgid") or "").strip(), "orgname": (org.get("orgname") or u.get("orgname") or "").strip(), "positionid": positionid, "positionname": positionname, } def _load_managers_from_env() -> List[Dict[str, str]]: raw = (os.getenv("LEAVE_APP_MANAGERS_JSON") or "").strip() if not raw: return [] try: data = json.loads(raw) if not isinstance(data, list): return [] out: List[Dict[str, str]] = [] for it in data: if not isinstance(it, dict): continue out.append({ "username": (it.get("username") or "").strip(), "email": (it.get("email") or "").strip(), "userid": (it.get("userid") or "").strip(), "positionname": (it.get("positionname") or "").strip(), "orgname": (it.get("orgname") or "").strip(), "role": (it.get("role") or "manager").strip(), }) # filtra mínimos return [m for m in out if m.get("username") or m.get("email")] except Exception: current_app.logger.exception("[leave_app] invalid LEAVE_APP_MANAGERS_JSON") return [] def _run_aroflo_users_list(pagesize: int = 200, page: int = 1) -> Dict[str, Any]: project_root = _project_root() cmd = [ sys.executable, "-m", "apps.aroflo_connector_app.cli", "users", "list", "--raw", "--pagesize", str(pagesize), "--page", str(page), ] env = os.environ.copy() env["PYTHONPATH"] = project_root + (":" + env["PYTHONPATH"] if env.get("PYTHONPATH") else "") env["PYTHONNOUSERSITE"] = "1" proc = subprocess.run(cmd, cwd=project_root, env=env, capture_output=True, text=True) if proc.returncode != 0: raise RuntimeError(f"AroFlo CLI failed rc={proc.returncode} stderr={proc.stderr.strip()[:800]}") out = (proc.stdout or "").strip() return json.loads(out) @users_bp.get("/users/managers") def users_managers(): """ Query params opcionales: - orgname: filtra por orgname si llega (útil si manejas multi-org) """ orgname = (request.args.get("orgname") or "").strip() # 1) Fallback manual por env (útil para demo / cuando CLI no está listo) managers_env = _load_managers_from_env() if managers_env: if orgname: managers_env = [m for m in managers_env if (m.get("orgname") or "") == orgname] return jsonify({"status": "ok", "source": "env", "managers": managers_env}) # 2) Intento automático por AroFlo CLI try: payload = _run_aroflo_users_list(pagesize=500) data = payload.get("data") or payload zr = data.get("zoneresponse") or {} users = zr.get("users") or [] if not isinstance(users, list): users = [] out: List[Dict[str, str]] = [] for u in users: if not isinstance(u, dict): continue norm = _normalize_user_from_list_item(u) role = _resolve_role_from_position(norm.get("positionid", ""), norm.get("positionname", "")) if role not in ("manager", "admin", "director"): continue if orgname and norm.get("orgname") != orgname: continue out.append({ "userid": norm.get("userid", ""), "email": norm.get("email", ""), "username": norm.get("username", ""), "full_name": norm.get("full_name", ""), "positionname": norm.get("positionname", ""), "orgname": norm.get("orgname", ""), "role": role, }) # orden simple por nombre out.sort(key=lambda x: (x.get("full_name") or x.get("username") or "").lower()) return jsonify({"status": "ok", "source": "aroflo_cli", "managers": out}) except Exception as e: current_app.logger.exception("[leave_app] users_managers failed") return jsonify({ "status": "error", "message": str(e), "hint": "If your CLI list command differs, update _run_aroflo_users_list() or set LEAVE_APP_MANAGERS_JSON.", }), 500