"""Read-only, multi-tenant Google Group inventory and reconciliation job. Google Workspace access reuses workspace.py's per-tenant, domain-wide- delegated credentials. The job never mutates a Google Group. A separate Cloud Run runtime identity reads/writes only the central extraction spreadsheet. """ from __future__ import annotations import json import logging import os import re import sys from dataclasses import dataclass from datetime import datetime, timezone from typing import Any, Callable from zoneinfo import ZoneInfo import google.auth from google.auth.transport.requests import Request as GoogleAuthRequest from googleapiclient.discovery import build from googleapiclient.errors import HttpError from google.oauth2 import service_account import workspace logger = logging.getLogger("google_workspace_group_inventory") SHEETS_SCOPE = "https://www.googleapis.com/auth/spreadsheets" SHEETS_READONLY_SCOPE = "https://www.googleapis.com/auth/spreadsheets.readonly" GROUP_MEMBER_READONLY_SCOPE = ( "https://www.googleapis.com/auth/admin.directory.group.member.readonly" ) EMAIL_RE = re.compile(r"^[^\s@]+@[^\s@]+\.[^\s@]+$") ACTUAL_SHEET = "_GW Actual" RECONCILIATION_SHEET = "_GW Reconciliation" STATUS_SHEET = "_GW Status" DISPLAY_TIME_ZONE = ZoneInfo("America/New_York") TIMESTAMP_BLANK_ROWS = 2 STATUS_HEADERS = [ "Group tab", "Sync status", "Last attempted", "Last successful sync", "Google members", "HR members", "Differences", "Details", ] STATUS_TAB_COLORS = { "SYNCED": {"red": 0.20, "green": 0.66, "blue": 0.33}, "PENDING": {"red": 0.95, "green": 0.75, "blue": 0.20}, "CONFIGURATION REQUIRED": {"red": 1.0, "green": 0.55, "blue": 0.0}, "GROUP NOT FOUND": {"red": 0.85, "green": 0.20, "blue": 0.20}, "ERROR": {"red": 0.85, "green": 0.20, "blue": 0.20}, } STATUS_CELL_COLORS = { "SYNCED": {"red": 0.72, "green": 0.88, "blue": 0.80}, "PENDING": {"red": 1.0, "green": 0.95, "blue": 0.70}, "CONFIGURATION REQUIRED": {"red": 0.99, "green": 0.85, "blue": 0.63}, "GROUP NOT FOUND": {"red": 0.96, "green": 0.78, "blue": 0.76}, "ERROR": {"red": 0.96, "green": 0.78, "blue": 0.76}, } @dataclass(frozen=True) class GroupTarget: group_email: str workspace_key: str = "" @dataclass(frozen=True) class GroupMember: email: str role: str member_type: str status: str def env_bool(name: str, default: bool = False) -> bool: value = os.getenv(name) if value is None: return default normalized = value.strip().lower() if normalized in {"1", "true", "yes", "on"}: return True if normalized in {"0", "false", "no", "off"}: return False raise RuntimeError(f"{name} must be true or false.") def normalize_email(value: Any) -> str: return str(value or "").strip().lower() def normalize_role(value: Any, default: str = "member") -> str: return str(value or default).strip().lower() def format_display_timestamp(value: datetime) -> str: return value.astimezone(DISPLAY_TIME_ZONE).strftime("%m/%d/%Y %I:%M %p %Z") def build_group_table( members: list[GroupMember], updated_at: datetime, ) -> list[list[str]]: rows = [["Email address", "Role"]] rows.extend([[member.email, member.role] for member in members]) rows.extend([["", ""] for _ in range(TIMESTAMP_BLANK_ROWS)]) rows.append(["Last updated on " + format_display_timestamp(updated_at), ""]) return rows def parse_targets(raw: str) -> list[GroupTarget]: try: parsed = json.loads(raw) except json.JSONDecodeError as exc: raise RuntimeError("GROUP_TARGETS_JSON is not valid JSON.") from exc if not isinstance(parsed, list): raise RuntimeError("GROUP_TARGETS_JSON must be a JSON array.") targets: list[GroupTarget] = [] seen: set[str] = set() for index, value in enumerate(parsed): if isinstance(value, str): email = normalize_email(value) route = "" elif isinstance(value, dict): email = normalize_email(value.get("group") or value.get("groupEmail")) route = str(value.get("workspace") or value.get("workspaceKey") or "").strip() else: raise RuntimeError(f"Invalid group target at index {index}.") if not EMAIL_RE.match(email): raise RuntimeError(f"Invalid group email at target index {index}: {email!r}.") if email not in seen: targets.append(GroupTarget(email, route)) seen.add(email) if not targets: raise RuntimeError("GROUP_TARGETS_JSON contains no groups.") return targets def quote_sheet_title(title: str) -> str: return "'" + title.replace("'", "''") + "'" def spreadsheet_titles(sheets_service: Any, spreadsheet_id: str) -> list[str]: try: response = ( sheets_service.spreadsheets() .get(spreadsheetId=spreadsheet_id, fields="sheets.properties.title") .execute(num_retries=3) ) except HttpError as exc: if getattr(exc.resp, "status", None) == 404: raise RuntimeError( "The destination spreadsheet was not found or is not shared with " "the Cloud Run runtime service account. Share the workbook with " "that identity and retry." ) from exc raise return [ str(sheet.get("properties", {}).get("title", "")) for sheet in response.get("sheets") or [] if sheet.get("properties", {}).get("title") ] def spreadsheet_sheet_ids( sheets_service: Any, spreadsheet_id: str, ) -> dict[str, int]: response = ( sheets_service.spreadsheets() .get( spreadsheetId=spreadsheet_id, fields="sheets.properties(sheetId,title)", ) .execute(num_retries=3) ) return { str(properties["title"]): int(properties["sheetId"]) for sheet in response.get("sheets") or [] if (properties := sheet.get("properties", {})).get("title") and properties.get("sheetId") is not None } def discover_targets(titles: list[str]) -> list[GroupTarget]: targets = { normalize_email(title) for title in titles if EMAIL_RE.match(normalize_email(title)) } if not targets: raise RuntimeError("The spreadsheet has no email-named group tabs.") return [GroupTarget(email) for email in sorted(targets)] def resolve_target(target: GroupTarget) -> tuple[GroupTarget, workspace.WorkspaceRoute]: route = workspace.resolve_route(target.workspace_key or target.group_email) return target, route def group_directory_service(route: workspace.WorkspaceRoute) -> Any: """Build the narrowest delegated Directory client needed by this job.""" info = workspace.service_account_info_for_route(route.key) credentials = service_account.Credentials.from_service_account_info( info, scopes=(GROUP_MEMBER_READONLY_SCOPE,), ).with_subject(route.admin_email) return build( "admin", "directory_v1", credentials=credentials, cache_discovery=False, ) def fetch_all_members( target: GroupTarget, route: workspace.WorkspaceRoute, *, include_derived: bool = False, service_factory: Callable[[str, str, str], Any] | None = None, ) -> list[GroupMember]: directory = ( service_factory(route.key, "admin", "directory_v1") if service_factory else group_directory_service(route) ) members: dict[str, GroupMember] = {} page_token = "" while True: params: dict[str, Any] = { "groupKey": target.group_email, "maxResults": 200, "includeDerivedMembership": include_derived, } if page_token: params["pageToken"] = page_token response = directory.members().list(**params).execute(num_retries=3) for value in response.get("members") or []: email = normalize_email(value.get("email")) if not EMAIL_RE.match(email): logger.warning( "Skipping member without an email: group=%s member_id=%s", target.group_email, value.get("id", ""), ) continue members[email] = GroupMember( email=email, role=normalize_role(value.get("role")), member_type=str(value.get("type") or "unknown").strip().lower(), status=str(value.get("status") or "unknown").strip().lower(), ) page_token = str(response.get("nextPageToken") or "") if not page_token: break return [members[email] for email in sorted(members)] def read_hr_check_members( sheets_service: Any, spreadsheet_id: str, target: GroupTarget, ) -> set[str]: try: result = ( sheets_service.spreadsheets() .values() .get( spreadsheetId=spreadsheet_id, range=f"{quote_sheet_title(target.group_email)}!C2:C", ) .execute(num_retries=3) ) except HttpError as exc: # Some legacy group tabs were created with only two physical columns. # Column C is an optional HR comparison input; its absence must not # prevent us from inventorying and publishing the authoritative A:B # Google Group membership. message = str(exc).lower() max_columns = re.search(r"max columns:\s*(\d+)", message) if ( getattr(exc.resp, "status", None) == 400 and "exceeds grid limits" in message and max_columns and int(max_columns.group(1)) < 3 ): logger.warning( "HR Check column is absent; treating it as empty: group=%s", target.group_email, ) return set() raise desired: set[str] = set() for row in result.get("values") or []: email = normalize_email(row[0] if row else "") if not EMAIL_RE.match(email): continue desired.add(email) return desired def reconcile_members( desired: set[str], actual: list[GroupMember], ) -> list[tuple[str, str, str]]: actual_by_email = {member.email: member for member in actual} rows: list[tuple[str, str, str]] = [] for email in sorted(set(desired) | set(actual_by_email)): actual_role = actual_by_email[email].role if email in actual_by_email else "" if not actual_role: result = "missing_in_google" elif email not in desired: result = "extra_in_google" else: result = "match" rows.append((email, actual_role, result)) return rows def classify_sync_error(exc: Exception) -> tuple[str, str]: message = str(exc).lower() if isinstance(exc, ValueError) and "unknown workspace" in message: return ( "CONFIGURATION REQUIRED", "Google Workspace tenant route is not configured.", ) if isinstance(exc, HttpError) and getattr(exc.resp, "status", None) == 404: return ( "GROUP NOT FOUND", "Group address was not found in the configured Workspace tenant.", ) return ("ERROR", f"{type(exc).__name__}: inventory attempt failed.") def read_previous_status( service: Any, spreadsheet_id: str, titles: list[str], ) -> dict[str, list[Any]]: if STATUS_SHEET not in titles: return {} result = ( service.spreadsheets() .values() .get( spreadsheetId=spreadsheet_id, range=f"{quote_sheet_title(STATUS_SHEET)}!A2:H", ) .execute(num_retries=3) ) return { normalize_email(row[0]): list(row) for row in result.get("values") or [] if row and EMAIL_RE.match(normalize_email(row[0])) } def failed_status_row( group_email: str, status: str, attempted_at: str, details: str, previous: list[Any] | None, ) -> list[Any]: prior = list(previous or []) + [""] * len(STATUS_HEADERS) return [ group_email, status, attempted_at, prior[3], prior[4], prior[5], prior[6], details, ] def sheets_service(write_enabled: bool) -> Any: scope = SHEETS_SCOPE if write_enabled else SHEETS_READONLY_SCOPE credentials, credential_project = google.auth.default(scopes=[scope]) credentials.refresh(GoogleAuthRequest()) logger.info( "Sheets credential ready: type=%s principal=%s credential_project=%s " "requested_scope=%s", type(credentials).__name__, getattr(credentials, "service_account_email", "unknown"), credential_project or "unknown", scope, ) return build("sheets", "v4", credentials=credentials, cache_discovery=False) def select_targets( titles: list[str], raw_targets: str, write_enabled: bool, ) -> list[GroupTarget]: if raw_targets: return parse_targets(raw_targets) if write_enabled: raise RuntimeError( "GROUP_TARGETS_JSON is required when GROUP_INVENTORY_WRITE_ENABLED=true. " "Automatic discovery is dry-run only." ) return discover_targets(titles) def ensure_output_sheets( service: Any, spreadsheet_id: str, existing_titles: list[str], ) -> None: missing = [ title for title in (STATUS_SHEET, ACTUAL_SHEET, RECONCILIATION_SHEET) if title not in existing_titles ] if not missing: return service.spreadsheets().batchUpdate( spreadsheetId=spreadsheet_id, body={ "requests": [ {"addSheet": {"properties": {"title": title}}} for title in missing ] }, ).execute(num_retries=3) def pad_rows(rows: list[list[Any]], row_count: int, width: int) -> list[list[Any]]: padded = [row[:width] + [""] * max(0, width - len(row)) for row in rows] padded.extend([[""] * width for _ in range(max(0, row_count - len(padded)))]) return padded def existing_row_counts( service: Any, spreadsheet_id: str, titles: list[str], ) -> dict[str, int]: result = ( service.spreadsheets() .values() .batchGet( spreadsheetId=spreadsheet_id, ranges=[f"{quote_sheet_title(title)}!A:Z" for title in titles], ) .execute(num_retries=3) ) value_ranges = result.get("valueRanges") or [] return { title: len( (value_ranges[index] if index < len(value_ranges) else {}).get("values") or [] ) for index, title in enumerate(titles) } def publish_results( service: Any, spreadsheet_id: str, titles: list[str], group_tables: dict[str, list[list[Any]]], actual_rows: list[list[Any]], reconciliation_rows: list[list[Any]], status_rows: list[list[Any]], group_statuses: dict[str, str], ) -> None: ensure_output_sheets(service, spreadsheet_id, titles) sheet_ids = spreadsheet_sheet_ids(service, spreadsheet_id) tables = [ (title, rows, 2) for title, rows in sorted(group_tables.items()) ] + [ (ACTUAL_SHEET, actual_rows, 7), (RECONCILIATION_SHEET, reconciliation_rows, 6), (STATUS_SHEET, status_rows, len(STATUS_HEADERS)), ] prior_row_counts = existing_row_counts( service, spreadsheet_id, [title for title, _rows, _width in tables], ) data = [] group_format_rows: dict[str, tuple[int, int]] = {} for title, rows, width in tables: row_count = max(len(rows), prior_row_counts.get(title, 0), 1) if title in group_tables: group_format_rows[title] = (row_count, len(rows) - 1) values = pad_rows(rows, row_count, width) end_column = chr(64 + width) data.append( { "range": ( f"{quote_sheet_title(title)}!A1:{end_column}{row_count}" ), "majorDimension": "ROWS", "values": values, } ) service.spreadsheets().values().batchUpdate( spreadsheetId=spreadsheet_id, body={"valueInputOption": "RAW", "data": data}, ).execute(num_retries=3) format_requests = [] for title in sorted(group_tables): sheet_id = sheet_ids[title] row_count, timestamp_row_index = group_format_rows[title] format_requests.extend([ { "repeatCell": { "range": { "sheetId": sheet_id, "startRowIndex": 1, "endRowIndex": row_count, "startColumnIndex": 0, "endColumnIndex": 1, }, "cell": { "userEnteredFormat": {"textFormat": {"bold": False}} }, "fields": "userEnteredFormat.textFormat.bold", } }, { "repeatCell": { "range": { "sheetId": sheet_id, "startRowIndex": timestamp_row_index, "endRowIndex": timestamp_row_index + 1, "startColumnIndex": 0, "endColumnIndex": 1, }, "cell": { "userEnteredFormat": {"textFormat": {"bold": True}} }, "fields": "userEnteredFormat.textFormat.bold", } }, ]) status_sheet_id = sheet_ids[STATUS_SHEET] format_requests.extend([ { "updateSheetProperties": { "properties": { "sheetId": status_sheet_id, "index": 1, "tabColor": {"red": 0.20, "green": 0.45, "blue": 0.80}, "gridProperties": {"frozenRowCount": 1}, }, "fields": "gridProperties.frozenRowCount,index,tabColor", } }, { "repeatCell": { "range": { "sheetId": status_sheet_id, "startRowIndex": 0, "endRowIndex": 1, "startColumnIndex": 0, "endColumnIndex": len(STATUS_HEADERS), }, "cell": { "userEnteredFormat": { "backgroundColor": {"red": 0.0, "green": 0.0, "blue": 0.0}, "textFormat": {"bold": True, "foregroundColor": {"red": 1.0, "green": 1.0, "blue": 1.0}}, } }, "fields": "userEnteredFormat(backgroundColor,textFormat)", } }, { "setBasicFilter": { "filter": { "range": { "sheetId": status_sheet_id, "startRowIndex": 0, "endRowIndex": max(len(status_rows), 1), "startColumnIndex": 0, "endColumnIndex": len(STATUS_HEADERS), } } } }, { "autoResizeDimensions": { "dimensions": { "sheetId": status_sheet_id, "dimension": "COLUMNS", "startIndex": 0, "endIndex": len(STATUS_HEADERS), } } }, ]) for row_index, row in enumerate(status_rows[1:], start=1): status = str(row[1]) format_requests.append({ "repeatCell": { "range": { "sheetId": status_sheet_id, "startRowIndex": row_index, "endRowIndex": row_index + 1, "startColumnIndex": 1, "endColumnIndex": 2, }, "cell": { "userEnteredFormat": { "backgroundColor": STATUS_CELL_COLORS.get( status, STATUS_CELL_COLORS["ERROR"] ), "textFormat": {"bold": True}, } }, "fields": "userEnteredFormat(backgroundColor,textFormat.bold)", } }) for title, status in sorted(group_statuses.items()): if title not in sheet_ids: continue format_requests.append({ "updateSheetProperties": { "properties": { "sheetId": sheet_ids[title], "tabColor": STATUS_TAB_COLORS.get( status, STATUS_TAB_COLORS["ERROR"] ), }, "fields": "tabColor", } }) if format_requests: service.spreadsheets().batchUpdate( spreadsheetId=spreadsheet_id, body={"requests": format_requests}, ).execute(num_retries=3) def run() -> dict[str, Any]: spreadsheet_id = os.getenv("DESTINATION_SPREADSHEET_ID", "").strip() if not spreadsheet_id: raise RuntimeError("DESTINATION_SPREADSHEET_ID is required.") write_enabled = env_bool("GROUP_INVENTORY_WRITE_ENABLED", False) include_derived = env_bool("GROUP_INVENTORY_INCLUDE_DERIVED", False) service = sheets_service(write_enabled) titles = spreadsheet_titles(service, spreadsheet_id) raw_targets = os.getenv("GROUP_TARGETS_JSON", "").strip() targets = select_targets(titles, raw_targets, write_enabled) previous_status = read_previous_status(service, spreadsheet_id, titles) attempted_at = format_display_timestamp(datetime.now(timezone.utc)) actual_table: list[list[Any]] = [[ "Group email", "Member email", "Role", "Type", "Status", "Workspace", "Fetched at (UTC)", ]] reconciliation_table: list[list[Any]] = [[ "Group email", "Member email", "Actual role", "Result", "Workspace", "Fetched at (UTC)", ]] group_tables: dict[str, list[list[Any]]] = {} status_by_group: dict[str, list[Any]] = {} group_statuses: dict[str, str] = {} errors: list[dict[str, str]] = [] result_counts = {key: 0 for key in ( "match", "missing_in_google", "extra_in_google" )} for target in targets: try: _, route = resolve_target(target) actual = fetch_all_members( target, route, include_derived=include_derived, ) desired = read_hr_check_members(service, spreadsheet_id, target) fetched_at_value = datetime.now(timezone.utc) fetched_at = fetched_at_value.isoformat() group_tables[target.group_email] = build_group_table( actual, fetched_at_value, ) for member in actual: actual_table.append([ target.group_email, member.email, member.role, member.member_type, member.status, route.key, fetched_at, ]) reconciliation = reconcile_members(desired, actual) for email, actual_role, result in reconciliation: result_counts[result] += 1 reconciliation_table.append([ target.group_email, email, actual_role, result, route.key, fetched_at, ]) differences = sum( 1 for _email, _role, result in reconciliation if result != "match" ) group_statuses[target.group_email] = "SYNCED" status_by_group[target.group_email] = [ target.group_email, "SYNCED", attempted_at, attempted_at, len(actual), len(desired), differences, "Updated successfully from Google Workspace.", ] logger.info( "Inventoried group=%s workspace=%s desired=%d actual=%d", target.group_email, route.key, len(desired), len(actual), ) except Exception as exc: # collect every tenant failure before stopping logger.exception("Group inventory failed: group=%s", target.group_email) status, details = classify_sync_error(exc) group_statuses[target.group_email] = status status_by_group[target.group_email] = failed_status_row( target.group_email, status, attempted_at, details, previous_status.get(target.group_email), ) errors.append({ "group": target.group_email, "error": f"{type(exc).__name__}: {exc}"[:300], }) completed_at = datetime.now(timezone.utc).isoformat() status_table = [STATUS_HEADERS] + [ status_by_group[target.group_email] for target in targets ] status_counts = { status: sum(1 for value in group_statuses.values() if value == status) for status in sorted(set(group_statuses.values())) } summary = { "dryRun": not write_enabled, "includeDerivedMembership": include_derived, "targetGroups": len(targets), "actualMembershipRows": len(actual_table) - 1, "reconciliationRows": len(reconciliation_table) - 1, "results": result_counts, "statusCounts": status_counts, "errors": errors, "completedAt": completed_at, } if errors: if write_enabled: publish_results( service, spreadsheet_id, titles, group_tables, actual_table, reconciliation_table, status_table, group_statuses, ) summary["published"] = False summary["partialPublish"] = True summary["publishedGroups"] = len(group_tables) summary["statusPublished"] = True logger.error( "Partial publish: successful group tabs were updated; failed tabs " "were preserved and marked in %s.", STATUS_SHEET, ) else: summary["published"] = False summary["partialPublish"] = False logger.error( "Dry run: no sheet output was changed because groups failed." ) return summary if write_enabled: publish_results( service, spreadsheet_id, titles, group_tables, actual_table, reconciliation_table, status_table, group_statuses, ) summary["published"] = True summary["partialPublish"] = False summary["publishedGroups"] = len(group_tables) summary["statusPublished"] = True else: summary["published"] = False summary["partialPublish"] = False return summary def main() -> int: logging.basicConfig( level=os.getenv("LOG_LEVEL", "INFO").upper(), format="%(asctime)s %(levelname)s %(name)s %(message)s", ) try: summary = run() except Exception as exc: logger.exception("Inventory job could not start.") print(json.dumps({"error": f"{type(exc).__name__}: {exc}"})) return 1 print(json.dumps(summary, indent=2, sort_keys=True)) return 2 if summary["errors"] else 0 if __name__ == "__main__": sys.exit(main())