code_stats.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

import argparse
import datetime
import requests
import sys
import time
from typing import Dict, List, Tuple
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
from requests.exceptions import SSLError, ConnectionError, Timeout


class CodeStats:
    def __init__(self, token: str = ""):
        self.headers = {"Accept": "application/vnd.github.v3+json"}
        if token:
            self.headers["Authorization"] = f"token {token}"

        self.base_url = "https://api.github.com"
        self.log_file = "log.md"

        self.session = self._create_retry_session()
        self.session.headers.update(self.headers)

        with open(self.log_file, 'w', encoding='utf-8') as f:
            f.write(f"# Code Statistics Log\n\n")
            f.write(f"Generated at: {datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n\n")

    def _create_retry_session(self):
        session = requests.Session()
        retry_strategy = Retry(
            total=8,
            backoff_factor=2,
            status_forcelist=[429, 500, 502, 503, 504],
            allowed_methods=["GET"],
            respect_retry_after_header=True,
        )
        adapter = HTTPAdapter(max_retries=retry_strategy)
        session.mount("https://", adapter)
        return session

    def log(self, message: str):
        print(message)
        with open(self.log_file, 'a', encoding='utf-8') as f:
            f.write(f"{message}\n")

    def handle_rate_limit(self, response) -> bool:
        if response.status_code == 403:
            remaining = int(response.headers.get("X-RateLimit-Remaining", 0))
            if remaining == 0:
                reset_time = int(response.headers.get("X-RateLimit-Reset", 0))
                wait_time = max(0, reset_time - int(time.time())) + 5
                self.log(f"Rate limit exhausted. Waiting {wait_time} seconds...")
                time.sleep(wait_time)
                return True
        return False

    def get_repos_from_file(self, file_path: str) -> List[str]:
        repos = []
        try:
            with open(file_path, 'r', encoding='utf-8') as f:
                for line in f:
                    repo = line.strip()
                    if repo and not repo.startswith('#'):
                        repos.append(repo)
        except Exception as e:
            self.log(f"Failed to read repository file: {str(e)}")
        return repos

    def get_merged_prs(self, owner: str, repo: str, start_date: str, end_date: str) -> int:
        count = 0
        page = 1

        while True:
            url = f"{self.base_url}/repos/{owner}/{repo}/pulls"
            params = {
                "state": "closed",
                "sort": "created",
                "direction": "desc",
                "per_page": 100,
                "page": page,
            }

            try:
                response = self.session.get(url, params=params, timeout=(15, 45))

                if response.status_code == 200:
                    prs = response.json()
                    if not prs:
                        break

                    should_stop = False
                    for pr in prs:
                        if pr.get("merged_at"):
                            merged_date = pr["merged_at"].split("T")[0]
                            if start_date <= merged_date <= end_date:
                                count += 1
                            elif merged_date < start_date:
                                should_stop = True
                                break

                    if should_stop:
                        break
                    page += 1

                elif self.handle_rate_limit(response):
                    continue
                else:
                    self.log(f"API Error fetching PRs: {response.status_code} | {owner}/{repo}")
                    break

            except (SSLError, ConnectionError, Timeout) as e:
                self.log(f"Network/SSL Error for PRs: {str(e)} | {owner}/{repo}")
                time.sleep(5)
                continue
            except Exception as e:
                self.log(f"Unexpected error fetching PRs: {str(e)} | {owner}/{repo}")
                break

        return count

    def get_closed_issues(self, owner: str, repo: str, start_date: str, end_date: str) -> int:
        count = 0
        page = 1

        while True:
            url = f"{self.base_url}/repos/{owner}/{repo}/issues"
            params = {
                "state": "closed",
                "sort": "created",
                "direction": "desc",
                "per_page": 100,
                "page": page,
            }

            try:
                response = self.session.get(url, params=params, timeout=(15, 45))

                if response.status_code == 200:
                    issues = response.json()
                    if not issues:
                        break

                    should_stop = False
                    for issue in issues:
                        if "pull_request" in issue:
                            continue
                        if issue.get("closed_at"):
                            closed_date = issue["closed_at"].split("T")[0]
                            if start_date <= closed_date <= end_date:
                                count += 1
                            elif closed_date < start_date:
                                should_stop = True
                                break

                    if should_stop:
                        break
                    page += 1

                elif self.handle_rate_limit(response):
                    continue
                else:
                    self.log(f"API Error fetching Issues: {response.status_code} | {owner}/{repo}")
                    break

            except (SSLError, ConnectionError, Timeout) as e:
                self.log(f"Network/SSL Error for Issues: {str(e)} | {owner}/{repo}")
                time.sleep(5)
                continue
            except Exception as e:
                self.log(f"Unexpected error fetching Issues: {str(e)} | {owner}/{repo}")
                break

        return count

    def calculate_stats(self, start_date: str, end_date: str, repo_file: str) -> Tuple[Dict, int, int]:
        repo_stats: Dict[str, Dict[str, int]] = {}
        total_prs = 0
        total_issues = 0

        self.log(f"Starting statistics: {start_date} → {end_date}")
        self.log("=" * 80)

        repos = self.get_repos_from_file(repo_file)
        total_repos = len(repos)
        self.log(f"Loaded {total_repos} repositories from: {repo_file}\n")

        for idx, full_name in enumerate(repos, 1):
            self.log(f"Processing {idx}/{total_repos}: {full_name}")
            time.sleep(1)
            try:
                owner, repo_name = full_name.split("/")
                pr_count = self.get_merged_prs(owner, repo_name, start_date, end_date)
                issue_count = self.get_closed_issues(owner, repo_name, start_date, end_date)

                repo_stats[full_name] = {"pr_count": pr_count, "issue_count": issue_count}
                total_prs += pr_count
                total_issues += issue_count

                self.log(f"  → {pr_count} PRs | {issue_count} Issues")
            except Exception as e:
                self.log(f"  → Failed to process: {str(e)}")

            time.sleep(0.8)
            self.log("")

        return repo_stats, total_prs, total_issues

    def print_stats(self, start_date: str, end_date: str, repo_file: str):
        repo_stats, total_prs, total_issues = self.calculate_stats(start_date, end_date, repo_file)

        self.log("=" * 80)
        self.log("FINAL STATISTICS SUMMARY")
        self.log("=" * 80)
        self.log(f"Date Range: {start_date} to {end_date}")
        self.log(f"Repositories Scanned: {len(repo_stats)}")
        self.log(f"Total Merged PRs: {total_prs}")
        self.log(f"Total Closed Issues: {total_issues}")
        self.log("=" * 80)

        self.log("\nTop 10 Repositories (Merged PRs):")
        top_prs = sorted(repo_stats.items(), key=lambda x: x[1]["pr_count"], reverse=True)[:10]
        for repo, stats in top_prs:
            self.log(f"  {repo}: {stats['pr_count']} PRs")

        self.log("\nTop 10 Repositories (Closed Issues):")
        top_issues = sorted(repo_stats.items(), key=lambda x: x[1]["issue_count"], reverse=True)[:10]
        for repo, stats in top_issues:
            self.log(f"  {repo}: {stats['issue_count']} Issues")

        self.log("\n" + "=" * 80)
        self.log("STATISTICS COMPLETED")
        self.log("=" * 80)


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="GitHub Repository Statistics Tool")
    parser.add_argument("--start-date", required=True, help="Start date (YYYY-MM-DD)")
    parser.add_argument("--end-date", required=True, help="End date (YYYY-MM-DD)")
    parser.add_argument("--token", default="", help="GitHub API token for higher rate limits")
    parser.add_argument("--repo-file", default="asf/repos.txt", help="Path to repository list file")
    args = parser.parse_args()

    try:
        datetime.datetime.strptime(args.start_date, "%Y-%m-%d")
        datetime.datetime.strptime(args.end_date, "%Y-%m-%d")
    except ValueError:
        print("Error: Invalid date format! Use YYYY-MM-DD", file=sys.stderr)
        sys.exit(1)

    if args.start_date > args.end_date:
        print("Error: Start date must be earlier than end date", file=sys.stderr)
        sys.exit(1)

    stats = CodeStats(args.token)
    stats.print_stats(args.start_date, args.end_date, args.repo_file)