# /apps/aroflo_connector_app/zones/timesheets/mutations.py from __future__ import annotations from typing import Any, Dict, List import sys import subprocess from ..base import ZoneOperation, ParamSpec # ------------------------- # Operation codes (WRITE via UI automation) # ------------------------- OP_UI_CREATE = "ui_create_timesheet_entries" OP_UI_DELETE = "ui_delete_timesheet_entries" def get_operations() -> List[ZoneOperation]: return [ ZoneOperation( code=OP_UI_CREATE, label="UI Create Timesheet entries", description=( "Crea entradas de timesheet usando UI automation (Playwright). " "Requiere storageState (bootstrap). " "Soporta rows opcional para definir filas dinámicas; " "si no se envía rows, el flujo usa filas por defecto." ), http_method="UI", side_effect="write", idempotent=False, default_params={ # requeridos a nivel de runner "timesheet_bu": None, # Business Unit name "timesheet_user_id": None, # preferido "timesheet_user_name": None, # fallback "timesheet_date": None, # YYYY-MM-DD "rows": None, # Optional[List[str]] "dry_run": False, "raw": False, }, params=[ # UI selection ParamSpec("timesheet_bu", "string", True, "Business Unit a seleccionar en AroFlo."), ParamSpec("timesheet_user_id", "string", False, "AroFlo User ID (preferido, único)."), ParamSpec("timesheet_user_name", "string", False, "Nombre completo del usuario (fallback)."), # date + rows ParamSpec("timesheet_date", "string", True, "Fecha objetivo (YYYY-MM-DD)."), ParamSpec( "rows", "array", False, ( "Lista de filas (opcional). Cada item es un string con formato: " "'hours=5;overhead=Admin Duties;worktype=NT;tracking=ADMIN;note=...'" ), items_schema={"type": "string"}, ), ParamSpec("dry_run", "boolean", False, "Si true, no ejecuta UI; solo preview del comando."), ParamSpec("raw", "boolean", False, "Si true, incluye stdout/stderr completos del runner."), ], category="timesheets", use_cases=["Crear overheads por UI", "Inserción controlada cuando API es READ ONLY"], risk_level="high", requires_confirmation=True, ), ZoneOperation( code=OP_UI_DELETE, label="UI Delete Timesheet entries", description=( "Elimina entradas (hours -> 0 + save) usando UI automation. " "Soporta delete_all e include_protected." ), http_method="UI", side_effect="write", idempotent=False, default_params={ # requeridos a nivel de runner "timesheet_bu": None, # Business Unit name "timesheet_user_id": None, # preferido "timesheet_user_name": None, # fallback "timesheet_date": None, # YYYY-MM-DD "delete_all": False, "include_protected": False, "dry_run": False, "raw": False, }, params=[ # UI selection (faltaban en tu operación delete) ParamSpec("timesheet_bu", "string", True, "Business Unit a seleccionar en AroFlo."), ParamSpec("timesheet_user_id", "string", False, "AroFlo User ID (preferido, único)."), ParamSpec("timesheet_user_name", "string", False, "Nombre completo del usuario (fallback)."), ParamSpec("timesheet_date", "string", True, "Fecha objetivo (YYYY-MM-DD)."), ParamSpec("delete_all", "boolean", False, "Borra TODO en la fecha (hours->0)."), ParamSpec("include_protected", "boolean", False, "Incluye filas protegidas (solo con delete_all)."), ParamSpec("dry_run", "boolean", False, "Si true, no ejecuta UI; solo preview del comando."), ParamSpec("raw", "boolean", False, "Si true, incluye stdout/stderr completos del runner."), ], category="timesheets", use_cases=["Borrar overheads por UI", "Limpieza de días antes de reinsertar"], risk_level="high", requires_confirmation=True, ), ] def supports(operation_code: str) -> bool: return operation_code in {OP_UI_CREATE, OP_UI_DELETE} def _preview(argv: List[str]) -> Dict[str, Any]: return { "dry_run": True, "invocation": "subprocess", "argv": argv, } def _run_ui(argv: List[str], *, raw: bool) -> Any: """ Ejecuta el runner de UI automation como subprocess para aislar Playwright del runtime principal de la app/zona. """ proc = subprocess.run( argv, capture_output=True, text=True, check=False, ) if raw: return { "returncode": proc.returncode, "stdout": proc.stdout, "stderr": proc.stderr, "argv": argv, } out_tail = (proc.stdout or "").strip().splitlines()[-30:] err_tail = (proc.stderr or "").strip().splitlines()[-30:] return { "returncode": proc.returncode, "stdout_tail": "\n".join(out_tail), "stderr_tail": "\n".join(err_tail), "argv": argv, } def execute(operation_code: str, client: Any, params: Dict[str, Any]) -> Any: # client no se usa: esta mutation opera vía UI automation raw = bool(params.get("raw", False)) dry_run = bool(params.get("dry_run", False)) # ------------------------- # Required inputs (date + BU + user selector) # ------------------------- timesheet_date = params.get("timesheet_date") or params.get("timesheet-date") if not timesheet_date: raise ValueError("timesheet_date es requerido (YYYY-MM-DD).") timesheet_bu = params.get("timesheet_bu") or params.get("timesheet-bu") if not timesheet_bu or not str(timesheet_bu).strip(): raise ValueError("timesheet_bu es requerido (ej: 'Utility Solutions Group').") timesheet_user_id = str(params.get("timesheet_user_id") or params.get("timesheet-user-id") or "").strip() timesheet_user_name = str(params.get("timesheet_user_name") or params.get("timesheet-user-name") or "").strip() if not timesheet_user_id and not timesheet_user_name: raise ValueError("Debes enviar timesheet_user_id o timesheet_user_name (al menos uno).") # ------------------------- # Base runner invocation # ------------------------- base_cmd = [ sys.executable, "-m", "apps.aroflo_connector_app.ui_automation.runner", "--timesheet-date", str(timesheet_date), "--timesheet-bu", str(timesheet_bu), ] # Preferir ID si existe if timesheet_user_id: base_cmd.extend(["--timesheet-user-id", timesheet_user_id]) else: base_cmd.extend(["--timesheet-user-name", timesheet_user_name]) # ------------------------- # UI CREATE # ------------------------- if operation_code == OP_UI_CREATE: argv = list(base_cmd) rows = params.get("rows") or [] if rows: if not isinstance(rows, list): raise ValueError("rows debe ser una lista de strings.") for r in rows: s = str(r).strip() if s: argv.extend(["--row", s]) argv.append("timesheet-create") if dry_run: return _preview(argv) return _run_ui(argv, raw=raw) # ------------------------- # UI DELETE # ------------------------- if operation_code == OP_UI_DELETE: argv = list(base_cmd) if bool(params.get("delete_all", False)): argv.append("--delete-all") if bool(params.get("include_protected", False)): argv.append("--include-protected") argv.append("timesheet-delete") if dry_run: return _preview(argv) return _run_ui(argv, raw=raw) raise ValueError(f"[Timesheets.mutations] Operación no soportada: {operation_code}")