Audit BMLT servers to find meetings with orphaned service body references

audit_meetings.py

#!/usr/bin/env python3
"""
Audit BMLT servers to find meetings with orphaned service body references.

This script checks all servers in serverList.json and reports meetings that
reference service body IDs that no longer exist in the server's service body list.

Usage:
    python3 audit_meetings.py              # Audit all servers
    python3 audit_meetings.py 102          # Audit only server with id=102
"""

import json
import sys
from typing import List, Dict, Set, Optional
from urllib.request import urlopen, Request
from urllib.error import URLError, HTTPError
import time


def load_server_list(use_github: bool = True) -> List[Dict]:
    """Load the list of BMLT servers from GitHub or local file."""
    if use_github:
        github_url = "https://raw.githubusercontent.com/bmlt-enabled/aggregator/refs/heads/main/serverList.json"
        try:
            servers = fetch_json(github_url)
            if servers:
                return servers
            print("⚠️  Failed to fetch from GitHub, falling back to local file", file=sys.stderr)
            return None
        except Exception as e:
            print(f"⚠️  Error fetching from GitHub: {e}, falling back to local file", file=sys.stderr)
            return None
    return None


def fetch_json(url: str, timeout: int = 30) -> List[Dict]:
    """Fetch JSON data from a URL with error handling."""
    try:
        req = Request(url)
        req.add_header('User-Agent', 'Mozilla/5.0 (X11; Linux x86_64; rv:146.0) Gecko/20100101 Firefox/146.0')
        with urlopen(req, timeout=timeout) as response:
            data = json.loads(response.read().decode('utf-8'))
            # Ensure we got a list, not a dict or other type
            if not isinstance(data, list):
                print(f"  ⚠️  Unexpected response type: {type(data)}", file=sys.stderr)
                return []
            return data
    except HTTPError as e:
        print(f"  ⚠️  HTTP Error {e.code}: {e.reason}", file=sys.stderr)
        return []
    except URLError as e:
        print(f"  ⚠️  URL Error: {e.reason}", file=sys.stderr)
        return []
    except json.JSONDecodeError as e:
        print(f"  ⚠️  JSON decode error: {e}", file=sys.stderr)
        return []
    except Exception as e:
        print(f"  ⚠️  Error: {str(e)}", file=sys.stderr)
        return []


def get_service_bodies(server_url: str) -> Set[str]:
    """Get all service body IDs from a BMLT server."""
    url = f"{server_url}client_interface/json/?switcher=GetServiceBodies"
    service_bodies = fetch_json(url)
    return {sb['id'] for sb in service_bodies if 'id' in sb}


def get_meetings(server_url: str) -> List[Dict]:
    """Get all meetings (including unpublished) with only id and service_body_bigint from a BMLT server."""
    url = f"{server_url}client_interface/json/?switcher=GetSearchResults&data_field_key=id_bigint,service_body_bigint&advanced_published=0"
    return fetch_json(url)


def audit_server(server: Dict) -> Dict:
    """Audit a single BMLT server for orphaned meeting references."""
    server_name = server['name']
    server_url = server['url']
    
    print(f"\n{'='*80}")
    print(f"Auditing: {server_name}")
    print(f"URL: {server_url}")
    print(f"{'='*80}")
    
    # Fetch service bodies
    print("  Fetching service bodies...", end=" ", flush=True)
    valid_service_body_ids = get_service_bodies(server_url)
    
    if not valid_service_body_ids:
        print("❌ No service bodies found or error occurred")
        return {
            'server': server_name,
            'url': server_url,
            'error': 'Failed to fetch service bodies',
            'orphaned_meetings': []
        }
    
    print(f"✓ Found {len(valid_service_body_ids)} service bodies")
    
    # Fetch meetings
    print("  Fetching meetings...", end=" ", flush=True)
    meetings = get_meetings(server_url)
    
    if not meetings:
        print("❌ No meetings found or error occurred")
        return {
            'server': server_name,
            'url': server_url,
            'error': 'Failed to fetch meetings',
            'orphaned_meetings': []
        }
    
    print(f"✓ Found {len(meetings)} meetings")
    
    # Check for orphaned service body references
    print("  Checking for orphaned meetings...", end=" ", flush=True)
    orphaned_meetings = []
    
    for meeting in meetings:
        service_body_id = meeting.get('service_body_bigint', '')
        
        if service_body_id and service_body_id not in valid_service_body_ids:
            orphaned_meetings.append({
                'id': meeting.get('id_bigint', 'unknown'),
                'service_body_id': service_body_id,
            })
    
    if orphaned_meetings:
        meeting_ids = [m['id'] for m in orphaned_meetings]
        print(f"⚠️  Found {len(orphaned_meetings)} orphaned meetings")
        print(f"     Meeting IDs: {', '.join(meeting_ids)}")
    else:
        print("✓ No orphaned meetings found")
    
    return {
        'server': server_name,
        'url': server_url,
        'total_service_bodies': len(valid_service_body_ids),
        'total_meetings': len(meetings),
        'orphaned_meetings': orphaned_meetings
    }


