TrendRadar：开源技术趋势追踪雷达

开源项目 5 小时前 0 1

在信息爆炸的时代，追踪开源技术趋势变得越来越重要。TrendRadar 是一个轻量级的开源项目，帮助开发者自动追踪 GitHub、HackerNews 等平台的热门项目，生成趋势报告，让你不错过任何技术风向。

项目背景

作为开发者，我们经常需要了解：

今天 GitHub 上什么项目最火？
哪些新框架值得关注？
某个技术领域有什么新动向？

手动刷 GitHub Trending、HackerNews 太耗时间。TrendRadar 就是为了解决这个痛点而生。

核心功能

1. 多源数据采集

支持从多个平台抓取热门项目：

GitHub Trending - 按语言、时间范围筛选
HackerNews - Show HN、Top Stories
ProductHunt - 每日热门产品
Reddit - r/programming、r/opensource 等

2. 智能分类与标签

自动识别项目类型并打标签：

AI/机器学习
Web 框架
DevOps 工具
CLI 工具
数据库
安全工具

3. 趋势分析报告

生成多维度趋势报告：

新星项目 - 本周新上榜的项目
增长最快 - Star 增速最高的项目
持续热门 - 连续多天上榜的项目
语言分布 - 热门项目的编程语言占比

4. 订阅通知

支持多种推送方式：

邮件日报/周报
Telegram Bot
Discord Webhook
Slack 集成
RSS 订阅

技术架构

TrendRadar/
├── trendradar/
│   ├── __init__.py
│   ├── cli.py              # 命令行入口
│   ├── config.py           # 配置管理
│   ├── collectors/         # 数据采集器
│   │   ├── github.py
│   │   ├── hackernews.py
│   │   ├── producthunt.py
│   │   └── reddit.py
│   ├── analyzers/          # 趋势分析
│   │   ├── trending.py
│   │   ├── growth.py
│   │   └── classifier.py
│   ├── reporters/          # 报告生成
│   │   ├── markdown.py
│   │   ├── html.py
│   │   └── json.py
│   ├── notifiers/          # 通知推送
│   │   ├── email.py
│   │   ├── telegram.py
│   │   └── discord.py
│   └── storage/            # 数据存储
│       ├── sqlite.py
│       └── models.py
├── config.yaml             # 配置文件
├── requirements.txt
└── README.md

快速开始

安装

# 使用 pip 安装
pip install trendradar

# 或从源码安装
git clone https://github.com/yourname/trendradar.git
cd trendradar
pip install -e .

基本用法

# 查看今日 GitHub Trending
trendradar github --lang python --since daily

# 查看 HackerNews 热门
trendradar hn --top 20

# 生成周报
trendradar report --format markdown --output weekly.md

# 启动定时任务
trendradar serve --interval 6h --notify telegram

配置文件示例

# config.yaml
sources:
  github:
    enabled: true
    languages:
      - python
      - go
      - rust
      - typescript
    since: daily
  
  hackernews:
    enabled: true
    min_score: 100
  
  reddit:
    enabled: true
    subreddits:
      - programming
      - opensource
      - devops

notify:
  telegram:
    enabled: true
    bot_token: "YOUR_BOT_TOKEN"
    chat_id: "YOUR_CHAT_ID"
  
  email:
    enabled: false
    smtp_host: "smtp.gmail.com"
    smtp_port: 587
    recipients:
      - "you@example.com"

schedule:
  daily_report: "08:00"
  weekly_report: "monday 09:00"

核心代码示例

GitHub Trending 采集器

import httpx
from bs4 import BeautifulSoup
from dataclasses import dataclass
from typing import List, Optional

@dataclass
class TrendingRepo:
    name: str
    url: str
    description: str
    language: Optional[str]
    stars: int
    stars_today: int
    forks: int

