Last active
August 20, 2025 23:02
-
-
Save samuelguebo/ee9cde5919e5977103149372de5c53ca to your computer and use it in GitHub Desktop.
Stats on Wikimedia's production groups from publicly availble data.yaml
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| #!/usr/bin/env python3 | |
| # ----------------------------------------------------------------------------- | |
| # MIT License | |
| # | |
| # Copyright (c) 2025 Samuel Guebo | |
| # | |
| # Permission is hereby granted, free of charge, to any person obtaining a copy | |
| # of this software and associated documentation files (the "Software"), to deal | |
| # in the Software without restriction, including without limitation the rights | |
| # to use, copy, modify, merge, publish, distribute, sublicense, and/or sell | |
| # copies of the Software, and to permit persons to whom the Software is | |
| # furnished to do so, subject to the following conditions: | |
| # | |
| # The above copyright notice and this permission notice shall be included in all | |
| # copies or substantial portions of the Software. | |
| # | |
| # THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR | |
| # IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, | |
| # FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE | |
| # AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER | |
| # LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, | |
| # OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE | |
| # SOFTWARE. | |
| # ----------------------------------------------------------------------------- | |
| """ | |
| wmf-production-access-groups-stats.py — stdlib-only | |
| Reads Wikimedia's production groups file and prints: | |
| - Per-group member counts (filterable by regex or group type) | |
| - Optional CSV/JSON output | |
| - Optional staff vs volunteer breakdown (based on @wikimedia.org emails) | |
| Usage: | |
| python3 wmf-production-access-groups-stats.py --categorize-staff-volunteers | |
| python3 wmf-production-access-groups-stats.py --group-type analytics --format csv | |
| python3 wmf-production-access-groups-stats.py --show-detailed-users | |
| """ | |
| from __future__ import annotations | |
| import argparse | |
| import csv | |
| import io | |
| import json | |
| import re | |
| import sys | |
| import urllib.request | |
| from datetime import datetime | |
| from typing import Dict, List, Tuple, Optional, Set | |
| DEFAULT_URL = "https://raw.githubusercontent.com/wikimedia/operations-puppet/master/modules/admin/data/data.yaml" | |
| DEFAULT_GROUP_REGEX = ( | |
| r"^(?:" | |
| r"analytics-privatedata-users|" | |
| r"analytics-(?:wmde|search|product|research|platform-eng)-users|" | |
| r"statistics-(?:admins|web-users)|" | |
| r"eventlogging-admins|" | |
| r"mw-log-readers" | |
| r")$" | |
| ) | |
| # Predefined group categories for easier filtering | |
| GROUP_CATEGORIES = { | |
| "analytics": r"^(analytics|statistics|eventlogging|airflow-analytics)", | |
| "admin": r"(admins?|deployers?|root)", | |
| "access": r"(access|users|readers|writers)", | |
| "wmde": r"(wmde|deutschland)", | |
| "search": r"(search|elasticsearch|solr)", | |
| "ml": r"(ml|machine-learning|ai)", | |
| "platform": r"(platform|infrastructure|ops)", | |
| "absent": r"^(absent)", | |
| "all": r".*" # matches everything | |
| } | |
| # ----------------------------- Minimal YAML reader ---------------------------- | |
| class MiniYAMLParseError(RuntimeError): | |
| pass | |
| def _rstrip_comment(line: str) -> str: | |
| """Remove inline comments (# …) unless the # is inside quotes (very simple).""" | |
| # This is intentionally simple; usernames shouldn't contain '#' anyway. | |
| if not line: | |
| return line | |
| out, in_s, in_d = [], False, False | |
| for i, ch in enumerate(line): | |
| if ch == "'" and not in_d: | |
| in_s = not in_s | |
| elif ch == '"' and not in_s: | |
| in_d = not in_d | |
| if ch == "#" and not in_s and not in_d: | |
| break | |
| out.append(ch) | |
| return "".join(out).rstrip() | |
| def _indent_width(line: str) -> int: | |
| return len(line) - len(line.lstrip(" ")) | |
| def parse_groups_members(text: str) -> Dict[str, List[str]]: | |
| """ | |
| Heuristic parser for the specific structure used in WMF data.yaml: | |
| groups: | |
| <groupname>: | |
| members: [&optional_anchor] | |
| - user1 | |
| - *alias_list | |
| - user2 | |
| <groupname2>: | |
| # may have no members key, etc. | |
| Returns: { group_name: [members...] } | |
| Supports: | |
| - sequence anchors on the 'members:' key: 'members: &anchor' | |
| - alias on members key: 'members: *alias' | |
| - alias list items: '- *alias' | |
| - plain items: '- username' | |
| - single-line array format: 'members: [user1, user2, user3]' | |
| - multi-line array format: 'members: [user1, user2,\n user3, user4]' | |
| - anchors in arrays: 'members: &anchor [user1, user2]' | |
| """ | |
| lines = text.splitlines() | |
| # Strip BOM if present | |
| if lines and lines[0].startswith("\ufeff"): | |
| lines[0] = lines[0].lstrip("\ufeff") | |
| # Preprocess: strip trailing comments and keep raw indentation | |
| proc = [ _rstrip_comment(l.rstrip("\n\r")) for l in lines ] | |
| # Find 'groups:' top-level | |
| groups_idx = None | |
| for i, l in enumerate(proc): | |
| if not l.strip(): | |
| continue | |
| if l.lstrip().startswith("groups:"): | |
| groups_idx = i | |
| break | |
| if groups_idx is None: | |
| raise MiniYAMLParseError("No top-level 'groups:' key found.") | |
| groups_indent = _indent_width(proc[groups_idx]) | |
| i = groups_idx + 1 | |
| groups: Dict[str, List[str]] = {} | |
| anchors: Dict[str, List[str]] = {} | |
| # Helper: parse single-line array format like "members: [user1, user2, user3]" | |
| def parse_array_members(array_str: str) -> List[str]: | |
| """Parse array string like '[user1, user2, user3]' into list of usernames.""" | |
| members = [] | |
| # Remove outer brackets and split by comma | |
| array_str = array_str.strip() | |
| if array_str.startswith('[') and array_str.endswith(']'): | |
| array_str = array_str[1:-1] | |
| if not array_str.strip(): | |
| return members | |
| # Split by comma and clean up each item | |
| for item in array_str.split(','): | |
| item = item.strip() | |
| # Remove quotes if present | |
| if ((item.startswith("'") and item.endswith("'")) or | |
| (item.startswith('"') and item.endswith('"'))): | |
| item = item[1:-1] | |
| if item: | |
| members.append(item) | |
| return members | |
| # Helper: parse multi-line array format like "members: [user1, user2,\n user3, user4]" | |
| def parse_multiline_array_members(start_index: int, base_indent: int) -> Tuple[int, List[str]]: | |
| """Parse multi-line array format where array spans multiple lines.""" | |
| members = [] | |
| idx = start_index | |
| # Get the first line content after "members:" | |
| first_line = proc[idx].strip() | |
| if not first_line.startswith("members:"): | |
| return idx, members | |
| # Extract the part after "members:" | |
| members_part = first_line[8:].strip() # Remove "members:" | |
| # Check for anchor: "members: &anchor [user1, user2" | |
| anchor_name = None | |
| if "&" in members_part and "[" in members_part: | |
| # Extract anchor name | |
| anchor_match = re.search(r'&([a-zA-Z_][a-zA-Z0-9_]*)', members_part) | |
| if anchor_match: | |
| anchor_name = anchor_match.group(1) | |
| # Remove anchor from the part to parse | |
| members_part = re.sub(r'&[a-zA-Z_][a-zA-Z0-9_]*\s*', '', members_part) | |
| # If it starts with '[', it's a multi-line array | |
| if members_part.startswith('['): | |
| # Collect all the array content | |
| array_content = members_part | |
| idx += 1 | |
| # Continue reading lines until we find the closing ']' | |
| while idx < len(proc): | |
| line = proc[idx] | |
| if not line.strip(): | |
| idx += 1 | |
| continue | |
| ind = _indent_width(line) | |
| if ind <= base_indent: | |
| # We've left the group block | |
| break | |
| # Add this line's content to the array | |
| array_content += " " + line.strip() | |
| # Check if we've found the closing bracket | |
| if ']' in line: | |
| break | |
| idx += 1 | |
| # Now parse the complete array content | |
| if array_content.startswith('[') and array_content.endswith(']'): | |
| array_content = array_content[1:-1] # Remove outer brackets | |
| # Split by comma and clean up each item | |
| for item in array_content.split(','): | |
| item = item.strip() | |
| # Remove quotes if present | |
| if ((item.startswith("'") and item.endswith("'")) or | |
| (item.startswith('"') and item.endswith('"'))): | |
| item = item[1:-1] | |
| if item: | |
| members.append(item) | |
| # Store anchor if defined | |
| if anchor_name is not None: | |
| anchors[anchor_name] = list(members) | |
| return idx, members | |
| # Helper: read a YAML list (indented block) into python list of members. | |
| def read_members(start_index: int, base_indent: int, anchor_name: Optional[str]) -> Tuple[int, List[str]]: | |
| members: List[str] = [] | |
| idx = start_index | |
| # If the 'members:' line had ' *alias' on same line, we handle before calling. | |
| while idx < len(proc): | |
| line = proc[idx] | |
| if not line.strip(): | |
| idx += 1 | |
| continue | |
| ind = _indent_width(line) | |
| if ind <= base_indent: | |
| # list ended | |
| break | |
| # Expect list item lines: "- something" | |
| stripped = line.strip() | |
| if not stripped.startswith("- "): | |
| # Not a list item; end of list block | |
| break | |
| item = stripped[2:].strip() | |
| # alias item? | |
| if item.startswith("*"): | |
| alias = item[1:].strip() | |
| members.extend(anchors.get(alias, [])) | |
| else: | |
| # plain scalar username (unquote if quoted) | |
| if ((item.startswith("'") and item.endswith("'")) or | |
| (item.startswith('"') and item.endswith('"'))): | |
| item = item[1:-1] | |
| if item: | |
| members.append(item) | |
| idx += 1 | |
| # Store anchor if defined on members: | |
| if anchor_name is not None: | |
| anchors[anchor_name] = list(members) | |
| return idx, members | |
| # Iterate group entries under 'groups:' | |
| while i < len(proc): | |
| line = proc[i] | |
| if not line.strip(): | |
| i += 1 | |
| continue | |
| ind = _indent_width(line) | |
| if ind <= groups_indent: | |
| # left the groups block | |
| break | |
| # group header line: "<name>:" | |
| m = re.match(r"^\s*([A-Za-z0-9._-]+)\s*:\s*$", line) | |
| if not m: | |
| i += 1 | |
| continue | |
| group = m.group(1) | |
| groups[group] = [] | |
| group_indent = ind | |
| i += 1 | |
| # scan inside the group block for 'members:' | |
| while i < len(proc): | |
| line2 = proc[i] | |
| if not line2.strip(): | |
| i += 1 | |
| continue | |
| ind2 = _indent_width(line2) | |
| if ind2 <= group_indent: | |
| # end of this group block | |
| break | |
| # look for 'members:' (possibly with &anchor or *alias on same line) | |
| mm = re.match(r"^\s*members\s*:\s*(.*)$", line2) | |
| if mm: | |
| tail = mm.group(1).strip() | |
| anchor_name: Optional[str] = None | |
| inline_alias: Optional[str] = None | |
| current_members: List[str] = [] | |
| if tail: | |
| # Check for anchor: "members: &anchor" | |
| anchor_match = re.search(r'&([a-zA-Z_][a-zA-Z0-9_]*)', tail) | |
| if anchor_match: | |
| anchor_name = anchor_match.group(1) | |
| # Check for alias: "members: *alias" | |
| alias_match = re.search(r'\*([a-zA-Z_][a-zA-Z0-9_]*)', tail) | |
| if alias_match: | |
| inline_alias = alias_match.group(1) | |
| current_members.extend(anchors.get(inline_alias, [])) | |
| # Check for array format: "members: [user1, user2" or "members: &anchor [user1, user2" | |
| if '[' in tail: | |
| # This is an array format - could be single-line or multi-line | |
| if tail.endswith(']'): | |
| # Single-line array format | |
| current_members = parse_array_members(tail) | |
| else: | |
| # Multi-line array format - the array starts on this line but continues | |
| new_i, current_members = parse_multiline_array_members(i, group_indent) | |
| groups[group] = current_members | |
| i = new_i # Update the index to where we left off | |
| break # Exit the inner loop since we've processed this group | |
| i += 1 | |
| # If we didn't find an array format, read multi-line list | |
| if not current_members: | |
| i, list_members = read_members(i, ind2, anchor_name) | |
| current_members.extend(list_members) | |
| groups[group] = current_members | |
| continue | |
| i += 1 | |
| return groups | |
| # ----------------------------- CLI and reporting ------------------------------ | |
| def fetch_text(src: str) -> str: | |
| """Fetch from URL or read from local file path.""" | |
| if re.match(r"^https?://", src): | |
| with urllib.request.urlopen(src, timeout=30) as resp: | |
| data = resp.read() | |
| try: | |
| return data.decode("utf-8") | |
| except UnicodeDecodeError: | |
| return data.decode("latin-1") | |
| else: | |
| with open(src, "r", encoding="utf-8") as f: | |
| return f.read() | |
| def parse_users_emails(text: str) -> Dict[str, str]: | |
| """ | |
| Parse the users section to extract username-to-email mappings. | |
| Returns: { username: email } | |
| """ | |
| lines = text.splitlines() | |
| # Strip BOM if present | |
| if lines and lines[0].startswith("\ufeff"): | |
| lines[0] = lines[0].lstrip("\ufeff") | |
| # Preprocess: strip trailing comments and keep raw indentation | |
| proc = [ _rstrip_comment(l.rstrip("\n\r")) for l in lines ] | |
| # Find 'users:' top-level | |
| users_idx = None | |
| for i, l in enumerate(proc): | |
| if not l.strip(): | |
| continue | |
| if l.lstrip().startswith("users:"): | |
| users_idx = i | |
| break | |
| if users_idx is None: | |
| return {} # No users section found | |
| users_indent = _indent_width(proc[users_idx]) | |
| i = users_idx + 1 | |
| username_to_email: Dict[str, str] = {} | |
| # Iterate through user entries | |
| while i < len(proc): | |
| line = proc[i] | |
| if not line.strip(): | |
| i += 1 | |
| continue | |
| ind = _indent_width(line) | |
| if ind <= users_indent: | |
| # left the users block | |
| break | |
| # Check if this is a username line (e.g., " username:") | |
| m = re.match(r"^\s*([A-Za-z0-9._-]+)\s*:\s*$", line) | |
| if m: | |
| username = m.group(1) | |
| user_indent = ind | |
| i += 1 | |
| # Look for email in this user's block | |
| while i < len(proc): | |
| line2 = proc[i] | |
| if not line2.strip(): | |
| i += 1 | |
| continue | |
| ind2 = _indent_width(line2) | |
| if ind2 <= user_indent: | |
| # end of this user block | |
| break | |
| # Look for email line | |
| email_match = re.match(r"^\s*email\s*:\s*(.+)$", line2) | |
| if email_match: | |
| email = email_match.group(1).strip() | |
| # Remove quotes if present | |
| if ((email.startswith("'") and email.endswith("'")) or | |
| (email.startswith('"') and email.endswith('"'))): | |
| email = email[1:-1] | |
| username_to_email[username] = email | |
| break | |
| i += 1 | |
| else: | |
| i += 1 | |
| return username_to_email | |
| def categorize_users_by_email(usernames: Set[str], username_to_email: Dict[str, str]) -> Tuple[Set[str], Set[str]]: | |
| """ | |
| Categorize users as staff (@wikimedia.org) or volunteers (other domains). | |
| Uses the email mapping to determine staff vs volunteers. | |
| Filters out YAML anchors/aliases and users without emails. | |
| """ | |
| staff = set() | |
| volunteers = set() | |
| for username in usernames: | |
| # Skip YAML anchors/aliases and users without emails | |
| if username.startswith(('&', '*')): | |
| continue | |
| email = username_to_email.get(username, "") | |
| if not email: | |
| continue | |
| if "@wikimedia.org" in email: | |
| staff.add(username) | |
| else: | |
| volunteers.add(username) | |
| return staff, volunteers | |
| def print_staff_volunteer_table(staff: Set[str], volunteers: Set[str], total_users: int): | |
| """Print a table showing staff vs volunteer counts for total unique users.""" | |
| print(f"\nTotal Unique Users - Staff vs Volunteers:") | |
| print(f"{'Category':<12} {'Count':<8} {'Percentage':<12}") | |
| print(f"{'-'*12} {'-'*8} {'-'*12}") | |
| total_with_emails = len(staff) + len(volunteers) | |
| if total_with_emails > 0: | |
| staff_pct = (len(staff) / total_with_emails) * 100 | |
| volunteer_pct = (len(volunteers) / total_with_emails) * 100 | |
| print(f"{'Staff':<12} {len(staff):<8} {staff_pct:>6.1f}%") | |
| print(f"{'Volunteers':<12} {len(volunteers):<8} {volunteer_pct:>6.1f}%") | |
| print(f"{'Total':<12} {total_with_emails:<8} {'100.0%':>12}") | |
| else: | |
| print("No users found in this category.") | |
| def print_detailed_user_table(staff: Set[str], volunteers: Set[str], username_to_email: Dict[str, str]): | |
| """Print a detailed table showing each user with username, email, and status.""" | |
| print(f"\nDetailed User List - Staff vs Volunteers:") | |
| print(f"{'Username':<20} {'Email':<35} {'Status':<12}") | |
| print(f"{'-'*20} {'-'*35} {'-'*12}") | |
| # Filter out invalid entries (YAML anchors/aliases) and only show users with valid emails | |
| valid_staff = {user for user in staff if username_to_email.get(user, "") and not user.startswith(('&', '*'))} | |
| valid_volunteers = {user for user in volunteers if username_to_email.get(user, "") and not user.startswith(('&', '*'))} | |
| # Sort users alphabetically for better readability | |
| all_valid_users = sorted(valid_staff | valid_volunteers) | |
| for username in all_valid_users: | |
| email = username_to_email.get(username, "") | |
| status = "Staff" if username in staff else "Volunteer" | |
| print(f"{username:<20} {email:<35} {status:<12}") | |
| # Show summary of filtered entries | |
| total_filtered = len(staff | volunteers) - len(all_valid_users) | |
| if total_filtered > 0: | |
| print(f"\nNote: {total_filtered} entries filtered out (YAML anchors/aliases or users without emails)") | |
| def print_results(groups: List[Tuple[str, List[str]]], | |
| total_excluding_absent: Optional[Set[str]], | |
| total_unique_users: Optional[Set[str]], | |
| total_staff: Set[str], | |
| total_volunteers: Set[str], | |
| user_emails: Dict[str, str], | |
| absent_group_usernames: Set[str], | |
| args): | |
| """Print results in the requested format with staff/volunteer breakdown.""" | |
| # Print group data | |
| if args.format == "csv": | |
| writer = csv.writer(sys.stdout) | |
| writer.writerow(["group", "member_count", "members_semicolon_joined"]) | |
| for name, members in groups: | |
| writer.writerow([name, len(members), ";".join(members)]) | |
| elif args.format == "json": | |
| payload = [{"group": name, "member_count": len(members), "members": members} | |
| for name, members in groups] | |
| json.dump(payload, sys.stdout, indent=2) | |
| sys.stdout.write("\n") | |
| else: | |
| # Table format | |
| if groups: | |
| name_w = max(5, max(len(name) for name, _ in groups)) | |
| print(f"{'Group':{name_w}} Members") | |
| print(f"{'-'*name_w} {'-'*7}") | |
| for name, members in groups: | |
| print(f"{name:{name_w}} {len(members):7d}") | |
| # Show member lists if requested | |
| if args.list_members and groups: | |
| for name, members in groups: | |
| print(f"\n[{name}]") | |
| if members: | |
| for m in members: | |
| print(m) | |
| else: | |
| print("(no members)") | |
| # Print totals | |
| if total_excluding_absent is not None: | |
| print(f"\nTotal unique users: {len(total_excluding_absent)}") | |
| print(f"Note: Excludes groups that start with 'absent' ({len(absent_group_usernames)} users in absent groups)") | |
| elif total_unique_users is not None: | |
| print(f"\nTotal unique users: {len(total_unique_users)}") | |
| print(f"Note: Includes all groups including absent groups") | |
| # Print staff/volunteer breakdown | |
| if args.categorize_staff_volunteers: | |
| total_users_count = len(total_excluding_absent) if total_excluding_absent is not None else len(total_unique_users) | |
| print_staff_volunteer_table(total_staff, total_volunteers, total_users_count) | |
| # Show detailed user table if requested | |
| if args.show_detailed_users: | |
| print_detailed_user_table(total_staff, total_volunteers, user_emails) | |
| # Print timestamp | |
| if args.format == "table": | |
| print(f"\nStats generated: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}") | |
| else: | |
| print(f"\n# Stats generated: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}") | |
| def main(): | |
| ap = argparse.ArgumentParser(description="Count/list production groups from Wikimedia data.yaml (stdlib-only).") | |
| ap.add_argument("--source", default=DEFAULT_URL, help="Raw data.yaml URL or local path") | |
| ap.add_argument("--all-groups", action="store_true", default=True, | |
| help="Show all groups (default)") | |
| ap.add_argument("--no-all-groups", action="store_true", | |
| help="Disable showing all groups (use with --group-regex or --group-type)") | |
| ap.add_argument("--group-regex", metavar="PATTERN", | |
| help="Show only groups matching regex pattern") | |
| ap.add_argument("--group-type", choices=list(GROUP_CATEGORIES.keys()), | |
| help="Show only groups of a specific type") | |
| ap.add_argument("--min-members", type=int, default=0, | |
| help="Show only groups with at least N members (default: 0)") | |
| ap.add_argument("--max-members", type=int, | |
| help="Show only groups with at most N members") | |
| ap.add_argument("--sort-by", choices=["group", "members"], default="group", | |
| help="Sort by group name or member count (default: group)") | |
| ap.add_argument("--asc", action="store_true", default=True, | |
| help="Sort in ascending order (default)") | |
| ap.add_argument("--desc", action="store_true", | |
| help="Sort in descending order") | |
| ap.add_argument("--format", choices=["table", "csv", "json"], default="table", | |
| help="Output format (default: table)") | |
| ap.add_argument("--list-members", action="store_true", | |
| help="Show member lists for each group") | |
| ap.add_argument("--show-total-excluding-absent", action="store_true", default=True, | |
| help="Show total unique users excluding absent groups (default)") | |
| ap.add_argument("--no-show-total-excluding-absent", action="store_true", | |
| help="Disable showing total excluding absent groups") | |
| ap.add_argument("--categorize-staff-volunteers", action="store_true", | |
| help="Categorize users as staff (@wikimedia.org) vs volunteers based on email addresses") | |
| ap.add_argument("--show-detailed-users", action="store_true", | |
| help="Show detailed table of users with username, email, and staff/volunteer status") | |
| args = ap.parse_args() | |
| # Load and parse | |
| try: | |
| text = fetch_text(args.source) | |
| except Exception as e: | |
| print(f"Failed to download/read YAML: {e}", file=sys.stderr) | |
| sys.exit(2) | |
| try: | |
| all_groups = parse_groups_members(text) | |
| except Exception as e: | |
| print(f"Failed to parse YAML (stdlib parser): {e}", file=sys.stderr) | |
| print("Hint: For full YAML fidelity (anchors, merges), consider PyYAML.", file=sys.stderr) | |
| sys.exit(3) | |
| # Parse users section to get email mappings | |
| try: | |
| user_emails = parse_users_emails(text) | |
| except Exception as e: | |
| print(f"Warning: Failed to parse users section: {e}", file=sys.stderr) | |
| user_emails = {} | |
| # Determine which regex to use | |
| if args.group_type or args.group_regex: | |
| # User specified a specific filter, so use that instead of all-groups | |
| if args.group_type: | |
| regex = GROUP_CATEGORIES[args.group_type] | |
| else: | |
| regex = args.group_regex | |
| else: | |
| # Default behavior: show all groups | |
| regex = None | |
| # The pick_groups function is removed, so we'll just filter by regex if provided | |
| # and by all_flag if not. | |
| selected = [] | |
| if regex: | |
| pat = re.compile(regex) | |
| selected = [(n, all_groups.get(n, [])) for n in all_groups.keys() if pat.search(n)] | |
| else: | |
| selected = [(n, all_groups.get(n, [])) for n in all_groups.keys()] | |
| # Apply member count filters | |
| if args.min_members > 0 or args.max_members is not None: | |
| filtered = [] | |
| for name, members in selected: | |
| member_count = len(members) | |
| if args.min_members > 0 and member_count < args.min_members: | |
| continue | |
| if args.max_members is not None and member_count > args.max_members: | |
| continue | |
| filtered.append((name, members)) | |
| selected = filtered | |
| # Sort the results | |
| if args.sort_by == "members": | |
| selected.sort(key=lambda x: len(x[1]), reverse=args.desc) | |
| else: # sort by group name | |
| selected.sort(key=lambda x: x[0], reverse=args.desc) | |
| # Calculate total excluding absent users if requested | |
| total_excluding_absent = None | |
| if not args.no_show_total_excluding_absent: | |
| # Get all usernames from non-absent groups (this is what we want to count) | |
| all_usernames = set() | |
| for group_name, members in all_groups.items(): | |
| if not group_name.startswith("absent"): | |
| all_usernames.update(members) | |
| # Filter out YAML anchors/aliases and users without emails | |
| valid_usernames = set() | |
| for username in all_usernames: | |
| if not username.startswith(('&', '*')): | |
| email = user_emails.get(username, "") | |
| if email: | |
| valid_usernames.add(username) | |
| # The total is valid users in non-absent groups | |
| total_excluding_absent = valid_usernames | |
| # For informational purposes, also get absent group users | |
| absent_group_usernames = set() | |
| for group_name, members in all_groups.items(): | |
| if group_name.startswith("absent"): | |
| absent_group_usernames.update(members) | |
| # Calculate total unique users for all groups if requested | |
| total_unique_users = None | |
| if not args.no_all_groups: | |
| # Get all usernames from all groups | |
| all_usernames = set() | |
| for group_name, members in all_groups.items(): | |
| all_usernames.update(members) | |
| # Filter out YAML anchors/aliases and users without emails | |
| valid_usernames = set() | |
| for username in all_usernames: | |
| if not username.startswith(('&', '*')): | |
| email = user_emails.get(username, "") | |
| if email: | |
| valid_usernames.add(username) | |
| total_unique_users = valid_usernames | |
| # Handle staff/volunteer categorization for total unique users | |
| total_staff = set() | |
| total_volunteers = set() | |
| if args.categorize_staff_volunteers: | |
| # Get total unique users (excluding absent) | |
| total_unique_users_set = set() | |
| if total_excluding_absent is not None: | |
| total_unique_users_set = total_excluding_absent | |
| elif total_unique_users is not None: | |
| total_unique_users_set = total_unique_users | |
| # Categorize total unique users as staff vs volunteers | |
| # Note: total_unique_users_set is already filtered to valid users | |
| total_staff, total_volunteers = categorize_users_by_email(total_unique_users_set, user_emails) | |
| print_results(selected, total_excluding_absent, total_unique_users, total_staff, total_volunteers, user_emails, absent_group_usernames, args) | |
| if __name__ == "__main__": | |
| try: | |
| main() | |
| except KeyboardInterrupt: | |
| print("\nInterrupted.", file=sys.stderr) | |
| sys.exit(130) |
samuelguebo
commented
Aug 20, 2025
Author
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment