利用开源项目 TrendRadar 打造一个属于自己的 “今日热榜”

TrendRadar/
├── 配置管理模块 (config/*.yaml, config/frequency_words.txt)
├── 数据获取模块 (DataFetcher)
├── 数据处理模块 (数据解析、统计分析)
├── 报告生成模块 (HTML/Markdown 生成)
├── 通知推送模块 (多渠道适配器)
├── AI 分析模块 (MCP 服务器)
└── 自动化部署 (GitHub Actions)

class DataFetcher:
    def __init__(self, proxy_url: Optional[str] = None):
        self.proxy_url = proxy_url

    def fetch_data(self, id_info: Union[str, Tuple[str, str]]) -> Tuple[Optional[str], str, str]:
        """获取指定ID数据，支持重试"""
        url = f"https://newsnow.busiyi.world/api/s?id={id_value}&latest"

        # 代理支持、重试机制、错误处理
        proxies = {"http": proxy_url, "https": proxy_url} if self.proxy_url else None
        headers = {
            "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
            "Accept": "application/json, text/plain, */*",
        }

        # 带重试的数据获取逻辑
        response = requests.get(url, proxies=proxies, headers=headers, timeout=10)
        return response.text, id_value, alias

def matches_word_groups(title: str, word_groups: List[Dict], filter_words: List[str]) -> bool:
    """检查标题是否匹配词组规则"""
    title_lower = title.lower()

    # 1. 过滤词检查（优先级最高）
    if any(filter_word.lower() in title_lower for filter_word in filter_words):
        return False

    # 2. 词组匹配检查
    for group in word_groups:
        required_words = group["required"]  # 必须词(+)
        normal_words = group["normal"]      # 普通词

        # 必须词检查：所有必须词都必须包含
        if required_words:
            all_required_present = all(
                req_word.lower() in title_lower for req_word in required_words
            )
            if not all_required_present:
                continue

        # 普通词检查：包含任意一个即可
        if normal_words:
            any_normal_present = any(
                normal_word.lower() in title_lower for normal_word in normal_words
            )
            if not any_normal_present:
                continue

        return True

    return False

def calculate_news_weight(title_data: Dict, rank_threshold: int = CONFIG["RANK_THRESHOLD"]) -> float:
    """计算新闻权重，用于排序"""
    ranks = title_data.get("ranks", [])
    count = title_data.get("count", len(ranks))
    weight_config = CONFIG["WEIGHT_CONFIG"]

    # 排名权重：Σ(11 - min(rank, 10)) / 出现次数
    rank_scores = [11 - min(rank, 10) for rank in ranks]
    rank_weight = sum(rank_scores) / len(ranks) if ranks else 0

    # 频次权重：min(出现次数, 10) × 10
    frequency_weight = min(count, 10) * 10

    # 热度加成：高排名次数 / 总出现次数 × 100
    high_rank_count = sum(1 for rank in ranks if rank <= rank_threshold)
    hotness_ratio = high_rank_count / len(ranks) if ranks else 0
    hotness_weight = hotness_ratio * 100

    # 综合权重计算
    total_weight = (
        rank_weight * weight_config["RANK_WEIGHT"] +      # 排名权重 60%
        frequency_weight * weight_config["FREQUENCY_WEIGHT"] + # 频次权重 30%
        hotness_weight * weight_config["HOTNESS_WEIGHT"]         # 热度权重 10%
    )

    return total_weight

report:
  mode: "daily"  # 当日汇总模式

if mode == "current":
    # 加载完整的历史数据
    analysis_data = self._load_analysis_data()
    # 只处理 last_time 等于最新时间的新闻
    for source_id, source_titles in results.items():
        filtered_titles = {}
        for title, title_data in source_titles.items():
            if title_info[source_id][title]["last_time"] == latest_time:
                filtered_titles[title] = title_data

def detect_latest_new_titles(current_platform_ids: Optional[List[str]] = None) -> Dict:
    """检测当日最新批次的新增标题"""
    # 解析最新文件
    latest_file = files[-1]
    latest_titles = parse_file_titles(latest_file)

    # 汇总历史标题
    historical_titles = set()
    for file_path in files[:-1]:
        historical_data = parse_file_titles(file_path)
        for source_id, titles_data in historical_data.items():
            for title in titles_data.keys():
                historical_titles.add(title)

    # 找出新增标题
    new_titles = {}
    for source_id, latest_source_titles in latest_titles.items():
        source_new_titles = {}
        for title in latest_source_titles.keys():
            if title not in historical_set:
                source_new_titles[title] = latest_source_titles[title]

