""" 个性化订阅（增强版） - 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())