Github Topics

"""
AI Daily - AI资讯日报自动生成器
自动从 smol.ai 获取 AI 资讯，通过 Claude 分析分类，生成精美 HTML 页面
"""

__version__ = "1.0.0"
__author__ = "AI Daily"

"""
配置模块 - GitHub Topics Trending 配置
"""
import os

# ============================================================================
# GitHub API 配置
# ============================================================================
GITHUB_TOKEN = os.getenv("GH_TOKEN")
TOPIC = os.getenv("TOPIC", "claude-code")
GITHUB_API_BASE = "https://api.github.com"
GITHUB_PER_PAGE = 100  # GitHub API max per page
GITHUB_MAX_PAGES = 10  # Maximum pages to fetch (1000 repos)

# GitHub 搜索配置
GITHUB_SEARCH_SORT = "stars"  # stars, forks, updated
GITHUB_SEARCH_ORDER = "desc"  # desc, asc

# ============================================================================
# 请求配置
# ============================================================================
FETCH_REQUEST_DELAY = 0.5  # API 请求间隔（秒）


def format_number(num: int) -> str:
    """格式化数字显示"""
    if num >= 1000000:
        return f"{num / 1000000:.1f}M"
    elif num >= 1000:
        return f"{num / 1000:.1f}K"
    return str(num)


def get_repo_url(owner: str, repo_name: str) -> str:
    """生成仓库 URL"""
    return f"https://github.com/{owner}/{repo_name}"

"""
GitHub Fetcher - 从 GitHub API 获取仓库数据
使用 GitHub Search API 按话题获取仓库
"""
import time
import requests
from typing import Dict, List, Optional
from datetime import datetime, timedelta

from src.config import (
    GITHUB_TOKEN, TOPIC, GITHUB_API_BASE,
    GITHUB_PER_PAGE, GITHUB_MAX_PAGES, GITHUB_SEARCH_SORT,
    GITHUB_SEARCH_ORDER, FETCH_REQUEST_DELAY
)


