Wikimedia projects stats

wiki-user-group-count.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.
# -----------------------------------------------------------------------------

"""
wiki-user-group-count.py — stdlib-only

Usage:
python3 wiki-user-group-count.py
python3 wiki-user-group-count.py --projects frwiki,enwiki  # only specific projects
python3 wiki-user-group-count.py --output-limit 50         # preview 50 rows
python3 wiki-user-group-count.py --format csv              # output as CSV
python3 wiki-user-group-count.py --groups sysop,checkuser  # only specific groups
python3 wiki-user-group-count.py --limit 10                # only scan first 10 wikis

groups format:
  - sysop (Administrators)
  - checkuser (CheckUsers)
  - suppress (Oversighters)
  - steward (Stewards; global, mostly on Meta-Wiki)
  - rollbacker (Rollbackers)

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 datetime import datetime
from concurrent.futures import ThreadPoolExecutor, as_completed
from typing import Dict, List, Optional, Any
from urllib.parse import urlencode, urlparse, urlunparse, ParseResult
from urllib.request import urlopen, Request
from urllib.error import URLError, HTTPError

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

# API Politeness and Rate Limiting Configuration
# Based on Wikimedia's API etiquette guidelines: https://www.mediawiki.org/wiki/API:Etiquette
# 
# Key principles:
# - "There is no hard speed limit on read requests, but be considerate"
# - "Try not to take a site down"
# - "Making requests in series rather than parallel should result in a safe request rate"
# - "Most system administrators reserve the right to block you if you endanger site stability"
#
# Our approach:
# - Use efficient usergroups API (single call per wiki instead of pagination)
# - Implement polite pauses between requests
# - Use batch processing with pauses between batches
# - Respect server resources while maintaining good performance

MAX_WORKERS = 20  # Parallel workers for processing wikis
BETWEEN_TASKS_SEC = 0.05  # Pause between parallel tasks
RETRIES = 3
RETRY_BACKOFF_SEC = 0.3  # Exponential backoff for retries (respects rate limits)

def fetch_json(url: str, params: Optional[Dict[str, str]] = None, tries: int = RETRIES, timeout: int = 30) -> Dict:
    """HTTP GET → JSON (stdlib only), with retries and polite error handling.
    
    Implements Wikimedia's API etiquette guidelines:
    - Sets informative User-Agent string (required by Wikimedia policy)
    - Uses exponential backoff for retries (respects rate limits)
    - Handles rate limiting gracefully
    """
    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:
            # Set informative User-Agent as required by Wikimedia's User-Agent policy
            req = Request(url, headers={"User-Agent": "wmf-group-counter/stdlib/1.1"})
            with urlopen(req, timeout=timeout) as resp:
                return json.loads(resp.read().decode("utf-8"))
        except (HTTPError, URLError, TimeoutError) as e:
            last_err = e
            if attempt < tries:
                # Exponential backoff: respect rate limits and be polite
                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[List[str]], 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
            
            # Project filtering using dbname - support multiple projects
            if project_filter and dbname.lower() not in project_filter:
                continue
            
            api = url.rstrip("/") + "/w/api.php"
            sites.append({"dbname": dbname, "api": api, "project": project, "lang": code})
            
            # Apply limit if specified
            if limit and len(sites) >= limit:
                return sites
    return sites

def fetch_group_counts_fast(api: str, groups: List[str]) -> Dict[str, int]:
    """Fetch group counts using the efficient usergroups API endpoint."""
    params = {
        "action": "query",
        "format": "json",
        "meta": "siteinfo",
        "siprop": "usergroups",
        "sinumberingroup": "1"
    }
    
    try:
        data = fetch_json(api, params, timeout=30)
        usergroups = data.get("query", {}).get("usergroups", [])
        
        # Create a mapping of group names to counts
        group_counts = {}
        for group_info in usergroups:
            group_name = group_info.get("name")
            group_count = group_info.get("number", 0)
            if group_name in groups:
                group_counts[group_name] = group_count
        
        return group_counts
        
    except Exception as e:
        print(f"Error fetching group counts from {api}: {e}", file=sys.stderr)
        return {}

def process_wiki(site: Dict[str, str], groups: List[str], progress_callback=None) -> Dict[str, Any]:
    """Process a single wiki and return results."""
    api = site["api"]
    dbname = site["dbname"]
    
    # Use the fast method to get group counts
    group_counts = fetch_group_counts_fast(api, groups)
    
    # Build result row
    row = {
        "wiki": dbname,
        "project": site.get("project", "Unknown"),
        "error": ""
    }
    
    # Add group counts
    for group in groups:
        row[group] = group_counts.get(group, 0)
    
    if progress_callback:
        progress_callback(dbname, "completed")
    
    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)}; raise --output-limit to see more)")
    return "\n".join(out)

def main():
    """Main function implementing Wikimedia API etiquette.
    
    This script follows Wikimedia's API usage guidelines:
    - Uses efficient usergroups API to minimize server load
    - Implements polite pauses and batch processing
    - Respects rate limits with exponential backoff
    - Sets proper User-Agent headers
    - Processes requests in manageable chunks
    
    See: https://www.mediawiki.org/wiki/API:Etiquette
    """
    ap = argparse.ArgumentParser(description="Count WMF per-wiki group memberships with deduplicated totals (stdlib-only).")
    ap.add_argument("--projects", help="Comma-separated list of projects to include (e.g., 'frwiki,enwiki,dewiki').")
    ap.add_argument("--output-limit", type=int, help="Limit number of printed rows.")
    ap.add_argument("--format", choices=["table", "csv"], default="table", help="Output format (default: table).")
    ap.add_argument("--groups", help="Comma-separated list of groups to scan (default: sysop,checkuser,rollbacker).")
    ap.add_argument("--limit", type=int, help="Limit number of wikis to scan (useful for testing).")
    args = ap.parse_args()

    # Start timing
    start_time = time.time()

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

    # Parse projects argument
    projects = None
    if args.projects:
        projects = [p.strip().lower() for p in args.projects.split(",")]

    print("Fetching SiteMatrix…", file=sys.stderr)
    sm = fetch_json(SITEMATRIX_URL)
    sites = extract_wikis_from_sitematrix(sm, projects, args.limit)
    print(f"Wikis to scan: {len(sites)}" + (f" (projects={args.projects})" if projects else "") + (f" (limited to {args.limit})" if args.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
        if completed_count % 10 == 0 or completed_count == total_count:  # Show every 10th wiki and the last one
            print(f"[{completed_count}/{total_count}] {wiki_name}: {status}", file=sys.stderr)
    
    # Process wikis with polite concurrency
    # This approach respects Wikimedia's API etiquette by:
    # - Using parallel processing for efficiency
    # - Maintaining polite request rates
    # - Processing in manageable chunks
    batch_size = 25  # Process 25 wikis at a time
    batch_pause = 2  # Pause 2 seconds between batches
    
    for batch_start in range(0, len(sites), batch_size):
        batch_end = min(batch_start + batch_size, len(sites))
        batch_sites = sites[batch_start:batch_end]
        
        with ThreadPoolExecutor(max_workers=MAX_WORKERS) as ex:
            futures = []
            for s in batch_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())
        
        # Pause between batches to respect server resources
        if batch_end < len(sites):
            time.sleep(batch_pause)

    results.sort(key=lambda r: str(r.get("wiki")))
    print()
    if args.format == "table":
        print(format_table(results, limit=args.output_limit))
    else:
        # CSV output
        # 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)
        
        w = csv.DictWriter(sys.stdout, fieldnames=headers)
        w.writeheader()
        for r in results:
            w.writerow({h: r.get(h, "") for h in headers})

    # Compute totals (sum of per-wiki counts)
    print()
    print("=== Totals (sum of per-wiki counts) ===")
    totals = {}
    for g in groups:
        total = sum(int(r.get(g, 0) or 0) for r in results)
        totals[g] = total
    print(", ".join(f"{g}={totals[g]}" for g in groups))

    # Add timestamp at the end
    print(f"\nStats generated: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")

    # End timing
    end_time = time.time()
    elapsed_time = end_time - start_time
    
    # Calculate performance statistics
    wikis_per_second = len(sites) / elapsed_time if elapsed_time > 0 else 0
    total_api_calls = len(sites) * len(groups)
    api_calls_per_second = total_api_calls / elapsed_time if elapsed_time > 0 else 0
    
    print(f"\nTotal execution time: {elapsed_time:.2f} seconds")
    print(f"Performance: {wikis_per_second:.1f} wikis/second, {api_calls_per_second:.1f} API calls/second")
    print(f"Processed {len(sites)} wikis with {len(groups)} groups each ({total_api_calls} total API calls)")

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

Usage:

python3 wiki-user-group-count.py
python3 wiki-user-group-count.py --projects frwiki,enwiki  # only specific projects
python3 wiki-user-group-count.py --output-limit 50         # preview 50 rows
python3 wiki-user-group-count.py --format csv              # output as CSV
python3 wiki-user-group-count.py --groups sysop,checkuser  # only specific groups
python3 wiki-user-group-count.py --limit 10                # only scan first 10 wikis

Test:

python3 wiki-user-group-count.py --groups sysop,steward,checkuser,rollbacker

=== Per-wiki counts ===
wiki                       project                                         sysop  steward  checkuser  rollbacker  error
-------------------------  ----------------------------------------------  -----  -------  ---------  ----------  -----
aawiki                     Wikipedia                                       1      0        0          0                
aawikibooks                Wikibooks                                       1      0        0          0                
aawiktionary               Wiktionary                                      1      0        0          0                
abwiki                     Авикипедиа                                      2      0        0          0                
abwiktionary               Wiktionary                                      1      0        0          0                
acewiki                    Wikipedia                                       3      0        0          0                
advisorswiki               Advisors                                        0      0        0          0                
advisorywiki               Advisory Board                                  1      0        0          0                
.......
zhwiki                     Wikipedia                                       62     0        0          198              
zhwikibooks                Wikibooks                                       8      0        0          0                
zhwikinews                 Wikinews                                        7      0        0          4                
zhwikiquote                Wikiquote                                       5      0        0          0                
zhwikisource               Wikisource                                      8      0        0          0                
zhwiktionary               Wiktionary                                      9      0        0          0                
zuwiki                     Wikipedia                                       1      0        0          0                
zuwikibooks                Wikibooks                                       1      0        0          0                
zuwiktionary               Wiktionary                                      1      0        0          0                

=== Totals (sum of per-wiki counts) ===
sysop=6496, steward=34, checkuser=238, rollbacker=15088

Note: Totals are sum of per-wiki counts (not unique usernames). Getting unique is a much slower process not supported by the script.

Stats generated: 2025-08-15 19:03:32
Total execution time: 155.97 seconds