Trending Skills

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

__version__ = "1.0.0"
__author__ = "AI Daily"

"""
配置模块 - Skills Trending 配置
"""
import os

# ============================================================================
# Skills.sh 配置
# ============================================================================
SKILLS_BASE_URL = os.getenv("SKILLS_BASE_URL", "https://skills.sh")
SKILLS_TRENDING_URL = f"{SKILLS_BASE_URL}/trending"

# ============================================================================
# 抓取配置
# ============================================================================
TOP_N_DETAILS = 20  # 抓取详情的数量
FETCH_REQUEST_DELAY = 2  # 抓取详情时的请求间隔（秒）

# ============================================================================
# 技能分类定义
# ============================================================================
CATEGORIES = {
    "frontend": {
        "name": "前端",
        "name_en": "Frontend",
        "icon": "🌐",
        "description": "React, Vue, CSS 等前端技术"
    },
    "backend": {
        "name": "后端",
        "name_en": "Backend",
        "icon": "⚙️",
        "description": "Node, Python, APIs 等后端技术"
    },
    "mobile": {
        "name": "移动",
        "name_en": "Mobile",
        "icon": "📱",
        "description": "iOS, Android, React Native"
    },
    "devops": {
        "name": "运维",
        "name_en": "DevOps",
        "icon": "🔧",
        "description": "CI/CD, Docker, Kubernetes"
    },
    "video": {
        "name": "视频",
        "name_en": "Video",
        "icon": "🎬",
        "description": "视频生成、编辑"
    },
    "animation": {
        "name": "动画",
        "name_en": "Animation",
        "icon": "✨",
        "description": "动画、动效设计"
    },
    "data": {
        "name": "数据",
        "name_en": "Data",
        "icon": "📊",
        "description": "数据分析、BI 工具"
    },
    "ai-ml": {
        "name": "AI/ML",
        "name_en": "AI/ML",
        "icon": "🤖",
        "description": "机器学习、人工智能"
    },
    "testing": {
        "name": "测试",
        "name_en": "Testing",
        "icon": "🧪",
        "description": "QA, E2E 测试"
    },
    "marketing": {
        "name": "营销",
        "name_en": "Marketing",
        "icon": "📢",
        "description": "SEO, 内容营销"
    },
    "docs": {
        "name": "文档",
        "name_en": "Docs",
        "icon": "📚",
        "description": "教程、文档"
    },
    "design": {
        "name": "设计",
        "name_en": "Design",
        "icon": "🎨",
        "description": "UI/UX 设计"
    },
    "database": {
        "name": "数据库",
        "name_en": "Database",
        "icon": "🗄️",
        "description": "SQL, NoSQL"
    },
    "security": {
        "name": "安全",
        "name_en": "Security",
        "icon": "🔒",
        "description": "安全工具"
    },
    "other": {
        "name": "其他",
        "name_en": "Other",
        "icon": "📁",
        "description": "未分类"
    }
}


def get_category_info(category_key: str) -> dict:
    """获取分类信息"""
    return CATEGORIES.get(category_key, CATEGORIES["other"])

"""
Detail Fetcher - 抓取技能详情页
"""
import re
import time
from typing import Dict, List, Optional
from bs4 import BeautifulSoup
import requests

from src.config import FETCH_REQUEST_DELAY, SKILLS_BASE_URL


