the-nexus/bin/check_duplicate_milestones.py

#!/usr/bin/env python3
"""
Check for duplicate milestones across repositories.
Part of Issue #1127 implementation.
"""

import json
import os
import sys
import urllib.request
from typing import Dict, List, Any, Optional
from collections import Counter

# Configuration
GITEA_BASE = "https://forge.alexanderwhitestone.com/api/v1"
TOKEN_PATH = os.path.expanduser("~/.config/gitea/token")


class MilestoneChecker:
    def __init__(self):
        self.token = self._load_token()
        self.org = "Timmy_Foundation"
        
    def _load_token(self) -> str:
        """Load Gitea API token."""
        try:
            with open(TOKEN_PATH, "r") as f:
                return f.read().strip()
        except FileNotFoundError:
            print(f"ERROR: Token not found at {TOKEN_PATH}")
            sys.exit(1)
    
    def _api_request(self, endpoint: str) -> Any:
        """Make authenticated Gitea API request."""
        url = f"{GITEA_BASE}{endpoint}"
        headers = {"Authorization": f"token {self.token}"}
        
        req = urllib.request.Request(url, headers=headers)
        
        try:
            with urllib.request.urlopen(req) as resp:
                return json.loads(resp.read())
        except urllib.error.HTTPError as e:
            if e.code == 404:
                return []
            error_body = e.read().decode() if e.fp else "No error body"
            print(f"API Error {e.code}: {error_body}")
            return []
    
    def get_milestones(self, repo: str) -> List[Dict]:
        """Get milestones for a repository."""
        endpoint = f"/repos/{self.org}/{repo}/milestones?state=all"
        return self._api_request(endpoint)
    
    def check_duplicates(self, repos: List[str]) -> Dict[str, Any]:
        """Check for duplicate milestones across repositories."""
        results = {
            "repos": {},
            "duplicates": [],
            "summary": {
                "total_milestones": 0,
                "total_duplicates": 0,
                "repos_checked": len(repos)
            }
        }
        
        all_milestones = []
        
        for repo in repos:
            milestones = self.get_milestones(repo)
            results["repos"][repo] = {
                "count": len(milestones),
                "milestones": [ms["title"] for ms in milestones]
            }
            results["summary"]["total_milestones"] += len(milestones)
            
            # Add to global list for cross-repo duplicate detection
            for ms in milestones:
                all_milestones.append({
                    "repo": repo,
                    "id": ms["id"],
                    "title": ms["title"],
                    "state": ms["state"],
                    "description": ms.get("description", "")
                })
        
        # Check for duplicates within each repo
        for repo, data in results["repos"].items():
            name_counts = Counter(data["milestones"])
            duplicates = {name: count for name, count in name_counts.items() if count > 1}
            
            if duplicates:
                results["duplicates"].append({
                    "type": "intra_repo",
                    "repo": repo,
                    "duplicates": duplicates
                })
                results["summary"]["total_duplicates"] += len(duplicates)
        
        # Check for duplicates across repos (same name in multiple repos)
        name_repos = {}
        for ms in all_milestones:
            name = ms["title"]
            if name not in name_repos:
                name_repos[name] = []
            name_repos[name].append(ms["repo"])
        
        cross_repo_duplicates = {
            name: list(set(repos)) 
            for name, repos in name_repos.items() 
            if len(set(repos)) > 1
        }
        
        if cross_repo_duplicates:
            results["duplicates"].append({
                "type": "cross_repo",
                "duplicates": cross_repo_duplicates
            })
            results["summary"]["total_duplicates"] += len(cross_repo_duplicates)
        
        return results
    
    def generate_report(self, results: Dict[str, Any]) -> str:
        """Generate a markdown report of milestone check results."""
        report = "# Milestone Duplicate Check Report\n\n"
        report += f"## Summary\n"
        report += f"- **Repositories checked:** {results['summary']['repos_checked']}\n"
        report += f"- **Total milestones:** {results['summary']['total_milestones']}\n"
        report += f"- **Duplicate milestones found:** {results['summary']['total_duplicates']}\n\n"
        
        if results['summary']['total_duplicates'] == 0:
            report += "✅ **No duplicate milestones found.**\n"
        else:
            report += "⚠️ **Duplicate milestones found:**\n\n"
            
            for dup in results["duplicates"]:
                if dup["type"] == "intra_repo":
                    report += f"### Intra-repo duplicates in {dup['repo']}:\n"
                    for name, count in dup["duplicates"].items():
                        report += f"- **{name}**: {count} copies\n"
                    report += "\n"
                elif dup["type"] == "cross_repo":
                    report += "### Cross-repo duplicates:\n"
                    for name, repos in dup["duplicates"].items():
                        report += f"- **{name}**: exists in {', '.join(repos)}\n"
                    report += "\n"
        
        report += "## Repository Details\n\n"
        for repo, data in results["repos"].items():
            report += f"### {repo}\n"
            report += f"- **Milestones:** {data['count']}\n"
            if data['count'] > 0:
                report += "- **Names:**\n"
                for name in data["milestones"]:
                    report += f"  - {name}\n"
            report += "\n"
        
        return report


def main():
    """Main entry point for milestone checker."""
    import argparse
    
    parser = argparse.ArgumentParser(description="Check for duplicate milestones")
    parser.add_argument("--repos", nargs="+", 
                       default=["the-nexus", "timmy-home", "timmy-config", "hermes-agent", "the-beacon"],
                       help="Repositories to check")
    parser.add_argument("--report", action="store_true", help="Generate report")
    parser.add_argument("--json", action="store_true", help="Output JSON instead of report")
    
    args = parser.parse_args()
    
    checker = MilestoneChecker()
    results = checker.check_duplicates(args.repos)
    
    if args.json:
        print(json.dumps(results, indent=2))
    elif args.report:
        report = checker.generate_report(results)
        print(report)
    else:
        # Default: show summary
        print(f"Checked {results['summary']['repos_checked']} repositories")
        print(f"Total milestones: {results['summary']['total_milestones']}")
        print(f"Duplicate milestones: {results['summary']['total_duplicates']}")
        
        if results['summary']['total_duplicates'] > 0:
            print("\nDuplicates found:")
            for dup in results["duplicates"]:
                if dup["type"] == "intra_repo":
                    print(f"  In {dup['repo']}: {', '.join(dup['duplicates'].keys())}")
                elif dup["type"] == "cross_repo":
                    for name, repos in dup["duplicates"].items():
                        print(f"  '{name}' in: {', '.join(repos)}")
            sys.exit(1)
        else:
            print("\n✅ No duplicate milestones found")
            sys.exit(0)


if __name__ == "__main__":
    main()