diff --git a/skills/daily-hot-news/.gitignore b/skills/daily-hot-news/.gitignore new file mode 100644 index 0000000..18c7a21 --- /dev/null +++ b/skills/daily-hot-news/.gitignore @@ -0,0 +1,53 @@ +# Dependencies +node_modules/ +.pnpm-store/ + +# Build outputs +dist/ +build/ + +# Environment variables +.env +.env.local +.env.*.local + +# IDE +.vscode/ +.idea/ +*.swp +*.swo +*~ + +# OS +.DS_Store +Thumbs.db + +# Logs +logs/ +*.log +npm-debug.log* + +# Testing +coverage/ +.nyc_output/ + +# Cache +.cache/ +.__pycache__/ +*.py[cod] +*$py.class + +# Data files (user data) +data/ +*.json + +# Temporary files +tmp/ +temp/ +*.tmp + +# OpenClaw specific +.clawdhub/ + +# Personal milestone (local only) +MILESTONE.md diff --git a/skills/daily-hot-news/CONTRIBUTORS b/skills/daily-hot-news/CONTRIBUTORS new file mode 100644 index 0000000..d8649da --- /dev/null +++ b/skills/daily-hot-news/CONTRIBUTORS @@ -0,0 +1 @@ +root diff --git a/skills/daily-hot-news/LICENSE b/skills/daily-hot-news/LICENSE new file mode 100644 index 0000000..4f92758 --- /dev/null +++ b/skills/daily-hot-news/LICENSE @@ -0,0 +1,21 @@ +MIT License + +Copyright (c) 2026 OpenClaw Contributors + +Permission is hereby granted, free of charge, to any person obtaining a copy +of this software and associated documentation files (the "Software"), to deal +in the Software without restriction, including without limitation the rights +to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +copies of the Software, and to permit persons to whom the Software is +furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in all +copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +SOFTWARE. diff --git a/skills/daily-hot-news/README.md b/skills/daily-hot-news/README.md new file mode 100644 index 0000000..9828d06 --- /dev/null +++ b/skills/daily-hot-news/README.md @@ -0,0 +1,360 @@ +
+ +# 🔥 每日热榜 - OpenClaw Skill + +[![GitHub stars](https://img.shields.io/github/stars/one-box-u/openclaw-daily-hot-news)](https://github.com/one-box-u/openclaw-daily-hot-news/stargazers) +[![GitHub forks](https://img.shields.io/github/forks/one-box-u/openclaw-daily-hot-news)](https://github.com/one-box-u/openclaw-daily-hot-news/network) +[![MIT License](https://img.shields.io/badge/License-MIT-blue.svg)](LICENSE) + +--- + +**📖 [中文说明](README.md)** | **🇺🇸 [English Readme](README_EN.md)** + +--- + +一个基于 DailyHotApi 的热榜聚合查询技能，支持 54 个平台热榜查询、跨平台聚合、舆情监控等功能。 + +
+ +## 🎯 功能特性 + +### 核心功能 +- **热榜查询**：查询任意 54 个平台的热榜数据 +- **分类浏览**：按类别快速定位特定平台 +- **实时获取**：每次请求都获取最新热榜数据 +- **历史记录**：自动保存每日热榜数据到本地 +- **智能清理**：启动时自动检查并提示清理7天前的旧数据 + +### 扩展功能 +- **热点摘要**：15 种标签分类，AI 引导选择 +- **行业垂直**：十大行业分类，科技/游戏/金融等 +- **个性化订阅**：自定义关键词和平台偏好 +- **跨平台聚合**：全网 TOP10 热点榜单 +- **舆情监控**：关键词监控和热度告警 + +## 📊 支持平台（54 个） + +### 🎬 视频/直播平台（5 个） +| 平台 | 接口 | 说明 | +|------|------|------| +| 哔哩哔哩 | bilibili | 热门榜 | +| AcFun | acfun | 排行榜 | +| 抖音 | douyin | 热点榜 | +| 快手 | kuaishou | 热点榜 | +| 酷安 | coolapk | 热榜 | + +### 💬 社交媒体（8 个） +| 平台 | 接口 | 说明 | +|------|------|------| +| 微博 | weibo | 热搜榜 | +| 知乎 | zhihu | 热榜 | +| 知乎日报 | zhihu-daily | 推荐榜 | +| 百度贴吧 | tieba | 热议榜 | +| 豆瓣讨论小组 | douban-group | 讨论精选 | +| V2EX | v2ex | 主题榜 | +| NGA | ngabbs | 热帖 | +| 虎扑 | hupu | 步行街热帖 | + +### 📰 新闻资讯（10 个） +| 平台 | 接口 | 说明 | +|------|------|------| +| 百度 | baidu | 热搜榜 | +| 澎湃新闻 | thepaper | 热榜 | +| 今日头条 | toutiao | 热榜 | +| 36氪 | 36kr | 热榜 | +| 腾讯新闻 | qq-news | 热点榜 | +| 新浪网 | sina | 热榜 | +| 新浪新闻 | sina-news | 热点榜 | +| 网易新闻 | netease-news | 热点榜 | +| 虎嗅 | huxiu | 24小时 | +| 爱范儿 | ifanr | 快讯 | + +### 💻 科技/技术社区（8 个） +| 平台 | 接口 | 说明 | +|------|------|------| +| IT之家 | ithome | 热榜 | +| IT之家「喜加一」 | ithome-xijiayi | 最新动态 | +| 少数派 | sspai | 热榜 | +| CSDN | csdn | 排行榜 | +| 稀土掘金 | juejin | 热榜 | +| 51CTO | 51cto | 推荐榜 | +| NodeSeek | nodeseek | 最新动态 | +| HelloGitHub | hellogithub | Trending | + +### 🎮 游戏/ACG（5 个） +| 平台 | 接口 | 说明 | +|------|------|------| +| 原神 | genshin | 最新消息 | +| 米游社 | miyoushe | 最新消息 | +| 崩坏3 | honkai | 最新动态 | +| 崩坏：星穹铁道 | starrail | 最新动态 | +| 英雄联盟 | lol | 更新公告 | + +### 📚 阅读/文化（4 个） +| 平台 | 接口 | 说明 | +|------|------|------| +| 简书 | jianshu | 热门推荐 | +| 果壳 | guokr | 热门文章 | +| 微信读书 | weread | 飙升榜 | +| 豆瓣电影 | douban-movie | 新片榜 | + +### 🔧 工具/其他（5 个） +| 平台 | 接口 | 说明 | +|------|------|------| +| 吾爱破解 | 52pojie | 榜单 | +| 全球主机交流 | hostloc | 榜单 | +| 中央气象台 | weatheralarm | 全国气象预警 | +| 中国地震台 | earthquake | 地震速报 | +| 历史上的今天 | history | 月-日 | + +## 🚀 快速开始 + +### 1. 部署后端服务 + +```bash +cd daily-hot-api +./deploy.sh +``` + +### 2. 安装依赖 + +```bash +pip install -r requirements.txt +``` + +### 3. 配置环境变量 + +```bash +export DAILY_HOT_API_URL=http://localhost:6688 +``` + +### 4. 运行 + +```bash +python3 daily_hot_news.py --query "微博热搜" +``` + +## 💬 使用案例 + +### 案例 1：查询单个平台热榜 + +```bash +# 查询微博热搜 +python3 daily_hot_news.py -q "微博热搜" + +# 查询知乎热榜 +python3 daily_hot_news.py -q "知乎热榜" + +# 查询 B站热门 +python3 daily_hot_news.py -q "B站热门" + +# 查询原神最新消息 +python3 daily_hot_news.py -q "原神" +``` + +**效果**： +``` +🔥 **微博热搜** +更新时间: 2026-02-05T19:00:00.000Z + +1. 王一博 中山装 520万 +2. 肖战 害羞笑 480万 +3. 微博之夜红毯 450万 +... +``` + +### 案例 2：浏览所有平台 + +```bash +# 查看所有支持的热榜源 +python3 daily_hot_news.py --list +``` + +**效果**： +``` +📊 **支持的热榜源（共 54 个）** + +🎬 视频/直播 +• 哔哩哔哩 (bilibili) +• 抖音 (douyin) +• 快手 (kuaishou) +• AcFun (acfun) +• 酷安 (coolapk) + +💬 社交媒体 +• 微博 (weibo) +• 知乎 (zhihu) +• V2EX (v2ex) +• NGA (ngabbs) +... +``` + +### 案例 3：跨平台聚合 + +```bash +# 查看全网热点 TOP10 +python3 daily_hot_news.py --cross-platform +``` + +**效果**： +``` +🌐 全网热点 TOP10 + +🥇 原神 · 热度 952万 【B站最高】 + 评分: 98分 - 🔥 超级爆款 + 跨 5 平台讨论 + +🥈 微博热搜 · 热度 876万 【微博最高】 + 评分: 95分 - 全网讨论 + +🥉 特朗普 · 热度 654万 【微博最高】 + 评分: 89分 - 国际话题 +... +``` + +### 案例 4：设置舆情监控 + +```bash +# 设置监控：AI 话题，热度超过 500 万通知 +python3 daily_hot_news.py --monitor "AI,500万" + +# 查看监控配置 +python3 daily_hot_news.py -q "查看我的监控" +``` + +**效果**： +``` +✅ **已设置监控！** + +监控关键词：AI +热度阈值：500万 +监控平台：全部 54 个 + +当有话题热度超过阈值时，我会立即通知您！ +``` + +### 案例 5：定时推送配置 + +```bash +# 配置每天早上 8 点推送微博热搜 +"每天早上 8 点推送微博热搜" +``` + +**效果**： +``` +⏰ **定时推送已设置！** + +推送时间：每天 08:00 +热榜源：微博热搜 +推送方式：飞书消息 + +每天早上 8 点会自动推送微博热榜到您的飞书！ +``` + +## 🎮 高级用法 + +### 按类别查询 + +```bash +# 查看科技类热榜 +"有什么科技热榜" + +# 查看游戏类热榜 +"游戏有什么热点" +``` + +### 搜索特定话题 + +```bash +# 搜索包含某个关键词的热榜 +"搜索 AI 相关热榜" +"查找 ChatGPT 热点" +``` + +### 查看历史数据 + +```bash +# 查看昨天的微博热榜 +"微博昨天" + +# 查看历史记录 +"微博历史" +``` + +### 查看已保存数据 + +```bash +# 查看已保存的热榜统计 +"已保存了哪些数据" +``` + +### 清理旧数据 + +```bash +# 回复"清理"或"是"可删除7天前的旧热榜数据 +# Skill启动时会自动检测并提示 +``` + +## 📁 文件结构 + +``` +daily-hot-news/ +├── daily_hot_news.py # 主入口 +├── news_digest.py # 热点摘要 +├── industry_hot.py # 行业垂直 +├── personalized.py # 个性化订阅 +├── cross_platform.py # 跨平台聚合 +├── sentiment_monitor.py # 舆情监控 +├── api_client.py # API 客户端 +├── formatter.py # 格式化器 +├── storage.py # 数据存储 +├── config.py # 配置 +├── requirements.txt # 依赖列表 +└── README.md # 本说明 +``` + +## ⚙️ 配置说明 + +| 环境变量 | 默认值 | 说明 | +|----------|--------|------| +| `DAILY_HOT_API_URL` | http://localhost:6688 | 后端 API 地址 | +| `DAILY_HOT_CACHE_TTL` | 3600 | 缓存时间（秒） | +| `DAILY_HOT_MAX_ITEMS` | 20 | 返回最大条数 | +| `DAILY_HOT_TIMEOUT` | 10 | 请求超时（秒） | + +### 🔧 技术架构 + +- **API 调用**：每次用户请求都实时调用 DailyHotApi，确保数据最新 +- **历史记录**：自动保存到 `data/{平台}/{日期}.json` +- **旧数据清理**：启动时自动检测 7 天前数据，提示用户清理 + +## 🛡️ 安全说明 + +- ✅ 不包含任何 API 密钥或密码 +- ✅ 所有敏感配置通过环境变量管理 +- ✅ 用户数据存储在本地，不上传云端 + +## 📝 更新日志 + +**v2.0.1** (2026-02-06) +- ✨ 优化数据获取策略：每次请求都获取最新数据 +- ✨ 自动保存历史记录到本地 +- ✨ 启动时自动检查并提示清理7天前旧数据 +- 🔧 修复 API 客户端兼容性问题 + +**v2.0.0** (2026-02-05) +- ✨ 新增 5 个扩展功能 +- ✨ 支持 15 种热点标签分类 +- ✨ 支持十大行业垂直热榜 +- ✨ 新增个性化订阅功能 +- ✨ 新增跨平台 TOP10 聚合 +- ✨ 新增舆情监控告警 + +## 📄 许可证 + +MIT License + +## 🤝 致谢 + +- [DailyHotApi](https://github.com/imsyy/DailyHotApi) - 提供 54 个热榜源 API +- [OpenClaw](https://github.com/openclaw/openclaw) - AI 助手平台 diff --git a/skills/daily-hot-news/README_EN.md b/skills/daily-hot-news/README_EN.md new file mode 100644 index 0000000..14110a6 --- /dev/null +++ b/skills/daily-hot-news/README_EN.md @@ -0,0 +1,360 @@ +
+ +# 🔥 Daily Hot News - OpenClaw Skill + +[![GitHub stars](https://img.shields.io/github/stars/one-box-u/openclaw-daily-hot-news)](https://github.com/one-box-u/openclaw-daily-hot-news/stargazers) +[![GitHub forks](https://img.shields.io/github/forks/one-box-u/openclaw-daily-hot-news)](https://github.com/one-box-u/openclaw-daily-hot-news/network) +[![MIT License](https://img.shields.io/badge/License-MIT-blue.svg)](LICENSE) + +--- + +**🇺🇸 English Readme** | **📖 [中文说明](README.md)** + +--- + +A hot news aggregation skill based on DailyHotApi, supporting 54 platform hot search queries, cross-platform aggregation, and sentiment monitoring. + +
+ +## 🎯 Features + +### Core Features +- **Hot Search Query**: Query hot search data from any of 54 platforms +- **Category Browse**: Quickly locate specific platforms by category +- **Real-time Fetch**: Always get the latest hot search data on each request +- **History**: Automatically save daily hot search data to local storage +- **Smart Cleanup**: Automatically check and prompt to clean data older than 7 days on startup + +### Extended Features +- **Hot News Digest**: 15-tag classification, AI-guided selection +- **Industry Vertical**: 10 industry categories (tech, gaming, finance, etc.) +- **Personalized Subscription**: Custom keywords and platform preferences +- **Cross-Platform Aggregation**: Top 10 hot search榜单 nationwide +- **Sentiment Monitoring**: Keyword monitoring and hot alerts + +## 📊 Supported Platforms (54) + +### 🎬 Video/Live Streaming (5) +| Platform | API | Description | +|----------|------|------| +| Bilibili | bilibili | Hot Ranking | +| AcFun | acfun | Ranking List | +| Douyin | douyin | Hot Topics | +| Kuaishou | kuaishou | Hot Topics | +| Coolapk | coolapk | Hot Ranking | + +### 💬 Social Media (8) +| Platform | API | Description | +|----------|------|------| +| Weibo | weibo | Hot Search | +| Zhihu | zhihu | Hot List | +| Zhihu Daily | zhihu-daily | Recommended | +| Tieba | tieba | Hot Discussion | +| Douban Group | douban-group | Discussion Picks | +| V2EX | v2ex | Topic Ranking | +| NGA | ngabbs | Hot Posts | +| Hupu | hupu | Street Hot Posts | + +### 📰 News & Media (10) +| Platform | API | Description | +|----------|------|------| +| Baidu | baidu | Hot Search | +| The Paper | thepaper | Hot List | +| Toutiao | toutiao | Hot List | +| 36kr | 36kr | Hot List | +| QQ News | qq-news | Hot Topics | +| Sina | sina | Hot List | +| Sina News | sina-news | Hot Topics | +| NetEase News | netease-news | Hot Topics | +| Huxiu | huxiu | 24 Hours | +| Ifanr | ifanr | Quick News | + +### 💻 Tech/Developer Communities (8) +| Platform | API | Description | +|----------|------|------| +| IT Home | ithome | Hot List | +| IT Home Xijiayi | ithome-xijiayi | Latest Updates | +| Sspai | sspai | Hot List | +| CSDN | csdn | Ranking List | +| Juejin | juejin | Hot List | +| 51CTO | 51cto | Recommended | +| NodeSeek | nodeseek | Latest Updates | +| HelloGitHub | hellogithub | Trending | + +### 🎮 Gaming/ACG (5) +| Platform | API | Description | +|----------|------|------| +| Genshin | genshin | Latest News | +| MiyouShe | miyoushe | Latest News | +| Honkai 3 | honkai | Latest Updates | +| StarRail | starrail | Latest Updates | +| LOL | lol | Update Notice | + +### 📚 Reading/Culture (4) +| Platform | API | Description | +|----------|------|------| +| Jianshu | jianshu | Popular Recommendations | +| Guokr | guokr | Popular Articles | +| WeRead | weread | Rising List | +| Douban Movie | douban-movie | New Movies | + +### 🔧 Tools/Other (5) +| Platform | API | Description | +|----------|------|------| +| 52pojie | 52pojie | Ranking List | +| HostLoc | hostloc | Ranking List | +| Weather Alarm | weatheralarm | National Warning | +| Earthquake | earthquake | Earthquake Report | +| History Today | history | Month-Day | + +## 🚀 Quick Start + +### 1. Deploy Backend Service + +```bash +cd daily-hot-api +./deploy.sh +``` + +### 2. Install Dependencies + +```bash +pip install -r requirements.txt +``` + +### 3. Configure Environment Variables + +```bash +export DAILY_HOT_API_URL=http://localhost:6688 +``` + +### 4. Run + +```bash +python3 daily_hot_news.py --query "weibo hot" +``` + +## 💬 Usage Examples + +### Example 1: Query Single Platform + +```bash +# Query Weibo hot search +python3 daily_hot_news.py -q "微博热搜" + +# Query Zhihu hot list +python3 daily_hot_news.py -q "知乎热榜" + +# Query Bilibili trending +python3 daily_hot_news.py -q "B站热门" + +# Query Genshin latest +python3 daily_hot_news.py -q "原神" +``` + +**Output**: +``` +🔥 **Weibo Hot Search** +Update Time: 2026-02-05T19:00:00.000Z + +1. Wang Yibo Zhongshan Suit 5.2M +2. Xiao Zhan Shy Smile 4.8M +3. Weibo Night Red Carpet 4.5M +... +``` + +### Example 2: Browse All Platforms + +```bash +# View all supported hot search sources +python3 daily_hot_news.py --list +``` + +**Output**: +``` +📊 **Supported Hot Search Sources (54)** + +🎬 Video/Live Streaming +• Bilibili (bilibili) +• Douyin (douyin) +• Kuaishou (kuaishou) +• AcFun (acfun) +• Coolapk (coolapk) + +💬 Social Media +• Weibo (weibo) +• Zhihu (zhihu) +• V2EX (v2ex) +• NGA (ngabbs) +... +``` + +### Example 3: Cross-Platform Aggregation + +```bash +# View nationwide Top 10 +python3 daily_hot_news.py --cross-platform +``` + +**Output**: +``` +🌐 Nationwide Hot Topics TOP10 + +🥇 Genshin Update · 9.52M 【Highest on Bilibili】 + Score: 98 - 🔥 Super Popular + Discussed on 5 platforms + +🥈 Weibo Hot · 8.76M 【Highest on Weibo】 + Score: 95 - Nationwide Discussion + +🥉 Trump · 6.54M 【Highest on Weibo】 + Score: 89 - International Topic +... +``` + +### Example 4: Set Sentiment Monitoring + +```bash +# Set monitoring: AI topic, alert when exceeds 5M +python3 daily_hot_news.py --monitor "AI,500万" + +# View monitoring configuration +python3 daily_hot_news.py -q "查看我的监控" +``` + +**Output**: +``` +✅ **Monitoring Set!** + +Monitoring Keyword: AI +Hot Threshold: 5M +Platforms: All 54 + +I'll notify you when any topic exceeds the threshold! +``` + +### Example 5: Scheduled Push Configuration + +```bash +# Configure daily 8 AM Weibo hot push +"每天早上 8 点推送微博热搜" +``` + +**Output**: +``` +⏰ **Scheduled Push Configured!** + +Push Time: Daily 08:00 +Hot Source: Weibo Hot Search +Push Method: Feishu Message + +I'll automatically push Weibo hot to your Feishu every morning at 8! +``` + +## 🎮 Advanced Usage + +### Query by Category + +```bash +# View tech hot searches +"有什么科技热榜" + +# View gaming hot topics +"游戏有什么热点" +``` + +### Search Specific Topics + +```bash +# Search for topics containing keywords +"搜索 AI 相关热榜" +"查找 ChatGPT 热点" +``` + +### View Historical Data + +```bash +# View yesterday's Weibo hot +"微博昨天" + +# View history records +"微博历史" +``` + +### View Saved Data + +```bash +# View saved hot search statistics +"已保存了哪些数据" +``` + +### Cleanup Old Data + +```bash +# Reply "清理" or "是" to delete data older than 7 days +# Skill will automatically detect and prompt on startup +``` + +## 📁 File Structure + +``` +daily-hot-news/ +├── daily_hot_news.py # Main Entry +├── news_digest.py # Hot News Digest +├── industry_hot.py # Industry Vertical +├── personalized.py # Personalized Subscription +├── cross_platform.py # Cross-Platform Aggregation +├── sentiment_monitor.py # Sentiment Monitoring +├── api_client.py # API Client +├── formatter.py # Formatter +├── storage.py # Data Storage +├── config.py # Configuration +├── requirements.txt # Dependencies +└── README.md # This Document +``` + +## ⚙️ Configuration + +| Environment Variable | Default | Description | +|---------------------|---------|-------------| +| `DAILY_HOT_API_URL` | http://localhost:6688 | Backend API URL | +| `DAILY_HOT_CACHE_TTL` | 3600 | Cache Time (seconds) | +| `DAILY_HOT_MAX_ITEMS` | 20 | Max Items Returned | +| `DAILY_HOT_TIMEOUT` | 10 | Request Timeout (seconds) | + +### 🔧 Technical Architecture + +- **API Calls**: Each user request triggers a real-time call to DailyHotApi for the latest data +- **History**: Automatically saved to `data/{platform}/{date}.json` +- **Old Data Cleanup**: Automatically detects data older than 7 days on startup and prompts for cleanup + +## 🛡️ Security Note + +- ✅ No API keys or passwords included +- ✅ All sensitive configurations managed via environment variables +- ✅ User data stored locally, not uploaded to cloud + +## 📝 Changelog + +**v2.0.1** (2026-02-06) +- ✨ Optimized data fetching: always fetch latest data on each request +- ✨ Auto-save history records to local storage +- ✨ Auto-check and prompt cleanup for data older than 7 days on startup +- 🔧 Fixed API client compatibility issues + +**v2.0.0** (2026-02-05) +- ✨ Added 5 extended features +- ✨ Support 15 hot tag classifications +- ✨ Support 10 industry vertical hot searches +- ✨ Added personalized subscription feature +- ✨ Added cross-platform TOP10 aggregation +- ✨ Added sentiment monitoring alerts + +## 📄 License + +MIT License + +## 🤝 Acknowledgements + +- [DailyHotApi](https://github.com/imsyy/DailyHotApi) - Providing 54 hot search source APIs +- [OpenClaw](https://github.com/openclaw/openclaw) - AI Assistant Platform diff --git a/skills/daily-hot-news/SKILL.md b/skills/daily-hot-news/SKILL.md new file mode 100644 index 0000000..4021081 --- /dev/null +++ b/skills/daily-hot-news/SKILL.md @@ -0,0 +1,323 @@ +--- +name: daily-hot-news +description: 每日热榜技能 - 查询微博、知乎、B站、抖音等54个平台的热榜数据，支持定时推送和分类浏览。 +categories: + - information-aggregation + - daily-utility + - news +emoji: 🔥 +metadata: + openclaw: + requires: + bins: ["python3"] + install: + - id: python-deps + kind: exec + command: "cd /root/.openclaw/workspace/skills/daily-hot-news && python3 -m pip install requests aiohttp" + label: "安装Python依赖" +--- + +# 🔥 每日热榜 + +## 🎯 概述 + +提供 **54 个热榜源** 的本地化查询服务，基于 [DailyHotApi](https://github.com/imsyy/DailyHotApi) 项目。 + +**核心功能**： +- 📊 热榜查询 - 查询任意平台的热榜数据 +- 📋 分类浏览 - 列出所有支持的热榜源 +- 💾 历史记录 - 自动保存每日热榜数据 +- ⏰ 定时推送 - 自动推送热榜到飞书 + +--- + +## 🏗️ 架构设计 + +``` +用户请求 → DailyHotApi Skill → 本地 DailyHotApi 服务 → 返回格式化结果 +``` + +### 组件说明 +| 组件 | 职责 | +|------|------| +| **DailyHotApi 服务** | 独立运行，抓取/聚合热榜数据 | +| **DailyHotApi Skill** | OpenClaw 插件，处理用户请求 | + +--- + +## 📡 支持的热榜源（54个） + +### 🎬 视频/直播平台 +| 接口 | 名称 | +|------|------| +| bilibili | 哔哩哔哩 | +| acfun | AcFun | +| douyin | 抖音 | +| kuaishou | 快手 | +| coolapk | 酷安 | + +### 💬 社交媒体 +| 接口 | 名称 | +|------|------| +| weibo | 微博 | +| zhihu | 知乎 | +| zhihu-daily | 知乎日报 | +| tieba | 百度贴吧 | +| douban-group | 豆瓣讨论小组 | +| v2ex | V2EX | +| ngabbs | NGA | +| hupu | 虎扑 | + +### 📰 新闻资讯 +| 接口 | 名称 | +|------|------| +| baidu | 百度热搜 | +| thepaper | 澎湃新闻 | +| toutiao | 今日头条 | +| 36kr | 36氪 | +| qq-news | 腾讯新闻 | +| sina | 新浪网 | +| sina-news | 新浪新闻 | +| netease-news | 网易新闻 | +| huxiu | 虎嗅 | +| ifanr | 爱范儿 | + +### 💻 科技/技术社区 +| 接口 | 名称 | +|------|------| +| ithome | IT之家 | +| ithome-xijiayi | IT之家「喜加一」 | +| sspai | 少数派 | +| csdn | CSDN | +| juejin | 稀土掘金 | +| 51cto | 51CTO | +| nodeseek | NodeSeek | +| hellogithub | HelloGitHub | + +### 🎮 游戏/ACG +| 接口 | 名称 | +|------|------| +| genshin | 原神 | +| miyoushe | 米游社 | +| honkai | 崩坏3 | +| starrail | 崩坏：星穹铁道 | +| lol | 英雄联盟 | + +### 📚 阅读/文化 +| 接口 | 名称 | +|------|------| +| jianshu | 简书 | +| guokr | 果壳 | +| weread | 微信读书 | +| douban-movie | 豆瓣电影 | + +### 🔧 工具/其他 +| 接口 | 名称 | +|------|------| +| 52pojie | 吾爱破解 | +| hostloc | 全球主机交流 | +| weatheralarm | 中央气象台 | +| earthquake | 中国地震台 | +| history | 历史上的今天 | + +--- + +## 🚀 部署说明 + +### 1. PM2 方式管理（推荐） + +DailyHotApi 服务使用 PM2 管理，确保稳定运行。 + +```bash +cd /root/.openclaw/workspace/skills/daily-hot-api + +# 部署并启动服务 +./deploy.sh + +# 查看状态 +./deploy.sh status + +# 重启服务 +./deploy.sh restart + +# 停止服务 +./deploy.sh stop + +# 查看日志 +./deploy.sh logs +``` + +**服务地址**: `http://localhost:6688` + +### 2. 配置环境变量 + +```bash +export DAILY_HOT_API_URL=http://localhost:6688 +``` + +### 3. 安装 Skill 依赖 + +```bash +cd /root/.openclaw/workspace/skills/daily-hot-news +pip install requests aiohttp +``` + +--- + +## 🎮 使用示例 + +### 查询热榜 +``` +用户: 微博热搜 +Skill: 调用 /weibo → 返回 Top 10 热榜 +``` + +### 查看所有热榜 +``` +用户: 有什么热榜 +Skill: 返回 54 个热榜源列表 +``` + +### 查询历史热榜 +``` +用户: 微博历史 +Skill: 显示之前保存的微博热榜记录 +``` + +### 查看已保存数据 +``` +用户: 已保存了哪些数据 +Skill: 返回所有已保存的热榜数据统计 +``` + +### 定时推送 +``` +用户: 每天早上8点推送B站热门 +Skill: 设置 cron 任务 → 每日调用 /bilibili → 推送到飞书 +``` + +--- + +## 💾 数据存储 + +### 存储位置 +所有热榜数据保存在： +``` +/root/.openclaw/workspace/skills/daily-hot-news/data/ +``` + +### 文件结构 +``` +data/ +├── weibo/ +│ ├── 2026-02-05.json +│ └── 2026-02-04.json +├── zhihu/ +│ └── 2026-02-05.json +└── ... +``` + +### 配置项 + +| 环境变量 | 默认值 | 说明 | +|----------|--------|------| +| `DAILY_HOT_DATA_DIR` | data/ | 数据存储目录 | +| `DAILY_HOT_AUTO_SAVE` | true | 是否自动保存热榜数据 | + +### 管理命令 +```bash +# 查看已保存的数据统计 +python3 storage.py + +# 清理 30 天前的旧数据 +python3 storage.py --clear 30 +``` + +--- + +## 📊 响应格式 + +### 热榜列表响应 +```json +{ + "platform": "微博", + "updateTime": "2026-02-05 17:00:00", + "data": [ + { + "rank": 1, + "title": "热搜标题", + "hot": "1234万", + "url": "https://..." + } + ] +} +``` + +--- + +## ⚙️ 配置项 + +| 环境变量 | 默认值 | 说明 | +|----------|--------|------| +| `DAILY_HOT_API_URL` | http://localhost:6688 | DailyHotApi 服务地址 | +| `DAILY_HOT_CACHE_TTL` | 3600 | 缓存时间（秒） | +| `DAILY_HOT_MAX_ITEMS` | 20 | 返回最大条数 | + +--- + +## 📊 资源占用 + +| 组件 | 内存 | CPU | +|------|------|-----| +| DailyHotApi 服务 | ~200MB | 极低 | +| DailyHotApi Skill | <10MB | 可忽略 | + +**总计**: <250MB，对服务器无压力 + +--- + +## 🔧 故障排查 + +### 问题: 服务无法连接 +```bash +# 检查 PM2 状态 +./deploy.sh status + +# 查看日志 +./deploy.sh logs + +# 重启服务 +./deploy.sh restart +``` + +--- + +## 📁 文件结构 + +``` +daily-hot-news/ +├── SKILL.md # 本说明书 +├── daily_hot_news.py # 核心 Skill 脚本 +├── api_client.py # API 客户端封装 +├── formatter.py # 响应格式化 +├── config.py # 配置管理 +├── storage.py # 数据存储模块 +├── data/ # 热榜数据存储目录 +├── README.md # 快速开始 +└── requirements.txt # 依赖列表 +``` + +--- + +## 📝 更新日志 + +**v1.1.0** (2026-02-05) +- ✨ 新增数据存储功能 +- ✨ 支持自动保存每日热榜 +- ✨ 支持查询历史记录 +- ✨ 新增数据统计命令 + +**v1.0.0** (2026-02-05) +- 初始版本 +- 支持 54 个热榜源 +- 基础查询和定时推送功能 diff --git a/skills/daily-hot-news/__init__.py b/skills/daily-hot-news/__init__.py new file mode 100644 index 0000000..20625ab --- /dev/null +++ b/skills/daily-hot-news/__init__.py @@ -0,0 +1,89 @@ +""" +每日热榜技能扩展模块 + +整合了以下功能： +- news_digest: 热点新闻摘要（15种标签分类） +- industry_hot: 行业热榜垂直（十大行业分类） +- personalized: 个性化订阅（关键词/平台/排除项配置） +- cross_platform: 跨平台聚合（TOP10聚合） +- sentiment_monitor: 舆情监控（关键词过滤） +""" + +from .news_digest import ( + NewsDigest, + DigestConfig, + DigestMode, + TAG_MAPPING, + ALL_TAGS, + create_digest as create_news_digest +) + +from .industry_hot import ( + IndustryHot, + IndustryConfig, + IndustryMode, + INDUSTRIES, + ALL_INDUSTRIES, + create_industry_hot +) + +from .personalized import ( + PersonalizedSubscription, + UserPreferences, + SubscriptionMode, + KEYWORD_OPTIONS, + PLATFORM_OPTIONS, + EXCLUDE_OPTIONS, + create_personalized +) + +from .cross_platform import ( + CrossPlatformAggregator, + CrossPlatformConfig, + create_cross_platform +) + +from .sentiment_monitor import ( + SentimentMonitor, + SentimentConfig, + SentimentType, + create_sentiment_monitor +) + +__all__ = [ + # News Digest + "NewsDigest", + "DigestConfig", + "DigestMode", + "TAG_MAPPING", + "ALL_TAGS", + "create_news_digest", + + # Industry Hot + "IndustryHot", + "IndustryConfig", + "IndustryMode", + "INDUSTRIES", + "ALL_INDUSTRIES", + "create_industry_hot", + + # Personalized + "PersonalizedSubscription", + "UserPreferences", + "SubscriptionMode", + "KEYWORD_OPTIONS", + "PLATFORM_OPTIONS", + "EXCLUDE_OPTIONS", + "create_personalized", + + # Cross Platform + "CrossPlatformAggregator", + "CrossPlatformConfig", + "create_cross_platform", + + # Sentiment Monitor + "SentimentMonitor", + "SentimentConfig", + "SentimentType", + "create_sentiment_monitor" +] diff --git a/skills/daily-hot-news/api_client.py b/skills/daily-hot-news/api_client.py new file mode 100644 index 0000000..9b923be --- /dev/null +++ b/skills/daily-hot-news/api_client.py @@ -0,0 +1,432 @@ +# -*- coding: utf-8 -*- +""" +DailyHotApi Skill - API 客户端封装 +""" + +import aiohttp +import asyncio +import subprocess +import os +from typing import Optional, Dict, List, Any +from datetime import datetime +from config import config +from storage import storage # 导入存储模块 + +# 部署状态存储 +_deployment_status = { + "is_deploying": False, + "last_check": None, + "needs_deployment": False, + "message": "" +} + + +class HotSource: + """热榜源定义""" + + def __init__(self, id: str, name: str, category: str, description: str = ""): + self.id = id + self.name = name + self.category = category + self.description = description + + +# 54 个热榜源定义 +HOT_SOURCES = { + # 视频/直播平台 + "bilibili": HotSource("bilibili", "哔哩哔哩", "video", "热门榜"), + "acfun": HotSource("acfun", "AcFun", "video", "排行榜"), + "douyin": HotSource("douyin", "抖音", "video", "热点榜"), + "kuaishou": HotSource("kuaishou", "快手", "video", "热点榜"), + "coolapk": HotSource("coolapk", "酷安", "video", "热榜"), + + # 社交媒体 + "weibo": HotSource("weibo", "微博", "social", "热搜榜"), + "zhihu": HotSource("zhihu", "知乎", "social", "热榜"), + "zhihu-daily": HotSource("zhihu-daily", "知乎日报", "social", "推荐榜"), + "tieba": HotSource("tieba", "百度贴吧", "social", "热议榜"), + "douban-group": HotSource("douban-group", "豆瓣讨论小组", "social", "讨论精选"), + "v2ex": HotSource("v2ex", "V2EX", "social", "主题榜"), + "ngabbs": HotSource("ngabbs", "NGA", "social", "热帖"), + "hupu": HotSource("hupu", "虎扑", "social", "步行街热帖"), + + # 新闻资讯 + "baidu": HotSource("baidu", "百度", "news", "热搜榜"), + "thepaper": HotSource("thepaper", "澎湃新闻", "news", "热榜"), + "toutiao": HotSource("toutiao", "今日头条", "news", "热榜"), + "36kr": HotSource("36kr", "36氪", "news", "热榜"), + "qq-news": HotSource("qq-news", "腾讯新闻", "news", "热点榜"), + "sina": HotSource("sina", "新浪网", "news", "热榜"), + "sina-news": HotSource("sina-news", "新浪新闻", "news", "热点榜"), + "netease-news": HotSource("netease-news", "网易新闻", "news", "热点榜"), + "huxiu": HotSource("huxiu", "虎嗅", "news", "24小时"), + "ifanr": HotSource("ifanr", "爱范儿", "news", "快讯"), + + # 科技/技术社区 + "ithome": HotSource("ithome", "IT之家", "tech", "热榜"), + "ithome-xijiayi": HotSource("ithome-xijiayi", "IT之家「喜加一」", "tech", "最新动态"), + "sspai": HotSource("sspai", "少数派", "tech", "热榜"), + "csdn": HotSource("csdn", "CSDN", "tech", "排行榜"), + "juejin": HotSource("juejin", "稀土掘金", "tech", "热榜"), + "51cto": HotSource("51cto", "51CTO", "tech", "推荐榜"), + "nodeseek": HotSource("nodeseek", "NodeSeek", "tech", "最新动态"), + "hellogithub": HotSource("hellogithub", "HelloGitHub", "tech", "Trending"), + + # 游戏/ACG + "genshin": HotSource("genshin", "原神", "game", "最新消息"), + "miyoushe": HotSource("miyoushe", "米游社", "game", "最新消息"), + "honkai": HotSource("honkai", "崩坏3", "game", "最新动态"), + "starrail": HotSource("starrail", "崩坏：星穹铁道", "game", "最新动态"), + "lol": HotSource("lol", "英雄联盟", "game", "更新公告"), + + # 阅读/文化 + "jianshu": HotSource("jianshu", "简书", "reading", "热门推荐"), + "guokr": HotSource("guokr", "果壳", "reading", "热门文章"), + "weread": HotSource("weread", "微信读书", "reading", "飙升榜"), + "douban-movie": HotSource("douban-movie", "豆瓣电影", "reading", "新片榜"), + + # 工具/其他 + "52pojie": HotSource("52pojie", "吾爱破解", "tool", "榜单"), + "hostloc": HotSource("hostloc", "全球主机交流", "tool", "榜单"), + "weatheralarm": HotSource("weatheralarm", "中央气象台", "tool", "全国气象预警"), + "earthquake": HotSource("earthquake", "中国地震台", "tool", "地震速报"), + "history": HotSource("history", "历史上的今天", "tool", "月-日"), +} + + +class DailyHotApiClient: + """DailyHotApi 客户端""" + + def __init__(self): + self.api_url = config.api_url + self.timeout = config.timeout + + async def fetch_hot_list(self, source_id: str, use_cache: bool = False) -> Optional[Dict[str, Any]]: + """ + 获取热榜数据 + + Args: + source_id: 热榜源 ID（如 weibo, zhihu, bilibili） + use_cache: 是否使用缓存（默认False，每次获取最新数据） + + Returns: + 热榜数据字典或 None（失败时） + """ + # 获取热榜源信息 + source = HOT_SOURCES.get(source_id) + if not source: + return None + + # 构建请求 URL + url = config.get_api_url(source_id) + + # 步骤1：检查API是否可用 + print(f"[DailyHotApi] 正在连接 {source.name}...") + api_available = await check_api_availability(url) + + if not api_available: + # API不可用，尝试部署 + print(f"[DailyHotApi] ⚠️ 后端服务不可用，尝试自动部署...") + + # 触发部署 + deploy_result = await deploy_daily_hot_api() + + if deploy_result["success"]: + # 部署成功，等待服务启动 + print(f"[DailyHotApi] ⏳ 等待服务启动 (5秒)...") + await asyncio.sleep(5) + + # 再次检查API + api_available = await check_api_availability(url) + + if not api_available: + # 仍然不可用，可能需要更多时间 + print(f"[DailyHotApi] ⏳ 服务可能需要更多时间启动，再次等待 (10秒)...") + await asyncio.sleep(10) + api_available = await check_api_availability(url) + + # 返回部署状态 + if deploy_result["success"]: + return { + "success": True, + "deploy_message": deploy_result["message"], + "is_deployed": True, + "data": None, + "message": "🎉 后端服务已部署成功！请稍后再次尝试获取热榜数据。" + } + else: + return { + "success": False, + "deploy_message": "❌ 自动部署失败", + "steps": deploy_result.get("steps", []), + "data": None, + "message": f"⚠️ 无法连接后端服务\n\n自动部署失败：{deploy_result['message']}\n\n请手动部署：\n1. cd /root/.openclaw\n2. git clone https://github.com/imsyy/DailyHotApi.git\n3. cd DailyHotApi\n4. bash deploy.sh" + } + + # API可用，获取数据 + try: + timeout_obj = aiohttp.ClientTimeout(total=self.timeout) + async with aiohttp.ClientSession(timeout=timeout_obj) as session: + async with session.get(url) as response: + if response.status != 200: + return None + + result = await response.json() + + # 检查返回格式 + if result.get("code") != 200: + return None + + # 格式化数据 + data = { + "platform": source.name, + "category": source.category, + "source_id": source_id, + "update_time": result.get("updateTime"), + "from_cache": result.get("fromCache", False), + "total": result.get("total", 0), + "data": self._format_items(result.get("data", [])), + } + + # 每次获取最新数据后，保存到本地历史记录 + save_result = storage.save_hot_list(source_id, data) + if save_result: + print(f"[DailyHotApi] ✅ 已保存 {source.name} 热榜数据到历史记录") + + return data + + except Exception as e: + print(f"[DailyHotApi] Error fetching {source_id}: {e}") + return None + + async def get_hot榜单(self, source_id: str, limit: int = 10) -> Optional[List[Dict]]: + """ + 获取热榜条目列表（兼容旧接口） + + Args: + source_id: 热榜源 ID + limit: 返回条目数限制 + + Returns: + 热榜条目列表或 None + """ + data = await self.fetch_hot_list(source_id) + if data: + return data.get("data", [])[:limit] + return None + + def _format_items(self, items: List[Dict]) -> List[Dict]: + """格式化热榜条目""" + formatted = [] + for i, item in enumerate(items[:config.max_items], 1): + formatted.append({ + "rank": i, + "title": item.get("title", ""), + "desc": item.get("desc", ""), + "hot": item.get("hot", ""), + "url": item.get("url", ""), + "mobile_url": item.get("mobileUrl", ""), + }) + return formatted + + def get_all_sources(self) -> Dict[str, Dict]: + """获取所有热榜源""" + return { + source_id: { + "id": source_id, + "name": source.name, + "category": source.category, + "description": source.description, + } + for source_id, source in HOT_SOURCES.items() + } + + def get_sources_by_category(self) -> Dict[str, List[Dict]]: + """按类别获取热榜源""" + categories = {} + for source_id, source in HOT_SOURCES.items(): + if source.category not in categories: + categories[source.category] = [] + categories[source.category].append({ + "id": source_id, + "name": source.name, + "description": source.description, + }) + return categories + + def search_sources(self, query: str) -> List[Dict]: + """搜索热榜源""" + query = query.lower() + results = [] + for source_id, source in HOT_SOURCES.items(): + if (query in source.name.lower() or + query in source.id.lower() or + query in source.category.lower()): + results.append({ + "id": source_id, + "name": source.name, + "category": source.category, + }) + return results + + +# 全局客户端实例 +api_client = DailyHotApiClient() + + +# ============================================ +# 自动检测和部署功能 +# ============================================ + +async def check_api_availability(url: str, timeout: int = 3) -> bool: + """ + 检查API是否可用 + + Args: + url: API地址 + timeout: 超时时间（秒） + + Returns: + True表示可用，False表示不可用 + """ + try: + timeout_obj = aiohttp.ClientTimeout(total=timeout) + async with aiohttp.ClientSession(timeout=timeout_obj) as session: + async with session.get(url, allow_redirects=True) as response: + return response.status == 200 + except Exception as e: + print(f"[DailyHotApi] API不可用: {e}") + return False + + +async def deploy_daily_hot_api() -> Dict[str, Any]: + """ + 自动部署DailyHotApi后端服务 + + Returns: + 部署结果字典 + """ + global _deployment_status + + if _deployment_status["is_deploying"]: + return { + "success": False, + "message": "部署正在进行中，请稍候...", + "is_deploying": True + } + + _deployment_status["is_deploying"] = True + _deployment_status["message"] = "🚀 正在自动部署DailyHotApi后端服务..." + + result = { + "success": False, + "message": "", + "steps": [] + } + + try: + # 检查是否已安装 + daily_hot_path = "/root/.openclaw/DailyHotApi" + if not os.path.exists(daily_hot_path): + # 步骤1：克隆仓库 + step_msg = "📦 正在克隆DailyHotApi仓库..." + print(step_msg) + result["steps"].append(step_msg) + + clone_cmd = ["git", "clone", "https://github.com/imsyy/DailyHotApi.git", daily_hot_path] + clone_proc = await asyncio.create_subprocess_exec( + *clone_cmd, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE + ) + stdout, stderr = await clone_proc.communicate() + + if clone_proc.returncode != 0: + error_msg = f"❌ 克隆失败: {stderr.decode()}" + print(error_msg) + result["steps"].append(error_msg) + result["message"] = "部署失败：无法克隆仓库" + _deployment_status["is_deploying"] = False + return result + + result["steps"].append("✅ 克隆成功") + else: + result["steps"].append("✅ DailyHotApi已存在，跳过克隆") + + # 步骤2：部署服务 + if os.path.exists(daily_hot_path): + step_msg = "🔧 正在部署DailyHotApi服务..." + print(step_msg) + result["steps"].append(step_msg) + + deploy_script = os.path.join(daily_hot_path, "deploy.sh") + deploy_cmd = ["bash", deploy_script] + + deploy_proc = await asyncio.create_subprocess_exec( + *deploy_cmd, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE + ) + stdout, stderr = await deploy_proc.communicate() + + deploy_output = stdout.decode() + stderr.decode() + + if deploy_proc.returncode == 0: + result["steps"].append("✅ 部署成功") + result["success"] = True + result["message"] = "🎉 DailyHotApi后端服务部署成功！正在启动..." + _deployment_status["message"] = result["message"] + else: + error_msg = f"❌ 部署失败: {deploy_output}" + print(error_msg) + result["steps"].append(error_msg) + result["message"] = "部署失败，请手动检查" + + except Exception as e: + error_msg = f"❌ 部署异常: {str(e)}" + print(error_msg) + result["steps"].append(error_msg) + result["message"] = f"部署异常: {str(e)}" + + finally: + _deployment_status["is_deploying"] = False + _deployment_status["last_check"] = datetime.now().isoformat() + + return result + + +def get_deployment_status() -> Dict[str, Any]: + """ + 获取当前部署状态 + + Returns: + 部署状态字典 + """ + return { + "is_deploying": _deployment_status["is_deploying"], + "last_check": _deployment_status["last_check"], + "message": _deployment_status["message"] + } + + +async def test_connection() -> bool: + """测试 API 连接""" + try: + timeout_obj = aiohttp.ClientTimeout(total=5) + async with aiohttp.ClientSession(timeout=timeout_obj) as session: + async with session.get(f"{config.api_url}/") as response: + return response.status == 200 + except Exception: + return False + + +if __name__ == "__main__": + import sys + print("Testing DailyHotApi connection...") + + if asyncio.run(test_connection()): + print("✓ Service is running") + else: + print("✗ Service not available") + print(f" URL: {config.api_url}") + sys.exit(1) diff --git a/skills/daily-hot-news/config.py b/skills/daily-hot-news/config.py new file mode 100644 index 0000000..2cfbc90 --- /dev/null +++ b/skills/daily-hot-news/config.py @@ -0,0 +1,75 @@ +# -*- coding: utf-8 -*- +""" +每日热榜 Skill - 配置管理 +""" + +import os +from typing import Optional + + +class Config: + """配置管理类""" + + def __init__(self): + # 每日热榜服务地址 + self.api_url: str = os.getenv( + "DAILY_HOT_API_URL", + "http://localhost:6688" + ) + + # 缓存时间（秒） + self.cache_ttl: int = int(os.getenv("DAILY_HOT_CACHE_TTL", "3600")) + + # 返回最大条数 + self.max_items: int = int(os.getenv("DAILY_HOT_MAX_ITEMS", "20")) + + # 请求超时时间（秒） + self.timeout: int = int(os.getenv("DAILY_HOT_TIMEOUT", "10")) + + # 启用调试模式 + self.debug: bool = os.getenv("DAILY_HOT_DEBUG", "false").lower() == "true" + + # 数据存储路径 + self.data_dir: str = os.getenv( + "DAILY_HOT_DATA_DIR", + "/root/.openclaw/workspace/skills/daily-hot-news/data" + ) + + # 是否自动保存每日热榜 + self.auto_save: bool = os.getenv("DAILY_HOT_AUTO_SAVE", "true").lower() == "true" + + def get_api_url(self, endpoint: str = "") -> str: + """获取完整的 API 地址""" + # 移除末尾斜杠 + base_url = self.api_url.rstrip("/") + if endpoint: + return f"{base_url}/{endpoint}" + return base_url + + def is_service_available(self) -> bool: + """检查服务是否配置""" + return bool(self.api_url) + + def get_data_path(self, source_id: str = "", date_str: str = "") -> str: + """获取数据文件路径""" + import os + from datetime import datetime + + if not date_str: + date_str = datetime.now().strftime("%Y-%m-%d") + + data_dir = self.data_dir + + if source_id: + # 按热榜源分目录存储 + source_dir = os.path.join(data_dir, source_id) + os.makedirs(source_dir, exist_ok=True) + return os.path.join(source_dir, f"{date_str}.json") + else: + # 主目录 + os.makedirs(data_dir, exist_ok=True) + return os.path.join(data_dir, f"all_{date_str}.json") + + +# 全局配置实例 +config = Config() diff --git a/skills/daily-hot-news/cross_platform.py b/skills/daily-hot-news/cross_platform.py new file mode 100644 index 0000000..b687ae4 --- /dev/null +++ b/skills/daily-hot-news/cross_platform.py @@ -0,0 +1,218 @@ +""" +跨平台聚合 - Cross Platform Aggregation Module + +功能： +- 直接用全量数据做TOP10聚合 +- 跨平台热点排行 +""" + +from typing import List, Dict, Optional, Any +from dataclasses import dataclass +import asyncio + + +@dataclass +class CrossPlatformConfig: + """跨平台聚合配置""" + total_items: int = 10 # 总条目数限制（默认TOP10） + min_hot_score: float = 0 # 最小热度阈值 + merge_strategy: str = "score" # 合并策略：score(按热度), time(按时间) + include_platforms: List[str] = None # 包含的平台（None表示全部） + exclude_platforms: List[str] = None # 排除的平台 + + +class CrossPlatformAggregator: + """跨平台聚合类""" + + def __init__(self, api_client=None, formatter=None): + """ + 初始化跨平台聚合 + + Args: + api_client: API客户端实例（可选） + formatter: 格式化器实例（可选） + """ + self.api_client = api_client + self.formatter = formatter + self.all_platforms = self._get_default_platforms() + + def _get_default_platforms(self) -> List[str]: + """获取默认的全部平台列表""" + return [ + "weibo", "zhihu", "douban-group", "douban-movie", + "ithome", "36kr", "sspai", "csdn", "juejin", + "genshin", "miyoushe", "bilibili", "hupu", + "sina-news", "netease-news", "qq-news", + "sina-money", "eastmoney", "xueqiu", + "autohome", "懂车帝", "mafengwo", "ctrip", + "dianping", "xiaohongshu", "weibo" + ] + + async def fetch_all_hot_data(self, limit_per_platform: int = 10) -> List[Dict[str, Any]]: + """ + 一次性获取全部平台的热榜数据 + + Args: + limit_per_platform: 每个平台获取的条目数 + + Returns: + 全部平台的热榜数据列表 + """ + all_items = [] + + if self.api_client: + for platform in self.all_platforms: + try: + items = await self.api_client.get_hot榜单(platform, limit=limit_per_platform) + if items: + for item in items: + item["source_platform"] = platform + all_items.extend(items) + except Exception as e: + print(f"Error fetching {platform}: {e}") + continue + + return all_items + + async def aggregate_top_hot(self, config: Optional[CrossPlatformConfig] = None) -> Dict[str, Any]: + """ + 聚合跨平台TOP热点（新版：先全部获取，再聚合） + + Args: + config: 配置对象（可选） + + Returns: + 聚合后的热榜数据 + """ + if config is None: + config = CrossPlatformConfig() + + # 步骤1：全部获取 + all_items = await self.fetch_all_hot_data() + + # 步骤2：聚合处理 + # 平台过滤 + if config.include_platforms: + all_items = [item for item in all_items if item.get("source_platform") in config.include_platforms] + if config.exclude_platforms: + all_items = [item for item in all_items if item.get("source_platform") not in config.exclude_platforms] + + # 热度过滤 + if config.min_hot_score > 0: + all_items = [item for item in all_items if (item.get("hot", 0) or item.get("score", 0)) >= config.min_hot_score] + + # 去重和排序 + merged_items = self._merge_and_sort(all_items, config.merge_strategy) + + # 限制总数 + merged_items = merged_items[:config.total_items] + + return { + "total_items": len(merged_items), + "items": merged_items + } + + def _merge_and_sort(self, items: List[Dict], strategy: str = "score") -> List[Dict]: + """ + 合并、去重和排序 + + Args: + items: 条目列表 + strategy: 排序策略 + + Returns: + 处理后的条目列表 + """ + if not items: + return [] + + # 按标题去重 + seen_titles = set() + unique_items = [] + + for item in items: + title = item.get("title", "").strip().lower() + if title and title not in seen_titles: + seen_titles.add(title) + unique_items.append(item) + + # 根据策略排序 + if strategy == "score": + unique_items.sort(key=lambda x: x.get("hot", 0) or x.get("score", 0), reverse=True) + elif strategy == "time": + unique_items.sort(key=lambda x: x.get("time", "") or "", reverse=True) + + return unique_items + + def format_aggregation_response(self, agg_data: Dict[str, Any]) -> str: + """ + 格式化聚合响应 + + Args: + agg_data: 聚合数据 + + Returns: + 格式化的响应文本 + """ + items = agg_data.get("items", []) + + if not items: + return "❌ 暂无热点数据" + + response = "🏆 **跨平台热点TOP10**\n" + response += f"共 {agg_data['total_items']} 条热点\n" + response += "-" * 40 + "\n\n" + + for i, item in enumerate(items, 1): + title = item.get("title", "无标题") + hot = item.get("hot", item.get("score", "")) + platform = item.get("source_platform", "") + + response += f"{i}. {title}\n" + if hot: + response += f" 🔥 热度: {hot}" + if platform: + response += f" | 📱 {platform}" + response += "\n\n" + + return response + + async def process_user_request(self, user_input: str = None) -> Dict[str, Any]: + """ + 处理用户请求 + + Args: + user_input: 用户输入（可选） + + Returns: + 处理结果 + """ + # 获取聚合数据 + agg_data = await self.aggregate_top_hot() + + # 格式化响应 + response_text = self.format_aggregation_response(agg_data) + + return { + "action": "show_top_hot", + "data": agg_data, + "message": response_text + } + + +# 便捷函数 +async def create_cross_platform(api_client=None, formatter=None) -> CrossPlatformAggregator: + """创建跨平台聚合实例""" + return CrossPlatformAggregator(api_client=api_client, formatter=formatter) + + +if __name__ == "__main__": + # 测试代码 + async def test(): + aggregator = await create_cross_platform() + + print("获取跨平台热点TOP10...") + result = await aggregator.process_user_request() + print(result["message"]) + + asyncio.run(test()) diff --git a/skills/daily-hot-news/daily_hot_news.py b/skills/daily-hot-news/daily_hot_news.py new file mode 100644 index 0000000..a24c737 --- /dev/null +++ b/skills/daily-hot-news/daily_hot_news.py @@ -0,0 +1,369 @@ +""" +每日热榜技能主入口示例 + +这个文件展示如何将三个扩展功能集成到主Skill中。 +在实际使用时，可以根据需要调整和整合。 +""" + +import asyncio +from typing import Dict, Any, Optional + +# 支持相对导入（作为包的一部分）和绝对导入（直接运行） +try: + from .news_digest import NewsDigest, DigestConfig, create_digest as create_news_digest + from .industry_hot import IndustryHot, IndustryConfig, create_industry_hot + from .personalized import PersonalizedSubscription, UserPreferences, create_personalized + from .storage import storage + from .api_client import api_client, check_api_availability, deploy_daily_hot_api, get_deployment_status + from .config import config +except ImportError: + from news_digest import NewsDigest, DigestConfig, create_digest as create_news_digest + from industry_hot import IndustryHot, IndustryConfig, create_industry_hot + from personalized import PersonalizedSubscription, UserPreferences, create_personalized + from storage import storage + from api_client import api_client, check_api_availability, deploy_daily_hot_api, get_deployment_status + from config import config + + +class DailyHotNewsSkill: + """每日热榜技能主类""" + + def __init__(self, api_client=None, formatter=None): + """ + 初始化技能 + + Args: + api_client: API客户端实例（用于获取热榜数据） + formatter: 格式化器实例（用于格式化输出） + """ + self.api_client = api_client + self.formatter = formatter + + # 初始化各功能模块 + self.news_digest: Optional[NewsDigest] = None + self.industry_hot: Optional[IndustryHot] = None + self.personalized: Optional[PersonalizedSubscription] = None + + # 标记是否已检查过旧数据（避免每次都提示） + self._old_data_checked = False + self._old_data_notification = None + + async def initialize(self): + """初始化各模块""" + self.news_digest = await create_news_digest( + api_client=self.api_client, + formatter=self.formatter + ) + self.industry_hot = await create_industry_hot( + api_client=self.api_client, + formatter=self.formatter + ) + self.personalized = await create_personalized( + api_client=self.api_client, + formatter=self.formatter + ) + + # 启动时检查是否有7天前的旧数据 + await self._check_old_data() + + async def _check_old_data(self): + """检查并提示用户清理7天前的旧数据""" + if self._old_data_checked: + return + + old_files = storage.get_old_data_files(days=7) + if old_files: + # 统计 + sources = set(f["source_id"] for f in old_files) + print(f"\n⚠️ 发现 {len(old_files)} 个旧数据文件（7天前）") + print(f" 涉及平台: {', '.join(list(sources)[:5])}...") + print(f" 示例文件: {old_files[0]['date_str']} - {old_files[0]['source_id']}") + + # 返回提示信息给用户 + self._old_data_notification = { + "has_old_data": True, + "count": len(old_files), + "sources": list(sources), + "message": f"""🗑️ **发现旧热榜数据** + +检测到 {len(old_files)} 个热榜数据文件已超过7天未清理，涉及 {len(sources)} 个平台。 + +是否需要清理这些旧数据？ +- 回复"**清理**"或"**是**"：删除7天前的所有旧数据 +- 回复"**跳过**"或"**否**"：保留数据，下次启动不再提醒""" + } + else: + self._old_data_notification = None + + self._old_data_checked = True + + def get_old_data_notification(self) -> Optional[Dict[str, Any]]: + """获取旧数据清理提示（如果有）""" + return self._old_data_notification + + async def handle_old_data_cleanup(self, confirm: bool = False) -> Dict[str, Any]: + """ + 处理旧数据清理 + + Args: + confirm: 是否确认清理 + + Returns: + 清理结果 + """ + if not confirm: + return { + "action": "ask_confirm", + "message": "请确认是否清理7天前的旧数据？\n回复\"清理\"确认，回复\"跳过\"取消。" + } + + old_files = storage.get_old_data_files(days=7) + if not old_files: + return { + "action": "show_message", + "message": "✅ 没有需要清理的旧数据" + } + + deleted = storage.cleanup_old_files(old_files) + return { + "action": "show_message", + "message": f"✅ 成功清理 {deleted} 个旧数据文件" + } + + async def handle_request(self, user_input: str, intent: str = None) -> Dict[str, Any]: + """ + 处理用户请求 + + Args: + user_input: 用户输入 + intent: 意图（可选，用于路由到对应功能） + + Returns: + 处理结果 + """ + user_input_lower = user_input.lower() + + # 检查是否是清理旧数据命令 + cleanup_keywords = ["清理", "删除旧数据", "清除缓存", "clean"] + if any(kw in user_input_lower for kw in cleanup_keywords): + return await self.handle_old_data_cleanup(confirm=True) + + # 检查是否是获取热榜的请求 + is_hot_request = any([ + intent == "news_digest", + self._is_news_digest_request(user_input_lower), + intent == "industry_hot", + self._is_industry_hot_request(user_input_lower), + ]) + + if is_hot_request: + # 先检查API是否可用 + api_url = config.api_url + print(f"\n[DailyHotSkill] 检查后端服务可用性: {api_url}") + + api_available = await check_api_availability(api_url) + + if not api_available: + print(f"[DailyHotSkill] ⚠️ 后端服务不可用，触发自动部署...") + deploy_result = await deploy_daily_hot_api() + + # 返回部署状态给用户 + if deploy_result["success"]: + # 构建部署步骤消息 + steps_text = "" + for step in deploy_result.get("steps", []): + steps_text += f"{step}\n" + + return { + "action": "show_deploying", + "message": f"""🚀 **正在自动部署后端服务** + +{steps_text} + +⏳ **请稍候...** + +后端服务正在启动，预计需要1-2分钟。 + +**部署完成后**，请再次发送请求获取热榜数据。 + +--- +💡 如果自动部署失败，请手动执行： +1. `cd /root/.openclaw` +2. `git clone https://github.com/imsyy/DailyHotApi.git` +3. `cd DailyHotApi` +4. `bash deploy.sh`""", + "deploy_success": True, + "steps": deploy_result.get("steps", []) + } + else: + # 部署失败，返回错误和手动部署指南 + steps_text = "" + for step in deploy_result.get("steps", []): + steps_text += f"{step}\n" + + return { + "action": "show_deploy_failed", + "message": f"""⚠️ **后端服务不可用** + +自动部署失败，需要手动部署。 + +**部署步骤：** +```bash +cd /root/.openclaw +git clone https://github.com/imsyy/DailyHotApi.git +cd DailyHotApi +bash deploy.sh +```""", + "deploy_success": False, + "steps": deploy_result.get("steps", []) + } + + # 路由到对应功能 + if intent == "news_digest" or self._is_news_digest_request(user_input_lower): + return await self._handle_news_digest(user_input) + + elif intent == "industry_hot" or self._is_industry_hot_request(user_input_lower): + return await self._handle_industry_hot(user_input) + + elif intent == "personalized" or self._is_personalized_request(user_input_lower): + return await self._handle_personalized(user_input) + + else: + # 默认返回功能选择引导 + return await self._show_main_menu() + + def _is_news_digest_request(self, user_input: str) -> bool: + """判断是否为新闻摘要请求""" + keywords = ["热点", "摘要", "标签", "科技", "游戏", "娱乐", "财经", "新闻"] + return any(kw in user_input for kw in keywords) + + def _is_industry_hot_request(self, user_input: str) -> bool: + """判断是否为行业热榜请求""" + keywords = ["行业", "汽车", "金融", "医疗", "旅游", "餐饮", "房产"] + return any(kw in user_input for kw in keywords) + + def _is_personalized_request(self, user_input: str) -> bool: + """判断是否为个性化请求""" + keywords = ["配置", "设置", "偏好", "关注", "个性化", "订阅"] + return any(kw in user_input for kw in keywords) + + async def _handle_news_digest(self, user_input: str) -> Dict[str, Any]: + """处理新闻摘要请求""" + if not self.news_digest: + return {"error": "模块未初始化"} + + return await self.news_digest.process_user_request(user_input) + + async def _handle_industry_hot(self, user_input: str) -> Dict[str, Any]: + """处理行业热榜请求""" + if not self.industry_hot: + return {"error": "模块未初始化"} + + return await self.industry_hot.process_user_request(user_input) + + async def _handle_personalized(self, user_input: str) -> Dict[str, Any]: + """处理个性化订阅请求""" + if not self.personalized: + return {"error": "模块未初始化"} + + return await self.personalized.process_user_request(user_input) + + async def _show_main_menu(self) -> Dict[str, Any]: + """显示主菜单""" + # 检查是否有旧数据清理提示 + old_data = self.get_old_data_notification() + + menu_text = """🎯 **每日热榜 - 功能选择** + +请选择您想使用的功能： + +1. **📰 热点新闻摘要** + 按标签浏览热点新闻（科技、游戏、娱乐、财经等） + +2. **🏭 行业热榜垂直** + 按行业分类查看热榜（汽车、金融、医疗、旅游等） + +3. **⚙️ 个性化订阅** + 配置您的偏好，获取定制化热榜 + +💡 您可以直接告诉我您想做什么，例如： +- "今天有什么科技热点" +- "看看汽车行业热榜" +- "配置个性化热榜" +""" + + result = { + "action": "show_menu", + "message": menu_text + } + + # 如果有旧数据，添加提示 + if old_data: + result["old_data_prompt"] = old_data["message"] + + return result + + # 便捷方法 + + async def get_news_digest_tags(self) -> str: + """获取新闻摘要标签选项""" + if self.news_digest: + return await self.news_digest.get_tag_options() + return "模块未初始化" + + async def get_industry_options(self) -> str: + """获取行业选项""" + if self.industry_hot: + return await self.industry_hot.get_industry_options() + return "模块未初始化" + + async def get_personalized_options(self) -> str: + """获取个性化配置选项""" + if self.personalized: + return await self.personalized.get_config_options() + return "模块未初始化" + + def get_current_config(self) -> Optional[UserPreferences]: + """获取当前个性化配置""" + if self.personalized: + return self.personalized.get_current_config() + return None + + +# 便捷函数 +async def create_skill(api_client=None, formatter=None) -> DailyHotNewsSkill: + """创建技能实例""" + skill = DailyHotNewsSkill(api_client=api_client, formatter=formatter) + await skill.initialize() + return skill + + +# 示例使用 +if __name__ == "__main__": + async def example(): + # 创建技能实例（不传入api_client时的模拟示例） + skill = await create_skill() + + # 示例1：展示主菜单 + result = await skill.handle_request("帮助") + print(result["message"]) + + print("\n" + "="*60 + "\n") + + # 示例2：展示标签选择 + result = await skill.get_news_digest_tags() + print(result) + + print("\n" + "="*60 + "\n") + + # 示例3：展示行业选择 + result = await skill.get_industry_options() + print(result) + + print("\n" + "="*60 + "\n") + + # 示例4：展示个性化配置选项 + result = await skill.get_personalized_options() + print(result) + + asyncio.run(example()) diff --git a/skills/daily-hot-news/formatter.py b/skills/daily-hot-news/formatter.py new file mode 100644 index 0000000..2833d05 --- /dev/null +++ b/skills/daily-hot-news/formatter.py @@ -0,0 +1,169 @@ +# -*- coding: utf-8 -*- +""" +DailyHotApi Skill - 响应格式化 +""" + +from typing import Dict, List, Any +from api_client import HOT_SOURCES + + +class ResponseFormatter: + """响应格式化器""" + + # 分类名称映射（中文） + CATEGORY_NAMES = { + "video": "🎬 视频/直播", + "social": "💬 社交媒体", + "news": "📰 新闻资讯", + "tech": "💻 科技/技术", + "game": "🎮 游戏/ACG", + "reading": "📚 阅读/文化", + "tool": "🔧 工具/其他", + } + + @staticmethod + def format_hot_list(data: Dict[str, Any]) -> str: + """格式化热榜列表为文本""" + lines = [] + platform = data.get("platform", "未知平台") + update_time = data.get("update_time", "") + + # 头部 + lines.append(f"🔥 **{platform}**") + if update_time: + lines.append(f"更新时间: {update_time}") + lines.append("") + + # 列表 + items = data.get("data", []) + if not items: + lines.append("暂无数据") + return "\n".join(lines) + + for item in items: + rank = item.get("rank", 0) + title = item.get("title", "") + hot = item.get("hot", "") + url = item.get("url", "") + + # 热度处理 + hot_str = f" {hot}" if hot else "" + + # 标题处理（过长截断） + if len(title) > 40: + title = title[:40] + "..." + + lines.append(f"{rank:2d}. {title}{hot_str}") + + # 底部 + lines.append("") + lines.append(f"共 {len(items)} 条") + + return "\n".join(lines) + + @staticmethod + def format_hot_list_compact(data: Dict[str, Any], max_items: int = 10) -> str: + """格式化热榜列表为紧凑格式""" + lines = [] + platform = data.get("platform", "未知平台") + + lines.append(f"🔥 **{platform}**") + lines.append("-" * 40) + + items = data.get("data", [])[:max_items] + for item in items: + rank = item.get("rank", 0) + title = item.get("title", "") + hot = item.get("hot", "") + + # 简化标题 + title = title.replace("\n", " ") + if len(title) > 30: + title = title[:30] + "..." + + hot_str = f" 📈 {hot}" if hot else "" + lines.append(f"{rank:2d}. {title}{hot_str}") + + return "\n".join(lines) + + @staticmethod + def format_all_sources() -> str: + """格式化所有热榜源列表""" + from api_client import api_client + + sources_by_cat = api_client.get_sources_by_category() + lines = [] + + lines.append("📊 **支持的热榜源（共 54 个）**") + lines.append("") + + for cat_key, cat_name in ResponseFormatter.CATEGORY_NAMES.items(): + if cat_key in sources_by_cat: + sources = sources_by_cat[cat_key] + lines.append(f"### {cat_name}") + lines.append(f"共 {len(sources)} 个") + + for source in sources: + lines.append(f"• **{source['name']}** (`{source['id']}`)") + + lines.append("") + + return "\n".join(lines) + + @staticmethod + def format_sources_by_category() -> str: + """按类别格式化热榜源""" + from api_client import api_client + + sources_by_cat = api_client.get_sources_by_category() + lines = [] + + for cat_key, cat_name in ResponseFormatter.CATEGORY_NAMES.items(): + if cat_key not in sources_by_cat: + continue + + lines.append(f"\n{cat_name}\n{'─' * 30}") + sources = sources_by_cat[cat_key] + for source in sources: + lines.append(f"• {source['name']} (`{source['id']}`)") + + return "\n".join(lines) + + @staticmethod + def format_search_results(results: List[Dict], query: str) -> str: + """格式化搜索结果""" + lines = [] + + if not results: + return f"❌ 没有找到与「{query}」相关的热榜源" + + lines.append(f"🔍 搜索「{query}」结果 ({len(results)} 个)") + lines.append("") + + for result in results: + lines.append(f"• **{result['name']}** (`{result['id']}`)") + + return "\n".join(lines) + + @staticmethod + def format_error(message: str, suggestion: str = "") -> str: + """格式化错误信息""" + lines = [f"❌ {message}"] + + if suggestion: + lines.append("") + lines.append(f"💡 {suggestion}") + + return "\n".join(lines) + + @staticmethod + def format_service_status(is_running: bool, url: str) -> str: + """格式化服务状态""" + if is_running: + return f"✅ 每日热榜服务运行中\n\n📡 API 地址: {url}" + else: + return f"❌ 每日热榜服务未运行\n\n📡 预期地址: {url}\n\n💡 请使用 `./deploy.sh status` 查看状态" + + +# 全局格式化器实例 +formatter = ResponseFormatter() diff --git a/skills/daily-hot-news/industry_hot.py b/skills/daily-hot-news/industry_hot.py new file mode 100644 index 0000000..4a4a41d --- /dev/null +++ b/skills/daily-hot-news/industry_hot.py @@ -0,0 +1,369 @@ +""" +行业热榜垂直（增强版） - Industry Hot Module + +功能： +- 十大行业分类：科技互联网、游戏、汽车、金融财经、数码消费、娱乐影视、房产家居、医疗健康、旅游出行、餐饮消费 +- 用户自主选择行业 +- 行业描述和平台标注 +""" + +from typing import List, Dict, Optional, Any +from dataclasses import dataclass +from enum import Enum +import asyncio + +# 十大行业映射 +INDUSTRIES: Dict[str, Dict[str, Any]] = { + "科技互联网": { + "platforms": ["ithome", "36kr", "csdn", "juejin", "oschina", "infoq"], + "description": "IT之家、36氪、CSDN、稀土掘金、开源中国、InfoQ" + }, + "游戏行业": { + "platforms": ["genshin", "miyoushe", "lol", "bilibili", "douyu", "huya", "netease-game"], + "description": "原神、米游社、英雄联盟、B站、斗鱼、虎牙、网易游戏" + }, + "汽车行业": { + "platforms": ["autohome", "car", "懂车帝", "bitauto", "soufun-auto"], + "description": "汽车之家、懂车帝、易车网、苏宁汽车" + }, + "金融财经": { + "platforms": ["sina-money", "eastmoney", "xueqiu", "jrj", "cnstock", "wallstreetcn", "money163"], + "description": "新浪财经、东方财富、雪球、金融界、中国财经网、华尔街见闻、网易财经" + }, + "数码消费": { + "platforms": ["coolapk", "ithome", "sspai", "geekpark", "少数派", "smzdm"], + "description": "酷安、IT之家、少数派、什么值得买" + }, + "娱乐影视": { + "platforms": ["weibo", "douban-group", "douban-movie", "mtime", "movie", "bilibili"], + "description": "微博、豆瓣、豆瓣电影、时光网、B站" + }, + "房产家居": { + "platforms": ["lfang", "soufunianjia", "anjuke", "house", "lianjia"], + "description": "链家、安居客、房天下、贝壳找房" + }, + "医疗健康": { + "platforms": ["zhihu", "知乎", "sina-health", "health", "baikemy", "丁香园"], + "description": "知乎、新浪健康、丁香园、百度健康" + }, + "旅游出行": { + "platforms": ["mafengwo", "ctrip", "qunar", "飞猪", "马蜂窝", "携程"], + "description": "马蜂窝、携程、去哪儿、飞猪" + }, + "餐饮消费": { + "platforms": ["dianping", "xiaohongshu", "大众点评", "ele.me", "meituan"], + "description": "大众点评、美团、饿了么、小红书" + } +} + +# 所有行业名称 +ALL_INDUSTRIES = list(INDUSTRIES.keys()) + + +class IndustryMode(Enum): + """行业模式""" + SINGLE = "single" # 单行业 + MULTI = "multi" # 多行业 + + +@dataclass +class IndustryConfig: + """行业配置""" + industries: List[str] # 选择的行业列表 + mode: IndustryMode = IndustryMode.SINGLE + items_per_platform: int = 10 # 每个平台显示的条目数 + total_items: int = 50 # 总条目数限制 + include_description: bool = True # 是否包含行业描述 + + +class IndustryHot: + """行业热榜类""" + + def __init__(self, api_client=None, formatter=None): + """ + 初始化行业热榜 + + Args: + api_client: API客户端实例（可选） + formatter: 格式化器实例（可选） + """ + self.api_client = api_client + self.formatter = formatter + self._all_platforms = self._get_all_platforms() + + def _get_all_platforms(self) -> List[str]: + """获取所有行业相关的平台""" + platforms = set() + for industry_info in INDUSTRIES.values(): + platforms.update(industry_info.get("platforms", [])) + return list(platforms) + + async def fetch_all_hot_data(self, limit_per_platform: int = 10) -> List[Dict[str, Any]]: + """ + 一次性获取全部平台的热榜数据 + + Args: + limit_per_platform: 每个平台获取的条目数 + + Returns: + 全部平台的热榜数据列表 + """ + all_items = [] + + if self.api_client: + for platform in self._all_platforms: + try: + items = await self.api_client.get_hot榜单(platform, limit=limit_per_platform) + if items: + for item in items: + item["source_platform"] = platform + all_items.extend(items) + except Exception as e: + print(f"Error fetching {platform}: {e}") + continue + + return all_items + + async def get_industry_options(self) -> str: + """ + 获取行业选择选项（AI引导话术） + + Returns: + 格式化的行业选择文本 + """ + options_text = "🏭 **请选择您关注的行业**\n\n" + + for i, (industry, info) in enumerate(INDUSTRIES.items(), 1): + platforms_count = len(info["platforms"]) + options_text += f"{i}. **{industry}**\n" + options_text += f" 📋 包含 {platforms_count} 个平台\n" + options_text += f" 🔗 {info['description']}\n\n" + + options_text += "-" * 50 + "\n" + options_text += "💡 您可以输入行业名称或数字编号，支持多选（如：1,3或汽车+金融）" + return options_text + + def parse_industries_from_input(self, user_input: str) -> List[str]: + """ + 解析用户输入的行业 + + Args: + user_input: 用户输入的文本 + + Returns: + 匹配的行业列表 + """ + user_input = user_input.strip().lower() + matched_industries = [] + + # 处理数字选择 + import re + number_pattern = re.findall(r'\d+', user_input) + for num in number_pattern: + idx = int(num) - 1 + if 0 <= idx < len(ALL_INDUSTRIES): + matched_industries.append(ALL_INDUSTRIES[idx]) + + # 行业关键词映射（行业名称 -> 包含的关键词） + industry_keywords = { + "科技互联网": ["科技", "互联网", "IT", "技术"], + "游戏行业": ["游戏", "手游", "网游"], + "汽车行业": ["汽车", "车", "车企", "新能源车"], + "金融财经": ["金融", "财经", "投资", "理财", "股票"], + "数码消费": ["数码", "手机", "电脑", "电子"], + "娱乐影视": ["娱乐", "影视", "电影", "综艺", "明星"], + "房产家居": ["房产", "房", "家居", "装修", "买房"], + "医疗健康": ["医疗", "健康", "医药", "养生"], + "旅游出行": ["旅游", "出行", "旅行", "机票", "酒店"], + "餐饮消费": ["餐饮", "美食", "外卖", "餐厅", "消费"] + } + + # 处理行业关键词 + for industry, keywords in industry_keywords.items(): + for keyword in keywords: + if keyword in user_input: + if industry not in matched_industries: + matched_industries.append(industry) + break + + # 处理完整行业名称（向后兼容） + for industry in ALL_INDUSTRIES: + if industry.lower() in user_input or industry in user_input: + if industry not in matched_industries: + matched_industries.append(industry) + + return matched_industries if matched_industries else [] + + async def get_industry_hot(self, industries: List[str], config: Optional[IndustryConfig] = None) -> Dict[str, Any]: + """ + 获取行业热榜（新版：先全部获取，再按行业筛选） + + Args: + industries: 行业列表 + config: 配置对象（可选） + + Returns: + 行业热榜数据 + """ + if config is None: + config = IndustryConfig(industries=industries) + + # 步骤1：全部获取 + all_items = await self.fetch_all_hot_data(limit_per_platform=config.items_per_platform) + + # 步骤2：按行业筛选 + # 获取所有相关平台 + target_platforms = set() + for industry in industries: + if industry in INDUSTRIES: + target_platforms.update(INDUSTRIES[industry]["platforms"]) + + filtered_items = [] + for item in all_items: + platform = item.get("source_platform", "") + if platform in target_platforms: + item["source_industry"] = next((ind for ind in industries if ind in INDUSTRIES and platform in INDUSTRIES[ind]["platforms"]), industries[0]) + filtered_items.append(item) + + # 去重和排序 + merged_items = self._merge_items(filtered_items) + + # 限制总数 + merged_items = merged_items[:config.total_items] + + return { + "industries": industries, + "industry_descriptions": {ind: INDUSTRIES[ind]["description"] for ind in industries}, + "total_items": len(merged_items), + "items": merged_items + } + + def _merge_items(self, items: List[Dict]) -> List[Dict]: + """ + 合并和去重条目 + + Args: + items: 条目列表 + + Returns: + 合并后的条目列表 + """ + if not items: + return [] + + # 按标题去重 + seen_titles = set() + unique_items = [] + + for item in items: + title = item.get("title", "").strip().lower() + if title and title not in seen_titles: + seen_titles.add(title) + unique_items.append(item) + + # 按热度排序 + unique_items.sort(key=lambda x: x.get("hot", 0) or x.get("score", 0), reverse=True) + + return unique_items + + def format_industry_response(self, hot_data: Dict[str, Any]) -> str: + """ + 格式化行业热榜响应 + + Args: + hot_data: 热榜数据 + + Returns: + 格式化的响应文本 + """ + if not hot_data.get("items"): + return "❌ 暂无行业热榜数据" + + industries = hot_data["industries"] + items = hot_data["items"] + + response = f"🏭 **行业热榜 - {', '.join(industries)}**\n\n" + + # 显示行业描述 + for industry in industries: + desc = hot_data["industry_descriptions"].get(industry, "") + response += f"📌 **{industry}**: {desc}\n" + + response += "-" * 50 + "\n" + response += f"共 {hot_data['total_items']} 条热榜\n\n" + + # 按行业分组显示 + items_by_industry = {} + for item in items: + industry = item.get("source_industry", "其他") + if industry not in items_by_industry: + items_by_industry[industry] = [] + items_by_industry[industry].append(item) + + for industry, ind_items in items_by_industry.items(): + response += f"\n📊 **{industry}**\n" + for i, item in enumerate(ind_items[:5], 1): # 每个行业显示5条 + title = item.get("title", "无标题") + hot = item.get("hot", item.get("score", "")) + platform = item.get("source_platform", "") + + response += f"{i}. {title}\n" + if hot: + response += f" 🔥 {hot}" + if platform: + response += f" | 📱 {platform}" + response += "\n" + + return response + + async def process_user_request(self, user_input: str) -> Dict[str, Any]: + """ + 处理用户请求 + + Args: + user_input: 用户输入 + + Returns: + 处理结果 + """ + # 解析行业 + industries = self.parse_industries_from_input(user_input) + + if not industries: + # 返回引导信息 + return { + "action": "ask_industry", + "message": await self.get_industry_options() + } + + # 获取热榜 + hot_data = await self.get_industry_hot(industries) + + # 格式化响应 + response_text = self.format_industry_response(hot_data) + + return { + "action": "show_industry_hot", + "data": hot_data, + "message": response_text + } + + +# 便捷函数 +async def create_industry_hot(api_client=None, formatter=None) -> IndustryHot: + """创建行业热榜实例""" + return IndustryHot(api_client=api_client, formatter=formatter) + + +if __name__ == "__main__": + # 测试代码 + async def test(): + industry_hot = await create_industry_hot() + print(await industry_hot.get_industry_options()) + + print("\n" + "="*50 + "\n") + + result = await industry_hot.process_user_request("汽车和金融") + print(result["message"]) + + asyncio.run(test()) diff --git a/skills/daily-hot-news/news_digest.py b/skills/daily-hot-news/news_digest.py new file mode 100644 index 0000000..a50572f --- /dev/null +++ b/skills/daily-hot-news/news_digest.py @@ -0,0 +1,320 @@ +""" +热点新闻摘要（增强版） - News Digest Module + +功能： +- 15种标签分类：科技、互联网、游戏、娱乐、社会、财经、汽车、体育、教育、健康、国际、房产、数码、时尚、美食 +- AI主动引导用户选择标签 +- 按标签获取和合并热榜 +""" + +from typing import List, Dict, Optional, Any +from dataclasses import dataclass +from enum import Enum +import asyncio + +# 标签到平台的映射 +TAG_MAPPING: Dict[str, List[str]] = { + "科技": ["ithome", "36kr", "sspai", "csdn", "juejin", "51cto", "oschina", "infoq"], + "互联网": ["sina-news", "netease-news", "qq-news", "sohu-news", "ifeng"], + "游戏": ["genshin", "miyoushe", "lol", "hupu", "bilibili", "douyu", "huya", "netease-game"], + "娱乐": ["weibo", "douban-group", "douban-movie", "mtime", "movie"], + "社会": ["sina-news", "netease-news", "qq-news", "sohu-news", "ifeng", "qq"], + "财经": ["sina-money", "eastmoney", "xueqiu", "jrj", "cnstock", "wallstreetcn"], + "汽车": ["autohome", "car", "懂车帝", "bitauto", "car1"], + "体育": ["hupu", "sports", "sina-sports", "qq-sports", "zhibo8"], + "教育": ["zhihu", "知乎", "bilibili", "jike", "dazhihui"], + "健康": ["zhihu", "知乎", "sina-health", "health", "baikemy"], + "国际": ["sina-news", "netease-news", "qq-news", "ifeng", "cnn", "bbc"], + "房产": ["lfang", "soufunianjia", "anjuke", "house"], + "数码": ["ithome", "coolapk", "sspai", "geekpark", "少数派"], + "时尚": ["mogujie", "meilishuo", "xiaohongshu", "微博时尚", "yoho"], + "美食": ["dianping", "xiaohongshu", "大众点评", "maoyan", "ele.me"] +} + +# 所有可用的标签 +ALL_TAGS = list(TAG_MAPPING.keys()) + + +class DigestMode(Enum): + """摘要模式""" + SINGLE = "single" # 单标签 + MULTI = "multi" # 多标签 + + +@dataclass +class DigestConfig: + """摘要配置""" + tags: List[str] # 选择的标签列表 + mode: DigestMode = DigestMode.MULTI + items_per_platform: int = 10 # 每个平台显示的条目数 + total_items: int = 50 # 总条目数限制 + merge_strategy: str = "score" # 合并策略：score(按热度), time(按时间), random(随机) + + +class NewsDigest: + """热点新闻摘要类""" + + def __init__(self, api_client=None, formatter=None): + """ + 初始化新闻摘要 + + Args: + api_client: API客户端实例（可选，如果不提供则需要外部传入） + formatter: 格式化器实例（可选） + """ + self.api_client = api_client + self.formatter = formatter + self.platforms = self._get_all_platforms() + + def _get_all_platforms(self) -> List[str]: + """获取所有可用的平台""" + platforms = set() + for tag, plats in TAG_MAPPING.items(): + platforms.update(plats) + return list(platforms) + + async def fetch_all_hot_data(self, limit_per_platform: int = 10) -> List[Dict[str, Any]]: + """ + 一次性获取全部54个平台的热榜数据 + + Args: + limit_per_platform: 每个平台获取的条目数 + + Returns: + 全部平台的热榜数据列表 + """ + all_items = [] + + if self.api_client: + for platform in self.platforms: + try: + items = await self.api_client.get_hot榜单(platform, limit=limit_per_platform) + if items: + for item in items: + item["source_platform"] = platform + all_items.extend(items) + except Exception as e: + print(f"Error fetching {platform}: {e}") + continue + + return all_items + + async def get_tag_options(self) -> str: + """ + 获取标签选择选项（AI引导话术） + + Returns: + 格式化的标签选择文本 + """ + options_text = "📊 **请选择您感兴趣的标签**\n\n" + + # 分两列显示 + tags_left = ALL_TAGS[:8] + tags_right = ALL_TAGS[8:] + + for i, tag in enumerate(tags_left): + right_tag = tags_right[i] if i < len(tags_right) else "" + left_plats = ", ".join(TAG_MAPPING[tag][:3]) + right_plats = f"│ {i+8+1}. {right_tag}: {', '.join(TAG_MAPPING[right_tag][:3])}" if right_tag else "" + options_text += f"{i+1}. {tag} ({left_plats}) {right_plats}\n" + + options_text += "\n💡 您可以输入标签名称或数字编号，支持多选（如：1,3或科技+游戏）" + return options_text + + def parse_tags_from_input(self, user_input: str) -> List[str]: + """ + 解析用户输入的标签 + + Args: + user_input: 用户输入的文本 + + Returns: + 匹配的标签列表 + """ + user_input = user_input.strip().lower() + matched_tags = [] + + # 处理数字选择 + numbers = [] + import re + number_pattern = re.findall(r'\d+', user_input) + for num in number_pattern: + idx = int(num) - 1 + if 0 <= idx < len(ALL_TAGS): + numbers.append(ALL_TAGS[idx]) + + # 处理标签名称 + for tag in ALL_TAGS: + if tag.lower() in user_input or tag in user_input: + if tag not in matched_tags and tag not in numbers: + matched_tags.append(tag) + + # 合并数字选择的结果 + matched_tags.extend([t for t in numbers if t not in matched_tags]) + + return matched_tags if matched_tags else [] + + async def get_digest_by_tags(self, tags: List[str], config: Optional[DigestConfig] = None) -> Dict[str, Any]: + """ + 按标签获取新闻摘要（新版：先全部获取，再按标签筛选） + + Args: + tags: 标签列表 + config: 配置对象（可选） + + Returns: + 合并后的热榜数据 + """ + if config is None: + config = DigestConfig(tags=tags) + + # 步骤1：全部获取 + all_items = await self.fetch_all_hot_data(limit_per_platform=config.items_per_platform) + + # 步骤2：按标签筛选 + # 获取所有相关平台 + target_platforms = set() + for tag in tags: + if tag in TAG_MAPPING: + target_platforms.update(TAG_MAPPING[tag]) + + filtered_items = [] + for item in all_items: + platform = item.get("source_platform", "") + if platform in target_platforms: + item["source_tag"] = next((t for t in tags if t in TAG_MAPPING and platform in TAG_MAPPING[t]), tags[0]) + filtered_items.append(item) + + # 去重和合并 + merged_items = self._merge_items(filtered_items, config.merge_strategy) + + # 限制总数 + merged_items = merged_items[:config.total_items] + + return { + "tags": tags, + "platforms": list(target_platforms), + "total_items": len(merged_items), + "items": merged_items + } + + def _merge_items(self, items: List[Dict], strategy: str = "score") -> List[Dict]: + """ + 合并和去重条目 + + Args: + items: 条目列表 + strategy: 合并策略 + + Returns: + 合并后的条目列表 + """ + if not items: + return [] + + # 按标题去重 + seen_titles = set() + unique_items = [] + + for item in items: + title = item.get("title", "").strip().lower() + if title and title not in seen_titles: + seen_titles.add(title) + unique_items.append(item) + + # 根据策略排序 + if strategy == "score": + # 按热度/分数排序 + unique_items.sort(key=lambda x: x.get("hot", 0) or x.get("score", 0), reverse=True) + elif strategy == "time": + # 按时间排序 + unique_items.sort(key=lambda x: x.get("time", "") or "", reverse=True) + + return unique_items + + def format_digest_response(self, digest_data: Dict[str, Any]) -> str: + """ + 格式化摘要响应 + + Args: + digest_data: 摘要数据 + + Returns: + 格式化的响应文本 + """ + if not digest_data.get("items"): + return "❌ 暂无热点数据" + + tags = digest_data["tags"] + items = digest_data["items"] + + response = f"📰 **热点摘要 - {', '.join(tags)}**\n" + response += f"来源平台: {', '.join(digest_data['platforms'])}\n" + response += f"共 {digest_data['total_items']} 条热点\n" + response += "-" * 40 + "\n\n" + + for i, item in enumerate(items, 1): + title = item.get("title", "无标题") + hot = item.get("hot", item.get("score", "")) + platform = item.get("source_platform", "") + + response += f"{i}. {title}\n" + if hot: + response += f" 🔥 热度: {hot}" + if platform: + response += f" | 📱 {platform}" + response += "\n\n" + + return response + + async def process_user_request(self, user_input: str) -> Dict[str, Any]: + """ + 处理用户请求 + + Args: + user_input: 用户输入 + + Returns: + 处理结果 + """ + # 解析标签 + tags = self.parse_tags_from_input(user_input) + + if not tags: + # 返回引导信息 + return { + "action": "ask_tag", + "message": await self.get_tag_options() + } + + # 获取摘要 + digest_data = await self.get_digest_by_tags(tags) + + # 格式化响应 + response_text = self.format_digest_response(digest_data) + + return { + "action": "show_digest", + "data": digest_data, + "message": response_text + } + + +# 便捷函数 +async def create_digest(api_client=None, formatter=None) -> NewsDigest: + """创建新闻摘要实例""" + return NewsDigest(api_client=api_client, formatter=formatter) + + +if __name__ == "__main__": + # 测试代码 + async def test(): + digest = await create_digest() + print(await digest.get_tag_options()) + + print("\n" + "="*50 + "\n") + + result = await digest.process_user_request("科技和游戏") + print(result["message"]) + + asyncio.run(test()) diff --git a/skills/daily-hot-news/personalized.py b/skills/daily-hot-news/personalized.py new file mode 100644 index 0000000..72a956b --- /dev/null +++ b/skills/daily-hot-news/personalized.py @@ -0,0 +1,557 @@ +""" +个性化订阅（增强版） - Personalized Subscription Module + +功能： +- 用户自主配置：关键词、平台、排除项 +- AI主动提供备选项 +- 关键词过滤和偏好排序 +- 用户配置存储 +""" + +from typing import List, Dict, Optional, Any, Set +from dataclasses import dataclass, field, asdict +from enum import Enum +from datetime import datetime +import json +import asyncio +import os + +# 可选的关键词标签 +KEYWORD_OPTIONS = [ + "AI", "ChatGPT", "人工智能", "大模型", "机器学习", + "游戏", "原神", "英雄联盟", "王者荣耀", "米哈游", + "科技产品", "iPhone", "华为", "小米", "特斯拉", + "新能源汽车", "比亚迪", "宁德时代", "蔚来", "小鹏", + "互联网", "字节跳动", "腾讯", "阿里巴巴", "美团", + "电商", "直播", "短视频", "网红", "明星八卦", + "影视", "电影", "电视剧", "综艺", "动漫", + "财经", "股票", "基金", "加密货币", "比特币", + "房产", "房价", "房地产", "房贷", "租房", + "美食", "餐厅", "外卖", "网红店", "探店", + "旅游", "出行", "机票", "酒店", "景点", + "时尚", "穿搭", "美妆", "护肤", "奢侈品", + "体育", "足球", "篮球", "NBA", "奥运会", + "教育", "高考", "考研", "留学", "职场" +] + +# 可选平台 +PLATFORM_OPTIONS = [ + "微博", "知乎", "B站", "抖音", "快手", + "原神", "米游社", "IT之家", "36氪", "虎嗅", + "豆瓣", "小红书", "今日头条", "澎湃新闻", "观察者网" +] + +# 排除关键词示例 +EXCLUDE_OPTIONS = [ + "广告", "推广", "营销号", "震惊", "必看", + "流量明星", "网红脸", "擦边", "引战" +] + + +class SubscriptionMode(Enum): + """订阅模式""" + INCLUDE = "include" # 包含模式 + EXCLUDE = "exclude" # 排除模式 + + +@dataclass +class UserPreferences: + """用户偏好配置""" + keywords: List[str] = field(default_factory=list) # 关注的关键词 + platforms: List[str] = field(default_factory=list) # 关注的平台 + exclude_keywords: List[str] = field(default_factory=list) # 排除的关键词 + subscription_mode: SubscriptionMode = SubscriptionMode.INCLUDE + items_per_platform: int = 10 + total_items: int = 30 + sort_by: str = "relevance" # relevance(相关性), hot(热度), time(时间) + created_at: str = field(default_factory=lambda: datetime.now().isoformat()) + updated_at: str = field(default_factory=lambda: datetime.now().isoformat()) + + def to_dict(self) -> Dict: + """转换为字典""" + data = asdict(self) + data["subscription_mode"] = self.subscription_mode.value + return data + + @classmethod + def from_dict(cls, data: Dict) -> "UserPreferences": + """从字典创建""" + if "subscription_mode" in data and isinstance(data["subscription_mode"], str): + data["subscription_mode"] = SubscriptionMode(data["subscription_mode"]) + return cls(**data) + + +class PersonalizedSubscription: + """个性化订阅类""" + + def __init__(self, api_client=None, formatter=None, storage_path: str = None): + """ + 初始化个性化订阅 + + Args: + api_client: API客户端实例（可选） + formatter: 格式化器实例（可选） + storage_path: 配置存储路径（可选） + """ + self.api_client = api_client + self.formatter = formatter + self.storage_path = storage_path or os.path.join( + os.path.dirname(__file__), + "personalized_config.json" + ) + self.current_config = None + self._all_platforms = self._get_all_platforms() + self._load_config() + + def _get_all_platforms(self) -> List[str]: + """获取所有可用的平台""" + return [self._platform_name_to_id(p) for p in PLATFORM_OPTIONS] + + def _platform_name_to_id(self, platform_name: str) -> str: + """将平台名称转换为API ID""" + mapping = { + "微博": "weibo", + "知乎": "zhihu", + "B站": "bilibili", + "抖音": "douyin", + "快手": "kuaishou", + "原神": "genshin", + "米游社": "miyoushe", + "IT之家": "ithome", + "36氪": "36kr", + "虎嗅": "huxiu", + "豆瓣": "douban-group", + "小红书": "xiaohongshu", + "今日头条": "jinritoutiao", + "澎湃新闻": "thepaper", + "观察者网": "guanchazhe" + } + return mapping.get(platform_name, platform_name.lower()) + + def _load_config(self) -> Optional[UserPreferences]: + """加载用户配置""" + if os.path.exists(self.storage_path): + try: + with open(self.storage_path, 'r', encoding='utf-8') as f: + data = json.load(f) + self.current_config = UserPreferences.from_dict(data) + return self.current_config + except Exception as e: + print(f"Error loading config: {e}") + return None + + def _save_config(self, config: UserPreferences) -> bool: + """保存用户配置""" + try: + # 更新修改时间 + config.updated_at = datetime.now().isoformat() + + with open(self.storage_path, 'w', encoding='utf-8') as f: + json.dump(config.to_dict(), f, ensure_ascii=False, indent=2) + + self.current_config = config + return True + except Exception as e: + print(f"Error saving config: {e}") + return False + + async def fetch_all_hot_data(self, limit_per_platform: int = 10) -> List[Dict[str, Any]]: + """ + 一次性获取全部平台的热榜数据 + + Args: + limit_per_platform: 每个平台获取的条目数 + + Returns: + 全部平台的热榜数据列表 + """ + all_items = [] + + if self.api_client: + for platform in self._all_platforms: + try: + items = await self.api_client.get_hot榜单(platform, limit=limit_per_platform) + if items: + for item in items: + item["source_platform"] = platform + all_items.extend(items) + except Exception as e: + print(f"Error fetching {platform}: {e}") + continue + + return all_items + + async def get_config_options(self) -> str: + """ + 获取配置选项（AI引导话术） + + Returns: + 格式化的配置引导文本 + """ + options_text = "⚙️ **个性化热榜配置**\n\n" + + # 关键词选项 + options_text += "**【关键词】**\n" + options_text += "可选标签：\n" + + # 分组显示关键词 + keyword_groups = [ + ("科技类", ["AI", "人工智能", "大模型", "科技产品", "iPhone", "华为", "小米"]), + ("游戏类", ["游戏", "原神", "英雄联盟", "王者荣耀", "米哈游"]), + ("汽车类", ["特斯拉", "新能源汽车", "比亚迪", "蔚来", "小鹏"]), + ("财经类", ["财经", "股票", "基金", "加密货币", "比特币"]), + ("娱乐类", ["影视", "综艺", "明星八卦", "网红"]), + ] + + for group_name, keywords in keyword_groups: + options_text += f" • {group_name}: {', '.join(keywords)}\n" + + options_text += "\n**【平台】**\n" + options_text += f"可选：{', '.join(PLATFORM_OPTIONS)}\n" + + options_text += "\n**【排除项】**\n" + options_text += f"可选：{', '.join(EXCLUDE_OPTIONS)}\n" + + options_text += "\n" + "-" * 50 + "\n" + options_text += "💡 请告诉我您的偏好设置，我会帮您定制热榜！\n" + options_text += "示例：关注AI和游戏，平台选微博、B站、IT之家，排除广告" + + return options_text + + def parse_config_from_input(self, user_input: str) -> Dict[str, Any]: + """ + 从用户输入解析配置 + + Args: + user_input: 用户输入 + + Returns: + 解析的配置信息 + """ + user_input = user_input.lower() + + # 解析关键词 + keywords = [] + for kw in KEYWORD_OPTIONS: + if kw.lower() in user_input: + keywords.append(kw) + + # 解析平台 + platforms = [] + platform_mapping = { + "微博": ["微博", "weibo"], + "知乎": ["知乎", "zhihu"], + "B站": ["b站", "bilibili", "B站"], + "抖音": ["抖音", "tiktok"], + "快手": ["快手"], + "原神": ["原神", "genshin"], + "米游社": ["米游社", "miyoushe"], + "IT之家": ["it之家", "ithome", "IT之家"], + "36氪": ["36氪", "36kr"], + "虎嗅": ["虎嗅", "huxiu"], + "豆瓣": ["豆瓣", "douban"], + "小红书": ["小红书", "xiaohongshu"], + "今日头条": ["今日头条", "头条"], + "澎湃新闻": ["澎湃", "澎湃新闻"], + "观察者网": ["观察者网", "guanchazhe"], + } + + for platform, aliases in platform_mapping.items(): + for alias in aliases: + if alias.lower() in user_input: + if platform not in platforms: + platforms.append(platform) + break + + # 解析排除关键词 + exclude_keywords = [] + for ex in EXCLUDE_OPTIONS: + if ex.lower() in user_input: + exclude_keywords.append(ex) + + return { + "keywords": keywords, + "platforms": platforms, + "exclude_keywords": exclude_keywords + } + + async def configure_subscription(self, user_input: str) -> Dict[str, Any]: + """ + 配置个性化订阅 + + Args: + user_input: 用户输入的配置信息 + + Returns: + 配置结果 + """ + parsed = self.parse_config_from_input(user_input) + + # 检查是否有配置信息 + if not parsed["keywords"] and not parsed["platforms"]: + # 返回引导信息 + return { + "action": "ask_config", + "message": await self.get_config_options() + } + + # 创建配置 + config = UserPreferences( + keywords=parsed["keywords"], + platforms=parsed["platforms"], + exclude_keywords=parsed["exclude_keywords"] + ) + + # 保存配置 + if self._save_config(config): + return { + "action": "config_saved", + "config": config, + "message": self._format_config_confirmation(config) + } + else: + return { + "action": "error", + "message": "❌ 配置保存失败，请重试" + } + + def _format_config_confirmation(self, config: UserPreferences) -> str: + """ + 格式化配置确认信息 + + Args: + config: 用户配置 + + Returns: + 确认信息文本 + """ + response = "✅ **配置完成！**\n\n" + + response += f"**关键词**: {', '.join(config.keywords) if config.keywords else '未设置'}\n" + response += f"**平台**: {', '.join(config.platforms) if config.platforms else '未设置'}\n" + response += f"**排除项**: {', '.join(config.exclude_keywords) if config.exclude_keywords else '无'}\n" + + response += "\n" + "-" * 40 + "\n" + response += "📊 您可以输入「查看热榜」或「刷新热榜」来获取个性化热榜内容" + + return response + + async def get_personalized_hot(self, config: Optional[UserPreferences] = None) -> Dict[str, Any]: + """ + 获取个性化热榜（新版：先全部获取，再按用户配置过滤） + + Args: + config: 配置对象（可选，默认使用当前配置） + + Returns: + 个性化热榜数据 + """ + if config is None: + config = self.current_config + + if not config: + return { + "error": "未配置个性化订阅", + "action": "ask_config", + "message": "请先配置您的个性化热榜偏好" + } + + # 步骤1：全部获取（如果用户配置了平台，则获取配置的平台；否则获取全部） + if config.platforms: + platform_ids = [self._platform_name_to_id(p) for p in config.platforms] + else: + platform_ids = self._all_platforms + + all_items = [] + if self.api_client: + for platform_id in platform_ids: + try: + items = await self.api_client.get_hot榜单(platform_id, limit=config.items_per_platform) + if items: + for item in items: + item["source_platform"] = platform_id + all_items.extend(items) + except Exception as e: + print(f"Error fetching {platform_id}: {e}") + continue + + # 步骤2：按用户配置过滤 + filtered_items = self._filter_items(all_items, config) + + # 限制总数 + filtered_items = filtered_items[:config.total_items] + + return { + "config": config, + "total_items": len(filtered_items), + "items": filtered_items + } + + def _filter_items(self, items: List[Dict], config: UserPreferences) -> List[Dict]: + """ + 过滤条目 + + Args: + items: 条目列表 + config: 用户配置 + + Returns: + 过滤后的条目列表 + """ + if not items: + return [] + + filtered = [] + + for item in items: + title = item.get("title", "").lower() + desc = item.get("description", "").lower() + + # 排除关键词过滤 + should_exclude = False + for ex_kw in config.exclude_keywords: + if ex_kw.lower() in title or ex_kw.lower() in desc: + should_exclude = True + break + + if should_exclude: + continue + + # 关键词匹配（如果设置了关键词） + if config.keywords: + matches_keyword = False + for kw in config.keywords: + if kw.lower() in title or kw.lower() in desc: + matches_keyword = True + break + + if not matches_keyword: + continue + + filtered.append(item) + + # 排序 + if config.sort_by == "hot": + filtered.sort(key=lambda x: x.get("hot", 0) or x.get("score", 0), reverse=True) + elif config.sort_by == "time": + filtered.sort(key=lambda x: x.get("time", "") or "", reverse=True) + # relevance保持原顺序 + + return filtered + + def format_personalized_response(self, hot_data: Dict[str, Any]) -> str: + """ + 格式化个性化热榜响应 + + Args: + hot_data: 热榜数据 + + Returns: + 格式化的响应文本 + """ + if "error" in hot_data: + return hot_data.get("message", "❌ 获取失败") + + config = hot_data.get("config") + items = hot_data.get("items", []) + + if not items: + return "❌ 暂无符合条件的热榜内容" + + response = "🎯 **个性化热榜**\n\n" + + if config and config.keywords: + response += f"关键词: {', '.join(config.keywords)}\n" + if config and config.platforms: + response += f"平台: {', '.join(config.platforms)}\n" + + response += "-" * 40 + "\n" + response += f"共 {hot_data['total_items']} 条\n\n" + + for i, item in enumerate(items, 1): + title = item.get("title", "无标题") + hot = item.get("hot", item.get("score", "")) + platform = item.get("source_platform", "") + + response += f"{i}. {title}\n" + if hot: + response += f" 🔥 {hot}" + if platform: + response += f" | 📱 {platform}" + response += "\n" + + return response + + async def process_user_request(self, user_input: str) -> Dict[str, Any]: + """ + 处理用户请求 + + Args: + user_input: 用户输入 + + Returns: + 处理结果 + """ + user_input_lower = user_input.lower() + + # 检查是否是查看热榜请求 + if "热榜" in user_input or "hot" in user_input_lower: + if self.current_config: + hot_data = await self.get_personalized_hot() + response_text = self.format_personalized_response(hot_data) + return { + "action": "show_hot", + "data": hot_data, + "message": response_text + } + else: + return { + "action": "ask_config", + "message": "请先配置个性化热榜偏好" + } + + # 检查是否是配置请求 + if "配置" in user_input or "设置" in user_input or "偏好" in user_input: + return await self.configure_subscription(user_input) + + # 默认当作配置处理 + return await self.configure_subscription(user_input) + + def get_current_config(self) -> Optional[UserPreferences]: + """获取当前配置""" + return self.current_config + + def clear_config(self) -> bool: + """清除配置""" + if self.storage_path and os.path.exists(self.storage_path): + try: + os.remove(self.storage_path) + self.current_config = None + return True + except Exception as e: + print(f"Error clearing config: {e}") + return False + return True + + +# 便捷函数 +async def create_personalized(api_client=None, formatter=None, storage_path: str = None) -> PersonalizedSubscription: + """创建个性化订阅实例""" + return PersonalizedSubscription(api_client=api_client, formatter=formatter, storage_path=storage_path) + + +if __name__ == "__main__": + # 测试代码 + async def test(): + ps = await create_personalized() + + print("配置选项：") + print(await ps.get_config_options()) + + print("\n" + "="*50 + "\n") + + # 测试配置 + result = await ps.process_user_request("关注AI和游戏，平台选微博、B站、IT之家") + print(result["message"]) + + asyncio.run(test()) diff --git a/skills/daily-hot-news/requirements.txt b/skills/daily-hot-news/requirements.txt new file mode 100644 index 0000000..36d7a1d --- /dev/null +++ b/skills/daily-hot-news/requirements.txt @@ -0,0 +1,9 @@ +# DailyHotApi Skill - 依赖列表 + +# 核心依赖 +requests>=2.28.0 +aiohttp>=3.8.0 + +# 可选依赖（用于测试） +pytest>=7.0.0 +pytest-asyncio>=0.20.0 diff --git a/skills/daily-hot-news/sentiment_monitor.py b/skills/daily-hot-news/sentiment_monitor.py new file mode 100644 index 0000000..d26f81d --- /dev/null +++ b/skills/daily-hot-news/sentiment_monitor.py @@ -0,0 +1,284 @@ +""" +舆情监控 - Sentiment Monitoring Module + +功能： +- 直接用全量数据做关键词过滤 +- 监控特定关键词的舆情 +""" + +from typing import List, Dict, Optional, Any +from dataclasses import dataclass, field +from enum import Enum +from datetime import datetime +import asyncio + + +class SentimentType(Enum): + """舆情类型""" + ALL = "all" # 全部 + POSITIVE = "positive" # 正面 + NEGATIVE = "negative" # 负面 + NEUTRAL = "neutral" # 中性 + + +@dataclass +class SentimentConfig: + """舆情监控配置""" + keywords: List[str] = field(default_factory=list) # 监控关键词 + sentiment_type: SentimentType = SentimentType.ALL # 舆情类型 + items_per_platform: int = 10 # 每个平台获取的条目数 + total_items: int = 50 # 总条目数限制 + include_platforms: List[str] = None # 包含的平台 + exclude_platforms: List[str] = None # 排除的平台 + + +class SentimentMonitor: + """舆情监控类""" + + def __init__(self, api_client=None, formatter=None): + """ + 初始化舆情监控 + + Args: + api_client: API客户端实例（可选） + formatter: 格式化器实例（可选） + """ + self.api_client = api_client + self.formatter = formatter + self.all_platforms = self._get_default_platforms() + + def _get_default_platforms(self) -> List[str]: + """获取默认的全部平台列表""" + return [ + "weibo", "zhihu", "douban-group", "douban-movie", + "ithome", "36kr", "sspai", "csdn", "juejin", + "genshin", "miyoushe", "bilibili", "hupu", + "sina-news", "netease-news", "qq-news", + "sina-money", "eastmoney", "xueqiu", + "autohome", "懂车帝", "mafengwo", "ctrip", + "dianping", "xiaohongshu", "weibo" + ] + + async def fetch_all_hot_data(self, limit_per_platform: int = 10) -> List[Dict[str, Any]]: + """ + 一次性获取全部平台的热榜数据 + + Args: + limit_per_platform: 每个平台获取的条目数 + + Returns: + 全部平台的热榜数据列表 + """ + all_items = [] + + if self.api_client: + for platform in self.all_platforms: + try: + items = await self.api_client.get_hot榜单(platform, limit=limit_per_platform) + if items: + for item in items: + item["source_platform"] = platform + all_items.extend(items) + except Exception as e: + print(f"Error fetching {platform}: {e}") + continue + + return all_items + + async def monitor_keywords(self, keywords: List[str], config: Optional[SentimentConfig] = None) -> Dict[str, Any]: + """ + 监控关键词舆情（新版：先全部获取，再关键词过滤） + + Args: + keywords: 关键词列表 + config: 配置对象（可选） + + Returns: + 舆情监控数据 + """ + if config is None: + config = SentimentConfig(keywords=keywords) + + # 步骤1：全部获取 + all_items = await self.fetch_all_hot_data(limit_per_platform=config.items_per_platform) + + # 步骤2：关键词过滤 + filtered_items = self._filter_by_keywords(all_items, config.keywords) + + # 平台过滤 + if config.include_platforms: + filtered_items = [item for item in filtered_items if item.get("source_platform") in config.include_platforms] + if config.exclude_platforms: + filtered_items = [item for item in filtered_items if item.get("source_platform") not in config.exclude_platforms] + + # 限制总数 + filtered_items = filtered_items[:config.total_items] + + return { + "keywords": config.keywords, + "total_items": len(filtered_items), + "items": filtered_items + } + + def _filter_by_keywords(self, items: List[Dict], keywords: List[str]) -> List[Dict]: + """ + 按关键词过滤 + + Args: + items: 条目列表 + keywords: 关键词列表 + + Returns: + 过滤后的条目列表 + """ + if not keywords: + return items + + filtered = [] + + for item in items: + title = item.get("title", "").lower() + desc = item.get("description", "").lower() + + # 检查是否匹配任何关键词 + matches = False + for kw in keywords: + if kw.lower() in title or kw.lower() in desc: + matches = True + break + + if matches: + filtered.append(item) + + # 按热度排序 + filtered.sort(key=lambda x: x.get("hot", 0) or x.get("score", 0), reverse=True) + + return filtered + + def format_monitoring_response(self, monitor_data: Dict[str, Any]) -> str: + """ + 格式化监控响应 + + Args: + monitor_data: 监控数据 + + Returns: + 格式化的响应文本 + """ + items = monitor_data.get("items", []) + keywords = monitor_data.get("keywords", []) + + if not items: + return f"❌ 暂无关于「{', '.join(keywords)}」的舆情数据" + + response = f"🔍 **舆情监控 - {', '.join(keywords)}**\n" + response += f"共 {monitor_data['total_items']} 条相关内容\n" + response += "-" * 40 + "\n\n" + + for i, item in enumerate(items, 1): + title = item.get("title", "无标题") + hot = item.get("hot", item.get("score", "")) + platform = item.get("source_platform", "") + + response += f"{i}. {title}\n" + if hot: + response += f" 🔥 热度: {hot}" + if platform: + response += f" | 📱 {platform}" + response += "\n\n" + + return response + + def parse_keywords_from_input(self, user_input: str) -> List[str]: + """ + 从用户输入解析关键词 + + Args: + user_input: 用户输入 + + Returns: + 关键词列表 + """ + user_input = user_input.strip() + + # 常用监控关键词 + common_keywords = [ + "AI", "人工智能", "ChatGPT", "大模型", + "特斯拉", "比亚迪", "新能源汽车", + "华为", "iPhone", "小米", + "直播", "电商", "网红", + "房价", "股票", "基金", + "高考", "考研", "留学" + ] + + matched_keywords = [] + + for kw in common_keywords: + if kw.lower() in user_input.lower(): + matched_keywords.append(kw) + + # 如果没有匹配到常用关键词，尝试提取用户输入的词 + if not matched_keywords: + # 按逗号、空格等分隔 + import re + words = re.split(r'[,，\s]+', user_input) + matched_keywords = [w for w in words if w.strip()] + + return matched_keywords + + async def process_user_request(self, user_input: str) -> Dict[str, Any]: + """ + 处理用户请求 + + Args: + user_input: 用户输入 + + Returns: + 处理结果 + """ + # 解析关键词 + keywords = self.parse_keywords_from_input(user_input) + + if not keywords: + # 返回引导信息 + return { + "action": "ask_keywords", + "message": "🔍 **舆情监控**\n\n" + "请输入您想监控的关键词，例如：\n" + "• AI、人工智能、ChatGPT\n" + "• 特斯拉、比亚迪、新能源汽车\n" + "• 华为、iPhone、小米\n" + "• 直播、电商、网红\n" + "• 房价、股票、基金\n\n" + "💡 您可以直接输入任意关键词进行监控" + } + + # 获取监控数据 + monitor_data = await self.monitor_keywords(keywords) + + # 格式化响应 + response_text = self.format_monitoring_response(monitor_data) + + return { + "action": "show_monitoring", + "data": monitor_data, + "message": response_text + } + + +# 便捷函数 +async def create_sentiment_monitor(api_client=None, formatter=None) -> SentimentMonitor: + """创建舆情监控实例""" + return SentimentMonitor(api_client=api_client, formatter=formatter) + + +if __name__ == "__main__": + # 测试代码 + async def test(): + monitor = await create_sentiment_monitor() + + print("测试舆情监控...") + result = await monitor.process_user_request("监控AI和特斯拉") + print(result["message"]) + + asyncio.run(test()) diff --git a/skills/daily-hot-news/storage.py b/skills/daily-hot-news/storage.py new file mode 100644 index 0000000..9cc0869 --- /dev/null +++ b/skills/daily-hot-news/storage.py @@ -0,0 +1,345 @@ +# -*- coding: utf-8 -*- +""" +每日热榜 Skill - 数据存储模块 + +功能： +- 自动保存每日热榜数据 +- 查询历史热榜记录 +- 管理数据文件 +""" + +import json +import os +from datetime import datetime, timedelta +from typing import Dict, List, Optional, Any +from config import config + + +class DataStorage: + """数据存储类""" + + def __init__(self): + self.data_dir = config.data_dir + self.auto_save = config.auto_save + os.makedirs(self.data_dir, exist_ok=True) + + def save_hot_list(self, source_id: str, data: Dict[str, Any]) -> bool: + """ + 保存热榜数据到文件 + + Args: + source_id: 热榜源 ID（如 weibo, zhihu） + data: 热榜数据字典 + + Returns: + 是否保存成功 + """ + if not self.auto_save: + return False + + try: + # 获取今日日期的文件路径 + file_path = config.get_data_path(source_id) + + # 构建存储结构 + storage_data = { + "source_id": source_id, + "save_time": datetime.now().isoformat(), + "update_time": data.get("update_time", ""), + "total": data.get("total", 0), + "data": data.get("data", []) + } + + # 读取现有数据（如果有） + existing_data = [] + if os.path.exists(file_path): + with open(file_path, 'r', encoding='utf-8') as f: + try: + existing_data = json.load(f) + if not isinstance(existing_data, list): + existing_data = [] + except: + existing_data = [] + + # 添加新数据到列表 + existing_data.append(storage_data) + + # 保存到文件 + with open(file_path, 'w', encoding='utf-8') as f: + json.dump(existing_data, f, ensure_ascii=False, indent=2) + + return True + + except Exception as e: + print(f"[DataStorage] 保存失败: {e}") + return False + + def load_history(self, source_id: str = "", days: int = 7) -> Dict[str, List[Dict]]: + """ + 加载历史热榜数据 + + Args: + source_id: 热榜源 ID（空则加载所有源） + days: 加载最近几天的数据 + + Returns: + 按日期组织的热榜数据 + """ + result = {} + + if source_id: + # 加载指定源的历史数据 + source_dir = os.path.join(self.data_dir, source_id) + if not os.path.exists(source_dir): + return {} + + for i in range(days): + date_str = (datetime.now() - timedelta(days=i)).strftime("%Y-%m-%d") + file_path = os.path.join(source_dir, f"{date_str}.json") + + if os.path.exists(file_path): + with open(file_path, 'r', encoding='utf-8') as f: + try: + data = json.load(f) + result[date_str] = data + except: + continue + + else: + # 加载所有源的历史数据 + if not os.path.exists(self.data_dir): + return {} + + for source in os.listdir(self.data_dir): + source_dir = os.path.join(self.data_dir, source) + if not os.path.isdir(source_dir): + continue + + result[source] = {} + for i in range(days): + date_str = (datetime.now() - timedelta(days=i)).strftime("%Y-%m-%d") + file_path = os.path.join(source_dir, f"{date_str}.json") + + if os.path.exists(file_path): + with open(file_path, 'r', encoding='utf-8') as f: + try: + data = json.load(f) + result[source][date_str] = data + except: + continue + + return result + + def get_saved_dates(self, source_id: str) -> List[str]: + """ + 获取指定热榜源已保存的日期列表 + + Args: + source_id: 热榜源 ID + + Returns: + 已保存的日期列表（降序） + """ + dates = [] + source_dir = os.path.join(self.data_dir, source_id) + + if not os.path.exists(source_dir): + return [] + + for filename in os.listdir(source_dir): + if filename.endswith('.json'): + date_str = filename.replace('.json', '') + dates.append(date_str) + + return sorted(dates, reverse=True) + + def get_old_data_files(self, days: int = 7) -> List[Dict[str, str]]: + """ + 获取指定天数之前的旧数据文件列表 + + Args: + days: 天数阈值（默认7天） + + Returns: + 旧文件列表 [{source_id, date_str, file_path}] + """ + from datetime import datetime, timedelta + + old_files = [] + cutoff_date = datetime.now() - timedelta(days=days) + + if not os.path.exists(self.data_dir): + return [] + + for source in os.listdir(self.data_dir): + source_dir = os.path.join(self.data_dir, source) + if not os.path.isdir(source_dir): + continue + + for filename in os.listdir(source_dir): + if not filename.endswith('.json'): + continue + + try: + file_date = datetime.strptime(filename.replace('.json', ''), "%Y-%m-%d") + if file_date < cutoff_date: + old_files.append({ + "source_id": source, + "date_str": filename.replace('.json', ''), + "file_path": os.path.join(source_dir, filename) + }) + except: + continue + + return old_files + + def cleanup_old_files(self, files: List[Dict[str, str]]) -> int: + """ + 删除指定的旧文件 + + Args: + files: 旧文件列表 + + Returns: + 删除的文件数量 + """ + deleted_count = 0 + for f in files: + try: + if os.path.exists(f["file_path"]): + os.remove(f["file_path"]) + deleted_count += 1 + except Exception as e: + print(f"[DataStorage] 删除失败 {f['file_path']}: {e}") + + return deleted_count + + def load_hot_list(self, source_id: str, date_str: str = "") -> Optional[List[Dict]]: + """ + 加载指定日期的热榜数据 + + Args: + source_id: 热榜源 ID + date_str: 日期（默认今天） + + Returns: + 热榜数据列表 + """ + if not date_str: + date_str = datetime.now().strftime("%Y-%m-%d") + + file_path = config.get_data_path(source_id, date_str) + + if not os.path.exists(file_path): + return None + + with open(file_path, 'r', encoding='utf-8') as f: + try: + data = json.load(f) + # 返回当天的最后一条记录 + if isinstance(data, list) and len(data) > 0: + return data[-1].get("data", []) + except: + pass + + return None + + def list_all_data(self) -> Dict[str, int]: + """ + 列出所有已保存的热榜数据统计 + + Returns: + 源ID -> 保存记录数 + """ + stats = {} + + if not os.path.exists(self.data_dir): + return {} + + for source in os.listdir(self.data_dir): + source_dir = os.path.join(self.data_dir, source) + if not os.path.isdir(source_dir): + continue + + count = 0 + for filename in os.listdir(source_dir): + if filename.endswith('.json'): + count += 1 + + if count > 0: + stats[source] = count + + return stats + + def clear_old_data(self, keep_days: int = 30) -> int: + """ + 清理旧数据 + + Args: + keep_days: 保留最近几天的数据 + + Returns: + 删除的文件数量 + """ + deleted_count = 0 + cutoff_date = datetime.now() - timedelta(days=keep_days) + + if not os.path.exists(self.data_dir): + return 0 + + for source in os.listdir(self.data_dir): + source_dir = os.path.join(self.data_dir, source) + if not os.path.isdir(source_dir): + continue + + for filename in os.listdir(source_dir): + if not filename.endswith('.json'): + continue + + try: + file_date = datetime.strptime(filename.replace('.json', ''), "%Y-%m-%d") + if file_date < cutoff_date: + file_path = os.path.join(source_dir, filename) + os.remove(file_path) + deleted_count += 1 + except: + continue + + return deleted_count + + +# 全局存储实例 +storage = DataStorage() + + +def auto_save_hot_list(source_id: str, data: Dict[str, Any]) -> bool: + """ + 自动保存热榜数据的便捷函数 + + 在获取热榜数据后调用此函数保存数据 + """ + return storage.save_hot_list(source_id, data) + + +if __name__ == "__main__": + # 测试存储功能 + print("📊 每日热榜数据存储管理") + print("=" * 50) + + # 列出所有数据 + stats = storage.list_all_data() + if stats: + print("\n已保存的热榜数据：") + for source, count in stats.items(): + print(f" • {source}: {count} 条记录") + else: + print("\n暂无保存的数据") + + # 列出已保存的日期 + print("\n已保存的日期（微博）：") + dates = storage.get_saved_dates("weibo") + if dates: + for date in dates[:7]: + print(f" • {date}") + else: + print(" 暂无数据")