class GitHubFetcher:
    """从 GitHub API 获取仓库数据"""

    def __init__(self, token: str = None, topic: str = None):
        """
        初始化

        Args:
            token: GitHub Personal Access Token
            topic: 要搜索的 GitHub Topic
        """
        self.token = token or GITHUB_TOKEN
        self.topic = topic or TOPIC
        self.api_base = GITHUB_API_BASE
        self.per_page = GITHUB_PER_PAGE
        self.max_pages = GITHUB_MAX_PAGES
        self.delay = FETCH_REQUEST_DELAY

        self.session = requests.Session()
        self.session.headers.update({
            "Accept": "application/vnd.github.v3+json",
            "User-Agent": "GitHub-Topics-Trending/1.0"
        })

        if self.token:
            self.session.headers.update({
                "Authorization": f"Bearer {self.token}"
            })

        self.rate_limit_remaining = 5000
        self.rate_limit_reset = None

    def fetch(self, sort_by: str = None, limit: int = None) -> List[Dict]:
        """
        获取指定话题下的仓库列表

        Args:
            sort_by: 排序方式 (stars, forks, updated)
            limit: 最大返回数量

        Returns:
            [
                {
                    "rank": 1,
                    "repo_name": "owner/repo",
                    "owner": "owner",
                    "stars": 1000,
                    "forks": 100,
                    "issues": 10,
                    "language": "Python",
                    "url": "https://github.com/owner/repo",
                    "description": "...",
                    "topics": ["topic1", "topic2"],
                    "created_at": "2024-01-01T00:00:00Z",
                    "updated_at": "2024-01-01T00:00:00Z"
                },
                ...
            ]
        """
        sort_by = sort_by or GITHUB_SEARCH_SORT
        limit = limit or (self.per_page * self.max_pages)

        print(f"📡 正在获取话题 '{self.topic}' 的仓库列表...")
        print(f"   排序方式: {sort_by}")

        repos = []
        page = 1

        while page <= self.max_pages and len(repos) < limit:
            # 检查速率限制
            if self.rate_limit_remaining < 10:
                self._wait_for_rate_limit()

            data = self._fetch_page(page, sort_by)

            if not data or "items" not in data:
                break

            items = data["items"]
            if not items:
                break

            for item in items:
                repo = self._parse_repo_item(item, len(repos) + 1)
                repos.append(repo)

                if len(repos) >= limit:
                    break

            # 更新速率限制信息
            self._update_rate_limit(data)

            print(f"   第 {page} 页: 获取 {len(items)} 个仓库 (累计 {len(repos)})")

            # 如果返回数量少于 per_page，说明已经到最后一页
            if len(items) < self.per_page:
                break

            page += 1

            # 请求间隔
            if page <= self.max_pages and len(repos) < limit:
                time.sleep(self.delay)

        print(f"✅ 成功获取 {len(repos)} 个仓库")
        return repos

    def _fetch_page(self, page: int, sort_by: str) -> Optional[Dict]:
        """
        获取单页数据

        Args:
            page: 页码
            sort_by: 排序方式

        Returns:
            API 响应数据
        """
        query = f"topic:{self.topic}"
        url = f"{self.api_base}/search/repositories"

        params = {
            "q": query,
            "sort": sort_by,
            "order": GITHUB_SEARCH_ORDER,
            "per_page": self.per_page,
            "page": page
        }

        try:
            response = self.session.get(url, params=params, timeout=30)
            response.raise_for_status()
            return response.json()

        except requests.RequestException as e:
            print(f"   ⚠️ 请求失败 (页 {page}): {e}")
            return None

    def _parse_repo_item(self, item: Dict, rank: int) -> Dict:
        """
        解析仓库数据

        Args:
            item: GitHub API 返回的仓库项
            rank: 排名

        Returns:
            仓库信息字典
        """
        owner_data = item.get("owner") or {}
        owner = owner_data.get("login", "")
        name = item.get("name", "")
        repo_name = f"{owner}/{name}"

        return {
            "rank": rank,
            "repo_name": repo_name,
            "owner": owner,
            "name": name,
            "stars": item.get("stargazers_count", 0),
            "forks": item.get("forks_count", 0),
            "issues": item.get("open_issues_count", 0),
            "language": item.get("language", ""),
            "url": item.get("html_url", ""),
            "description": item.get("description", ""),
            "topics": item.get("topics", []),
            "created_at": item.get("created_at", ""),
            "updated_at": item.get("updated_at", ""),
            "pushed_at": item.get("pushed_at", ""),
            "homepage": item.get("homepage", ""),
            "archived": item.get("archived", False),
        }

    def _update_rate_limit(self, response_data: Dict):
        """
        更新速率限制信息

        Args:
            response_data: API 响应数据
        """
        # 注意：这些信息在实际请求中从响应头获取
        # 这里只是一个简化版本
        pass

    def _wait_for_rate_limit(self):
        """等待速率限制重置"""
        if self.rate_limit_reset:
            now = int(time.time())
            wait_time = self.rate_limit_reset - now + 1

            if wait_time > 0:
                print(f"⏳ 速率限制已用尽，等待 {wait_time} 秒后重试...")
                time.sleep(wait_time)

    def fetch_new_repos(self, days: int = 7) -> List[Dict]:
        """
        获取最近创建的仓库

        Args:
            days: 最近多少天

        Returns:
            仓库列表
        """
        cutoff_date = (datetime.now() - timedelta(days=days)).strftime("%Y-%m-%d")
        query = f"topic:{self.topic} created:>{cutoff_date}"

        print(f"📡 正在获取最近 {days} 天创建的仓库...")

        repos = []
        page = 1

        while page <= self.max_pages:
            url = f"{self.api_base}/search/repositories"
            params = {
                "q": query,
                "sort": "stars",
                "order": "desc",
                "per_page": self.per_page,
                "page": page
            }

            try:
                response = self.session.get(url, params=params, timeout=30)
                response.raise_for_status()
                data = response.json()

                if not data or "items" not in data:
                    break

                items = data["items"]
                if not items:
                    break

                for item in items:
                    repo = self._parse_repo_item(item, len(repos) + 1)
                    repos.append(repo)

                print(f"   第 {page} 页: 获取 {len(items)} 个仓库")

                if len(items) < self.per_page:
                    break

                page += 1
                time.sleep(self.delay)

            except requests.RequestException as e:
                print(f"   ⚠️ 请求失败: {e}")
                break

        print(f"✅ 获取到 {len(repos)} 个新仓库")
        return repos

    def fetch_repo_details(self, owner: str, repo: str) -> Optional[Dict]:
        """
        获取单个仓库的详细信息

        Args:
            owner: 仓库拥有者
            repo: 仓库名称

        Returns:
            仓库详细信息
        """
        url = f"{self.api_base}/repos/{owner}/{repo}"

        try:
            response = self.session.get(url, timeout=30)
            response.raise_for_status()
            return response.json()

        except requests.RequestException as e:
            print(f"   ⚠️ 获取仓库详情失败 {owner}/{repo}: {e}")
            return None


