""" 行业热榜垂直（增强版） - 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())