class DetailFetcher:
    """抓取技能详情页"""

    def __init__(self, timeout: int = 30, delay: float = None):
        """
        初始化

        Args:
            timeout: 请求超时时间（秒）
            delay: 请求间隔（秒），默认使用配置中的值
        """
        self.base_url = SKILLS_BASE_URL
        self.timeout = timeout
        self.delay = delay if delay is not None else FETCH_REQUEST_DELAY
        self.session = requests.Session()
        self.session.headers.update({
            "User-Agent": "Mozilla/5.0 (compatible; SkillsTrendingBot/1.0)"
        })

    def fetch_top20_details(self, skills: List[Dict]) -> List[Dict]:
        """
        批量抓取 Top 20 详情

        Args:
            skills: Top 20 技能列表

        Returns:
            [
                {
                    "name": "remotion-best-practices",
                    "owner": "remotion-dev/skills",
                    "url": "...",
                    "html_content": "<html>...</html>",
                    "when_to_use": "Use this skills whenever...",
                    "rules": [
                        {"file": "3d.md", "desc": "3D content in Remotion..."},
                        ...
                    ],
                    "rules_count": 27
                },
                ...
            ]
        """
        results = []
        top_n = min(20, len(skills))

        print(f"📥 开始抓取 Top {top_n} 详情...")

        for i, skill in enumerate(skills[:top_n], 1):
            url = skill.get("url", "")
            if not url:
                # 尝试构建 URL
                name = skill.get("name", "")
                owner = skill.get("owner", "")
                url = f"{self.base_url}/{owner}/{name}"

            print(f"  [{i}/{top_n}] 抓取: {skill.get('name')}")

            detail = self.fetch_detail_page(url, skill)
            if detail:
                results.append(detail)
            else:
                # 即使失败也保留基本信息
                results.append({
                    "name": skill.get("name"),
                    "owner": skill.get("owner"),
                    "url": url,
                    "when_to_use": "",
                    "rules": [],
                    "rules_count": 0,
                    "error": "Failed to fetch details"
                })

            # 限速
            if i < top_n:
                time.sleep(self.delay)

        print(f"✅ 成功抓取 {len(results)} 个技能详情")
        return results

    def fetch_detail_page(self, url: str, skill_info: Dict = None) -> Optional[Dict]:
        """
        获取单个技能详情

        Args:
            url: 技能详情页 URL
            skill_info: 技能基本信息

        Returns:
            技能详情字典或 None
        """
        if not skill_info:
            skill_info = {}

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

            html_content = response.text

            # 解析页面
            detail = self.parse_detail_page(html_content, url, skill_info)
            return detail

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

    def parse_detail_page(self, html_content: str, url: str, skill_info: Dict) -> Dict:
        """
        解析技能详情页

        Args:
            html_content: 页面 HTML
            url: 页面 URL
            skill_info: 技能基本信息

        Returns:
            技能详情字典
        """
        soup = BeautifulSoup(html_content, "lxml")

        # 提取 "When to use" 部分
        when_to_use = self._extract_when_to_use(soup)

        # 提取规则列表
        rules = self._extract_rules(soup)

        # 提取技能名称（如果未提供）
        name = skill_info.get("name")
        if not name:
            name = self._extract_name_from_soup(soup, url)

        # 提取拥有者
        owner = skill_info.get("owner", "unknown")

        return {
            "name": name,
            "owner": owner,
            "url": url,
            "html_content": html_content,
            "when_to_use": when_to_use,
            "rules": rules,
            "rules_count": len(rules)
        }

    def _extract_when_to_use(self, soup: BeautifulSoup) -> str:
        """
        提取 "When to use" 部分

        Args:
            soup: BeautifulSoup 对象

        Returns:
            when_to_use 文本
        """
        # 尝试不同的选择器
        selectors = [
            "h2#when-to-use",
            '[id="when-to-use"]',
            "h2:contains('When to use')",
            "h2:contains('When to Use')",
        ]

        for selector in selectors:
            element = soup.select_one(selector)
            if element:
                # 获取下一个兄弟元素（通常是内容）
                content = element.find_next_sibling()
                if content:
                    return content.get_text(strip=True)

        # 如果找不到标题，尝试在整个页面中搜索
        text = soup.get_text()
        match = re.search(r'When to use\s*\n\s*(.+?)(?:\n\s*##|\n\s*###|\Z)', text, re.DOTALL | re.IGNORECASE)
        if match:
            return match.group(1).strip()

        return ""

    def _extract_rules(self, soup: BeautifulSoup) -> List[Dict]:
        """
        提取规则列表

        Args:
            soup: BeautifulSoup 对象

        Returns:
            规则列表
        """
        rules = []

        # 尝试找到规则列表
        # 通常在 "How to use" 或类似的标题下
        list_selectors = [
            "ul",
            "ol",
            '[class*="rules"]',
            '[class*="list"]',
        ]

        for selector in list_selectors:
            lists = soup.select(selector)

            for lst in lists:
                items = lst.find_all("li", recursive=False)
                if len(items) >= 3:  # 至少 3 项才认为是规则列表
                    for item in items:
                        link = item.find("a", href=True)
                        if link:
                            href = link.get("href", "")
                            text = link.get_text(strip=True)
                            # 描述可能在链接后面
                            desc = item.get_text(strip=True).replace(text, "", 1).strip()

                            rules.append({
                                "file": href.split("/")[-1] if href else text,
                                "desc": desc or text
                            })

                    if rules:
                        return rules

        # 备用方案：从 HTML 中提取所有看起来像规则的链接
        pattern = r'rules/([a-z0-9_-]+)\.md'
        matches = re.finditer(pattern, str(soup))

        for match in matches:
            rules.append({
                "file": f"{match.group(1)}.md",
                "desc": f"Rule: {match.group(1)}"
            })

        return rules

    def _extract_name_from_soup(self, soup: BeautifulSoup, url: str) -> str:
        """
        从 URL 或页面中提取技能名称

        Args:
            soup: BeautifulSoup 对象
            url: 页面 URL

        Returns:
            技能名称
        """
        # 从 URL 提取
        parts = url.strip("/").split("/")
        if len(parts) >= 1:
            return parts[-1]

        # 从页面标题提取
        title = soup.find("title")
        if title:
            title_text = title.get_text()
            # 通常格式是 "skill-name by owner"
            name = title_text.split(" by ")[0].strip()
            return name

        return "unknown"

    def get_skill_detail_summary(self, detail: Dict) -> str:
        """
        获取技能详情摘要（用于 AI 分析）

        Args:
            detail: 技能详情

        Returns:
            摘要文本
        """
        lines = [
            f"【技能名称】{detail.get('name')}",
            f"【拥有者】{detail.get('owner')}",
            f"【URL】{detail.get('url')}",
        ]

        if detail.get("when_to_use"):
            lines.append(f"\n【用途说明】")
            lines.append(detail.get("when_to_use"))

        if detail.get("rules"):
            lines.append(f"\n【规则列表】({len(detail.get('rules'))} 条)")
            for rule in detail.get("rules")[:10]:  # 最多显示 10 条
                lines.append(f"  - {rule.get('file')}: {rule.get('desc')}")

            if len(detail.get("rules")) > 10:
                lines.append(f"  ... 还有 {len(detail.get('rules')) - 10} 条规则")

        return "\n".join(lines)