def send_to_notifications(stats: List[Dict], report_data: Dict,
                         report_type: str, update_info: Optional[Dict] = None,
                         proxy_url: Optional[str] = None) -> Dict[str, bool]:
    """多渠道统一推送接口"""
    results = {}

    # 各平台调用
    if feishu_url:
        results["feishu"] = send_to_feishu(feishu_url, report_data, report_type)
    if dingtalk_url:
        results["dingtalk"] = send_to_dingtalk(dingtalk_url, report_data, report_type)
    if telegram_token and telegram_chat_id:
        results["telegram"] = send_to_telegram(
            telegram_token, telegram_chat_id, report_data, report_type
        )
    # ... 其他平台

    return results

def split_content_into_batches(report_data: Dict, format_type: str,
                              update_info: Optional[Dict] = None,
                              max_bytes: int = None) -> List[str]:
    """分批处理消息内容，确保词组标题+至少第一条新闻的完整性"""

    # 确保词组标题+第一条新闻的原子性
    for word_group in report_data["stats"]:
        word_header = f"🔥 {word_group['word']} : {word_group['count']} 条"
        first_news = word_group["titles"][0]

        # 检查词组标题+第一条新闻是否超出限制
        test_content = current_batch + word_header + first_news
        if len(test_content.encode("utf-8")) >= max_bytes:
            # 开启新批次
            batches.append(current_batch)
            current_batch = word_header + first_news

        # 继续添加剩余新闻
        for news in word_group["titles"][1:]:
            if current_batch_can_fit(news):
                current_batch += news
            else:
                batches.append(current_batch)
                current_batch = news

# 创建 FastMCP 2.0 应用
mcp = FastMCP('trendradar-news')

# 全局工具实例（在第一次请求时初始化）
_tools_instances = {}


def _get_tools(project_root: Optional[str] = None):
    """获取或创建工具实例（单例模式）"""
    if not _tools_instances:
        _tools_instances['data'] = DataQueryTools(project_root)
        _tools_instances['analytics'] = AnalyticsTools(project_root)
        _tools_instances['search'] = SearchTools(project_root)
        _tools_instances['config'] = ConfigManagementTools(project_root)
        _tools_instances['system'] = SystemManagementTools(project_root)
    return _tools_instances


# ==================== 数据查询工具 ====================

@mcp.tool
async def get_latest_news(
    platforms: Optional[List[str]] = None,
    limit: int = 50,
    include_url: bool = False
) -> str:

# 示例对话场景
[
    "查询今天热搜前10条新闻",
    "分析'人工智能'这个话题最近一周的热度趋势",
    "对比知乎和微博平台对'比特币'的关注度差异",
    "搜索'华为'相关的正面新闻",
    "生成今天的科技新闻摘要报告"
]

# .github/workflows/crawler.yml
name: Hot News Crawler
on:
  schedule:
    - cron: '0 * * * *'  # 每小时运行一次
  workflow_dispatch:  # 支持手动触发

jobs:
  crawl:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v3
      - name: Set up Python
        uses: actions/setup-python@v4
        with:
          python-version: '3.9'

      - name: Install dependencies
        run: |
          pip install pyyaml requests

      - name: Run crawler
        env:
          FEISHU_WEBHOOK_URL: ${{ secrets.FEISHU_WEBHOOK_URL }}
          # ... 其他环境变量
        run: python main.py

docker run -d --name trend-radar \
  -v ./config:/app/config:ro \
  -v ./output:/app/output \
  -e FEISHU_WEBHOOK_URL="你的webhook" \
  wantcat/trendradar:latest

def _setup_proxy(self) -> None:
    """设置代理配置"""
    if not self.is_github_actions and CONFIG["USE_PROXY"]:
        self.proxy_url = CONFIG["DEFAULT_PROXY"]
        print("本地环境，使用代理")
    else:
        print("GitHub Actions环境，不使用代理")

# 添加新平台仅需配置修改
platforms:
  - id: "new-platform"
    name: "新平台名称"

AI
机器学习
+技术

# 新主题
金融
股市
+投资

# 推送模式选择
report:
  mode: "daily" # 可选: "daily"|"incremental"|"current"
  rank_threshold: 5 # 排名高亮阈值

# 🕐 推送时间窗口控制（可选功能）
  # 用途：限制推送的时间范围，避免非工作时间打扰
  # 适用场景：
  #   - 只想在工作日白天接收推送（如 09:00-18:00）
  #   - 希望在晚上固定时间收到汇总（如 20:00-22:00）
  push_window:
    enabled: true  # 是否启用推送时间窗口控制，默认关闭
    # 注意：GitHub Actions 执行时间不稳定，时间范围建议至少留足 2 小时
    # 如果想要精准的定时推送，建议使用 Docker 部署在个人服务器上
    time_range:
      start: "08:00"  # 推送时间窗口开始（北京时间）
      end: "18:00"    # 推送时间窗口结束（北京时间）
    once_per_day: false  # 每天在时间窗口内只推送一次，如果 false，则窗口内每次执行都推送
    push_record_retention_days: 7  # 推送记录保留天数