# apps/aroflo_connector_app/agent/cheap/cheap_cli.py from __future__ import annotations import json import os import logging import re from typing import Any, Dict, Optional, List import click from dotenv import load_dotenv from ..det_runner import execute_deterministic from ..det_policy import DeterministicPolicy from ..pending_store import load_pending, save_pending, clear_pending, match_token from .router import build_plan, RouterError from .context_bridge import remember_any_if_present from .extractors import extract_kv_params log = logging.getLogger("aroflo_cheap_cli") load_dotenv(os.path.join(os.getcwd(), ".env")) def _dump(obj: Any) -> str: return json.dumps(obj, indent=2, ensure_ascii=False) def _echo_result(result: Any) -> None: if isinstance(result, (dict, list)): click.echo(_dump(result)) else: click.echo(str(result)) _JSON_LIKE_RE = re.compile(r"^\s*[\[\{].*[\]\}]\s*$", re.DOTALL) _BRACKET_LIST_RE = re.compile(r"^\s*\[(.*)\]\s*$", re.DOTALL) _TRUE_SET = {"true", "1", "yes", "y", "on"} _FALSE_SET = {"false", "0", "no", "n", "off"} def _safe_json_loads(s: str) -> Optional[Any]: try: return json.loads(s) except Exception: return None def _parse_bool(v: Any) -> Any: if isinstance(v, bool): return v if isinstance(v, (int, float)): return bool(v) if isinstance(v, str): t = v.strip().lower() if t in _TRUE_SET: return True if t in _FALSE_SET: return False return v def _parse_number(v: Any) -> Any: if isinstance(v, (int, float)): return v if isinstance(v, str): t = v.strip() if re.fullmatch(r"-?\d+", t): try: return int(t) except Exception: return v if re.fullmatch(r"-?\d+(\.\d+)?", t): try: return float(t) except Exception: return v return v def _parse_array_from_string(s: str) -> List[str]: """ Soporta: - "[a,b,c]" (no JSON estricto) - "a,b,c" - "a" """ s = s.strip() m = _BRACKET_LIST_RE.match(s) if m: inner = m.group(1).strip() if not inner: return [] parts = [p.strip() for p in inner.split(",")] out: List[str] = [] for p in parts: p2 = p.strip() if not p2: continue if (p2.startswith('"') and p2.endswith('"')) or (p2.startswith("'") and p2.endswith("'")): p2 = p2[1:-1] out.append(p2.strip()) return out if "," in s: return [p.strip() for p in s.split(",") if p.strip()] return [s] if s else [] def _normalize_pagesize(params: Dict[str, Any]) -> Dict[str, Any]: """ Normaliza el tamaño de página a la convención de AroFlo: pageSize (S mayúscula). Acepta pagesize (minúscula) por compatibilidad con UX. """ p = dict(params or {}) # Prioridad: si el usuario mandó pageSize explícito, se respeta. if "pageSize" in p and p["pageSize"] is not None: try: p["pageSize"] = int(p["pageSize"]) except Exception: pass return p if "pagesize" in p and p["pagesize"] is not None: try: p["pageSize"] = int(p["pagesize"]) except Exception: p["pageSize"] = p["pagesize"] # opcional: eliminar la clave minúscula para evitar ambigüedad p.pop("pagesize", None) return p def _coerce_param_by_name(name: str, value: Any) -> Any: """ Coerción heurística (sin tocar la arquitectura de zonas): - taskids / *_ids / ids: string -> list - notes/materials (array de objetos): JSON si aplica; si texto plano -> MVP [{"content":...}] o [{"item":...}] - page/pageSize/pagesize: number - dry_run/raw/meta: boolean """ if value is None: return value key = (name or "").strip() lkey = key.lower() # booleans comunes if lkey in {"dry_run", "raw", "meta"}: return _parse_bool(value) # numbers comunes (paging) # - 'page' (número de página) # - 'pagesize' (alias UX) # - 'pagesize' también captura 'pageSize' por lkey if lkey in {"page", "pagesize"}: return _parse_number(value) # arrays de objetos: notes / materials if lkey in {"notes", "materials"}: if isinstance(value, list): return value if isinstance(value, str): s = value.strip() # JSON real if _JSON_LIKE_RE.match(s): j = _safe_json_loads(s) if isinstance(j, list): return j if isinstance(j, dict): return [j] # texto plano -> MVP object if lkey == "notes": return [{"content": s}] if lkey == "materials": return [{"item": s}] # scalar no string if lkey == "notes": return [{"content": str(value)}] if lkey == "materials": return [{"item": str(value)}] # arrays de IDs: taskids, ids, *_ids, *ids if lkey == "taskids" or lkey == "ids" or lkey.endswith("_ids") or lkey.endswith("ids"): if isinstance(value, list): return value if isinstance(value, str): s = value.strip() # JSON list if _JSON_LIKE_RE.match(s): j = _safe_json_loads(s) if isinstance(j, list): return j return _parse_array_from_string(s) return [value] # default return value def _parse_bracket_list(s: str) -> Optional[List[str]]: s = s.strip() if not (s.startswith("[") and s.endswith("]")): return None inner = s[1:-1].strip() if not inner: return [] parts = [p.strip() for p in inner.split(",")] # quitar comillas si vienen parts = [p.strip("'\"") for p in parts if p] return parts def _coerce_params(params: Dict[str, Any]) -> Dict[str, Any]: p = dict(params or {}) for k, v in list(p.items()): if isinstance(v, str): # lista tipo [a,b,c] (aunque no sea JSON) maybe = _parse_bracket_list(v) if maybe is not None: p[k] = maybe continue # int simple if v.isdigit(): if k.lower() in {"mobile", "phone", "fax"}: # conservar como string (mantener ceros a la izquierda) continue p[k] = int(v) continue # float simple try: if "." in v: p[k] = float(v) except Exception: pass return p @click.group() @click.option("--debug", is_flag=True, help="Enable debug logs.") def cli(debug: bool) -> None: logging.basicConfig(level=logging.DEBUG if debug else logging.INFO) @cli.command("ask") @click.argument("question", nargs=-1, required=True) @click.option("--meta", is_flag=True, help="Include deterministic meta wrapper.") @click.option("--default-pagesize", type=int, default=10, show_default=True) @click.option("--max-pagesize", type=int, default=50, show_default=True) def ask_cmd( question: tuple[str, ...], meta: bool, default_pagesize: int, max_pagesize: int, ) -> None: """ Cheap deterministic agent: - Parses a minimal intent (English-oriented). - Builds a deterministic plan for a single zone/op. - Executes via execute_deterministic with safe policy. Writes: - Always runs in dry_run=True first (policy enforced). - Produces a CONFIRM token to apply. """ q = " ".join(question).strip() if not q: raise SystemExit("Empty question.") try: plan = build_plan(q) except RouterError as e: raise SystemExit(f"❌ {e}") # Always extract key=value params from the raw question (MVP for writes like create_task) kv_params = extract_kv_params(q) # Merge: plan params first, then user kv overrides merged_params = {**(plan.params or {}), **(kv_params or {})} # Coerce params (arrays, json, ids, booleans/numbers) for better UX merged_params = _coerce_params(merged_params) merged_params = _normalize_pagesize(merged_params) # ✅ Zone-specific param sanitation (after normalize_pagesize) if plan.zone_code == "permissiongroups": # permissiongroups list: doc only shows page; pagesize/pageSize is not documented merged_params.pop("pagesize", None) merged_params.pop("pageSize", None) # IMPORTANT: keep plan.params in sync for the rest of the flow (missing checks, pending payload, etc.) plan.params = merged_params if not plan.is_ready(): click.echo("Missing required inputs:") for m in plan.missing: click.echo(f"- {m.name}: {m.hint}") raise SystemExit(2) # Policy: force dry_run on writes policy = DeterministicPolicy( force_dry_run_for_writes=True, default_page_size=default_pagesize, max_page_size=max_pagesize, ) click.echo(f"PLAN: {plan.zone_code}.{plan.op_code} ({plan.side_effect})") if plan.summary: click.echo(f"NOTE: {plan.summary}") click.echo(f"PARAMS: {json.dumps(plan.params, ensure_ascii=False)}") click.echo("") result = execute_deterministic( zone_code=plan.zone_code, op_code=plan.op_code, params=plan.params, policy=policy, include_meta=meta, ) # Save snapshots for tasks lists (for "task #N") try: remember_any_if_present(plan.zone_code, result) except Exception: pass _echo_result(result) # If write: create a pending confirmation token with dry_run=False payload if plan.needs_confirmation: # In confirm run, we disable force_dry_run_for_writes and we keep same params payload = { "zone_code": plan.zone_code, "op_code": plan.op_code, "params": dict(plan.params or {}), "meta": bool(meta), "default_pagesize": default_pagesize, "max_pagesize": max_pagesize, } pending = save_pending(payload) click.echo("") click.echo("WRITE detected: dry_run was enforced by policy.") click.echo("To apply the change, run:") click.echo( f' python -m apps.aroflo_connector_app.agent.cheap.cheap_cli confirm "{pending["token"]}"' ) @cli.command("confirm") @click.argument("token", required=True) def confirm_cmd(token: str) -> None: pending = load_pending() hit = match_token(token, pending) if not hit: click.echo("No valid pending action for this token (expired or not found).") raise SystemExit(2) payload = hit.get("payload") or {} zone_code = payload.get("zone_code") op_code = payload.get("op_code") # ✅ Primero asignar params params = payload.get("params") or {} meta = bool(payload.get("meta", False)) default_pagesize = int(payload.get("default_pagesize", 10)) max_pagesize = int(payload.get("max_pagesize", 50)) if not zone_code or not op_code: clear_pending() raise SystemExit("Invalid pending payload (missing zone_code/op_code).") # ✅ Luego coerce/normalize params = _coerce_params(params) params = _normalize_pagesize(params) # ✅ CRÍTICO: en confirm SIEMPRE se aplica (no dry-run) params.pop("dry_run", None) policy = DeterministicPolicy( force_dry_run_for_writes=False, default_page_size=default_pagesize, max_page_size=max_pagesize, ) click.echo(f"CONFIRM: executing {zone_code}.{op_code} with dry_run disabled.") click.echo(f"PARAMS: {json.dumps(params, ensure_ascii=False)}") click.echo("") try: result = execute_deterministic( zone_code=zone_code, op_code=op_code, params=params, policy=policy, include_meta=meta, ) _echo_result(result) # No afirmar aplicado si aún es dry_run if isinstance(result, dict) and result.get("dry_run") is True: raise SystemExit("❌ Not applied: still in dry_run (check confirm override).") click.echo("") click.echo("✅ Change applied.") finally: clear_pending() if __name__ == "__main__": cli()