""" Load UserListWithReportingDept.xlsx into routing maps. Tab 1 ("user_export_ATIME"): one row per user/extension. - "extension" -> the extension number - "department" (Col I) -> AUTHORITATIVE: mapped to a Reporting Division via config.DEPT_CROSSWALK (the matrix) - "Reporting Department" (Col J) -> legacy fallback, used only when the department is blank or not in the matrix Tab 2 ("Groups and Members"): queues / auto-attendants / conference bridges. Real header is on row 13; data starts row 14. - Col B (idx 1) "Reporting Division" -> division for the queue - Col C (idx 2) "Queues/AA/Bridge" -> the queue / AA / bridge extension(s) - Col D (idx 3) "Ext/Members" -> member (agent) extensions [not used for routing] Produces: queue_map : ext(str) -> canonical division (route_type = "queue") direct_map : ext(str) -> canonical division or "Admin" (route_type = "direct") """ import re import pandas as pd import config _EXT_RE = re.compile(r"\b(\d{2,5})\b") def _normalize_division(raw): """Map a Tab-2 division label to a canonical helpline division name.""" if raw is None: return None s = str(raw).strip() if not s or s.lower() == "nan": return None return config.DIVISION_NORMALIZE.get(s, s) def _crosswalk_dept(raw): """Map a raw department string to a division (or Admin), else None.""" if raw is None: return None s = str(raw).strip() if not s or s.lower() == "nan": return None return config.DEPT_CROSSWALK.get(config.normalize_dept(s)) def _label_from_desc(cell): """Turn a Tab-2 description ('Heller-256,9256', 'Mansour -212 (Sephardic)') into a label.""" if cell is None: return "" s = str(cell) if not s or s.lower() == "nan": return "" # take the leading non-digit text before the first extension number / dash s = re.split(r"[\d]", s, 1)[0] s = s.strip(" -–/,()").strip() # drop generic words that aren't names if s.lower() in {"", "extensions", "ext", "queue", "member", "members"}: return "" return s def _extract_exts(cell): """Pull all extension-like integers out of a free-text cell.""" if cell is None: return [] s = str(cell) if not s or s.lower() == "nan": return [] return _EXT_RE.findall(s) def load_maps(xlsx_path=None): """Return (queue_map, direct_map, meta) built from the mapping workbook.""" xlsx_path = xlsx_path or config.MAPPING_XLSX # ---- Tab 1: direct extension -> division -------------------------------- t1 = pd.read_excel(xlsx_path, sheet_name="user_export_ATIME", header=0, engine="openpyxl", dtype=str) direct_map = {} direct_rd_fallback = 0 user_names = {} # ext -> "First Last" user_dept = {} # ext -> raw department for _, r in t1.iterrows(): ext = (r.get("extension") or "").strip() if not ext: continue fn = (r.get("first name") or "").strip() ln = (r.get("last name") or "").strip() user_names[ext] = (fn + " " + ln).strip() dept_raw = r.get("department") user_dept[ext] = ("" if pd.isna(dept_raw) else str(dept_raw).strip()) # Department is authoritative: department -> Reporting Division matrix. div = _crosswalk_dept(r.get("department")) if div is None: # Legacy fallback: the hand-maintained "Reporting Department" column. div = _normalize_division(r.get("Reporting Department")) if div is not None: direct_rd_fallback += 1 if div is not None: direct_map[ext] = div # ---- Tab 2: queue / AA / bridge extension -> division ------------------- t2 = pd.read_excel(xlsx_path, sheet_name="Groups and Members", header=None, engine="openpyxl", dtype=str) queue_map = {} queue_conflicts = [] desc_labels = {} # ext -> friendly label parsed from the row description (Col A) for i in range(14, len(t2)): # data starts at row index 14 row = t2.iloc[i] div = _normalize_division(row[1]) # Col B if div is None: continue col_c = _extract_exts(row[2]) # Col C = Queues/AA/Bridge col_d = _extract_exts(row[3]) # Col D = Ext/Members (agents) label = _label_from_desc(row[0]) # Col A description -> "Heller", "Mansour", ... for ext in col_c: if ext in queue_map and queue_map[ext] != div: queue_conflicts.append((ext, queue_map[ext], div)) else: queue_map[ext] = div # Prefer the full Tab-1 name of the single underlying member, else the desc label, # so a hunt pilot (9256) reads the same as its agent ext (256 = "Leah Heller"). named_members = [e for e in col_d if user_names.get(e)] row_name = user_names[named_members[0]] if len(named_members) == 1 else label if row_name: for ext in col_c + col_d: desc_labels.setdefault(ext, row_name) # Combined display name per extension: real Tab-1 person wins; else Tab-2 label. ext_names = dict(desc_labels) for ext, nm in user_names.items(): if nm: ext_names[ext] = nm meta = { "n_direct": len(direct_map), "n_queue": len(queue_map), "direct_rd_fallback": direct_rd_fallback, "queue_conflicts": queue_conflicts, "user_names": user_names, "ext_names": ext_names, "user_dept": user_dept, "divisions_seen_direct": sorted(set(direct_map.values())), "divisions_seen_queue": sorted(set(queue_map.values())), } return queue_map, direct_map, meta def load_live_departments(json_path=None): """Load the live NS ext -> department cache (written by pull.pull_departments) and map each extension to a Reporting Division through the department matrix. Returns (live_map, live_queues, live_names): live_map : ext -> division (users + call queues with a mappable department) live_queues : extensions that are call queues (route_type labeling) live_names : ext -> display name from the live pull Auto attendants are never mapped (the IVR-leg exclusion rules stay in force). Returns empty structures if the cache file does not exist yet. """ import json import os json_path = json_path or config.DEPARTMENTS_JSON if not os.path.exists(json_path): return {}, set(), {} with open(json_path) as f: raw = json.load(f) live_map, live_queues, live_names = {}, set(), {} for ext, info in raw.items(): typ = info.get("type", "") if info.get("name"): live_names[ext] = info["name"] if typ == "callqueue": live_queues.add(ext) if typ == "autoattendant": continue div = config.DEPT_CROSSWALK.get(config.normalize_dept(info.get("department", ""))) if div: live_map[ext] = div return live_map, live_queues, live_names if __name__ == "__main__": q, d, m = load_maps() print("queue_map:", m["n_queue"], "entries ->", m["divisions_seen_queue"]) print("direct_map:", m["n_direct"], "entries ->", m["divisions_seen_direct"]) print("Reporting-Department-column fallback used for", m["direct_rd_fallback"], "direct users") print("queue conflicts:", m["queue_conflicts"]) print("\nsample queue_map:", dict(list(q.items())[:20])) print("\nsample direct_map:", dict(list(d.items())[:20]))