# apps/aroflo_connector_app/ui_automation_zones/runner.py import argparse import sys import shlex from dataclasses import replace from pathlib import Path from dotenv import load_dotenv from .core.config import build_config # Cargar SIEMPRE el .env del root del repo (flask_server/.env) CORE_ROOT = Path(__file__).resolve().parents[3] load_dotenv(CORE_ROOT / ".env") def build_parser() -> argparse.ArgumentParser: p = argparse.ArgumentParser("AroFlo UI Automation (Zones)") # Connection/auth p.add_argument("--base-url") p.add_argument("--username") p.add_argument("--password") # Multi-tenant / session runtime p.add_argument("--tenant-id", default="", help="Tenant/Org identifier for session pooling (e.g. orgid)") p.add_argument("--reuse-session", action="store_true", help="Reuse a persistent Playwright session for this tenant") p.add_argument("--keep-open", action="store_true", help="Keep browser/session open after running the command") # Files/paths p.add_argument("--state-file", help="Override storageState.json path (advanced)") p.add_argument("--artifacts-dir", help="Override artifacts directory") # Playwright options p.add_argument("--headless", action=argparse.BooleanOptionalAction, default=None) p.add_argument("--slow-mo-ms", type=int) p.add_argument("--step", action="store_true") p.add_argument("--pause-on-mfa", action="store_true") p.add_argument("--mfa-code", default="") # Timesheets targeting p.add_argument("--timesheet-date", default="") p.add_argument("--timesheet-bu", default="") p.add_argument("--bu", default="", help="General BU selector (alias for --timesheet-bu)") p.add_argument("--timesheet-user-id", default="") p.add_argument("--timesheet-user-name", default="") p.add_argument("--row", action="append", default=[], help="Timesheet row spec: hours=5;overhead=...;worktype=NT;tracking=ADMIN;note=...") p.add_argument("--delete-all", action="store_true") p.add_argument("--include-protected", action="store_true") p.add_argument("--user-email", default="", help="Target user email for Users navigation") p.add_argument("--doc", action="append", default=[], help="User doc spec: file=/a.pdf;comment=...;filter=Show All") p.add_argument("--file", action="append", default=[], help="Legacy: file path (repeatable)") p.add_argument("--comment", action="append", default=[], help="Legacy: comment (index-based)") p.add_argument("--filter", action="append", default=[], help="Legacy: filter (index-based)") sub = p.add_subparsers(dest="cmd", required=True) # Runtime control sub.add_parser("shell", help="Interactive session shell (keeps Playwright alive)") sub.add_parser("status", help="Show session status for the tenant (if any)") sub.add_parser("close-session", help="Close session for the tenant (if open)") # Maintenance c = sub.add_parser("cleanup-artifacts", help="Delete old artifacts (png/html) under artifacts dir") c.add_argument("--days", type=int, default=14, help="Delete artifacts older than N days (default: 14)") c.add_argument("--dry-run", action="store_true", help="Only show what would be deleted") # Timesheets ts = sub.add_parser("timesheets-nav", help="Navigate: Dashboard -> Manage -> Timesheets (zones)") sub.add_parser("timesheets-create", help="Create timesheet rows (zones)") sub.add_parser("timesheets-delete", help="Delete timesheet rows (zones)") sub.add_parser("users-nav", help="Navigate: Dashboard -> Manage -> Users (zones)") sub.add_parser("users-upload-docs", help="Upload documents to a user (zones)") sub.add_parser("siteadmin-home", help="Navigate directly to SiteAdministration home (zones)") sub.add_parser("siteadmin-business-units", help="SiteAdministration: Business Units") sub.add_parser("siteadmin-users", help="SiteAdministration: Users") sub.add_parser("siteadmin-settings", help="SiteAdministration: Settings") sub.add_parser("siteadmin-integration", help="SiteAdministration: Integration") sub.add_parser("siteadmin-settings-general", help="SiteAdmin Settings: General group") sub.add_parser("siteadmin-settings-timesheets", help="SiteAdmin Settings: Timesheets group") sub.add_parser("siteadmin-timesheets-read-overheads", help="SiteAdmin Settings: Timesheets > Overheads (read JSON)") return p def main(argv=None) -> int: args = build_parser().parse_args(argv) cfg = build_config(args) if args.cmd == "shell": from .core.runtime import SESSION_POOL, MFARequired from .core.artifacts import create_run_dir, shot tenant_id = cfg.tenant_id or "" run_dir = create_run_dir(cfg, "shell") print(f"[UIZ][shell] tenant={tenant_id}") print("[UIZ][shell] commands: status | connect | timesheets-nav | timesheets-create | timesheets-delete | users-nav | users-upload-docs | siteadmin-home | siteadmin-business-units | siteadmin-users | siteadmin-settings | siteadmin-settings-general | siteadmin-settings-timesheets | siteadmin-timesheets-read-overheads | siteadmin-integration | close | exit") def _allow_mfa() -> bool: # In shell mode, allow MFA by default if stdin is interactive. try: interactive = sys.stdin.isatty() except Exception: interactive = False return ( bool((args.mfa_code or "").strip()) or bool(getattr(cfg, "pause_on_mfa", False)) or interactive ) def _connect(mfa_code: str = "") -> bool: lock = SESSION_POOL.tenant_lock(tenant_id) with lock: try: rt = SESSION_POOL.get( cfg=cfg, tenant_id=tenant_id, run_dir=run_dir, mfa_code=mfa_code, allow_mfa=_allow_mfa(), ) # evidencia visual try: shot(rt.page, run_dir, "shell-connected") except Exception: pass rt.touch() return True except MFARequired: print("[UIZ] MFA_REQUIRED: reintenta con --pause-on-mfa o pasa --mfa-code 123456 (o usa comando mfa).") return False while True: try: cmdline = input("ui> ").strip() except EOFError: cmdline = "exit" # Respeta comillas en shell try: tokens = shlex.split(cmdline) except Exception: tokens = cmdline.split() cmd = (tokens[:1] or [""])[0].lower() rest = tokens[1:] if cmd in ("exit", "quit"): print("[UIZ][shell] bye") break if cmd == "status": print(SESSION_POOL.status(tenant_id)) continue if cmd == "connect": # Permite: connect 123456 code = rest[0].strip() if rest else (args.mfa_code or "") ok = _connect(code) if ok: print("[UIZ] connected") print(SESSION_POOL.status(tenant_id)) continue if cmd == "timesheets-nav": lock = SESSION_POOL.tenant_lock(tenant_id) with lock: rt = SESSION_POOL.get( cfg=cfg, tenant_id=tenant_id, run_dir=run_dir, mfa_code=(args.mfa_code or ""), allow_mfa=_allow_mfa(), ) from .zones.timesheets.flows import nav as flow_nav try: # Permite: timesheets-nav cfg_run = cfg if rest: cfg_run = replace(cfg, timesheet_bu=" ".join(rest).strip()) flow_nav.run(rt.page, cfg_run, run_dir, select_user=False) shot(rt.page, run_dir, "shell-timesheets-nav-ok") rt.save_state() rt.touch() print("[UIZ] timesheets-nav OK") except Exception as e: try: shot(rt.page, run_dir, "shell-timesheets-nav-error") except Exception: pass print(f"[UIZ] timesheets-nav ERROR: {e}") continue if cmd == "timesheets-create": lock = SESSION_POOL.tenant_lock(tenant_id) with lock: rt = SESSION_POOL.get( cfg=cfg, tenant_id=tenant_id, run_dir=run_dir, mfa_code=(args.mfa_code or ""), allow_mfa=_allow_mfa(), ) from .zones.timesheets.flows import nav as flow_nav from .zones.timesheets.flows import select_date as flow_date from .zones.timesheets.flows import select_user as flow_user from .zones.timesheets.flows import create as flow_create from .zones.timesheets.flows.create import NewRowSpec try: cfg_run = cfg rows = list(cfg.timesheet_rows) if getattr(cfg, "timesheet_rows", None) else [] def _parse_shell_flags(tokens: list[str]) -> dict: out = {"bu": "", "date": "", "user_name": "", "user_id": "", "rows": []} i = 0 while i < len(tokens): t = tokens[i] if t in ("--bu",): out["bu"] = tokens[i + 1] if i + 1 < len(tokens) else "" i += 2 continue if t in ("--timesheet-date", "--date"): out["date"] = tokens[i + 1] if i + 1 < len(tokens) else "" i += 2 continue if t in ("--timesheet-user-name", "--user", "--user-name"): out["user_name"] = tokens[i + 1] if i + 1 < len(tokens) else "" i += 2 continue if t in ("--timesheet-user-id", "--user-id"): out["user_id"] = tokens[i + 1] if i + 1 < len(tokens) else "" i += 2 continue if t == "--row": out["rows"].append(tokens[i + 1] if i + 1 < len(tokens) else "") i += 2 continue i += 1 return out if rest: if any(r.startswith("--") for r in rest): parsed = _parse_shell_flags(rest) if parsed["bu"] or parsed["date"] or parsed["user_name"] or parsed["user_id"]: cfg_run = replace( cfg, timesheet_bu=parsed["bu"] or cfg.timesheet_bu, timesheet_date=parsed["date"] or cfg.timesheet_date, timesheet_user_name=parsed["user_name"] or cfg.timesheet_user_name, timesheet_user_id=parsed["user_id"] or cfg.timesheet_user_id, ) for spec in parsed["rows"]: if spec: rows.append(spec) else: # Positional: timesheets-create bu = "" date = "" user_name = "" for token in rest: if "-" in token and len(token) == 10 and token[4] == "-" and token[7] == "-": date = token if date: rest_no_date = [t for t in rest if t != date] else: rest_no_date = rest if rest_no_date: bu = rest_no_date[0] user_name = " ".join(rest_no_date[1:]).strip() cfg_run = replace( cfg, timesheet_bu=bu or cfg.timesheet_bu, timesheet_date=date or cfg.timesheet_date, timesheet_user_name=user_name or cfg.timesheet_user_name, ) # Convert row specs to NewRowSpec if provided rows = None if rows: # parse string specs parsed_rows = [] for s in rows: spec = s if isinstance(s, dict): parsed_rows.append(s) continue # reuse config parser via build_config in core, but here we parse inline out = {} for part in str(spec).split(";"): part = part.strip() if not part: continue if "=" not in part: raise RuntimeError(f"Invalid --row token (expected key=value): {part!r}") k, v = part.split("=", 1) out[k.strip().lower()] = v.strip() if not out.get("hours") or not out.get("overhead"): raise RuntimeError(f"--row missing hours/overhead: {spec}") out.setdefault("worktype", "NT") out.setdefault("tracking", "ADMIN") out.setdefault("note", "") parsed_rows.append(out) rows = [ NewRowSpec( hours=r["hours"], overhead=r["overhead"], note=(r.get("note") or None), worktype_label=r.get("worktype", "NT"), tracking_label=r.get("tracking", "ADMIN"), ) for r in parsed_rows ] elif getattr(cfg_run, "timesheet_rows", None): rows = [ NewRowSpec( hours=r["hours"], overhead=r["overhead"], note=(r.get("note") or None), worktype_label=r.get("worktype", "NT"), tracking_label=r.get("tracking", "ADMIN"), ) for r in cfg_run.timesheet_rows ] flow_nav.run(rt.page, cfg_run, run_dir) if cfg_run.timesheet_user_id or cfg_run.timesheet_user_name: flow_user.run( rt.page, cfg_run, run_dir, user_id=cfg_run.timesheet_user_id, user_name=cfg_run.timesheet_user_name, ) if cfg_run.timesheet_date: flow_date.run(rt.page, cfg_run, run_dir, target_date=cfg_run.timesheet_date) flow_create.run(rt.page, cfg_run, run_dir, rows=rows) shot(rt.page, run_dir, "shell-timesheets-create-ok") rt.save_state() rt.touch() print("[UIZ] timesheets-create OK") except Exception as e: try: shot(rt.page, run_dir, "shell-timesheets-create-error") except Exception: pass print(f"[UIZ] timesheets-create ERROR: {e}") continue if cmd == "timesheets-delete": lock = SESSION_POOL.tenant_lock(tenant_id) with lock: rt = SESSION_POOL.get( cfg=cfg, tenant_id=tenant_id, run_dir=run_dir, mfa_code=(args.mfa_code or ""), allow_mfa=_allow_mfa(), ) from .zones.timesheets.flows import nav as flow_nav from .zones.timesheets.flows import select_date as flow_date from .zones.timesheets.flows import select_user as flow_user from .zones.timesheets.flows import delete as flow_delete from .zones.timesheets.flows.delete import DeleteRule try: cfg_run = cfg def _parse_shell_flags(tokens: list[str]) -> dict: out = { "bu": "", "date": "", "user_name": "", "user_id": "", "delete_all": False, "include_protected": False, } i = 0 while i < len(tokens): t = tokens[i] if t in ("--bu",): out["bu"] = tokens[i + 1] if i + 1 < len(tokens) else "" i += 2 continue if t in ("--timesheet-date", "--date"): out["date"] = tokens[i + 1] if i + 1 < len(tokens) else "" i += 2 continue if t in ("--timesheet-user-name", "--user", "--user-name"): out["user_name"] = tokens[i + 1] if i + 1 < len(tokens) else "" i += 2 continue if t in ("--timesheet-user-id", "--user-id"): out["user_id"] = tokens[i + 1] if i + 1 < len(tokens) else "" i += 2 continue if t == "--delete-all": out["delete_all"] = True i += 1 continue if t == "--include-protected": out["include_protected"] = True i += 1 continue i += 1 return out if rest: if any(r.startswith("--") for r in rest): parsed = _parse_shell_flags(rest) cfg_run = replace( cfg, timesheet_bu=parsed["bu"] or cfg.timesheet_bu, timesheet_date=parsed["date"] or cfg.timesheet_date, timesheet_user_name=parsed["user_name"] or cfg.timesheet_user_name, timesheet_user_id=parsed["user_id"] or cfg.timesheet_user_id, delete_all=parsed["delete_all"] or cfg.delete_all, include_protected=parsed["include_protected"] or cfg.include_protected, ) else: # Positional: timesheets-delete bu = "" date = "" user_name = "" for token in rest: if "-" in token and len(token) == 10 and token[4] == "-" and token[7] == "-": date = token if date: rest_no_date = [t for t in rest if t != date] else: rest_no_date = rest if rest_no_date: bu = rest_no_date[0] user_name = " ".join(rest_no_date[1:]).strip() cfg_run = replace( cfg, timesheet_bu=bu or cfg.timesheet_bu, timesheet_date=date or cfg.timesheet_date, timesheet_user_name=user_name or cfg.timesheet_user_name, ) rule = DeleteRule( overheads_to_delete=[ "Admin Duties", "Admin Duties - Telecommunications", ], protected_overheads=[ "Lunch Break - Unpaid", ], match_mode="exact", delete_all=bool(getattr(cfg_run, "delete_all", False)), include_protected=bool(getattr(cfg_run, "include_protected", False)), ) flow_nav.run(rt.page, cfg_run, run_dir) if cfg_run.timesheet_user_id or cfg_run.timesheet_user_name: flow_user.run( rt.page, cfg_run, run_dir, user_id=cfg_run.timesheet_user_id, user_name=cfg_run.timesheet_user_name, ) if cfg_run.timesheet_date: flow_date.run(rt.page, cfg_run, run_dir, target_date=cfg_run.timesheet_date) flow_delete.run(rt.page, cfg_run, run_dir, rule=rule) shot(rt.page, run_dir, "shell-timesheets-delete-ok") rt.save_state() rt.touch() print("[UIZ] timesheets-delete OK") except Exception as e: try: shot(rt.page, run_dir, "shell-timesheets-delete-error") except Exception: pass print(f"[UIZ] timesheets-delete ERROR: {e}") continue if cmd == "users-nav": lock = SESSION_POOL.tenant_lock(tenant_id) with lock: rt = SESSION_POOL.get( cfg=cfg, tenant_id=tenant_id, run_dir=run_dir, mfa_code=(args.mfa_code or ""), allow_mfa=_allow_mfa(), ) from .zones.users.flows import nav as flow_nav try: cfg_run = cfg if rest: # Permite: users-nav OR users-nav email = "" bu = "" for token in rest: if "@" in token and not email: email = token if email: bu = " ".join([t for t in rest if t != email]).strip() else: bu = " ".join(rest).strip() if email or bu: cfg_run = replace( cfg, user_email=email or cfg.user_email, timesheet_bu=bu or cfg.timesheet_bu, ) flow_nav.run(rt.page, cfg_run, run_dir) shot(rt.page, run_dir, "shell-users-nav-ok") rt.save_state() rt.touch() print("[UIZ] users-nav OK") except Exception as e: try: shot(rt.page, run_dir, "shell-users-nav-error") except Exception: pass print(f"[UIZ] users-nav ERROR: {e}") continue if cmd == "siteadmin-home": lock = SESSION_POOL.tenant_lock(tenant_id) with lock: rt = SESSION_POOL.get( cfg=cfg, tenant_id=tenant_id, run_dir=run_dir, mfa_code=(args.mfa_code or ""), allow_mfa=_allow_mfa(), ) from .common.selectors import siteadministration as siteadmin_sel from .core.artifacts import shot try: siteadmin_sel.ensure_siteadministration(rt.page) shot(rt.page, run_dir, "shell-siteadmin-home-ok") rt.save_state() rt.touch() print("[UIZ] siteadmin-home OK") except Exception as e: try: shot(rt.page, run_dir, "shell-siteadmin-home-error") except Exception: pass print(f"[UIZ] siteadmin-home ERROR: {e}") continue if cmd == "siteadmin-business-units": lock = SESSION_POOL.tenant_lock(tenant_id) with lock: rt = SESSION_POOL.get( cfg=cfg, tenant_id=tenant_id, run_dir=run_dir, mfa_code=(args.mfa_code or ""), allow_mfa=_allow_mfa(), ) from .common.siteadministration import bussiness_units as sa_bu from .core.artifacts import shot try: sa_bu.run(rt.page, cfg, run_dir) shot(rt.page, run_dir, "shell-siteadmin-business-units-ok") rt.save_state() rt.touch() print("[UIZ] siteadmin-business-units OK") except Exception as e: try: shot(rt.page, run_dir, "shell-siteadmin-business-units-error") except Exception: pass print(f"[UIZ] siteadmin-business-units ERROR: {e}") continue if cmd == "siteadmin-users": lock = SESSION_POOL.tenant_lock(tenant_id) with lock: rt = SESSION_POOL.get( cfg=cfg, tenant_id=tenant_id, run_dir=run_dir, mfa_code=(args.mfa_code or ""), allow_mfa=_allow_mfa(), ) from .common.siteadministration import users as sa_users from .core.artifacts import shot try: sa_users.run(rt.page, cfg, run_dir) shot(rt.page, run_dir, "shell-siteadmin-users-ok") rt.save_state() rt.touch() print("[UIZ] siteadmin-users OK") except Exception as e: try: shot(rt.page, run_dir, "shell-siteadmin-users-error") except Exception: pass print(f"[UIZ] siteadmin-users ERROR: {e}") continue if cmd == "siteadmin-settings": lock = SESSION_POOL.tenant_lock(tenant_id) with lock: rt = SESSION_POOL.get( cfg=cfg, tenant_id=tenant_id, run_dir=run_dir, mfa_code=(args.mfa_code or ""), allow_mfa=_allow_mfa(), ) from .common.siteadministration.settings import settings as sa_settings from .core.artifacts import shot try: sa_settings.run(rt.page, cfg, run_dir) shot(rt.page, run_dir, "shell-siteadmin-settings-ok") rt.save_state() rt.touch() print("[UIZ] siteadmin-settings OK") except Exception as e: try: shot(rt.page, run_dir, "shell-siteadmin-settings-error") except Exception: pass print(f"[UIZ] siteadmin-settings ERROR: {e}") continue if cmd == "siteadmin-settings-general": lock = SESSION_POOL.tenant_lock(tenant_id) with lock: rt = SESSION_POOL.get( cfg=cfg, tenant_id=tenant_id, run_dir=run_dir, mfa_code=(args.mfa_code or ""), allow_mfa=_allow_mfa(), ) from .common.siteadministration.settings.general import general as sa_general from .core.artifacts import shot try: sa_general.run(rt.page, cfg, run_dir) shot(rt.page, run_dir, "shell-siteadmin-settings-general-ok") rt.save_state() rt.touch() print("[UIZ] siteadmin-settings-general OK") except Exception as e: try: shot(rt.page, run_dir, "shell-siteadmin-settings-general-error") except Exception: pass print(f"[UIZ] siteadmin-settings-general ERROR: {e}") continue if cmd == "siteadmin-settings-timesheets": lock = SESSION_POOL.tenant_lock(tenant_id) with lock: rt = SESSION_POOL.get( cfg=cfg, tenant_id=tenant_id, run_dir=run_dir, mfa_code=(args.mfa_code or ""), allow_mfa=_allow_mfa(), ) from .common.siteadministration.settings.timesheets import timesheets as sa_ts from .core.artifacts import shot try: sa_ts.run(rt.page, cfg, run_dir) shot(rt.page, run_dir, "shell-siteadmin-settings-timesheets-ok") rt.save_state() rt.touch() print("[UIZ] siteadmin-settings-timesheets OK") except Exception as e: try: shot(rt.page, run_dir, "shell-siteadmin-settings-timesheets-error") except Exception: pass print(f"[UIZ] siteadmin-settings-timesheets ERROR: {e}") continue if cmd == "siteadmin-timesheets-read-overheads": lock = SESSION_POOL.tenant_lock(tenant_id) with lock: rt = SESSION_POOL.get( cfg=cfg, tenant_id=tenant_id, run_dir=run_dir, mfa_code=(args.mfa_code or ""), allow_mfa=_allow_mfa(), ) from .common.siteadministration.settings.timesheets import overheads as sa_ov from .core.artifacts import shot try: sa_ov.run(rt.page, cfg, run_dir) shot(rt.page, run_dir, "shell-siteadmin-timesheets-read-overheads-ok") rt.save_state() rt.touch() print("[UIZ] siteadmin-timesheets-read-overheads OK") except Exception as e: try: shot(rt.page, run_dir, "shell-siteadmin-timesheets-read-overheads-error") except Exception: pass print(f"[UIZ] siteadmin-timesheets-read-overheads ERROR: {e}") continue if cmd == "siteadmin-integration": lock = SESSION_POOL.tenant_lock(tenant_id) with lock: rt = SESSION_POOL.get( cfg=cfg, tenant_id=tenant_id, run_dir=run_dir, mfa_code=(args.mfa_code or ""), allow_mfa=_allow_mfa(), ) from .common.siteadministration import integration as sa_integration from .core.artifacts import shot try: sa_integration.run(rt.page, cfg, run_dir) shot(rt.page, run_dir, "shell-siteadmin-integration-ok") rt.save_state() rt.touch() print("[UIZ] siteadmin-integration OK") except Exception as e: try: shot(rt.page, run_dir, "shell-siteadmin-integration-error") except Exception: pass print(f"[UIZ] siteadmin-integration ERROR: {e}") continue if cmd == "users-upload-docs": lock = SESSION_POOL.tenant_lock(tenant_id) with lock: rt = SESSION_POOL.get( cfg=cfg, tenant_id=tenant_id, run_dir=run_dir, mfa_code=(args.mfa_code or ""), allow_mfa=_allow_mfa(), ) from .zones.users.flows import nav as flow_nav from .zones.users.flows import select_user as flow_user from .zones.users.flows import upload_docs as flow_docs from .zones.users.flows.upload_docs import DocSpec try: cfg_run = cfg docs_specs = list(getattr(cfg, "user_docs", []) or []) files = list(getattr(cfg, "user_files", []) or []) comments = list(getattr(cfg, "user_comments", []) or []) filters = list(getattr(cfg, "user_filters", []) or []) def _parse_shell_flags(tokens: list[str]) -> dict: out = { "bu": "", "email": "", "docs": [], "files": [], "comments": [], "filters": [], } i = 0 while i < len(tokens): t = tokens[i] if t in ("--bu",): out["bu"] = tokens[i + 1] if i + 1 < len(tokens) else "" i += 2 continue if t in ("--user-email", "--email"): out["email"] = tokens[i + 1] if i + 1 < len(tokens) else "" i += 2 continue if t == "--doc": out["docs"].append(tokens[i + 1] if i + 1 < len(tokens) else "") i += 2 continue if t == "--file": out["files"].append(tokens[i + 1] if i + 1 < len(tokens) else "") i += 2 continue if t == "--comment": out["comments"].append(tokens[i + 1] if i + 1 < len(tokens) else "") i += 2 continue if t == "--filter": out["filters"].append(tokens[i + 1] if i + 1 < len(tokens) else "") i += 2 continue i += 1 return out if rest: if any(r.startswith("--") for r in rest): parsed = _parse_shell_flags(rest) cfg_run = replace( cfg, timesheet_bu=parsed["bu"] or cfg.timesheet_bu, user_email=parsed["email"] or cfg.user_email, ) docs_specs.extend([d for d in parsed["docs"] if d]) files.extend([f for f in parsed["files"] if f]) comments.extend(parsed["comments"]) filters.extend(parsed["filters"]) else: # Positional: users-upload-docs ... bu = rest[0] if rest else "" email = rest[1] if len(rest) > 1 else "" extra_files = rest[2:] if len(rest) > 2 else [] cfg_run = replace( cfg, timesheet_bu=bu or cfg.timesheet_bu, user_email=email or cfg.user_email, ) files.extend(extra_files) # Build DocSpec list (prefer --doc) docs = [] if docs_specs: for s in docs_specs: parts = [p.strip() for p in str(s).split(";") if p.strip()] kv = {} for part in parts: if "=" not in part: raise RuntimeError(f"Invalid --doc chunk (expected key=value): {part!r}") k, v = part.split("=", 1) kv[k.strip().lower()] = v.strip() if not kv.get("file"): raise RuntimeError(f"--doc missing required 'file=': {s!r}") docs.append(DocSpec(file=kv["file"], comment=kv.get("comment", ""), filter=kv.get("filter", ""))) else: if not files: raise RuntimeError("Missing files. Use --doc ... OR --file /path/to/doc.pdf") for i, f in enumerate(files): c = comments[i] if i < len(comments) else "" flt = filters[i] if i < len(filters) else "" docs.append(DocSpec(file=f, comment=c, filter=flt)) if not (getattr(cfg_run, "user_email", "") or "").strip(): raise RuntimeError("Missing --user-email (required).") flow_nav.run(rt.page, cfg_run, run_dir) if "thisUser=1" not in (rt.page.url or "") and rt.page.locator("table#tblUsers").count() > 0: result = flow_user.select_user_by_email( rt.page, cfg_run.user_email, run_dir=run_dir, ) if not result.found: raise RuntimeError(f"User not found by email: {cfg_run.user_email}") flow_docs.users_upload_documents( rt.page, docs=docs, cfg=cfg_run, run_dir=run_dir, timeout_ms=45_000, ) shot(rt.page, run_dir, "shell-users-upload-docs-ok") rt.save_state() rt.touch() print("[UIZ] users-upload-docs OK") except Exception as e: try: shot(rt.page, run_dir, "shell-users-upload-docs-error") except Exception: pass print(f"[UIZ] users-upload-docs ERROR: {e}") continue if cmd in ("exit", "quit"): print("[UIZ][shell] bye") return 0 if cmd == "close": SESSION_POOL.close(tenant_id) print("[UIZ][shell] session closed") continue print(f"[UIZ][shell] unknown: {cmdline}") if args.cmd == "cleanup-artifacts": from .maintenance.cleanup_artifacts import cleanup_artifacts deleted = cleanup_artifacts( artifacts_dir=cfg.artifacts_dir, days=args.days, dry_run=bool(args.dry_run), ) # cleanup_artifacts retorna int (cantidad) o lista; tolerante: if isinstance(deleted, int): print(f"[UIZ] cleanup-artifacts done. deleted={deleted} dry_run={args.dry_run}") else: print(f"[UIZ] cleanup-artifacts done. items={len(deleted)} dry_run={args.dry_run}") return 0 # Los comandos runtime necesitan tenant_id si quieres operar pool. # Si no hay tenant_id, igual podemos mostrar algo útil. from .core.runtime import SESSION_POOL if args.cmd == "status": status = SESSION_POOL.status(cfg.tenant_id or "") print(status) return 0 if args.cmd == "close-session": SESSION_POOL.close(cfg.tenant_id or "") print("[UIZ] session closed") return 0 if args.cmd == "timesheets-nav": from .zones.timesheets.commands.nav import cmd_timesheets_nav return cmd_timesheets_nav(cfg, mfa_code=args.mfa_code) if args.cmd == "timesheets-create": from .zones.timesheets.commands.create import cmd_timesheets_create return cmd_timesheets_create(cfg, mfa_code=args.mfa_code) if args.cmd == "timesheets-delete": from .zones.timesheets.commands.delete import cmd_timesheets_delete return cmd_timesheets_delete(cfg, mfa_code=args.mfa_code) if args.cmd == "users-nav": from .zones.users.commands.nav import cmd_users_nav return cmd_users_nav(cfg, mfa_code=args.mfa_code) if args.cmd == "users-upload-docs": from .zones.users.commands.upload_docs import cmd_users_upload_documents return cmd_users_upload_documents(cfg, mfa_code=args.mfa_code) if args.cmd == "siteadmin-home": from .core.runtime import run_once from .common.selectors import siteadministration as siteadmin_sel def _run(page, _cfg, _run_dir): siteadmin_sel.ensure_siteadministration(page) return run_once(cfg, _run, run_dir, mfa_code=args.mfa_code, allow_mfa=allow_mfa) if args.cmd == "siteadmin-business-units": from .core.runtime import run_once from .common.siteadministration import bussiness_units as sa_bu return run_once(cfg, sa_bu.run, run_dir, mfa_code=args.mfa_code, allow_mfa=allow_mfa) if args.cmd == "siteadmin-users": from .core.runtime import run_once from .common.siteadministration import users as sa_users return run_once(cfg, sa_users.run, run_dir, mfa_code=args.mfa_code, allow_mfa=allow_mfa) if args.cmd == "siteadmin-settings": from .core.runtime import run_once from .common.siteadministration.settings import settings as sa_settings return run_once(cfg, sa_settings.run, run_dir, mfa_code=args.mfa_code, allow_mfa=allow_mfa) if args.cmd == "siteadmin-settings-general": from .core.runtime import run_once from .common.siteadministration.settings.general import general as sa_general return run_once(cfg, sa_general.run, run_dir, mfa_code=args.mfa_code, allow_mfa=allow_mfa) if args.cmd == "siteadmin-settings-timesheets": from .core.runtime import run_once from .common.siteadministration.settings.timesheets import timesheets as sa_ts return run_once(cfg, sa_ts.run, run_dir, mfa_code=args.mfa_code, allow_mfa=allow_mfa) if args.cmd == "siteadmin-timesheets-read-overheads": from .core.runtime import run_once from .common.siteadministration.settings.timesheets import overheads as sa_ov return run_once(cfg, sa_ov.run, run_dir, mfa_code=args.mfa_code, allow_mfa=allow_mfa) if args.cmd == "siteadmin-integration": from .core.runtime import run_once from .common.siteadministration import integration as sa_integration return run_once(cfg, sa_integration.run, run_dir, mfa_code=args.mfa_code, allow_mfa=allow_mfa) raise SystemExit("Unknown command") if __name__ == "__main__": raise SystemExit(main())