class GitHubCollector:
    BASE_URL = "https://github.com/trending"
    
    def __init__(self, language: str = None, since: str = "daily"):
        self.language = language
        self.since = since
    
    def collect(self) -> List[TrendingRepo]:
        url = self._build_url()
        response = httpx.get(url)
        response.raise_for_status()
        return self._parse(response.text)
    
    def _build_url(self) -> str:
        url = self.BASE_URL
        if self.language:
            url += f"/{self.language}"
        url += f"?since={self.since}"
        return url
    
    def _parse(self, html: str) -> List[TrendingRepo]:
        soup = BeautifulSoup(html, "html.parser")
        repos = []
        
        for article in soup.select("article.Box-row"):
            name = article.select_one("h2 a").text.strip().replace("\n", "").replace(" ", "")
            url = "https://github.com" + article.select_one("h2 a")["href"]
            
            desc_elem = article.select_one("p")
            description = desc_elem.text.strip() if desc_elem else ""
            
            lang_elem = article.select_one("[itemprop='programmingLanguage']")
            language = lang_elem.text.strip() if lang_elem else None
            
            stars_text = article.select_one("a[href$='/stargazers']").text.strip()
            stars = self._parse_number(stars_text)
            
            today_elem = article.select_one("span.d-inline-block.float-sm-right")
            stars_today = self._parse_number(today_elem.text) if today_elem else 0
            
            forks_text = article.select_one("a[href$='/forks']").text.strip()
            forks = self._parse_number(forks_text)
            
            repos.append(TrendingRepo(
                name=name,
                url=url,
                description=description,
                language=language,
                stars=stars,
                stars_today=stars_today,
                forks=forks
            ))
        
        return repos
    
    def _parse_number(self, text: str) -> int:
        text = text.strip().replace(",", "")
        if "k" in text.lower():
            return int(float(text.lower().replace("k", "")) * 1000)
        return int(text) if text.isdigit() else 0

趋势分析器

from collections import defaultdict
from datetime import datetime, timedelta
from typing import List, Dict
from .storage import Storage

class TrendAnalyzer:
    def __init__(self, storage: Storage):
        self.storage = storage
    
    def get_rising_stars(self, days: int = 7) -> List[dict]:
        """获取新上榜项目"""
        since = datetime.now() - timedelta(days=days)
        recent = self.storage.get_repos_since(since)
        
        # 找出首次出现的项目
        first_seen = {}
        for repo in recent:
            if repo.name not in first_seen:
                first_seen[repo.name] = repo
        
        # 按 stars_today 排序
        rising = sorted(
            first_seen.values(),
            key=lambda r: r.stars_today,
            reverse=True
        )
        return rising[:20]
    
    def get_fastest_growing(self, days: int = 7) -> List[dict]:
        """获取增长最快的项目"""
        since = datetime.now() - timedelta(days=days)
        repos = self.storage.get_repos_since(since)
        
        # 计算每个项目的累计增长
        growth = defaultdict(int)
        latest = {}
        
        for repo in repos:
            growth[repo.name] += repo.stars_today
            latest[repo.name] = repo
        
        # 排序
        sorted_repos = sorted(
            growth.items(),
            key=lambda x: x[1],
            reverse=True
        )
        
        return [
            {"repo": latest[name], "growth": g}
            for name, g in sorted_repos[:20]
        ]
    
    def get_language_distribution(self, days: int = 7) -> Dict[str, int]:
        """获取语言分布"""
        since = datetime.now() - timedelta(days=days)
        repos = self.storage.get_repos_since(since)
        
        lang_count = defaultdict(int)
        for repo in repos:
            if repo.language:
                lang_count[repo.language] += 1
        
        return dict(sorted(
            lang_count.items(),
            key=lambda x: x[1],
            reverse=True
        ))

Telegram 通知器

import httpx
from typing import List

class TelegramNotifier:
    API_BASE = "https://api.telegram.org/bot{token}"
    
    def __init__(self, bot_token: str, chat_id: str):
        self.bot_token = bot_token
        self.chat_id = chat_id
        self.api_url = self.API_BASE.format(token=bot_token)
    
    def send_report(self, repos: List[dict], title: str = "📡 TrendRadar 日报"):
        message = self._format_message(repos, title)
        self._send(message)
    
    def _format_message(self, repos: List[dict], title: str) -> str:
        lines = [f"{title}\n"]
        
        for i, repo in enumerate(repos[:10], 1):
            lines.append(
                f"{i}. {repo.name}\n"
                f"   ⭐ {repo.stars:,} (+{repo.stars_today} today)\n"
                f"   {repo.description[:80]}...\n"
            )
        
        return "\n".join(lines)
    
    def _send(self, message: str):
        response = httpx.post(
            f"{self.api_url}/sendMessage",
            json={
                "chat_id": self.chat_id,
                "text": message,
                "parse_mode": "HTML",
                "disable_web_page_preview": True
            }
        )
        response.raise_for_status()

CLI 命令行工具

import click
from rich.console import Console
from rich.table import Table
from .collectors import GitHubCollector, HackerNewsCollector
from .analyzers import TrendAnalyzer
from .reporters import MarkdownReporter

console = Console()

@click.group()
def cli():
    """TrendRadar - 开源技术趋势追踪雷达"""
    pass

