wikimedia-user-group-stats.py

#!/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.
# -----------------------------------------------------------------------------

"""
wikimedia-user-group-stats.py — stdlib-only

Usage:
python3 wikimedia-user-group-stats.py
python3 wikimedia-user-group-stats.py --full                # print all rows
python3 wikimedia-user-group-stats.py --project frwiki      # only French Wikipedia
python3 wikimedia-user-group-stats.py --project enwiki      # only English Wikipedia
python3 wikimedia-user-group-stats.py --limit 50            # preview 50 rows
python3 wikimedia-user-group-stats.py --csv out.csv         # also write CSV
python3 wikimedia-user-group-stats.py --groups sysop,checkuser  # only specific groups
python3 wikimedia-user-group-stats.py --wiki-limit 10       # only scan first 10 wikis

Counts per-wiki memberships for:
  - sysop (Administrators)
  - checkuser (CheckUsers)
  - suppress (Oversighters)
  - steward (Stewards; global, mostly on Meta-Wiki)

Adds *deduplicated* (unique) totals across all wikis for each group.

Features:
  - Enumerates wikis via SiteMatrix
  - Skips closed/private/fishbowl sites
  - Follows continuation to fetch full membership
  - Modest concurrency + retries
  - Prints per-wiki table + summed totals + unique-user totals
  - Optional CSV
  - Configurable groups to scan
  - Progress reporting during scanning

Project names format:
  - frwiki (French Wikipedia)
  - enwiki (English Wikipedia)
  - dewiki (German Wikipedia)
  - eswiki (Spanish Wikipedia)
  - etc.

"""

from __future__ import annotations
import argparse
import csv
import json
import sys
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
from typing import Dict, List, Optional, Set
from urllib.parse import urlencode, urlparse, urlunparse, ParseResult
from urllib.request import urlopen, Request
from urllib.error import URLError, HTTPError

DEFAULT_GROUPS = ["sysop", "checkuser", "suppress", "steward"]
SITEMATRIX_URL = "https://meta.wikimedia.org/w/api.php?action=sitematrix&format=json"

# Politeness knobs
MAX_WORKERS = 6
PER_SITE_PAUSE_SEC = 0.08
BETWEEN_TASKS_SEC = 0.15
RETRIES = 3
RETRY_BACKOFF_SEC = 0.4

def fetch_json(url: str, params: Optional[Dict[str, str]] = None, tries: int = RETRIES) -> Dict:
    """HTTP GET → JSON (stdlib only), with retries."""
    if params:
        qs = urlencode(params)
        parsed = urlparse(url)
        url = urlunparse(ParseResult(parsed.scheme, parsed.netloc, parsed.path, parsed.params, qs, parsed.fragment))
    last_err = None
    for attempt in range(1, tries + 1):
        try:
            req = Request(url, headers={"User-Agent": "wmf-group-counter/stdlib/1.1"})
            with urlopen(req, timeout=30) as resp:
                return json.loads(resp.read().decode("utf-8"))
        except (HTTPError, URLError, TimeoutError) as e:
            last_err = e
            if attempt < tries:
                time.sleep(RETRY_BACKOFF_SEC * attempt)
            else:
                raise
    if last_err:
        raise last_err
    return {}

def extract_wikis_from_sitematrix(sm: Dict, project_filter: Optional[str], wiki_limit: Optional[int]) -> List[Dict[str, str]]:
    """Flatten SiteMatrix into [{dbname, api, project, lang}, ...]."""
    sites: List[Dict[str, str]] = []
    matrix = sm.get("sitematrix", {})
    
    # Debug: Show available projects if filtering
    if project_filter:
        available_projects = set()
        for key, entry in matrix.items():
            if key == "count":
                continue
            bucket_sites = []
            if isinstance(entry, dict) and "site" in entry:
                bucket_sites = entry["site"]
            elif isinstance(entry, list):
                bucket_sites = entry
            else:
                continue
            for s in bucket_sites:
                if isinstance(s, dict) and s.get("dbname"):
                    available_projects.add(s.get("dbname", ""))
        
        print(f"Available projects (dbnames): {sorted(available_projects)}", file=sys.stderr)
    
    for key, entry in matrix.items():
        if key == "count":
            continue
        bucket_sites = []
        if isinstance(entry, dict) and "site" in entry:
            bucket_sites = entry["site"]
        elif isinstance(entry, list):
            bucket_sites = entry
        else:
            continue
        for s in bucket_sites:
            if not isinstance(s, dict):
                continue
            if s.get("closed") or s.get("private") or s.get("fishbowl"):
                continue
            url = s.get("url")
            dbname = s.get("dbname")
            project = s.get("sitename") or ""
            code = s.get("code") or ""
            if not url or not dbname:
                continue
            
            # Simple project filtering using dbname
            if project_filter and dbname.lower() != project_filter.lower():
                continue
            
            api = url.rstrip("/") + "/w/api.php"
            sites.append({"dbname": dbname, "api": api, "project": project, "lang": code})
            
            # Apply wiki limit if specified
            if wiki_limit and len(sites) >= wiki_limit:
                return sites
    return sites

def fetch_group_usernames(api: str, group: str) -> List[str]:
    """Return all usernames in a specific group on a given wiki."""
    names: List[str] = []
    cont: Dict[str, str] = {}
    while True:
        params = {
            "action": "query",
            "format": "json",
            "list": "allusers",
            "augroup": group,
            "auprop": "",
            "aulimit": "max",
        }
        params.update(cont)
        data = fetch_json(api, params)
        users = data.get("query", {}).get("allusers", [])
        for u in users:
            name = u.get("name")
            if name:
                names.append(name)
        cont = data.get("continue") or {}
        if not cont:
            break
    return names

