#!/usr/bin/env python3 """CLI for the `alis` skill. ALIS (Medtelligent) staff are READ via the API and CHANGED via a web-UI bulk import. This CLI does both halves: - read: auth-test, communities, staff, roles, role-map (reference for setup) - build: template, build-import, inspect (produce/check the import workbook) The API is read-only for staff - there is no write subcommand because no staff write endpoint exists. The end deliverable is an .xls you upload in ALIS. Examples: python alis.py auth-test python alis.py communities python alis.py staff [--status Hired] [--limit 20] [--json] python alis.py roles # live security/job role vocab python alis.py role-map [--refresh] # jobRole -> securityRole reference python alis.py template --out new_staff.xls # blank, ready-to-fill template python alis.py build-import --input hires.csv --out import.xls [--gen-passwords] python alis.py inspect import.xls """ from __future__ import annotations import argparse import json import os import subprocess import sys from pathlib import Path from alis_client import ALISClient, ALISError import import_builder as ib # --- errorlog (soft-fail, per CLAUDE.md) -------------------------------------- def _log_skill_error(skill: str, msg: str, context: str = "") -> None: try: root = os.environ.get("CLAUDETOOLS_ROOT") or os.path.abspath( os.path.join(os.path.dirname(__file__), "..", "..", "..", "..") ) h = os.path.join(root, ".claude", "scripts", "log-skill-error.sh") if not os.path.exists(h): return a = ["bash", h, skill, msg] if context: a += ["--context", context] subprocess.run(a, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, timeout=10) except Exception: pass # 404/"not found" on a probe is expected; a real auth/transport failure is not. _EXPECTED = ("http 404", "not found", "http 429", "too many requests") def _should_log(msg: str) -> bool: if os.environ.get("ALIS_SUPPRESS_ERRORLOG"): return False m = (msg or "").lower() return not any(x in m for x in _EXPECTED) def _emit(obj, as_json: bool) -> None: print(json.dumps(obj, indent=2, default=str)) def _trunc(s, n): s = "" if s is None else str(s) return s if len(s) <= n else s[: n - 1] + "…" # --- read subcommands --------------------------------------------------------- def cmd_auth_test(args) -> int: c = ALISClient() c.authenticate() comms = c.list_communities() print("[OK] authenticated to", c.api_base_url) print("[INFO] communities:", [(x.get("communityId"), x.get("communityName")) for x in comms]) return 0 def cmd_communities(args) -> int: c = ALISClient() comms = c.list_communities() if args.json: _emit(comms, True) else: for x in comms: print(f" {x.get('communityId'):>6} {x.get('communityName')}") return 0 def cmd_staff(args) -> int: c = ALISClient() staff = c.list_staff(community_id=args.community, status=args.status) if args.json: _emit(staff[: args.limit] if args.limit else staff, True) return 0 print(f"[INFO] {len(staff)} staff" + (f" (status={args.status})" if args.status else "")) shown = staff[: args.limit] if args.limit else staff for s in shown: name = f"{s.get('firstName','')} {s.get('lastName','')}".strip() sr = s.get("securityRoles") or [] sr = ", ".join(sr) if isinstance(sr, list) else str(sr) print(f" {s.get('staffId'):>7} {_trunc(name,26):26} " f"{_trunc(s.get('status',''),11):11} " f"{_trunc(s.get('jobRole') or '-',26):26} [{_trunc(sr,40)}]") if args.limit and len(staff) > args.limit: print(f" ... {len(staff)-args.limit} more (raise --limit or use --json)") return 0 def cmd_roles(args) -> int: c = ALISClient() rm = c.build_role_map(community_id=args.community) if args.json: _emit(rm, True) return 0 print(f"[INFO] from {rm['sourceStaffCount']} staff " f"({rm['hiredCount']} Hired), community {rm['community']['id']}") print("\n=== Security Roles in use ===") for r in rm["securityRoleVocabulary"]: print(f" {r}") print("\n=== Job Roles in use ===") for r in rm["jobRoleVocabulary"]: print(f" {r}") return 0 def cmd_role_map(args) -> int: if args.refresh: c = ALISClient() rm = c.build_role_map(community_id=args.community) path = Path(__file__).resolve().parent.parent / "references" / "role-map.json" path.write_text(json.dumps(rm, indent=2), encoding="utf-8") print(f"[OK] refreshed role-map.json from live ({rm['sourceStaffCount']} staff)") else: rm = ib.load_role_map() if not rm: print("[WARNING] no cached role-map.json; run with --refresh", file=sys.stderr) return 1 combo = rm.get("jobRoleToSecurityRolesCombo") or { k: ", ".join(v) for k, v in rm.get("jobRoleToSecurityRoles", {}).items()} if args.json: _emit(combo, True) return 0 print("=== Job Role -> typical Security Roles (learned from current staff) ===") for jr, sr in sorted(combo.items()): print(f" {_trunc(jr,34):34} -> {sr}") return 0 # --- build subcommands -------------------------------------------------------- def cmd_template(args) -> int: out = ib.write_workbook([], args.out) print(f"[OK] blank ALIS staff-import template -> {out}") print("[INFO] columns:", " | ".join(ib.TEMPLATE_HEADERS)) return 0 def cmd_build_import(args) -> int: rows_in = ib.read_input_rows(args.input) if not rows_in: print("[ERROR] no rows read from input", file=sys.stderr) return 1 role_map = ib.load_role_map() if args.refresh_roles: try: role_map = ALISClient().build_role_map(community_id=args.community) except ALISError as exc: print(f"[WARNING] live role refresh failed ({exc}); using cached map", file=sys.stderr) rows, report = ib.enrich_and_validate( rows_in, role_map, default_status=args.default_status, suggest_roles=not args.no_suggest) sidecar = None if args.gen_passwords: if args.format == "update": print("[WARNING] --gen-passwords ignored: the update format has no " "Password column. Use --format create for new logins.", file=sys.stderr) else: generated = ib.fill_passwords(rows) sidecar = ib.write_password_sidecar(generated, args.out) out = ib.write_workbook(rows, args.out, fmt=args.format) # Report (never prints passwords) warn_rows = [r for r in report if r["warnings"]] print(f"[OK] wrote {len(rows)} staff -> {out} (format={args.format})") print("[INFO] columns:", " | ".join(ib.HEADERS_BY_FORMAT[args.format])) for r in report: for n in r["notes"]: print(f" [INFO] row {r['row']} ({r['name']}): {n}") for r in warn_rows: for w in r["warnings"]: print(f" [WARNING] row {r['row']} ({r['name']}): {w}") if sidecar: print(f"[CRITICAL] generated passwords written PLAINTEXT to {sidecar}") print(" Distribute to staff, then DELETE or vault it. Never commit it.") print(f"\n[NEXT] Upload the .xls in ALIS: Staff -> Import. Dates use " f"{ib.DATE_FORMAT_HINT}. Rows without an ALIS ID are CREATED; the update " "format (with ALIS ID) edits existing staff. Test with ONE row first.") return 0 def cmd_inspect(args) -> int: data = ib.read_workbook(args.path) if args.json: _emit(data, True) return 0 print(f"[INFO] format: {data['format']}") for name, rows in data["sheets"].items(): print(f"=== {name} ({len(rows)} rows shown) ===") for i, row in enumerate(rows): trimmed = list(row) while trimmed and trimmed[-1] in ("", None): trimmed.pop() print(f" row{i}: {trimmed}") return 0 def build_parser() -> argparse.ArgumentParser: p = argparse.ArgumentParser(prog="alis", description="ALIS staff read + import-builder") p.add_argument("--json", action="store_true", help="emit raw JSON") p.add_argument("--community", type=int, default=None, help="communityId (default: vault/env, Cascades=622)") sub = p.add_subparsers(dest="cmd", required=True) sub.add_parser("auth-test", help="mint a token and list communities") sub.add_parser("communities", help="list communities in scope") sp = sub.add_parser("staff", help="staff roster") sp.add_argument("--status", help="filter: Applicant|Hired|Discharged|Rejected") sp.add_argument("--limit", type=int, default=25) sub.add_parser("roles", help="security + job role vocabulary (live)") sp = sub.add_parser("role-map", help="job-role -> security-role reference") sp.add_argument("--refresh", action="store_true", help="rebuild from live + cache") sp = sub.add_parser("template", help="write a blank import template .xls") sp.add_argument("--out", required=True) sp = sub.add_parser("build-import", help="build an import .xls from a CSV/JSON") sp.add_argument("--input", required=True, help="CSV/JSON of new staff") sp.add_argument("--out", required=True, help="output .xls path") sp.add_argument("--format", choices=["create", "update"], default="create", help="create = new staff (has Password); update = edit " "existing (leads with ALIS ID)") sp.add_argument("--default-status", default="Hired") sp.add_argument("--no-suggest", action="store_true", help="do not infer Security Roles from Job Role") sp.add_argument("--refresh-roles", action="store_true", help="pull live role-map instead of cached") sp.add_argument("--gen-passwords", action="store_true", help="generate passwords for Login-enabled rows lacking one " "(writes a plaintext sidecar CSV)") sp = sub.add_parser("inspect", help="dump an existing import workbook") sp.add_argument("path") return p _DISPATCH = { "auth-test": cmd_auth_test, "communities": cmd_communities, "staff": cmd_staff, "roles": cmd_roles, "role-map": cmd_role_map, "template": cmd_template, "build-import": cmd_build_import, "inspect": cmd_inspect, } def main(argv=None) -> int: args = build_parser().parse_args(argv) try: return _DISPATCH[args.cmd](args) except (ALISError, ib.ImportBuilderError) as exc: print(f"[ERROR] {exc}", file=sys.stderr) if _should_log(str(exc)): _log_skill_error("alis", str(exc), context=f"cmd={args.cmd}") return 1 if __name__ == "__main__": raise SystemExit(main())