@cli.command()
@click.option("--lang", "-l", default=None, help="编程语言筛选")
@click.option("--since", "-s", default="daily", type=click.Choice(["daily", "weekly", "monthly"]))
@click.option("--limit", "-n", default=25, help="显示数量")
def github(lang, since, limit):
    """查看 GitHub Trending"""
    collector = GitHubCollector(language=lang, since=since)
    repos = collector.collect()[:limit]
    
    table = Table(title=f"GitHub Trending ({lang or 'All'} - {since})")
    table.add_column("#", style="dim")
    table.add_column("项目", style="cyan")
    table.add_column("语言", style="green")
    table.add_column("Stars", justify="right")
    table.add_column("Today", justify="right", style="yellow")
    table.add_column("描述")
    
    for i, repo in enumerate(repos, 1):
        table.add_row(
            str(i),
            repo.name,
            repo.language or "-",
            f"{repo.stars:,}",
            f"+{repo.stars_today}",
            repo.description[:50] + "..." if len(repo.description) > 50 else repo.description
        )
    
    console.print(table)

@cli.command()
@click.option("--format", "-f", default="markdown", type=click.Choice(["markdown", "html", "json"]))
@click.option("--output", "-o", default=None, help="输出文件路径")
@click.option("--days", "-d", default=7, help="分析天数")
def report(format, output, days):
    """生成趋势报告"""
    analyzer = TrendAnalyzer()
    
    data = {
        "rising_stars": analyzer.get_rising_stars(days),
        "fastest_growing": analyzer.get_fastest_growing(days),
        "language_dist": analyzer.get_language_distribution(days)
    }
    
    reporter = MarkdownReporter() if format == "markdown" else HtmlReporter()
    content = reporter.generate(data)
    
    if output:
        with open(output, "w") as f:
            f.write(content)
        console.print(f"[green]报告已保存到 {output}[/green]")
    else:
        console.print(content)

if __name__ == "__main__":
    cli()

运行效果

执行 trendradar github --lang python 后的输出效果：

┏━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┓
┃               GitHub Trending (Python - daily)                          ┃
┣━━━┳━━━━━━━━━━━━━━━━━━━━━━━┳━━━━━━━━━┳━━━━━━━━━┳━━━━━━━━┳━━━━━━━━━━━━━━━━━┫
┃ # ┃ 项目                  ┃ 语言    ┃  Stars  ┃ Today  ┃ 描述            ┃
┡━━━╇━━━━━━━━━━━━━━━━━━━━━━━╇━━━━━━━━━╇━━━━━━━━━╇━━━━━━━━╇━━━━━━━━━━━━━━━━━┩
│ 1 │ openai/gpt-5          │ Python  │ 45,230  │ +2,341 │ Next gen AI...  │
│ 2 │ langchain-ai/langflow │ Python  │ 28,102  │ +1,876 │ Visual LLM...   │
│ 3 │ microsoft/autogen     │ Python  │ 35,670  │ +1,203 │ Multi-agent...  │
│ 4 │ yt-dlp/yt-dlp         │ Python  │ 82,450  │ +987   │ Video downlo... │
│ 5 │ fastapi/fastapi       │ Python  │ 78,320  │ +654   │ Modern web...   │
└───┴───────────────────────┴─────────┴─────────┴────────┴─────────────────┘

部署方式

Docker 部署

# Dockerfile
FROM python:3.11-slim

WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .
RUN pip install -e .

CMD ["trendradar", "serve", "--interval", "6h"]

# docker-compose.yaml
version: '3.8'
services:
  trendradar:
    build: .
    volumes:
      - ./config.yaml:/app/config.yaml
      - ./data:/app/data
    environment:
      - TZ=Asia/Shanghai
    restart: unless-stopped

Cron 定时任务

# 每天早上 8 点发送日报
0 8 * * * /usr/local/bin/trendradar report --notify telegram

# 每周一发送周报
0 9 * * 1 /usr/local/bin/trendradar report --days 7 --notify email

未来规划

v1.1 - 添加 AI 摘要功能，自动总结项目亮点
v1.2 - Web Dashboard，可视化趋势图表
v1.3 - 个性化推荐，根据你的 Star 历史推荐相似项目
v2.0 - 支持自定义数据源插件

总结

TrendRadar 是一个简单但实用的开源趋势追踪工具。通过自动化采集和分析，帮助开发者节省时间，不错过任何值得关注的开源项目。

项目完全开源，欢迎 Star ⭐ 和贡献代码！

项目地址：https://github.com/cnbugs/trendradar

声明：本站所有文章，如无特殊说明或标注，均为本站原创发布。任何个人或组织，在未征得本站同意时，禁止复制、盗用、采集、发布本站内容到任何网站、书籍等各类媒体平台。如若本站内容侵犯了原著者的合法权益，可联系我们进行处理。