def fetch_details(skills: List[Dict]) -> List[Dict]:
    """便捷函数：获取技能详情"""
    fetcher = DetailFetcher()
    return fetcher.fetch_top20_details(skills)

"""
Skills Fetcher - 从 skills.sh/trending 获取技能排行榜
使用 Playwright 处理动态渲染页面
"""
import re
import asyncio
from typing import Dict, List
from playwright.async_api import async_playwright

from src.config import SKILLS_TRENDING_URL, SKILLS_BASE_URL


class SkillsFetcher:
    """从 skills.sh/trending 获取排行榜"""

    def __init__(self, timeout: int = 30000):
        """初始化"""
        self.base_url = SKILLS_BASE_URL
        self.trending_url = SKILLS_TRENDING_URL
        self.timeout = timeout

    def fetch(self) -> List[Dict]:
        """
        获取 Top 100 技能列表

        Returns:
            [
                {
                    "rank": 1,
                    "name": "remotion-best-practices",
                    "owner": "remotion-dev/skills",
                    "installs": 5600,
                    "url": "https://skills.sh/remotion-dev/skills/remotion-best-practices"
                },
                ...
            ]
        """
        print(f"📡 正在获取榜单: {self.trending_url}")

        # 运行异步方法
        return asyncio.run(self._fetch_async())

    async def _fetch_async(self) -> List[Dict]:
        """异步获取数据 - 带重试机制"""
        max_retries = 3
        retry_delay = 5

        for attempt in range(max_retries):
            try:
                async with async_playwright() as p:
                    # 启动浏览器 - CI 环境使用 headless 模式
                    browser = await p.chromium.launch(
                        headless=True,
                        args=[
                            '--disable-dev-shm-usage',
                            '--no-sandbox',
                            '--disable-setuid-sandbox',
                            '--disable-blink-features=AutomationControlled',
                        ]
                    )

                    # 创建页面
                    page = await browser.new_page()

                    # 设置用户代理，避免被识别为机器人
                    await page.set_extra_http_headers({
                        'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36'
                    })

                    # 导航到页面
                    print(f"  正在加载页面... (尝试 {attempt + 1}/{max_retries})")
                    await page.goto(self.trending_url, wait_until="domcontentloaded", timeout=60000)

                    # 等待页面稳定
                    await asyncio.sleep(5)

                    # 尝试滚动页面以确保内容加载
                    try:
                        await page.evaluate("window.scrollTo(0, document.body.scrollHeight / 2)")
                        await asyncio.sleep(2)
                    except:
                        pass

                    # 获取页面文本内容
                    content = await page.evaluate("() => document.body.innerText")

                    # 调试：检查内容
                    print(f"  页面内容长度: {len(content)} 字符")

                    # 检查是否包含关键内容
                    if "Skills Leaderboard" not in content and "Leaderboard" not in content:
                        print(f"  ⚠️ 未找到排行榜标题，等待更长时间...")
                        await asyncio.sleep(10)
                        content = await page.evaluate("() => document.body.innerText")

                    await browser.close()

                    # 解析排行榜
                    skills = self.parse_leaderboard(content)

                    if skills:
                        print(f"✅ 成功获取 {len(skills)} 个技能")
                        return skills

                    raise Exception("无法从页面解析技能列表")

            except Exception as e:
                print(f"  ⚠️ 尝试 {attempt + 1} 失败: {e}")
                if attempt < max_retries - 1:
                    await asyncio.sleep(retry_delay)
                else:
                    raise

        raise Exception("获取失败：已达最大重试次数")

    def parse_leaderboard(self, html_content: str) -> List[Dict]:
        """
        解析排行榜 - skills.sh 页面使用文本格式

        格式:
        SKILLS LEADERBOARD
        ...
        1
        remotion-best-practices
        remotion-dev/skills
        7.0K
        ...
        """
        skills = []

        # 查找排行榜开始位置 - 支持多种格式
        for marker in ["SKILLS LEADERBOARD", "Skills Leaderboard", "LEADERBOARD", "Leaderboard"]:
            leaderboard_start = html_content.find(marker)
            if leaderboard_start != -1:
                print(f"  找到标记: '{marker}'")
                break

        if leaderboard_start == -1:
            # 调试：打印页面内容的前1000字符
            preview = html_content[:1000] if html_content else "(空内容)"
            print(f"  ⚠️ 页面内容预览:\n{preview}")
            raise Exception("未找到 Skills Leaderboard 标题")

        # 提取排行榜部分
        content = html_content[leaderboard_start:]

        # 新格式: 没有###前缀，直接是 rank\nname\nowner\ninstalls
        # 格式示例:
        # 1
        # remotion-best-practices
        # remotion-dev/skills
        # 7.0K

        patterns = [
            # 模式1: 新格式 (无###前缀)
            r'(\d+)\s*\n\s*([a-z0-9-]+)\s*\n\s*([\w-]+/[\w-]+)\s*\n\s*([\d.]+K?)',
            # 模式2: 允许更多字符
            r'(\d+)\s*\n\s*([a-zA-Z0-9_-]+)\s*\n\s*([\w-]+/[\w-]+)\s*\n\s*([\d.]+K?)',
            # 模式3: 旧格式 (有###前缀)
            r'(\d+)\s*\n\s*###\s*([\w-]+)\s*\n\s*([\w-]+/[\w-]+)\s*\n\s*([\d.]+K?)',
            # 模式4: 最宽松
            r'(\d+)\s+([a-zA-Z0-9_-]+)\s+([\w-]+/[\w-]+)\s+([\d.]+K?)',
        ]

        skills_dict = {}  # 用于去重，保留最新排名

        for i, pattern in enumerate(patterns):
            matches = re.finditer(pattern, content, re.MULTILINE)

            for match in matches:
                rank = int(match.group(1))
                name = match.group(2)
                owner = match.group(3)
                installs_str = match.group(4)

                # 处理安装量
                installs = self._parse_installs(installs_str)

                # 只保留每个技能的最高排名（第一次出现）
                if name not in skills_dict or skills_dict[name]["rank"] > rank:
                    skills_dict[name] = {
                        "rank": rank,
                        "name": name,
                        "owner": owner,
                        "installs": installs,
                        "url": f"{self.base_url}/{owner}/{name}"
                    }

            if skills_dict:
                print(f"  使用模式 {i+1} 匹配到 {len(skills_dict)} 个技能")
                break

        # 按排名排序
        skills = sorted(skills_dict.values(), key=lambda x: x["rank"])

        return skills

    def _parse_installs(self, installs_str: str) -> int:
        """解析安装量字符串"""
        if not installs_str:
            return 0

        installs_str = installs_str.strip().upper()

        if "K" in installs_str:
            try:
                return int(float(installs_str.replace("K", "")) * 1000)
            except ValueError:
                return 0

        try:
            return int(installs_str)
        except ValueError:
            return 0

    def get_date_range(self) -> tuple:
        """获取可用日期范围"""
        return None, None


def fetch_skills() -> List[Dict]:
    """便捷函数：获取技能列表"""
    fetcher = SkillsFetcher()
    return fetcher.fetch()