Add 39 shared skills from local agent inventory

Sources: ~/.openclaw/skills, ~/.agents/skills, workshop-skills, workspace/skills

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())