def process_wiki(site: Dict[str, str], groups: List[str], progress_callback=None) -> Dict[str, object]:
    """Fetch members for all target groups for a single wiki."""
    row: Dict[str, object] = {"wiki": site["dbname"], "project": site["project"], "api": site["api"]}
    try:
        for g in groups:
            members = fetch_group_usernames(site["api"], g)
            row[g] = len(members)
            row[f"{g}_members"] = members
            time.sleep(PER_SITE_PAUSE_SEC)
        
        # Report progress
        if progress_callback:
            progress_callback(site["dbname"], "completed")
            
    except Exception as e:
        row["error"] = f"{type(e).__name__}: {e}"
        if progress_callback:
            progress_callback(site["dbname"], f"error: {e}")
    return row

def format_table(rows: List[Dict[str, object]], limit: Optional[int]) -> str:
    """Simple fixed-width console table."""
    if limit is not None:
        rows = rows[:limit]
    
    # Get groups from the first row that has group data
    groups = []
    for row in rows:
        for key in row.keys():
            if key not in ["wiki", "project", "api", "error"] and not key.endswith("_members"):
                if key not in groups:
                    groups.append(key)
        if groups:
            break
    
    headers = ["wiki", "project", *groups, "error"]
    def cell(r, k):
        v = r.get(k, "")
        return "" if v is None else str(v)
    widths = {h: max(len(h), max((len(cell(r, h)) for r in rows), default=0)) for h in headers}
    line = lambda r: "  ".join(cell(r, h).ljust(widths[h]) for h in headers)
    sep = "  ".join("-" * widths[h] for h in headers)
    out = []
    out.append("=== Per-wiki counts ===")
    out.append(line({h: h for h in headers}))
    out.append(sep)
    for r in rows:
        out.append(line(r))
    if limit is not None and limit < len(rows):
        out.append(f"(showing {limit} of {len(rows)}; use --full or raise --limit)")
    return "\n".join(out)

def main():
    ap = argparse.ArgumentParser(description="Count WMF per-wiki group memberships with deduplicated totals (stdlib-only).")
    ap.add_argument("--project", help="Only include a specific project (e.g., 'frwiki' for French Wikipedia, 'enwiki' for English Wikipedia).")
    ap.add_argument("--full", action="store_true", help="Print all per-wiki rows.")
    ap.add_argument("--limit", type=int, help="Limit number of printed rows.")
    ap.add_argument("--csv", help="Path to write CSV output.")
    ap.add_argument("--workers", type=int, default=MAX_WORKERS, help="Concurrency (default 6).")
    ap.add_argument("--groups", help="Comma-separated list of groups to scan (default: sysop,checkuser,suppress,steward).")
    ap.add_argument("--wiki-limit", type=int, help="Limit number of wikis to scan (useful for testing).")
    args = ap.parse_args()

    # Parse groups argument
    if args.groups:
        groups = [g.strip() for g in args.groups.split(",")]
    else:
        groups = DEFAULT_GROUPS

    print("Fetching SiteMatrix…", file=sys.stderr)
    sm = fetch_json(SITEMATRIX_URL)
    sites = extract_wikis_from_sitematrix(sm, args.project, args.wiki_limit)
    print(f"Wikis to scan: {len(sites)}" + (f" (project={args.project})" if args.project else "") + (f" (limited to {args.wiki_limit})" if args.wiki_limit else ""), file=sys.stderr)
    print(f"Groups to scan: {', '.join(groups)}", file=sys.stderr)

    results: List[Dict[str, object]] = []
    completed_count = 0
    total_count = len(sites)
    
    def progress_callback(wiki_name: str, status: str):
        nonlocal completed_count
        completed_count += 1
        print(f"[{completed_count}/{total_count}] {wiki_name}: {status}", file=sys.stderr)
    
    with ThreadPoolExecutor(max_workers=max(1, args.workers)) as ex:
        futures = []
        for s in sites:
            futures.append(ex.submit(process_wiki, s, groups, progress_callback))
            time.sleep(BETWEEN_TASKS_SEC)
        
        for fut in as_completed(futures):
            results.append(fut.result())

    results.sort(key=lambda r: str(r.get("wiki")))
    preview_limit = None if args.full else (args.limit or 20)
    print()
    print(format_table(results, limit=preview_limit))

    # Compute totals and deduplicated totals
    totals = {g: sum(int(r.get(g, 0) or 0) for r in results) for g in groups}
    unique_totals = {}
    for g in groups:
        seen: Set[str] = set()
        for r in results:
            seen.update(r.get(f"{g}_members", []))
        unique_totals[g] = len(seen)

    print("\n=== Totals across all scanned wikis (sum of per-wiki counts) ===")
    print(", ".join(f"{g}={totals[g]}" for g in groups))
    print("=== Totals across all scanned wikis (unique usernames) ===")
    print(", ".join(f"{g}={unique_totals[g]}" for g in groups))

    if args.csv:
        # Get all possible headers from results
        all_headers = set()
        for r in results:
            all_headers.update(r.keys())
        headers = ["wiki", "project", *groups, "error"]
        # Add any additional headers that might exist
        for h in sorted(all_headers):
            if h not in headers and not h.endswith("_members"):
                headers.append(h)
        
        with open(args.csv, "w", newline="", encoding="utf-8") as f:
            w = csv.DictWriter(f, fieldnames=headers)
            w.writeheader()
            for r in results:
                w.writerow({h: r.get(h, "") for h in headers})
        print(f"\nWrote CSV to {args.csv}")

if __name__ == "__main__":
    try:
        main()
    except KeyboardInterrupt:
        print("\nInterrupted.", file=sys.stderr)
        sys.exit(130)