NightStar
3d3dd79df0
Sources: ~/.openclaw/skills, ~/.agents/skills, workshop-skills, workspace/skills
370 lines
13 KiB
Python
370 lines
13 KiB
Python
"""
|
|
行业热榜垂直（增强版） - 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())
|