#!/usr/bin/env python3 """ Pull workspace member counts from the Slack Web API (users.list) and write data/slack.json. This mirrors what you see under Workspace → Members in Slack admin; it does not scrape the browser UI. Setup: 1. Create a Slack app at https://api.slack.com/apps → install to workspace → Bot User OAuth Token. 2. Bot Token Scopes: users:read (add users:read.email for emails; team.billing:read for plan). 3. User Token Scopes: admin — for team.billableInfo (billing_active per member). 4. cp env.example .env — set SLACK_BOT_TOKEN=xoxb-... and SLACK_USER_TOKEN=xoxp-... 5. Optional: SLACK_WORKSPACE_URL=https://g2media.slack.com (for links if auth.test fails) 6. Optional: SLACK_DETAIL_MAX (default 10000) — max rows per category in members*Detail arrays. 7. python3 app/src/jobs/fetch_slack_workspace.py Notes: - "Active members" = not deleted, not bot, not app user (Slack users.list flags). - "Deactivated" = billing inactive (team.billableInfo) plus identified deactivated accounts (user.deleted with a real name/email — bulk billableInfo omits many of these, e.g. Corissa). - "Billable" = billing_active true (same API). - membersDeletedApi = user.deleted count including anonymized placeholder accounts (not on dashboard). """ from __future__ import annotations import json import os import sys from datetime import datetime, timezone from pathlib import Path import requests from store import write_json_snapshot from env_config import env_str, jira_base_url from paths import DATA_DIR, REPO_ROOT, load_env_file ROOT = REPO_ROOT OUT = Path(os.environ.get("SLACK_OUT", str(DATA_DIR / "slack.json"))) def load_env_file() -> None: path = REPO_ROOT / ".env" if not path.is_file(): return for raw in path.read_text(encoding="utf-8-sig").splitlines(): line = raw.strip() if not line or line.startswith("#"): continue if line.startswith("export "): line = line[7:].strip() if "=" not in line: continue key, _, val = line.partition("=") key = key.strip() val = val.strip().strip('"').strip("'") if not key: continue if val: os.environ[key] = val continue prev = os.environ.get(key) if prev is None or not str(prev).strip(): os.environ[key] = val def _slack_member_row(m: dict, billing_active: bool | None = None) -> dict[str, object]: prof = m.get("profile") if isinstance(m.get("profile"), dict) else {} real = (m.get("real_name") or prof.get("real_name") or "") or "" row: dict[str, object] = { "id": m.get("id"), "name": (m.get("name") or "") or "", "realName": str(real).strip(), "displayName": str(prof.get("display_name") or "").strip(), "email": str(prof.get("email") or "").strip(), "isBot": bool(m.get("is_bot")), "isAppUser": bool(m.get("is_app_user")), "deleted": bool(m.get("deleted")), "isSingleChannelGuest": bool(m.get("is_restricted")) and not bool(m.get("is_ultra_restricted")), "isMultiChannelGuest": bool(m.get("is_ultra_restricted")), } if billing_active is not None: row["billingActive"] = billing_active return row def _is_human(m: dict) -> bool: return not m.get("is_bot") and not m.get("is_app_user") def _is_anonymized_deleted(m: dict) -> bool: """Slack replaces old deactivated accounts with deactivateduser* placeholders.""" name = str(m.get("name") or "") prof = m.get("profile") if isinstance(m.get("profile"), dict) else {} email = str(prof.get("email") or "") return name.startswith("deactivateduser") or "@slack-corp.com" in email def _is_billing_deactivated(m: dict, billable_map: dict[str, bool]) -> bool: """ Match Slack admin billing status Deactivated: - billing_active false in bulk billableInfo, or - account deleted with identifiable profile (omitted from bulk billableInfo). """ if not _is_human(m): return False uid = str(m.get("id") or "").strip() if not uid: return False if billable_map and uid in billable_map: return not billable_map[uid] if m.get("deleted") and not _is_anonymized_deleted(m): return True return False def _slack_token_line_hint() -> str: path = REPO_ROOT / ".env" if not path.is_file(): return f"No {path.name} next to the project (expected {path})." for raw in path.read_text(encoding="utf-8-sig").splitlines(): line = raw.strip() if not line or line.startswith("#") or "=" not in line: continue if line.startswith("export "): line = line[7:].strip() key, _, val = line.partition("=") if key.strip() != "SLACK_BOT_TOKEN": continue if not val.strip(): return ( "SLACK_BOT_TOKEN is present in .env but the value after '=' is empty on disk. " "Paste your xoxb- token and save the file (⌘S / Ctrl+S). " "If it looks filled in the editor, it may be unsaved." ) return "" return "No SLACK_BOT_TOKEN= line found in .env — add it next to the other keys." def _format_slack_api_err(data: dict, fallback: str = "unknown") -> str: err = str(data.get("error") or fallback) if err == "missing_scope": return f"{err}: add the required Slack scope and reinstall the app to the workspace." if err in ("token_revoked", "invalid_auth", "not_allowed_token_type"): return f"{err}: token invalid, revoked, or wrong token type for this API." return err def fetch_billing_plan(session: requests.Session) -> tuple[str | None, str | None]: try: r = session.post("https://slack.com/api/team.billing.info", timeout=30) r.raise_for_status() data = r.json() if not data.get("ok"): return None, _format_slack_api_err(data) plan = data.get("plan") return (str(plan).strip() if plan is not None else None), None except Exception as ex: # noqa: BLE001 return None, str(ex)[:400] def fetch_billable_map(session: requests.Session) -> tuple[dict[str, bool], str | None]: out: dict[str, bool] = {} cursor: str | None = "" try: while True: params: dict[str, str] = {"limit": "200"} if cursor: params["cursor"] = cursor r = session.get( "https://slack.com/api/team.billableInfo", params=params, timeout=120 ) r.raise_for_status() data = r.json() if not data.get("ok"): return {}, _format_slack_api_err(data) for uid, info in (data.get("billable_info") or {}).items(): if not isinstance(info, dict): continue out[str(uid)] = bool(info.get("billing_active")) cursor = (data.get("response_metadata") or {}).get("next_cursor") or "" if not cursor: break return out, None except Exception as ex: # noqa: BLE001 return out if out else {}, str(ex)[:400] def fetch_users_list(session: requests.Session) -> tuple[list[dict], str | None]: members: list[dict] = [] cursor: str | None = "" try: while True: params: dict[str, str] = {"limit": "200"} if cursor: params["cursor"] = cursor r = session.get("https://slack.com/api/users.list", params=params, timeout=120) r.raise_for_status() data = r.json() if not data.get("ok"): err = _format_slack_api_err(data) if data.get("error") == "missing_scope": err = ( "missing_scope: add Bot Token Scope users:read on the Slack app " "and reinstall the app to the workspace." ) return members, err members.extend(data.get("members") or []) cursor = (data.get("response_metadata") or {}).get("next_cursor") or "" if not cursor: break return members, None except Exception as ex: # noqa: BLE001 return members, str(ex)[:400] def main() -> int: load_env_file() token = os.environ.get("SLACK_BOT_TOKEN", "").strip() user_token = os.environ.get("SLACK_USER_TOKEN", "").strip() fallback_url = env_str("SLACK_WORKSPACE_URL") if not token: hint = _slack_token_line_hint() print( "Missing SLACK_BOT_TOKEN in environment (.env). See env.example.", file=sys.stderr, ) if hint: print(hint, file=sys.stderr) return 1 if not (token.startswith("xoxb-") or token.startswith("xoxp-")): print( "SLACK_BOT_TOKEN must be the OAuth token from Slack (starts with xoxb- for bots).\n" " Slack app → OAuth & Permissions → OAuth Tokens → Bot User OAuth Token.\n" " Do not use the Signing Secret, App ID, or verification token.", file=sys.stderr, ) return 1 bot_session = requests.Session() bot_session.headers.update({"Authorization": f"Bearer {token}"}) workspace_url = fallback_url try: r = bot_session.post("https://slack.com/api/auth.test", timeout=30) r.raise_for_status() auth = r.json() if auth.get("ok") and auth.get("url"): workspace_url = str(auth["url"]).strip().rstrip("/") except Exception as ex: # noqa: BLE001 print(f"[warn] auth.test: {ex}", file=sys.stderr) members, err = fetch_users_list(bot_session) billing_plan, billing_plan_err = fetch_billing_plan(bot_session) billable_map: dict[str, bool] = {} billable_err: str | None = None if user_token: if not user_token.startswith("xoxp-"): billable_err = "SLACK_USER_TOKEN must start with xoxp- (User OAuth Token with admin scope)." else: user_session = requests.Session() user_session.headers.update({"Authorization": f"Bearer {user_token}"}) billable_map, billable_err = fetch_billable_map(user_session) else: billable_err = ( "SLACK_USER_TOKEN not set — add User OAuth Token (admin scope) for billable / non-billable counts." ) payload: dict = { "updatedAt": datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"), "workspaceUrl": workspace_url, "billingPlan": billing_plan, "billingPlanError": billing_plan_err, "billableInfoError": billable_err, "membersActive": None, "membersBillable": None, "membersDeactivated": None, "membersDeletedApi": None, "membersSingleChannelGuests": None, "membersMultiChannelGuests": None, "membersActiveDetail": [], "membersBillableDetail": [], "membersDeactivatedDetail": [], "membersSingleChannelGuestsDetail": [], "membersMultiChannelGuestsDetail": [], "slackError": err, } if not err and members: try: detail_cap = int(os.environ.get("SLACK_DETAIL_MAX", "10000") or "10000") except ValueError: detail_cap = 10000 if detail_cap < 1: detail_cap = 1 deleted_api = sum(1 for m in members if m.get("deleted")) active_members = sum( 1 for m in members if not m.get("deleted") and _is_human(m) ) scg = sum( 1 for m in members if not m.get("deleted") and m.get("is_restricted") and not m.get("is_ultra_restricted") ) mcg = sum( 1 for m in members if not m.get("deleted") and m.get("is_ultra_restricted") ) payload["membersActive"] = active_members payload["membersDeletedApi"] = deleted_api payload["membersSingleChannelGuests"] = scg payload["membersMultiChannelGuests"] = mcg active_d: list[dict] = [] deact_d: list[dict] = [] scg_d: list[dict] = [] mcg_d: list[dict] = [] billable_d: list[dict] = [] billable_n = 0 deactivated_n = 0 for m in members: if not isinstance(m, dict): continue if m.get("is_ultra_restricted") and not m.get("deleted"): if len(mcg_d) < detail_cap: mcg_d.append(_slack_member_row(m)) elif m.get("is_restricted") and not m.get("deleted"): if len(scg_d) < detail_cap: scg_d.append(_slack_member_row(m)) if _is_human(m) and not m.get("deleted"): if len(active_d) < detail_cap: active_d.append(_slack_member_row(m)) if billable_map and _is_human(m): uid = str(m.get("id") or "").strip() if uid in billable_map and billable_map[uid]: billable_n += 1 if len(billable_d) < detail_cap: billable_d.append(_slack_member_row(m, True)) if billable_map and _is_billing_deactivated(m, billable_map): deactivated_n += 1 if len(deact_d) < detail_cap: deact_d.append(_slack_member_row(m, False)) payload["membersActiveDetail"] = active_d payload["membersSingleChannelGuestsDetail"] = scg_d payload["membersMultiChannelGuestsDetail"] = mcg_d if billable_map: payload["membersBillable"] = billable_n payload["membersDeactivated"] = deactivated_n payload["membersBillableDetail"] = billable_d payload["membersDeactivatedDetail"] = deact_d n_m = len(members) if ( len(active_d) >= detail_cap or len(deact_d) >= detail_cap or len(scg_d) >= detail_cap or len(mcg_d) >= detail_cap or len(billable_d) >= detail_cap ): print( f"[info] Slack member detail lists capped at {detail_cap} rows per category " f"(SLACK_DETAIL_MAX). Workspace returned {n_m} user object(s).", file=sys.stderr, ) if err: print(f"[warn] Slack API: {err}", file=sys.stderr) if billable_err and not billable_map: print(f"[warn] Slack billableInfo: {billable_err}", file=sys.stderr) if billing_plan_err: print(f"[warn] Slack billing.info: {billing_plan_err}", file=sys.stderr) out_path = write_json_snapshot(OUT, payload) print(f"Wrote {out_path}") print(f"Wrote {OUT}") return 0 if __name__ == "__main__": raise SystemExit(main())