1#!/usr/bin/env python3
2"""
3小红书热点跟踪工具
4 
5用法:
6    python track-topic.py <话题> [--limit N] [--feishu] [--output FILE]
7 
8示例:
9    python track-topic.py "DeepSeek" --limit 5 --feishu
10    python track-topic.py "春节旅游" --limit 10 --output report.md
11"""
12 
13import argparse
14import json
15import subprocess
16import sys
17import os
18from datetime import datetime
19from pathlib import Path
20 
21# 获取脚本目录
22SCRIPT_DIR = Path(__file__).parent.resolve()
23XHS_SCRIPTS = SCRIPT_DIR  # 现在就在 xiaohongshu/scripts 目录下
24 
25# 飞书 skill 路径（支持多种可能的位置）
26def find_feishu_scripts() -> Path:
27    """查找 feishu-docs skill 的 scripts 目录"""
28    # 只允许在已知的 skill 目录中查找
29    allowed_roots = [
30        SCRIPT_DIR.parent.parent,  # 同级 skill 目录
31        Path.home() / ".openclaw" / "workspace" / "skills",
32        Path.home() / ".claude" / "skills",
33    ]
34    for root in allowed_roots:
35        candidate = (root / "feishu-docs" / "scripts").resolve()
36        # 校验解析后的路径仍在允许的根目录下（防止符号链接逃逸）
37        if candidate.is_dir() and any(
38            str(candidate).startswith(str(r.resolve()) + os.sep) for r in allowed_roots
39        ):
40            return candidate
41    return allowed_roots[0] / "feishu-docs" / "scripts"  # 返回默认路径（可能不存在）
42 
43FEISHU_SCRIPTS = find_feishu_scripts()
44 
45 
46def call_xhs_mcp(tool: str, args: dict) -> dict:
47    """调用小红书 MCP 工具"""
48    mcp_call = XHS_SCRIPTS / "mcp-call.sh"
49    if not mcp_call.exists():
50        print(f"❌ 找不到 xiaohongshu skill: {mcp_call}", file=sys.stderr)
51        sys.exit(1)
52    
53    result = subprocess.run(
54        [str(mcp_call), tool, json.dumps(args)],
55        capture_output=True, text=True, timeout=120
56    )
57    
58    if result.returncode != 0:
59        print(f"❌ MCP 调用失败: {result.stderr}", file=sys.stderr)
60        return {}
61    
62    try:
63        response = json.loads(result.stdout)
64        if "result" in response and "content" in response["result"]:
65            text = response["result"]["content"][0].get("text", "{}")
66            return json.loads(text) if text else {}
67        elif "error" in response:
68            print(f"⚠️ MCP 错误: {response['error'].get('message', 'Unknown')}", file=sys.stderr)
69            return {}
70        return response
71    except json.JSONDecodeError:
72        return {}
73 
74 
75def search_feeds(keyword: str) -> list:
76    """搜索小红书内容"""
77    print(f"🔍 搜索: {keyword}")
78    result = call_xhs_mcp("search_feeds", {"keyword": keyword})
79    feeds = result.get("feeds", [])
80    # 过滤掉 hot_query 类型
81    return [f for f in feeds if f.get("modelType") == "note"]
82 
83 
84def get_feed_detail(feed_id: str, xsec_token: str, load_comments: bool = True) -> dict:
85    """获取帖子详情"""
86    args = {
87        "feed_id": feed_id,
88        "xsec_token": xsec_token,
89        "load_all_comments": load_comments
90    }
91    result = call_xhs_mcp("get_feed_detail", args)
92    return result.get("data", {})
93 
94 
95def format_timestamp(ts: int) -> str:
96    """格式化时间戳"""
97    if not ts:
98        return "未知"
99    try:
100        dt = datetime.fromtimestamp(ts / 1000)
101        return dt.strftime("%Y-%m-%d %H:%M")
102    except:
103        return "未知"
104 
105 
106def get_comments_list(post: dict) -> list:
107    """安全地获取评论列表"""
108    comments = post.get("comments", {})
109    if isinstance(comments, dict):
110        return comments.get("list", [])
111    elif isinstance(comments, list):
112        return comments
113    return []
114 
115 
116def generate_report(keyword: str, posts: list) -> str:
117    """生成 Markdown 报告"""
118    now = datetime.now().strftime("%Y-%m-%d %H:%M")
119    
120    report = f"""# 🔥 小红书热点跟踪报告
121 
122**话题:** {keyword}  
123**生成时间:** {now}  
124**收录帖子:** {len(posts)} 篇
125 
126---
127 
128## 📊 概览
129 
130"""
131    
132    # 统计信息
133    total_likes = sum(int(p.get("note", {}).get("interactInfo", {}).get("likedCount", 0) or 0) for p in posts)
134    total_comments = sum(len(get_comments_list(p)) for p in posts)
135    
136    report += f"""| 指标 | 数值 |
137|------|------|
138| 总帖子数 | {len(posts)} |
139| 总点赞数 | {total_likes:,} |
140| 总评论数 | {total_comments} |
141 
142---
143 
144## 📝 热帖详情
145 
146"""
147    
148    for i, post in enumerate(posts, 1):
149        note = post.get("note", {})
150        comments = get_comments_list(post)
151        
152        title = note.get("title", "无标题")
153        desc = note.get("desc", "")
154        user = note.get("user", {}).get("nickname", "匿名")
155        time_str = format_timestamp(note.get("time"))
156        interact = note.get("interactInfo", {})
157        likes = interact.get("likedCount", "0")
158        collected = interact.get("collectedCount", "0")
159        
160        report += f"""### {i}. {title}
161 
162**作者:** {user}  
163**时间:** {time_str}  
164**互动:** ❤️ {likes} 赞 · ⭐ {collected} 收藏
165 
166**正文:**
167 
168> {desc[:500]}{"..." if len(desc) > 500 else ""}
169 
170"""
171        
172        if comments:
173            report += f"""**热门评论 ({len(comments)} 条):**
174 
175"""
176            for j, comment in enumerate(list(comments)[:5], 1):
177                c_user = comment.get("userInfo", {}).get("nickname", "匿名")
178                c_content = comment.get("content", "")
179                c_likes = comment.get("likeCount", 0)
180                report += f"- **{c_user}** ({c_likes}赞): {c_content[:100]}\n"
181            
182            if len(comments) > 5:
183                report += f"- *... 还有 {len(comments) - 5} 条评论*\n"
184        
185        report += "\n---\n\n"
186    
187    # 评论区热点总结
188    report += """## 💬 评论区热点关键词
189 
190"""
191    
192    # 简单的关键词提取（统计高频词）
193    all_comments = []
194    for post in posts:
195        for c in get_comments_list(post):
196            all_comments.append(c.get("content", ""))
197    
198    if all_comments:
199        report += f"共 {len(all_comments)} 条评论，主要讨论方向：\n\n"
200        # 这里可以做更复杂的 NLP 分析，暂时简化
201        report += "- 用户对该话题的关注度较高\n"
202        report += "- 评论区互动活跃\n"
203    else:
204        report += "暂无足够评论数据进行分析\n"
205    
206    report += """
207---
208 
209## 📈 趋势分析
210 
211基于以上热帖和评论数据，该话题在小红书上呈现以下特点：
212 
2131. **热度指数**: """ + ("🔥🔥🔥 高" if total_likes > 1000 else "🔥🔥 中" if total_likes > 100 else "🔥 低") + f"""
2142. **互动活跃度**: """ + ("活跃" if total_comments > 50 else "一般" if total_comments > 10 else "较低") + """
2153. **内容类型**: 以图文笔记为主
216 
217---
218 
219*报告由 OpenClaw 小红书热点跟踪工具自动生成*
220"""
221    
222    return report
223 
224 
225def export_to_feishu(title: str, content: str) -> str:
226    """导出到飞书文档"""
227    import_script = FEISHU_SCRIPTS / "doc-import.sh"
228    if not import_script.exists():
229        print(f"❌ 找不到 feishu-docs skill: {import_script}", file=sys.stderr)
230        return ""
231    
232    print("📤 导出到飞书文档...")
233    
234    # 写入临时文件
235    tmp_file = Path("/tmp/xhs_report.md")
236    tmp_file.write_text(content, encoding="utf-8")
237    
238    result = subprocess.run(
239        [str(import_script), title, "--file", str(tmp_file)],
240        capture_output=True, text=True, timeout=60
241    )
242    
243    if result.returncode != 0:
244        print(f"⚠️ 飞书导出失败: {result.stderr}", file=sys.stderr)
245        return ""
246    
247    # 解析返回的文档链接
248    output = result.stdout
249    print(output)
250    return output
251 
252 
253def main():
254    parser = argparse.ArgumentParser(description="小红书热点跟踪工具")
255    parser.add_argument("keyword", help="要跟踪的话题/关键词")
256    parser.add_argument("--limit", "-n", type=int, default=10, help="获取帖子数量 (默认 10)")
257    parser.add_argument("--feishu", "-f", action="store_true", help="导出到飞书文档")
258    parser.add_argument("--output", "-o", help="输出 Markdown 文件路径")
259    parser.add_argument("--no-comments", action="store_true", help="不获取评论")
260    
261    args = parser.parse_args()
262    
263    # 1. 搜索帖子
264    feeds = search_feeds(args.keyword)
265    if not feeds:
266        print("❌ 未找到相关帖子")
267        sys.exit(1)
268    
269    print(f"✅ 找到 {len(feeds)} 条帖子")
270    
271    # 2. 获取详情
272    posts = []
273    for i, feed in enumerate(feeds[:args.limit]):
274        feed_id = feed.get("id")
275        xsec_token = feed.get("xsecToken")
276        title = feed.get("noteCard", {}).get("displayTitle", "")
277        
278        print(f"📖 [{i+1}/{min(len(feeds), args.limit)}] 获取: {title[:30]}...")
279        
280        detail = get_feed_detail(feed_id, xsec_token, not args.no_comments)
281        if detail:
282            posts.append(detail)
283    
284    if not posts:
285        print("❌ 未能获取帖子详情")
286        sys.exit(1)
287    
288    print(f"✅ 成功获取 {len(posts)} 篇帖子详情")
289    
290    # 3. 生成报告
291    print("📝 生成报告...")
292    report = generate_report(args.keyword, posts)
293    
294    # 4. 输出
295    if args.output:
296        output_path = Path(args.output)
297        output_path.write_text(report, encoding="utf-8")
298        print(f"✅ 报告已保存: {output_path}")
299    
300    if args.feishu:
301        doc_title = f"小红书热点跟踪: {args.keyword} ({datetime.now().strftime('%m-%d')})"
302        export_to_feishu(doc_title, report)
303    
304    if not args.output and not args.feishu:
305        # 默认输出到 stdout
306        print("\n" + "="*60 + "\n")
307        print(report)
308    
309    return report
310 
311 
312if __name__ == "__main__":
313    main()
314