azcomputerguru/claudetools
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

sync: auto-sync from HOWARD-HOME at 2026-06-29 16:55:22

Browse Source 
Author: Howard Enos
Machine: HOWARD-HOME
Timestamp: 2026-06-29 16:55:22
This commit is contained in:
Howard Enos
2026-06-29 16:55:55 -07:00
parent e692bff2bc
commit 31f2bdb84f
9 changed files with 1271 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

289
.claude/skills/alis/scripts/alis.py Normal file
View File

@@ -0,0 +1,289 @@
#!/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())

307
.claude/skills/alis/scripts/alis_client.py Normal file
View File

@@ -0,0 +1,307 @@
#!/usr/bin/env python3
"""ALIS (Medtelligent assisted-living EHR) REST API client for the `alis` skill.
ALIS is a partner/App-Store integration API at https://api.alisonline.com. A
tenant (e.g. Cascades of Tucson) lives at <tenant>.alisonline.com but ALL API
traffic goes to the shared api.alisonline.com host, scoped by the logged-in
user's company + a communityId.
Auth (verified live 2026-06-29):
POST /user/tokens {"username":"<user>@<tenantKey>","password":"..."}
-> {accessToken (JWT, ~1h), expiresIn, refreshToken}
Send Authorization: Bearer <accessToken> on every call.
Refresh: POST /user/tokens/refresh {accessToken, refreshToken}.
CRITICAL: the username MUST be tenant-qualified (e.g. howard.enos@cascadestucson)
or /user/tokens returns 400. The bare login name alone fails.
Global API security is OR(Bearer | BasicAuth | VendorKey) - a user JWT alone
authorizes the integration reads we use.
Scope reality: this API is READ-ONLY for staff (only GET endpoints exist - no
create/update/delete). Staff are CHANGED via the ALIS web-UI bulk import (see
import_builder.py). This client's job is to READ current staff so new staff can
be set up the same way (the job-role -> security-role reference map).
Credentials load from the SOPS vault; env overrides exist for testing.
Transport: prefers httpx if installed, else stdlib urllib (no hard dependency).
"""
from __future__ import annotations
import json
import os
import subprocess
import sys
import urllib.error
import urllib.parse
import urllib.request
from collections import Counter, defaultdict
from pathlib import Path
from typing import Any, Optional
try:
import httpx # type: ignore
_HAS_HTTPX = True
except ImportError: # pragma: no cover - depends on environment
_HAS_HTTPX = False
ERROR_BODY_MAX_CHARS = 500
ALIS_BASE_URL = os.environ.get("ALIS_BASE_URL", "https://api.alisonline.com")
TIMEOUT_SECONDS = 60.0
CONNECT_TIMEOUT_SECONDS = 10.0
# Cascades of Tucson is the only tenant this credential sees today.
DEFAULT_COMMUNITY_ID = int(os.environ.get("ALIS_COMMUNITY_ID", "622"))
VAULT_ENTRY = "clients/cascades-tucson/alis-api-howard-user.sops.yaml"
VAULT_USERNAME_FIELD = "credentials.username"
VAULT_PASSWORD_FIELD = "credentials.password"
SKILL_DIR = Path(__file__).resolve().parent.parent
class ALISError(RuntimeError):
"""Raised for transport or API errors."""
# --- credential loading -------------------------------------------------------
def _resolve_claudetools_root() -> Path:
derived_root = SKILL_DIR.parent.parent.parent # skills/alis -> repo root
env_root = os.environ.get("CLAUDETOOLS_ROOT")
if env_root:
return Path(env_root)
identity_path = derived_root / ".claude" / "identity.json"
if identity_path.exists():
try:
data = json.loads(identity_path.read_text(encoding="utf-8"))
root = data.get("claudetools_root")
if root:
return Path(root)
except (json.JSONDecodeError, OSError):
pass
return derived_root
def _vault_get(field: str) -> str:
root = _resolve_claudetools_root()
vault_script = root / ".claude" / "scripts" / "vault.sh"
if not vault_script.exists():
raise ALISError(f"vault wrapper not found at {vault_script}")
try:
completed = subprocess.run(
["bash", str(vault_script), "get-field", VAULT_ENTRY, field],
capture_output=True, text=True, timeout=60,
)
except FileNotFoundError as exc:
raise ALISError("'bash' not found on PATH; install Git Bash.") from exc
except subprocess.TimeoutExpired as exc:
raise ALISError("vault call timed out.") from exc
if completed.returncode != 0:
raise ALISError(
f"vault read failed for {field} (exit {completed.returncode}): "
f"{completed.stderr.strip()}"
)
val = completed.stdout.strip()
if not val:
raise ALISError(f"vault returned empty value for {field}.")
return val
def load_credentials() -> tuple[str, str]:
"""Return (username, password). Env overrides ALIS_USERNAME/ALIS_PASSWORD,
else the SOPS vault. Username must already be tenant-qualified."""
user = os.environ.get("ALIS_USERNAME")
pw = os.environ.get("ALIS_PASSWORD")
if not user:
user = _vault_get(VAULT_USERNAME_FIELD)
if not pw:
pw = _vault_get(VAULT_PASSWORD_FIELD)
return user, pw
# --- client -------------------------------------------------------------------
class ALISClient:
def __init__(
self,
username: Optional[str] = None,
password: Optional[str] = None,
api_base_url: str = ALIS_BASE_URL,
community_id: int = DEFAULT_COMMUNITY_ID,
timeout: float = TIMEOUT_SECONDS,
connect_timeout: float = CONNECT_TIMEOUT_SECONDS,
):
self.api_base_url = api_base_url.rstrip("/")
self.community_id = community_id
self._username = username
self._password = password
self._access_token: Optional[str] = None
self._refresh_token: Optional[str] = None
self.timeout = timeout
self.connect_timeout = connect_timeout
# -- auth ------------------------------------------------------------------
def _ensure_credentials(self) -> None:
if not self._username or not self._password:
self._username, self._password = load_credentials()
@property
def access_token(self) -> str:
if not self._access_token:
self.authenticate()
return self._access_token # type: ignore[return-value]
def authenticate(self) -> None:
"""Mint a fresh JWT via POST /user/tokens."""
self._ensure_credentials()
body = {"username": self._username, "password": self._password}
url = f"{self.api_base_url}/user/tokens"
data = json.dumps(body).encode("utf-8")
result = self._send("POST", url, data, with_auth=False)
if not isinstance(result, dict) or "accessToken" not in result:
raise ALISError(f"Unexpected token response: {str(result)[:200]}")
self._access_token = result["accessToken"]
self._refresh_token = result.get("refreshToken")
# -- core transport --------------------------------------------------------
def _request(self, method: str, path: str,
params: Optional[dict] = None,
body: Optional[dict] = None) -> Any:
url = f"{self.api_base_url}/{path.lstrip('/')}"
if params:
url = f"{url}?{urllib.parse.urlencode(params)}"
data = json.dumps(body).encode("utf-8") if body is not None else None
return self._send(method, url, data, with_auth=True)
def _send(self, method: str, url: str, data: Optional[bytes],
with_auth: bool) -> Any:
headers = {"Content-Type": "application/json", "Accept": "application/json"}
if with_auth:
headers["Authorization"] = f"Bearer {self.access_token}"
if _HAS_HTTPX:
try:
timeout = httpx.Timeout(self.timeout, connect=self.connect_timeout)
with httpx.Client(timeout=timeout) as client:
resp = client.request(method, url, content=data, headers=headers)
resp.raise_for_status()
return resp.json() if resp.content else None
except httpx.TimeoutException as exc:
raise ALISError(f"ALIS request timed out: {exc}") from exc
except httpx.HTTPStatusError as exc:
detail = (exc.response.text or "")[:ERROR_BODY_MAX_CHARS]
raise ALISError(
f"ALIS HTTP {exc.response.status_code} [{method} {url}]: {detail}"
) from exc
except httpx.HTTPError as exc:
raise ALISError(f"ALIS request failed: {exc}") from exc
req = urllib.request.Request(url, data=data, method=method, headers=headers)
try:
with urllib.request.urlopen(req, timeout=self.timeout) as resp:
raw = resp.read()
return json.loads(raw.decode("utf-8")) if raw else None
except urllib.error.HTTPError as exc:
detail = exc.read().decode("utf-8", errors="replace")[:ERROR_BODY_MAX_CHARS]
raise ALISError(f"ALIS HTTP {exc.code} [{method} {url}]: {detail}") from exc
except urllib.error.URLError as exc:
raise ALISError(f"ALIS request failed: {exc}") from exc
# ======================================================================
# READ METHODS (this API is read-only for staff)
# ======================================================================
def list_communities(self) -> list[dict]:
"""Communities the logged-in user can see. VERIFIED LIVE."""
return self._request("GET", "/v1/integration/communities") or []
def list_staff(self, community_id: Optional[int] = None,
status: Optional[str] = None,
include_associated: bool = False) -> list[dict]:
"""Staff roster for a community. VERIFIED LIVE.
communityId is REQUIRED - omitting it 403s 'Not authorized for facility 0'.
Each record: staffId, firstName, lastName, staffRecordNumber, primaryEmail,
mobilePhoneNumber, dateOfBirth, status, hireDate, dischargeDate, jobRole,
securityRoles (list)."""
params: dict = {"communityId": community_id or self.community_id}
if status:
params["status"] = status
if include_associated:
params["includeAssociatedStaff"] = "true"
res = self._request("GET", "/v1/integration/staff", params=params)
return res if isinstance(res, list) else (res or {}).get("data", []) or []
def get_staff(self, staff_id: int) -> dict:
"""One staff member's full record. VERIFIED LIVE."""
return self._request("GET", f"/v1/integration/staff/{staff_id}") or {}
def get_staff_basic_info(self, staff_id: int) -> dict:
"""Staff basicInfo: address, license, jobRole, securityRoles, etc."""
return self._request(
"GET", f"/v1/integration/staff/{staff_id}/basicInfo") or {}
# -- derived reference model -----------------------------------------------
def build_role_map(self, community_id: Optional[int] = None) -> dict:
"""Derive the setup-reference from LIVE staff: the security-role and
job-role vocabularies actually in use, and a job-role -> security-role(s)
map learned from HIRED staff. This is how a new hire of a given job role
should be configured to match existing staff."""
staff = self.list_staff(community_id=community_id)
hired = [s for s in staff if s.get("status") == "Hired"]
sr_all: Counter = Counter()
jr_all: Counter = Counter()
jr_to_sr: dict[str, Counter] = defaultdict(Counter)
for s in staff:
sr = s.get("securityRoles") or []
if isinstance(sr, str):
sr = [sr]
for x in sr:
sr_all[x] += 1
jr = (s.get("jobRole") or "").strip()
if jr:
jr_all[jr] += 1
for s in hired:
jr = (s.get("jobRole") or "").strip()
if not jr:
continue
sr = s.get("securityRoles") or []
if isinstance(sr, str):
sr = [sr]
for x in sr:
jr_to_sr[jr][x] += 1
return {
"community": {"id": community_id or self.community_id},
"sourceStaffCount": len(staff),
"hiredCount": len(hired),
"securityRoleVocabulary": sorted(sr_all),
"jobRoleVocabulary": sorted(jr_all),
"jobRoleToSecurityRoles": {
jr: [r for r, _ in c.most_common()] for jr, c in jr_to_sr.items()
},
}
# ======================================================================
# POWER TOOL
# ======================================================================
def raw(self, method: str, path: str,
params: Optional[dict] = None, body: Optional[dict] = None) -> Any:
"""Call any endpoint directly (read-only API - no staff writes exist)."""
return self._request(method.upper(), path, params=params, body=body)
def main() -> int:
"""Self-check: mint a token (live) and confirm community scope."""
try:
client = ALISClient()
client.authenticate()
print("[OK] authenticated; transport =",
"httpx" if _HAS_HTTPX else "urllib")
comms = client.list_communities()
print("[INFO] base =", client.api_base_url)
print("[INFO] communities:",
[(c.get("communityId"), c.get("communityName")) for c in comms])
return 0
except ALISError as exc:
print(f"[ERROR] {exc}", file=sys.stderr)
return 1
if __name__ == "__main__":
raise SystemExit(main())

