#/app/aroflo_connector_app/zones/timesheets/cli.py from __future__ import annotations import json from typing import Any, Dict, List, Optional, Callable import click DEFAULT_WHERE = "and|workdate|>|2001-01-01" def _echo(result: Any) -> None: if isinstance(result, (dict, list)): click.echo(json.dumps(result, indent=2, ensure_ascii=False)) else: click.echo(str(result)) def _available_ops(zone: Any) -> List[str]: ops = getattr(zone, "operations", []) return sorted([o.code for o in ops]) def _run(zone: Any, op_code: str, params: Dict[str, Any]) -> Any: available = set(_available_ops(zone)) if op_code not in available: click.echo(f"❌ Operación '{op_code}' no existe en zona '{zone.code}'.") click.echo("Operaciones disponibles:") for c in _available_ops(zone): click.echo(f" - {c}") raise SystemExit(2) return zone.execute(op_code, params=params) def _require_confirm_if_needed(*, confirm: bool, dry_run: bool, op_label: str) -> None: if dry_run: return if not confirm: raise SystemExit(f"❌ Falta --confirm. Operación de escritura cancelada ({op_label}).") def _add_optional(params: Dict[str, Any], *, pagesize: Optional[int], order: Optional[str]) -> Dict[str, Any]: if pagesize is not None: params["pageSize"] = pagesize if order: params["order"] = order return params def _common_list_options(fn: Callable[..., Any]) -> Callable[..., Any]: fn = click.option("--page", default=1, type=int, show_default=True)(fn) # 👇 importante: multiple=True fn = click.option( "--where", "where_", multiple=True, help="Cláusula WHERE estilo AroFlo. Repetible: --where ... --where ...", )(fn) fn = click.option("--order", default=None, show_default=True, help="Ej: createdutc|desc")(fn) fn = click.option("--pagesize", type=int, default=None, show_default=True)(fn) fn = click.option("--raw", is_flag=True)(fn) return fn def register_cli(root: click.Group, zone: Any) -> None: @root.group(name=zone.code) def timesheets_group(): """Operaciones de la zona timesheets (READ + UI mutations).""" @timesheets_group.command("list") @_common_list_options def list_cmd(page: int, where_: tuple[str, ...], order: Optional[str], pagesize: Optional[int], raw: bool): params = _add_optional({"page": page, "where": list(where_) or [DEFAULT_WHERE], "raw": raw}, pagesize=pagesize, order=order) _echo(_run(zone, "list_timesheets", params)) @timesheets_group.command("get") @click.option("--timesheetid", required=True, help="TimesheetID (AroFlo).") @click.option("--raw", is_flag=True) def get_cmd(timesheetid: str, raw: bool): _echo(_run(zone, "get_timesheet", {"timesheetid": timesheetid, "raw": raw})) @timesheets_group.command("by-user") @click.option("--userid", required=True, help="UserID (AroFlo).") @_common_list_options def by_user_cmd(userid: str, page: int, where_: tuple[str, ...], order: Optional[str], pagesize: Optional[int], raw: bool): params = _add_optional({"userid": userid, "page": page, "where": list(where_) or [DEFAULT_WHERE], "raw": raw}, pagesize=pagesize, order=order) _echo(_run(zone, "get_timesheets_by_user", params)) @timesheets_group.command("by-task") @click.option("--taskid", required=True) @_common_list_options def by_task_cmd(taskid: str, page: int, where_: tuple[str, ...], order, pagesize, raw): params = _add_optional( {"taskid": taskid, "page": page, "where": list(where_) or [DEFAULT_WHERE], "raw": raw}, pagesize=pagesize, order=order ) _echo(_run(zone, "get_timesheets_by_task", params)) @timesheets_group.command("by-type") @click.option("--type", "type_", required=True) @_common_list_options def by_type_cmd(type_: str, page: int, where_: tuple[str, ...], order, pagesize, raw): params = _add_optional( {"type": type_, "page": page, "where": list(where_) or [DEFAULT_WHERE], "raw": raw}, pagesize=pagesize, order=order ) _echo(_run(zone, "get_timesheets_by_type", params)) @timesheets_group.command("after-workdate") @click.option("--workdate", required=True) @_common_list_options def after_workdate_cmd(workdate: str, page: int, where_: tuple[str, ...], order, pagesize, raw): params = _add_optional( {"workdate": workdate, "page": page, "where": list(where_) or [DEFAULT_WHERE], "raw": raw}, pagesize=pagesize, order=order ) _echo(_run(zone, "get_timesheets_after_workdate", params)) @timesheets_group.command("by-user-range") @click.option("--userid", required=True, help="UserID AroFlo") @click.option("--from-date", "from_date", required=True, help="YYYY-MM-DD") @click.option("--to-date", "to_date", required=True, help="YYYY-MM-DD") @click.option("--pagesize", default=None, type=int) @click.option("--raw", is_flag=True, default=False) def by_user_range_cmd(userid: str, from_date: str, to_date: str, pagesize: int | None, raw: bool) -> None: params = { "userid": userid, "from_date": from_date, "to_date": to_date, "pageSize": pagesize, "raw": raw, } _echo(_run(zone, "get_timesheets_by_user_range", params)) # ------------------------- # MUTATIONS (write via UI) # ------------------------- @timesheets_group.command("ui-create") @click.option("--timesheet-date", required=True, help="YYYY-MM-DD") @click.option("--timesheet-bu", required=True, help="Business Unit name (e.g. Telecommunications)") @click.option("--timesheet-user-id", default="", help="Timesheet UI user_id (numeric) OR API userid if your runner resolves it") @click.option("--timesheet-user-name", default="", help="User name as shown in Timesheets dropdown (fallback)") @click.option( "--row", "rows_", multiple=True, help="Fila a crear. Repetible. Formato: hours=5;overhead=Admin Duties;worktype=NT;tracking=ADMIN;note=texto", ) @click.option("--raw", is_flag=True, help="Devuelve stdout/stderr completos.") @click.option("--dry-run", is_flag=True, help="No ejecuta UI; muestra el comando.") @click.option("--confirm", is_flag=True, help="Confirmo que deseo CREAR entries (UI automation).") def ui_create_cmd(timesheet_bu: str, timesheet_user_id: str, timesheet_user_name: str, timesheet_date: str, rows_: tuple[str, ...], raw: bool, dry_run: bool, confirm: bool): _require_confirm_if_needed(confirm=confirm, dry_run=dry_run, op_label="ui-create") params: Dict[str, Any] = { "timesheet_bu": timesheet_bu, "timesheet_user_id": timesheet_user_id, "timesheet_user_name": timesheet_user_name, "timesheet_date": timesheet_date, "rows": list(rows_), # <-- NUEVO: lista de strings --row "raw": raw, "dry_run": dry_run, } _echo(_run(zone, "ui_create_timesheet_entries", params)) @timesheets_group.command("ui-delete") @click.option("--timesheet-bu", required=True, help="Business Unit name") @click.option("--timesheet-user-id", default="", help="Timesheet UI user_id (numeric) OR API userid if your runner resolves it") @click.option("--timesheet-user-name", default="", help="User name as shown in Timesheets dropdown (fallback)") @click.option("--timesheet-date", required=True, help="YYYY-MM-DD") @click.option("--delete-all", is_flag=True, help="Borra TODO en la fecha (hours->0).") @click.option("--include-protected", is_flag=True, help="Incluye filas protegidas (solo con --delete-all).") @click.option("--raw", is_flag=True, help="Devuelve stdout/stderr completos.") @click.option("--dry-run", is_flag=True, help="No ejecuta UI; muestra el comando.") @click.option("--confirm", is_flag=True, help="Confirmo que deseo BORRAR entries (UI automation).") def ui_delete_cmd(timesheet_bu: str, timesheet_user_id: str, timesheet_user_name: str, timesheet_date: str, delete_all: bool, include_protected: bool, raw: bool, dry_run: bool, confirm: bool): _require_confirm_if_needed(confirm=confirm, dry_run=dry_run, op_label="ui-delete") params: Dict[str, Any] = { "timesheet_bu": timesheet_bu, "timesheet_user_id": timesheet_user_id, "timesheet_user_name": timesheet_user_name, "timesheet_date": timesheet_date, "delete_all": delete_all, "include_protected": include_protected, "raw": raw, "dry_run": dry_run, } _echo(_run(zone, "ui_delete_timesheet_entries", params))