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