350
.claude/skills/alis/scripts/import_builder.py Normal file
View File

@@ -0,0 +1,350 @@
#!/usr/bin/env python3
"""Build / inspect the ALIS Staff Import workbook for the `alis` skill.
ALIS has NO staff-write API. Staff (and their logins) are created/changed in
bulk by uploading an .xls on the ALIS web UI (Staff -> Import). This module
produces a workbook that exactly matches Medtelligent's template so it uploads
cleanly, and validates each row against the template's controlled values and
against the live job-role -> security-role reference (role-map.json).
Template (Sheet1 header, exact order - DO NOT reorder/rename):
First Name | Last Name | Staff Record Number | Security Roles | Staff Status |
Hire Date | Login Enabled | Email | Password | Date of Birth | Gender |
Job Role | Cell Phone
Sheet2 holds the dropdown lists:
Staff Status: Applicant / Discharged / Hired / Rejected
Login Enabled: Yes / No
Gender: Female / Male
Write uses xlwt (legacy .xls, matching the template format). Read uses xlrd.
"""
from __future__ import annotations
import csv
import json
import secrets
import string
from pathlib import Path
from typing import Any, Optional
# ALIS has TWO import layouts (column order matters for the importer):
# - CREATE (blank "new staff" template): has Password, NO ALIS ID. Rows WITHOUT
# an ALIS ID are created as new staff.
# - UPDATE (export of current staff): leads with ALIS ID, NO Password. Rows WITH
# an ALIS ID update the matching existing staff member.
# We build CREATE files for new hires by default.
TEMPLATE_HEADERS = [
"First Name", "Last Name", "Staff Record Number", "Security Roles",
"Staff Status", "Hire Date", "Login Enabled", "Email", "Password",
"Date of Birth", "Gender", "Job Role", "Cell Phone",
]
UPDATE_TEMPLATE_HEADERS = [
"ALIS ID", "First Name", "Last Name", "Staff Record Number", "Security Roles",
"Staff Status", "Hire Date", "Login Enabled", "Email",
"Date of Birth", "Gender", "Job Role", "Cell Phone",
]
HEADERS_BY_FORMAT = {"create": TEMPLATE_HEADERS, "update": UPDATE_TEMPLATE_HEADERS}
# ALIS dates render as MM/DD/YYYY (confirmed from the real staff export).
DATE_FORMAT_HINT = "MM/DD/YYYY"
# Sheet2 dropdown lists (mirror the official template).
STATUS_VALUES = ["Applicant", "Discharged", "Hired", "Rejected"]
LOGIN_VALUES = ["Yes", "No"]
GENDER_VALUES = ["Female", "Male"]
# Accepted input aliases -> canonical header. Lower-cased, stripped, non-alnum
# removed for matching, so "first_name", "FirstName", "First Name" all map.
_ALIASES = {
"alisid": "ALIS ID", "alis": "ALIS ID", "staffaliasid": "ALIS ID",
"firstname": "First Name", "first": "First Name", "fname": "First Name",
"lastname": "Last Name", "last": "Last Name", "lname": "Last Name",
"staffrecordnumber": "Staff Record Number", "recordnumber": "Staff Record Number",
"recordno": "Staff Record Number", "employeeid": "Staff Record Number",
"empid": "Staff Record Number", "staffid": "Staff Record Number",
"securityroles": "Security Roles", "securityrole": "Security Roles",
"role": "Security Roles", "accessrole": "Security Roles",
"staffstatus": "Staff Status", "status": "Staff Status",
"hiredate": "Hire Date", "datehired": "Hire Date", "startdate": "Hire Date",
"loginenabled": "Login Enabled", "login": "Login Enabled",
"loginaccess": "Login Enabled", "portalaccess": "Login Enabled",
"email": "Email", "emailaddress": "Email", "workemail": "Email",
"password": "Password", "pass": "Password", "pwd": "Password",
"dateofbirth": "Date of Birth", "dob": "Date of Birth", "birthdate": "Date of Birth",
"gender": "Gender", "sex": "Gender",
"jobrole": "Job Role", "jobtitle": "Job Role", "title": "Job Role",
"position": "Job Role",
"cellphone": "Cell Phone", "cell": "Cell Phone", "mobile": "Cell Phone",
"phone": "Cell Phone", "mobilephone": "Cell Phone", "cellphonenumber": "Cell Phone",
}
class ImportBuilderError(RuntimeError):
pass
def _norm(s: str) -> str:
return "".join(ch for ch in str(s).lower() if ch.isalnum())
def _canon_header(raw: str) -> Optional[str]:
raw_s = str(raw).strip()
for h in TEMPLATE_HEADERS:
if _norm(h) == _norm(raw_s):
return h
return _ALIASES.get(_norm(raw_s))
def load_role_map(path: Optional[Path] = None) -> dict:
"""Load the cached job-role -> security-role reference (role-map.json)."""
if path is None:
path = Path(__file__).resolve().parent.parent / "references" / "role-map.json"
if not path.exists():
return {}
try:
return json.loads(path.read_text(encoding="utf-8"))
except (json.JSONDecodeError, OSError):
return {}
# --- input parsing ------------------------------------------------------------
def read_input_rows(path: str) -> list[dict]:
"""Read a CSV or JSON file into a list of {canonical-header: value} dicts.
Unknown columns are ignored (with the raw key preserved under '_extra')."""
p = Path(path)
if not p.exists():
raise ImportBuilderError(f"input file not found: {path}")
text = p.read_text(encoding="utf-8-sig")
raw_rows: list[dict]
if p.suffix.lower() == ".json":
data = json.loads(text)
raw_rows = data if isinstance(data, list) else data.get("staff") or data.get("rows") or []
else: # CSV/TSV
delim = "\t" if p.suffix.lower() in (".tsv", ".tab") else ","
raw_rows = list(csv.DictReader(text.splitlines(), delimiter=delim))
rows: list[dict] = []
for raw in raw_rows:
row: dict = {}
extra: dict = {}
for k, v in raw.items():
if k is None:
continue
canon = _canon_header(k)
if canon:
row[canon] = "" if v is None else str(v).strip()
else:
extra[k] = v
if extra:
row["_extra"] = extra
rows.append(row)
return rows
# --- validation + enrichment --------------------------------------------------
def _suggest_security_role(job_role: str, role_map: dict) -> Optional[str]:
"""Return the typical FULL Security Roles string (comma-separated, as ALIS
stores it) for a job role, learned from current staff. Prefers the modal
exact combo (jobRoleToSecurityRolesCombo); falls back to the older
single-role map for compatibility."""
if not job_role:
return None
combo = role_map.get("jobRoleToSecurityRolesCombo", {})
for k, v in combo.items():
if k.lower() == job_role.lower() and v:
return v
# legacy fallback: list of roles -> join
mapping = role_map.get("jobRoleToSecurityRoles", {})
for k, v in mapping.items():
if k.lower() == job_role.lower() and v:
return ", ".join(v) if isinstance(v, list) else str(v)
return None
def enrich_and_validate(rows: list[dict], role_map: dict,
default_status: str = "Hired",
suggest_roles: bool = True) -> tuple[list[dict], list[dict]]:
"""Apply defaults, infer Security Roles from Job Role, validate enums.
Returns (processed_rows, report) where report has per-row notes/warnings."""
sec_vocab = set(role_map.get("securityRoleVocabulary", []))
report: list[dict] = []
out: list[dict] = []
for i, r in enumerate(rows):
notes: list[str] = []
warns: list[str] = []
# Keep every known header (union of both layouts) so ALIS ID survives for
# update rows; write_workbook selects the columns per chosen format.
all_headers = ["ALIS ID"] + TEMPLATE_HEADERS
row = {h: str(r.get(h, "") or "").strip() for h in all_headers}
if not row["First Name"] or not row["Last Name"]:
warns.append("missing First/Last Name (required)")
# Status
if not row["Staff Status"]:
row["Staff Status"] = default_status
notes.append(f"Staff Status defaulted to {default_status}")
elif row["Staff Status"] not in STATUS_VALUES:
warns.append(f"Staff Status '{row['Staff Status']}' not in {STATUS_VALUES}")
# Gender
if row["Gender"]:
g = row["Gender"].strip().capitalize()
if g in GENDER_VALUES:
row["Gender"] = g
else:
warns.append(f"Gender '{row['Gender']}' not in {GENDER_VALUES}")
# Security Roles - infer from Job Role if blank
if not row["Security Roles"] and suggest_roles:
sug = _suggest_security_role(row["Job Role"], role_map)
if sug:
row["Security Roles"] = sug
notes.append(f"Security Roles inferred '{sug}' from Job Role "
f"'{row['Job Role']}' (review)")
elif row["Job Role"]:
warns.append(f"no reference Security Role for Job Role "
f"'{row['Job Role']}' - set manually")
elif row["Security Roles"] and sec_vocab:
unknown = [p.strip() for p in row["Security Roles"].split(",")
if p.strip() and p.strip() not in sec_vocab]
if unknown:
warns.append(f"Security Role(s) {unknown} not seen in current staff "
"- confirm they're real ALIS roles")
# Login Enabled - default from presence of email+password
if not row["Login Enabled"]:
if row["Email"] and row["Password"]:
row["Login Enabled"] = "Yes"
notes.append("Login Enabled defaulted to Yes (email+password present)")
else:
row["Login Enabled"] = "No"
notes.append("Login Enabled defaulted to No (no email/password)")
elif row["Login Enabled"] not in LOGIN_VALUES:
warns.append(f"Login Enabled '{row['Login Enabled']}' not in {LOGIN_VALUES}")
if row["Login Enabled"] == "Yes" and not row["Email"]:
warns.append("Login Enabled=Yes but no Email (login needs an email)")
out.append(row)
report.append({
"row": i + 1,
"name": f"{row['First Name']} {row['Last Name']}".strip(),
"notes": notes, "warnings": warns,
})
return out, report
def generate_password(length: int = 14) -> str:
"""Strong random password (letters+digits+symbols), avoids ambiguous chars."""
alphabet = (string.ascii_uppercase.replace("O", "").replace("I", "")
+ string.ascii_lowercase.replace("l", "")
+ string.digits.replace("0", "").replace("1", "")
+ "!@#$%*?")
while True:
pw = "".join(secrets.choice(alphabet) for _ in range(length))
if (any(c.islower() for c in pw) and any(c.isupper() for c in pw)
and any(c.isdigit() for c in pw) and any(c in "!@#$%*?" for c in pw)):
return pw
def fill_passwords(rows: list[dict]) -> list[dict]:
"""For Login Enabled=Yes rows missing a Password, generate one in place.
Returns the list of {name, email, password} that were generated."""
generated = []
for row in rows:
if row.get("Login Enabled") == "Yes" and not row.get("Password"):
pw = generate_password()
row["Password"] = pw
generated.append({
"name": f"{row['First Name']} {row['Last Name']}".strip(),
"email": row.get("Email", ""), "password": pw,
})
return generated
# --- workbook writing ---------------------------------------------------------
def write_workbook(rows: list[dict], out_path: str, fmt: str = "create") -> str:
"""Write the rows to an .xls matching the ALIS template (Sheet1 data +
Sheet2 dropdown lists). fmt='create' (new staff, has Password) or 'update'
(existing staff, leads with ALIS ID). Returns the output path."""
headers = HEADERS_BY_FORMAT.get(fmt)
if headers is None:
raise ImportBuilderError(f"unknown format '{fmt}' (use create|update)")
try:
import xlwt # type: ignore
except ImportError as exc:
raise ImportBuilderError(
"xlwt is required to write .xls. Install with: pip install xlwt"
) from exc
wb = xlwt.Workbook(encoding="utf-8")
header_style = xlwt.easyxf("font: bold on; align: wrap on")
s1 = wb.add_sheet("Sheet1")
for c, h in enumerate(headers):
s1.write(0, c, h, header_style)
for ri, row in enumerate(rows, start=1):
for ci, h in enumerate(headers):
s1.write(ri, ci, row.get(h, ""))
# Sheet2 = validation lists, laid out exactly like the official template:
# col0 Status, col1 Login Enabled, col2 Gender.
s2 = wb.add_sheet("Sheet2")
lists = [STATUS_VALUES, LOGIN_VALUES, GENDER_VALUES]
for col, values in enumerate(lists):
for r, v in enumerate(values):
s2.write(r, col, v)
out = Path(out_path)
out.parent.mkdir(parents=True, exist_ok=True)
wb.save(str(out))
return str(out)
def write_password_sidecar(generated: list[dict], xls_path: str) -> Optional[str]:
"""Write generated logins to a sidecar CSV next to the workbook. Contains
PLAINTEXT passwords - caller must warn + tell the user to distribute then
delete/vault it. Returns the sidecar path, or None if nothing generated."""
if not generated:
return None
side = Path(xls_path).with_name(Path(xls_path).stem + "_passwords.csv")
with side.open("w", newline="", encoding="utf-8") as f:
w = csv.writer(f)
w.writerow(["Name", "Email", "Password"])
for g in generated:
w.writerow([g["name"], g["email"], g["password"]])
return str(side)
# --- workbook reading (inspect) -----------------------------------------------
def read_workbook(path: str, max_rows: int = 50) -> dict:
"""Read an existing .xls (or .xlsx) staff-import workbook for inspection."""
p = Path(path)
if not p.exists():
raise ImportBuilderError(f"workbook not found: {path}")
if p.suffix.lower() == ".xlsx":
try:
import openpyxl # type: ignore
except ImportError as exc:
raise ImportBuilderError("openpyxl required for .xlsx") from exc
wb = openpyxl.load_workbook(str(p), read_only=True, data_only=True)
sheets = {}
for sh in wb.worksheets:
rows = []
for ri, row in enumerate(sh.iter_rows(values_only=True)):
if ri >= max_rows:
break
rows.append([("" if c is None else c) for c in row])
sheets[sh.title] = rows
return {"format": "xlsx", "sheets": sheets}
try:
import xlrd # type: ignore
except ImportError as exc:
raise ImportBuilderError("xlrd required for .xls") from exc
wb = xlrd.open_workbook(str(p))
sheets = {}
for sh in wb.sheets():
rows = []
for r in range(min(sh.nrows, max_rows)):
rows.append([sh.cell_value(r, c) for c in range(sh.ncols)])
sheets[sh.name] = rows
return {"format": "xls", "sheets": sheets}