def print_report(results: List[Dict]):
    """Print a summary report of the audit."""
    print("\n" + "="*80)
    print("AUDIT SUMMARY")
    print("="*80)
    
    total_servers = len(results)
    servers_with_issues = sum(1 for r in results if r['orphaned_meetings'])
    total_orphaned = sum(len(r['orphaned_meetings']) for r in results)
    
    print(f"\nServers audited: {total_servers}")
    print(f"Servers with orphaned meetings: {servers_with_issues}")
    print(f"Total orphaned meetings: {total_orphaned}")
    
    if total_orphaned > 0:
        print("\n" + "="*80)
        print("DETAILED RESULTS")
        print("="*80)
        
        for result in results:
            if result['orphaned_meetings']:
                print(f"\n{result['server']} ({result['url']})")
                print(f"  {len(result['orphaned_meetings'])} orphaned meeting(s):")
                
                # Print comma-separated list of meeting IDs
                meeting_ids = [meeting['id'] for meeting in result['orphaned_meetings']]
                print(f"  Meeting IDs: {', '.join(meeting_ids)}")
                
                print("\n  Details:")
                for meeting in result['orphaned_meetings']:
                    print(f"    Meeting ID: {meeting['id']} → Invalid Service Body ID: {meeting['service_body_id']}")


def save_results(results: List[Dict], filepath: str = "audit_meetings_results.json"):
    """Save audit results to a JSON file."""
    with open(filepath, 'w') as f:
        json.dump(results, f, indent=2)
    print(f"\n✓ Results saved to {filepath}")


def main():
    """Main execution function."""
    print("BMLT Orphaned Meetings Audit")
    print("="*80)
    
    # Load server list
    try:
        all_servers = load_server_list()
    except FileNotFoundError:
        print("❌ Error: serverList.json not found", file=sys.stderr)
        sys.exit(1)
    except json.JSONDecodeError as e:
        print(f"❌ Error parsing serverList.json: {e}", file=sys.stderr)
        sys.exit(1)
    
    # Check if specific server ID was provided
    servers = all_servers
    if len(sys.argv) > 1:
        server_id = sys.argv[1]
        servers = [s for s in all_servers if s['id'] == server_id]
        if not servers:
            print(f"❌ Error: Server with id '{server_id}' not found in serverList.json", file=sys.stderr)
            print(f"Available server IDs: {', '.join([s['id'] for s in all_servers])}", file=sys.stderr)
            sys.exit(1)
        print(f"Auditing single server: {servers[0]['name']} (ID: {server_id})\n")
    else:
        print(f"Loaded {len(servers)} servers from serverList.json\n")
    
    # Audit each server
    results = []
    for i, server in enumerate(servers, 1):
        print(f"\n[{i}/{len(servers)}]", end=" ")
        result = audit_server(server)
        results.append(result)
        
        # Be respectful with rate limiting
        if i < len(servers):
            time.sleep(1)
    
    # Print and save results
    print_report(results)
    save_results(results)
    
    # Exit with appropriate code
    sys.exit(1 if any(r['orphaned_meetings'] for r in results) else 0)


if __name__ == "__main__":
    main()