def fetch_repos(sort_by: str = "stars", limit: int = 100) -> List[Dict]:
    """便捷函数：获取仓库列表"""
    fetcher = GitHubFetcher()
    return fetcher.fetch(sort_by=sort_by, limit=limit)

"""
README Fetcher - 获取仓库 README 内容
使用 GitHub API 获取仓库的 README 文件
"""
import time
import re
import requests
from typing import Dict, List, Optional

from src.config import GITHUB_TOKEN, GITHUB_API_BASE, FETCH_REQUEST_DELAY


class ReadmeFetcher:
    """获取仓库 README 内容"""

    def __init__(self, token: str = None):
        """
        初始化

        Args:
            token: GitHub Personal Access Token
        """
        self.token = token or GITHUB_TOKEN
        self.api_base = GITHUB_API_BASE
        self.delay = FETCH_REQUEST_DELAY

        self.session = requests.Session()
        self.session.headers.update({
            "Accept": "application/vnd.github.v3+json",
            "User-Agent": "GitHub-Topics-Trending/1.0"
        })

        if self.token:
            self.session.headers.update({
                "Authorization": f"Bearer {self.token}"
            })

    def fetch_readme(self, owner: str, repo: str, html: bool = False) -> Optional[str]:
        """
        获取仓库 README 内容

        Args:
            owner: 仓库拥有者
            repo: 仓库名称
            html: 是否返回 HTML 格式

        Returns:
            README 内容
        """
        url = f"{self.api_base}/repos/{owner}/{repo}/readme"

        if html:
            self.session.headers["Accept"] = "application/vnd.github.html"

        try:
            response = self.session.get(url, timeout=30)
            response.raise_for_status()

            # GitHub 返回的是 base64 编码的内容
            data = response.json()

            if data.get("encoding") == "base64":
                import base64
                content = base64.b64decode(data.get("content", "")).decode("utf-8", errors="ignore")
                return content
            else:
                return data.get("content", "")

        except requests.RequestException as e:
            print(f"   ⚠️ 获取 README 失败 {owner}/{repo}: {e}")
            return None

    def fetch_readme_summary(self, owner: str, repo: str, max_length: int = 500) -> Optional[str]:
        """
        获取 README 摘要

        Args:
            owner: 仓库拥有者
            repo: 仓库名称
            max_length: 最大长度

        Returns:
            README 摘要文本
        """
        readme = self.fetch_readme(owner, repo)

        if not readme:
            return None

        # 移除 Markdown 标记，提取纯文本
        summary = self._extract_text_from_markdown(readme)

        # 截断到指定长度
        if len(summary) > max_length:
            summary = summary[:max_length].rsplit(" ", 1)[0] + "..."

        return summary

    def _extract_text_from_markdown(self, markdown: str) -> str:
        """
        从 Markdown 中提取纯文本

        Args:
            markdown: Markdown 内容

        Returns:
            纯文本
        """
        # 移除代码块
        markdown = re.sub(r'```.*?```', '', markdown, flags=re.DOTALL)
        markdown = re.sub(r'`.*?`', '', markdown)

        # 移除链接
        markdown = re.sub(r'\[([^\]]+)\]\([^\)]+\)', r'\1', markdown)

        # 移除图片
        markdown = re.sub(r'!\[([^\]]*)\]\([^\)]+\)', '', markdown)

        # 移除标题标记
        markdown = re.sub(r'^#+\s+', '', markdown, flags=re.MULTILINE)

        # 移除加粗/斜体
        markdown = re.sub(r'\*\*([^*]+)\*\*', r'\1', markdown)
        markdown = re.sub(r'\*([^*]+)\*', r'\1', markdown)
        markdown = re.sub(r'__([^_]+)__', r'\1', markdown)
        markdown = re.sub(r'_([^_]+)_', r'\1', markdown)

        # 移除水平线
        markdown = re.sub(r'^---+$', '', markdown, flags=re.MULTILINE)
        markdown = re.sub(r'^\*\*\*+$', '', markdown, flags=re.MULTILINE)

        # 移除多余的空行
        lines = [line.strip() for line in markdown.split('\n')]
        lines = [line for line in lines if line]

        return ' '.join(lines)

    def batch_fetch_readmes(self, repos: List[Dict], delay: float = None) -> Dict[str, str]:
        """
        批量获取 README 内容

        Args:
            repos: 仓库列表
            delay: 请求间隔

        Returns:
            {repo_name: readme_summary} 字典
        """
        delay = delay if delay is not None else self.delay
        summaries = {}

        print(f"📥 开始批量获取 README...")

        for i, repo in enumerate(repos, 1):
            repo_name = repo.get("repo_name") or repo.get("name", "")

            if not repo_name or "/" not in repo_name:
                continue

            owner, name = repo_name.split("/", 1)

            print(f"  [{i}/{len(repos)}] {repo_name}")

            summary = self.fetch_readme_summary(owner, name)
            if summary:
                summaries[repo_name] = summary

            # 请求间隔
            if i < len(repos):
                time.sleep(delay)

        print(f"✅ 成功获取 {len(summaries)} 个 README 摘要")
        return summaries

    def fetch_from_github_raw(self, owner: str, repo: str, branch: str = "main") -> Optional[str]:
        """
        直接从 GitHub raw 内容获取 README

        Args:
            owner: 仓库拥有者
            repo: 仓库名称
            branch: 分支名

        Returns:
            README 内容
        """
        # 尝试常见的 README 文件名
        readme_names = ["README.md", "README.markdown", "README.rst", "README.txt"]

        for name in readme_names:
            url = f"https://raw.githubusercontent.com/{owner}/{repo}/{branch}/{name}"

            try:
                response = requests.get(url, timeout=10)
                if response.status_code == 200:
                    return response.text
            except requests.RequestException:
                continue

        # 尝试 master 分支
        if branch == "main":
            return self.fetch_from_github_raw(owner, repo, "master")

        return None


def fetch_readme_summary(owner: str, repo: str) -> Optional[str]:
    """便捷函数：获取 README 摘要"""
    fetcher = ReadmeFetcher()
    return fetcher.fetch_readme_summary(owner, repo)


def batch_fetch_readmes(repos: List[Dict]) -> Dict[str, str]:
    """便捷函数：批量获取 README"""
    fetcher = ReadmeFetcher()
    return fetcher.batch_fetch_readmes(repos)