compounding-intelligence/scripts/update_checker.py

#!/usr/bin/env python3
"""
5.3: Update Checker — Compare installed vs latest package versions

Check if dependencies have newer versions available. Query PyPI for each
installed package, compare versions, and generate an update report with
major/minor/patch classification and breaking change flags.

Usage:
    python3 scripts/update_checker.py
    python3 scripts/update_checker.py --json
    python3 scripts/update_checker.py --output updates.md
    python3 scripts/update_checker.py --package requests,pytest
"""

import argparse
import json
import subprocess
import sys
from datetime import datetime
from pathlib import Path
from typing import Dict, List, Optional, Tuple
from urllib.request import urlopen
from urllib.error import URLError, HTTPError


def get_installed_packages() -> Dict[str, str]:
    """Get all installed packages via pip list --format=json."""
    try:
        result = subprocess.run(
            ['pip', 'list', '--format=json'],
            capture_output=True, text=True, timeout=30
        )
        if result.returncode != 0:
            print(f"Warning: pip list failed: {result.stderr}", file=sys.stderr)
            return {}
        packages = json.loads(result.stdout)
        return {p['name'].lower(): p['version'] for p in packages}
    except (json.JSONDecodeError, subprocess.TimeoutExpired, KeyError) as e:
        print(f"Warning: failed to parse pip list: {e}", file=sys.stderr)
        return {}


def get_latest_version(package_name: str) -> Optional[str]:
    """Query PyPI JSON API for the latest version of a package."""
    url = f"https://pypi.org/pypi/{package_name}/json"
    try:
        with urlopen(url, timeout=10) as resp:
            if resp.status == 200:
                data = json.loads(resp.read())
                return data.get('info', {}).get('version')
    except (URLError, HTTPError, json.JSONDecodeError, TimeoutError):
        pass
    return None


def parse_version(version_str: str) -> Tuple[int, int, int]:
    """Parse semantic version string into (major, minor, patch)."""
    # Strip any extras like dev, post, rc
    cleaned = version_str.split('.')[0:3]
    # Pad to 3 parts
    while len(cleaned) < 3:
        cleaned.append('0')
    try:
        major = int(cleaned[0]) if cleaned[0].isdigit() else 0
        minor = int(cleaned[1]) if len(cleaned) > 1 and cleaned[1].isdigit() else 0
        patch = int(cleaned[2]) if len(cleaned) > 2 and cleaned[2].isdigit() else 0
        return (major, minor, patch)
    except (ValueError, IndexError):
        return (0, 0, 0)


def classify_update(installed: str, latest: str) -> Optional[Dict]:
    """Determine update type between installed and latest versions."""
    if not latest:
        return None
    
    inst_ver = parse_version(installed)
    latest_ver = parse_version(latest)
    
    if inst_ver == latest_ver:
        return None  # Already up to date
    
    # Calculate delta
    major_diff = latest_ver[0] - inst_ver[0]
    minor_diff = latest_ver[1] - inst_ver[1]
    patch_diff = latest_ver[2] - inst_ver[2]
    
    # Determine update type
    if major_diff > 0:
        update_type = 'major'
        breaking = True
        severity = 'high'
    elif minor_diff > 0:
        update_type = 'minor'
        breaking = False
        severity = 'medium'
    elif patch_diff > 0:
        update_type = 'patch'
        breaking = False
        severity = 'low'
    else:
        # Shouldn't happen but handle weird cases
        return None
    
    return {
        'package': None,  # filled by caller
        'installed': installed,
        'latest': latest,
        'update_type': update_type,
        'breaking_change': breaking,
        'severity': severity,
    }


def check_updates(packages: Dict[str, str] = None,
                  filter_packages: List[str] = None) -> List[Dict]:
    """
    Check all installed packages (or filtered subset) for updates.
    
    Args:
        packages: Dict of {name: version}. If None, queries pip list.
        filter_packages: Optional list of package names to check only.
    
    Returns:
        List of update report dicts sorted by severity.
    """
    if packages is None:
        packages = get_installed_packages()
    
    if filter_packages:
        packages = {k: v for k, v in packages.items() 
                    if k.lower() in [p.lower() for p in filter_packages]}
    
    updates = []
    print(f"Checking {len(packages)} packages...", file=sys.stderr)
    
    for pkg_name, installed_ver in packages.items():
        latest_ver = get_latest_version(pkg_name)
        if not latest_ver:
            continue
        
        update_info = classify_update(installed_ver, latest_ver)
        if update_info:
            update_info['package'] = pkg_name
            updates.append(update_info)
    
    # Sort: breaking first, then severity, then package name
    updates.sort(key=lambda u: (
        -1 if u['breaking_change'] else 0,
        {'high': 0, 'medium': 1, 'low': 2}[u['severity']],
        u['package']
    ))
    
    return updates


def generate_text_report(updates: List[Dict]) -> str:
    """Generate human-readable text report."""
    lines = []
    lines.append("=" * 60)
    lines.append("DEPENDENCY UPDATE REPORT")
    lines.append(f"Generated: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
    lines.append("=" * 60)
    lines.append("")
    
    if not updates:
        lines.append("✓ All packages are up to date.")
        return "\n".join(lines)
    
    lines.append(f"Found {len(updates)} package(s) with available updates:")
    lines.append("")
    
    for u in updates:
        breaking_marker = " ⚠ BREAKING CHANGE" if u['breaking_change'] else ""
        lines.append(f"  {u['package']}:")
        lines.append(f"    Installed: {u['installed']}")
        lines.append(f"    Latest:    {u['latest']}")
        lines.append(f"    Update:    {u['update_type'].upper()}{breaking_marker}")
        lines.append(f"    Severity:  {u['severity'].upper()}")
        lines.append("")
    
    lines.append("=" * 60)
    lines.append("Recommendation: Review breaking changes carefully before upgrading.")
    lines.append("Consider pinning versions or using a virtual environment.")
    return "\n".join(lines)


def generate_json_report(updates: List[Dict]) -> str:
    """Generate JSON report compatible with machine consumption."""
    report = {
        'generated_at': datetime.now().isoformat(),
        'total_updates': len(updates),
        'updates': updates,
        'summary': {
            'major': sum(1 for u in updates if u['update_type'] == 'major'),
            'minor': sum(1 for u in updates if u['update_type'] == 'minor'),
            'patch': sum(1 for u in updates if u['update_type'] == 'patch'),
            'breaking': sum(1 for u in updates if u['breaking_change']),
        }
    }
    return json.dumps(report, indent=2)


def main():
    parser = argparse.ArgumentParser(
        description="Check dependencies for available updates"
    )
    parser.add_argument(
        '--json', action='store_true',
        help='Output JSON report for machine consumption'
    )
    parser.add_argument(
        '--output', '-o', type=str,
        help='Write report to file instead of stdout'
    )
    parser.add_argument(
        '--package', '-p', type=str,
        help='Comma-separated list of specific packages to check'
    )
    args = parser.parse_args()

    # Build filter list if provided
    filter_list = None
    if args.package:
        filter_list = [p.strip() for p in args.package.split(',') if p.strip()]

    # Run checks
    updates = check_updates(filter_packages=filter_list)

    # Generate report
    if args.json:
        report = generate_json_report(updates)
    else:
        report = generate_text_report(updates)

    # Output
    if args.output:
        Path(args.output).write_text(report)
        print(f"Report written to {args.output}", file=sys.stderr)
    else:
        print(report)


if __name__ == '__main__':
    main()