github_api.py

"""GitHub API module for fetching commit history."""

import requests
from datetime import datetime, timezone
from typing import Optional
from loguru import logger


def get_commit_stats(username: str, year: int, token: Optional[str] = None) -> dict[str, int]:
    """
    获取指定用户某一年的contribution统计数据（使用GraphQL API）。

    Args:
        username: GitHub用户名
        year: 年份
        token: GitHub Personal Access Token (GraphQL API必需)

    Returns:
        字典，key为日期字符串(YYYY-MM-DD)，value为当天的contribution数量
    """
    if not token:
        raise ValueError(
            'GitHub GraphQL API 需要 Personal Access Token。\n'
            '请在项目根目录创建 .env 文件，并添加: GITHUB_TOKEN=your_token\n'
            '获取token: https://github.com/settings/tokens'
        )

    # GraphQL查询
    query = """
    query($username: String!, $from: DateTime!, $to: DateTime!) {
      user(login: $username) {
        contributionsCollection(from: $from, to: $to) {
          contributionCalendar {
            totalContributions
            weeks {
              contributionDays {
                contributionCount
                date
              }
            }
          }
        }
      }
    }
    """

    # 日期范围
    from_date = f'{year}-01-01T00:00:00Z'
    to_date = f'{year}-12-31T23:59:59Z'

    variables = {'username': username, 'from': from_date, 'to': to_date}

    headers = {'Authorization': f'Bearer {token}', 'Content-Type': 'application/json'}

    logger.info(f'正在查询用户 {username} 在 {year} 年的contribution数据...')

    # 发送GraphQL请求
    response = requests.post(
        'https://api.github.com/graphql',
        json={'query': query, 'variables': variables},
        headers=headers,
    )

    if response.status_code != 200:
        raise Exception(f'GitHub GraphQL API请求失败: {response.status_code}\n{response.text}')

    data = response.json()

    # 检查错误
    if 'errors' in data:
        errors = data['errors']
        error_messages = [e.get('message', str(e)) for e in errors]
        raise Exception(f'GraphQL查询错误: {", ".join(error_messages)}')

    # 解析数据
    try:
        user_data = data['data']['user']
        if not user_data:
            raise Exception(f'未找到用户: {username}')

        contribution_calendar = user_data['contributionsCollection']['contributionCalendar']
        total_contributions = contribution_calendar['totalContributions']

        logger.info(f'总计: {total_contributions} 个contributions')

        # 构建日期->contribution数量的映射
        commit_counts = {}
        active_days = 0

        for week in contribution_calendar['weeks']:
            for day in week['contributionDays']:
                date_str = day['date']
                count = day['contributionCount']

                if count > 0:
                    commit_counts[date_str] = count
                    active_days += 1

        logger.info(f'活跃天数: {active_days} 天')

        return commit_counts

    except (KeyError, TypeError) as e:
        raise Exception(f'解析GraphQL响应失败: {e}\n响应数据: {data}')


def get_daily_commits_for_year(year: int, commit_stats: dict[str, int]) -> list[tuple[datetime, int]]:
    """
    将commit统计数据转换为完整年份的每日列表。

    Args:
        year: 年份
        commit_stats: commit统计字典

    Returns:
        列表，包含(日期, commit数量)元组，覆盖整年365/366天
    """
    daily_commits = []

    # 判断是否闰年
    is_leap = (year % 4 == 0 and year % 100 != 0) or (year % 400 == 0)
    total_days = 366 if is_leap else 365

    for day in range(total_days):
        date = datetime(year, 1, 1, tzinfo=timezone.utc)
        from datetime import timedelta

        date = date + timedelta(days=day)
        date_key = date.strftime('%Y-%m-%d')

        commit_count = commit_stats.get(date_key, 0)
        daily_commits.append((date, commit_count))

    return daily_commits