|
|
@@ -1,5 +1,15 @@
|
|
|
-# -*- coding: utf-8 -*-
|
|
|
-"""小红书评论检测脚本 - 检测有评论的笔记，推送带回复建议的通知"""
|
|
|
+"""小红书评论检测脚本 - 记录每篇笔记评论数，发现新增评论时读取具体评论并推送给回复建议
|
|
|
+
|
|
|
+运行逻辑:
|
|
|
+1. CDP 网络捕获 note-manager 的 posted API，获取每篇笔记的 id/xsec_token/comments_count/标题
|
|
|
+2. 与状态文件 xhs_comment_state.json 对比: 评论数比上次增加的笔记 → 打开详情页读取评论区
|
|
|
+3. 用 createTime > last_check_time 识别新增评论(含楼中楼回复)
|
|
|
+4. 每条新评论用 opencode 生成针对性回复建议
|
|
|
+5. 推送飞书(逐条评论+建议); 无新增评论则不推送
|
|
|
+6. 更新状态文件。首次运行只建立基线, 不推送
|
|
|
+"""
|
|
|
+import argparse
|
|
|
+import datetime
|
|
|
import io
|
|
|
import json
|
|
|
import os
|
|
|
@@ -18,6 +28,7 @@ if not sys.stdout.isatty():
|
|
|
SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))
|
|
|
OPERATION_DIR = SCRIPT_DIR
|
|
|
CONFIG_PATH = os.path.join(OPERATION_DIR, "xhs_daily_config.json")
|
|
|
+STATE_PATH = os.path.join(OPERATION_DIR, "xhs_comment_state.json")
|
|
|
|
|
|
XHS_SKILLS_PATH = r"C:\code\XiaohongshuSkills\scripts"
|
|
|
sys.path.insert(0, XHS_SKILLS_PATH)
|
|
|
@@ -25,6 +36,18 @@ sys.path.insert(0, OPERATION_DIR)
|
|
|
|
|
|
from cdp_publish import XiaohongshuPublisher
|
|
|
from xhs_feishu import _get_token, FEISHU_MSG_URL
|
|
|
+import opencode_client
|
|
|
+
|
|
|
+POSTED_API_PATH = "/api/galaxy/v2/creator/note/user/posted"
|
|
|
+NOTE_MANAGER_URL = "https://creator.xiaohongshu.com/new/note-manager"
|
|
|
+
|
|
|
+FALLBACK_SUGGESTIONS = [
|
|
|
+ "谢谢你这么用心地评论，看到你的留言真的很暖心～",
|
|
|
+ "哈哈哈你这个角度很特别，欢迎常来聊～",
|
|
|
+ "你的想法很有意思，我记下了，后面单独展开说说",
|
|
|
+ "谢谢支持！你提的这个点我改天专门写一篇",
|
|
|
+ "被你说中了，看来咱们想到一块儿去了 😄",
|
|
|
+]
|
|
|
|
|
|
|
|
|
def _load_config():
|
|
|
@@ -32,140 +55,246 @@ def _load_config():
|
|
|
return json.load(f)
|
|
|
|
|
|
|
|
|
-def extract_comment_counts(publisher, limit=10):
|
|
|
- """从笔记管理页提取每篇笔记的评论数。
|
|
|
+def _load_state():
|
|
|
+ if not os.path.exists(STATE_PATH):
|
|
|
+ return {"notes": {}}
|
|
|
+ try:
|
|
|
+ with open(STATE_PATH, "r", encoding="utf-8") as f:
|
|
|
+ return json.load(f)
|
|
|
+ except Exception:
|
|
|
+ return {"notes": {}}
|
|
|
|
|
|
- 页面数字顺序：阅读 / 评论 / 点赞 / 收藏 / 分享
|
|
|
- """
|
|
|
- publisher._navigate("https://creator.xiaohongshu.com/new/note-manager")
|
|
|
- time.sleep(6)
|
|
|
-
|
|
|
- text = publisher._evaluate("document.body.innerText") or ""
|
|
|
- lines = [ln.strip() for ln in text.split("\n")]
|
|
|
-
|
|
|
- notes = []
|
|
|
- i = 0
|
|
|
- while i < len(lines) and len(notes) < limit:
|
|
|
- ln = lines[i]
|
|
|
- if re.match(r"^\d{4}-\d{2}-\d{2} \d{2}:\d{2}$", ln) and i + 5 < len(lines):
|
|
|
- title = lines[i - 1].strip() if i >= 1 else ""
|
|
|
- nums = []
|
|
|
- j = i + 1
|
|
|
- while j < len(lines) and len(nums) < 5 and lines[j].isdigit():
|
|
|
- nums.append(int(lines[j]))
|
|
|
- j += 1
|
|
|
- if len(nums) >= 5 and title:
|
|
|
- notes.append({
|
|
|
- "title": title,
|
|
|
- "publish_date": ln[:10],
|
|
|
- "read_count": nums[0],
|
|
|
- "comment_count": nums[1],
|
|
|
- "like_count": nums[2],
|
|
|
- "collect_count": nums[3],
|
|
|
- "share_count": nums[4],
|
|
|
- })
|
|
|
- i = j if j > i + 1 else i + 1
|
|
|
- else:
|
|
|
- i += 1
|
|
|
- return notes
|
|
|
|
|
|
+def _save_state(state):
|
|
|
+ tmp = STATE_PATH + ".tmp"
|
|
|
+ with open(tmp, "w", encoding="utf-8") as f:
|
|
|
+ json.dump(state, f, ensure_ascii=False, indent=1)
|
|
|
+ os.replace(tmp, STATE_PATH)
|
|
|
+
|
|
|
+
|
|
|
+def _now_ms() -> int:
|
|
|
+ return int(time.time() * 1000)
|
|
|
|
|
|
-# ─── 基于笔记标题生成针对性回复建议 ──────────────────────────────────────
|
|
|
-
|
|
|
-REPLY_TEMPLATES = {
|
|
|
- # 教育/育儿类
|
|
|
- "孩子": [
|
|
|
- "你家孩子也这样吗？来评论区聊聊你的经历 😊",
|
|
|
- "有没有同款娃？说说你家娃的情况～",
|
|
|
- "这个方法你家试过吗？效果怎么样？",
|
|
|
- ],
|
|
|
- "家长": [
|
|
|
- "作为家长，你们是怎么处理类似情况的？",
|
|
|
- "大家在育儿路上遇到过什么难题？",
|
|
|
- "评论区分享一下你们的经验吧 💬",
|
|
|
- ],
|
|
|
- "学习": [
|
|
|
- "你们家孩子学习时有什么好习惯？",
|
|
|
- "有没有其他方法可以试试？",
|
|
|
- "大家在学习方法上有什么心得？",
|
|
|
- ],
|
|
|
- "小学": [
|
|
|
- "小学生家长你们有什么建议？",
|
|
|
- "有没有类似经历？分享一下～",
|
|
|
- "大家怎么看待小学阶段的教育？",
|
|
|
- ],
|
|
|
- # 家庭/关系类
|
|
|
- "家": [
|
|
|
- "你家是怎么处理类似情况的？",
|
|
|
- "评论区聊聊你们的家庭故事 🏠",
|
|
|
- "大家有什么好的建议？",
|
|
|
- ],
|
|
|
- # 健康/生活类
|
|
|
- "温度": [
|
|
|
- "你们家室温一般控制在多少度？",
|
|
|
- "有没有其他方法保持室内舒适？",
|
|
|
- "评论区交流一下经验 🌡️",
|
|
|
- ],
|
|
|
- "睡": [
|
|
|
- "你家宝宝睡觉怎么样？",
|
|
|
- "有没有好的睡眠建议？",
|
|
|
- "大家是怎么解决睡眠问题的？",
|
|
|
- ],
|
|
|
- # 消费/决策类
|
|
|
- "决定": [
|
|
|
- "你们做过什么让自己不后悔的决定？",
|
|
|
- "评论区聊聊你的重要决定 💭",
|
|
|
- "有没有类似经历？分享一下",
|
|
|
- ],
|
|
|
- "教育": [
|
|
|
- "你们的教育理念是什么？",
|
|
|
- "评论区交流一下育儿心得 📚",
|
|
|
- "大家觉得最好的教育方式是什么？",
|
|
|
- ],
|
|
|
- "博主": [
|
|
|
- "你们有没有被割韭菜的经历？",
|
|
|
- "评论区说说你们的经验 💡",
|
|
|
- "大家是怎么避坑的？",
|
|
|
- ],
|
|
|
- # 通用
|
|
|
- "话题": [
|
|
|
- "你们怎么看这个问题？",
|
|
|
- "评论区说说你的想法 💬",
|
|
|
- "大家有什么不同的看法？",
|
|
|
- ],
|
|
|
-}
|
|
|
-
|
|
|
-GENERAL_SUGGESTIONS = [
|
|
|
- "你觉得呢？来评论区聊聊～",
|
|
|
- "大家有什么想说的？欢迎分享 🙌",
|
|
|
- "欢迎在评论区分享你的经历",
|
|
|
- "有没有同款经历？评论区见 👋",
|
|
|
- "说说你的看法吧～",
|
|
|
-]
|
|
|
|
|
|
+def _fmt_ms(ms: int) -> str:
|
|
|
+ try:
|
|
|
+ return datetime.datetime.fromtimestamp(ms / 1000).strftime("%m-%d %H:%M")
|
|
|
+ except Exception:
|
|
|
+ return ""
|
|
|
|
|
|
-def generate_reply_suggestions(title, comment_count):
|
|
|
- """根据笔记标题生成针对性的回复建议。"""
|
|
|
- suggestions = []
|
|
|
- title_lower = title.lower()
|
|
|
|
|
|
- for keywords, hints in REPLY_TEMPLATES.items():
|
|
|
- if keywords in title:
|
|
|
- # 根据评论数选择建议数量（最多3条）
|
|
|
- num = min(comment_count, 3)
|
|
|
- suggestions.extend(hints[:num])
|
|
|
+def _parse_posted_body(body_text: str) -> list:
|
|
|
+ """解析 posted API 响应, 返回笔记列表。"""
|
|
|
+ try:
|
|
|
+ payload = json.loads(body_text)
|
|
|
+ except json.JSONDecodeError:
|
|
|
+ return []
|
|
|
+ notes = ((payload.get("data") or {}).get("notes")) or []
|
|
|
+ result = []
|
|
|
+ for n in notes:
|
|
|
+ if not isinstance(n, dict):
|
|
|
+ continue
|
|
|
+ note_id = str(n.get("id") or "").strip()
|
|
|
+ if not note_id:
|
|
|
+ continue
|
|
|
+ result.append({
|
|
|
+ "id": note_id,
|
|
|
+ "title": str(n.get("display_title") or "(无标题)").strip(),
|
|
|
+ "comments_count": int(n.get("comments_count") or 0),
|
|
|
+ "view_count": int(n.get("view_count") or 0),
|
|
|
+ "xsec_token": str(n.get("xsec_token") or "").strip(),
|
|
|
+ "publish_time": str(n.get("time") or "").strip(),
|
|
|
+ })
|
|
|
+ return result
|
|
|
+
|
|
|
+
|
|
|
+def capture_posted_notes(publisher, max_wait: float = 30.0) -> list:
|
|
|
+ """通过 CDP 网络捕获 note-manager 的 posted API, 滚动加载全部笔记。
|
|
|
+
|
|
|
+ 注意: 滚动必须用裸 ws.send 发 Runtime.evaluate, 不能走 publisher._evaluate
|
|
|
+ (其内部 _send 会 recv 吞掉网络事件, 导致漏抓后续页面的 API 响应)。
|
|
|
+ """
|
|
|
+ publisher._send("Page.enable")
|
|
|
+ publisher._send("Network.enable", {"maxPostDataSize": 65536})
|
|
|
+ publisher._navigate(NOTE_MANAGER_URL)
|
|
|
+ time.sleep(3.0)
|
|
|
+
|
|
|
+ request_url_by_id = {}
|
|
|
+ target_request_ids = set()
|
|
|
+ start = time.time()
|
|
|
+ last_scroll_at = 0.0
|
|
|
+ last_new_at = time.time()
|
|
|
+ _scroll_cmd_id = 9000
|
|
|
+
|
|
|
+ while time.time() - start < max_wait:
|
|
|
+ # 捕获到请求后空闲 5s → 提前结束
|
|
|
+ if target_request_ids and time.time() - last_new_at > 5.0:
|
|
|
break
|
|
|
|
|
|
- # 如果没有匹配到模板，使用通用建议
|
|
|
- if not suggestions:
|
|
|
- suggestions = GENERAL_SUGGESTIONS[:min(comment_count, 2)]
|
|
|
+ timeout = min(0.8, max(0.1, max_wait - (time.time() - start)))
|
|
|
+ try:
|
|
|
+ raw = publisher.ws.recv(timeout=timeout)
|
|
|
+ except TimeoutError:
|
|
|
+ # 空闲期滚动触发下一页加载(裸 ws.send, 不 recv)
|
|
|
+ if time.time() - last_scroll_at > 2.5:
|
|
|
+ _scroll_cmd_id += 1
|
|
|
+ publisher.ws.send(json.dumps({
|
|
|
+ "id": _scroll_cmd_id,
|
|
|
+ "method": "Runtime.evaluate",
|
|
|
+ "params": {"expression": "window.scrollTo(0, document.body.scrollHeight)"},
|
|
|
+ }))
|
|
|
+ last_scroll_at = time.time()
|
|
|
+ continue
|
|
|
|
|
|
- return suggestions[:3] # 最多3条
|
|
|
+ try:
|
|
|
+ message = json.loads(raw)
|
|
|
+ except json.JSONDecodeError:
|
|
|
+ continue
|
|
|
+ method = message.get("method")
|
|
|
+ params = message.get("params", {})
|
|
|
+
|
|
|
+ if method == "Network.requestWillBeSent":
|
|
|
+ rid = params.get("requestId")
|
|
|
+ req = params.get("request", {})
|
|
|
+ if isinstance(rid, str):
|
|
|
+ request_url_by_id[rid] = req.get("url", "")
|
|
|
+ elif method == "Network.responseReceived":
|
|
|
+ rid = params.get("requestId")
|
|
|
+ if not isinstance(rid, str):
|
|
|
+ continue
|
|
|
+ url = request_url_by_id.get(rid, "")
|
|
|
+ if POSTED_API_PATH not in url:
|
|
|
+ continue
|
|
|
+ if params.get("response", {}).get("status") != 200:
|
|
|
+ continue
|
|
|
+ if rid not in target_request_ids:
|
|
|
+ target_request_ids.add(rid)
|
|
|
+ last_new_at = time.time()
|
|
|
+
|
|
|
+ # 空闲太久且已无新请求 → 停止
|
|
|
+ if not target_request_ids:
|
|
|
+ raise RuntimeError("未捕获到笔记列表 API 响应")
|
|
|
+
|
|
|
+ notes_by_id = {}
|
|
|
+ for rid in target_request_ids:
|
|
|
+ body_result = publisher._send("Network.getResponseBody", {"requestId": rid})
|
|
|
+ body_text = body_result.get("body", "")
|
|
|
+ if body_result.get("base64Encoded"):
|
|
|
+ import base64
|
|
|
+ body_text = base64.b64decode(body_text).decode("utf-8", errors="replace")
|
|
|
+ for n in _parse_posted_body(body_text):
|
|
|
+ notes_by_id[n["id"]] = n
|
|
|
+
|
|
|
+ notes = list(notes_by_id.values())
|
|
|
+ print(f"[capture] pages={len(target_request_ids)} notes={len(notes)}")
|
|
|
+ return notes
|
|
|
|
|
|
|
|
|
-def check_comments():
|
|
|
- """检查评论并推送到飞书群"""
|
|
|
+def _norm_comment(c: dict, is_sub: bool) -> dict:
|
|
|
+ """规范化一条评论(父评论或楼中楼)。"""
|
|
|
+ content = str(c.get("content") or "").strip()
|
|
|
+ pictures = c.get("pictures") or []
|
|
|
+ if not content and pictures:
|
|
|
+ content = "[图片]"
|
|
|
+ user = c.get("userInfo") or {}
|
|
|
+ create_ms = int(c.get("createTime") or 0)
|
|
|
+ return {
|
|
|
+ "id": str(c.get("id") or "").strip(),
|
|
|
+ "content": content,
|
|
|
+ "nickname": str(user.get("nickname") or "").strip() or "匿名用户",
|
|
|
+ "create_ms": create_ms,
|
|
|
+ "is_sub": is_sub,
|
|
|
+ }
|
|
|
+
|
|
|
+
|
|
|
+def fetch_note_comments(publisher, note_id: str, xsec_token: str) -> list:
|
|
|
+ """打开笔记详情页读取评论(父评论+楼中楼)。"""
|
|
|
+ if not xsec_token:
|
|
|
+ return []
|
|
|
+ res = publisher.get_feed_detail(
|
|
|
+ feed_id=note_id,
|
|
|
+ xsec_token=xsec_token,
|
|
|
+ load_all_comments=True,
|
|
|
+ limit=60,
|
|
|
+ click_more_replies=True,
|
|
|
+ reply_limit=10,
|
|
|
+ )
|
|
|
+ detail = res.get("detail") or {}
|
|
|
+ comments_raw = detail.get("comments") or {}
|
|
|
+ parents = comments_raw.get("list") or []
|
|
|
+
|
|
|
+ out = []
|
|
|
+ for p in parents:
|
|
|
+ if not isinstance(p, dict):
|
|
|
+ continue
|
|
|
+ out.append(_norm_comment(p, is_sub=False))
|
|
|
+ for s in (p.get("subComments") or []):
|
|
|
+ if isinstance(s, dict):
|
|
|
+ out.append(_norm_comment(s, is_sub=True))
|
|
|
+ return out
|
|
|
+
|
|
|
+
|
|
|
+def generate_suggestions(title: str, new_comments: list) -> list:
|
|
|
+ """用 opencode 批量生成每条新评论的回复建议; 失败则回退模板。"""
|
|
|
+ lines = []
|
|
|
+ for i, c in enumerate(new_comments, 1):
|
|
|
+ lines.append(f"{i}. 用户「{c.get('nickname')}」：{c.get('content') or '(图片)'}")
|
|
|
+
|
|
|
+ prompt = (
|
|
|
+ "你是小红书博主「72年冻龄妈妈的家的算法」(家庭教育博主) 的运营助手。\n"
|
|
|
+ f"笔记《{title}》收到以下新评论({len(lines)}条)，请为每一条写一条中文回复建议。\n"
|
|
|
+ "要求: 每条 15-40 字, 口语化、真诚、像真人博主, 避免套话和模板感, 可适当互动引导。\n\n"
|
|
|
+ "评论列表:\n" + "\n".join(lines) +
|
|
|
+ "\n\n请严格按以下格式输出, 每条一行:\n1. <回复内容>\n2. <回复内容>\n..."
|
|
|
+ )
|
|
|
+ try:
|
|
|
+ reply = opencode_client.ask("xhs-comment-suggest", prompt, timeout=120)
|
|
|
+ except Exception as e:
|
|
|
+ print(f"[suggest] opencode 失败, 使用模板回退: {e}")
|
|
|
+ return [FALLBACK_SUGGESTIONS[i % len(FALLBACK_SUGGESTIONS)] for i in range(len(new_comments))]
|
|
|
+
|
|
|
+ parsed = []
|
|
|
+ for ln in reply.splitlines():
|
|
|
+ ln = ln.strip()
|
|
|
+ m = re.match(r"^(\d+)\s*[.、):：]\s*(.+)$", ln)
|
|
|
+ if m and m.group(2).strip():
|
|
|
+ parsed.append(m.group(2).strip())
|
|
|
+
|
|
|
+ if not parsed:
|
|
|
+ # 严格格式未命中, 按非空行盲取
|
|
|
+ parsed = [ln.strip() for ln in reply.splitlines() if ln.strip()]
|
|
|
+
|
|
|
+ if len(parsed) < len(new_comments):
|
|
|
+ for i in range(len(parsed), len(new_comments)):
|
|
|
+ parsed.append(FALLBACK_SUGGESTIONS[i % len(FALLBACK_SUGGESTIONS)])
|
|
|
+ return parsed[:len(new_comments)]
|
|
|
+
|
|
|
+
|
|
|
+def push_feishu(cfg, text: str):
|
|
|
+ token = _get_token(cfg["feishu_app_id"], cfg["feishu_app_secret"])
|
|
|
+ url = f"{FEISHU_MSG_URL}?receive_id_type=chat_id"
|
|
|
+ r = requests.post(
|
|
|
+ url,
|
|
|
+ headers={"Authorization": f"Bearer {token}"},
|
|
|
+ json={
|
|
|
+ "receive_id": cfg["feishu_chat_id"],
|
|
|
+ "msg_type": "text",
|
|
|
+ "content": json.dumps({"text": text}, ensure_ascii=False),
|
|
|
+ },
|
|
|
+ timeout=20,
|
|
|
+ )
|
|
|
+ d = r.json()
|
|
|
+ if d.get("code") != 0:
|
|
|
+ raise RuntimeError(f"推送失败: {d.get('msg')}")
|
|
|
+ return True
|
|
|
+
|
|
|
+
|
|
|
+def check_comments(dry_run: bool = False):
|
|
|
+ """检查新增评论并推送飞书"""
|
|
|
cfg = _load_config()
|
|
|
+ state = _load_state()
|
|
|
+ notes_state = state.setdefault("notes", {})
|
|
|
+
|
|
|
publisher = XiaohongshuPublisher()
|
|
|
publisher.connect(reuse_existing_tab=True)
|
|
|
|
|
|
@@ -173,54 +302,96 @@ def check_comments():
|
|
|
if not publisher.check_login():
|
|
|
raise RuntimeError("未登录小红书")
|
|
|
|
|
|
- notes = extract_comment_counts(publisher, limit=10)
|
|
|
- noted_with_comments = [n for n in notes if n["comment_count"] > 0]
|
|
|
-
|
|
|
- if not noted_with_comments:
|
|
|
- print("没有发现有评论的笔记")
|
|
|
+ notes = capture_posted_notes(publisher)
|
|
|
+ now = _now_ms()
|
|
|
+ first_run = not notes_state
|
|
|
+ changed = []
|
|
|
+
|
|
|
+ for n in notes:
|
|
|
+ nid = n["id"]
|
|
|
+ prev = notes_state.get(nid)
|
|
|
+
|
|
|
+ if prev is None:
|
|
|
+ # 新笔记: 建立基线
|
|
|
+ notes_state[nid] = {
|
|
|
+ "title": n["title"],
|
|
|
+ "comment_count": n["comments_count"],
|
|
|
+ "last_check_time": now,
|
|
|
+ }
|
|
|
+ continue
|
|
|
+
|
|
|
+ prev_count = int(prev.get("comment_count") or 0)
|
|
|
+ if n["comments_count"] <= prev_count:
|
|
|
+ continue
|
|
|
+
|
|
|
+ # 评论数增加 → 打开评论区
|
|
|
+ try:
|
|
|
+ comments = fetch_note_comments(publisher, nid, n.get("xsec_token") or "")
|
|
|
+ except Exception as e:
|
|
|
+ print(f"[fetch] 笔记 {nid} 评论区读取失败, 跳过: {e}")
|
|
|
+ continue
|
|
|
+
|
|
|
+ last_check_ms = int(prev.get("last_check_time") or 0)
|
|
|
+ new = [c for c in comments if c["create_ms"] > last_check_ms and c["id"]]
|
|
|
+ if not new:
|
|
|
+ # 评论被删除/替换, 只更新计数
|
|
|
+ notes_state[nid]["comment_count"] = n["comments_count"]
|
|
|
+ continue
|
|
|
+
|
|
|
+ suggestions = generate_suggestions(n["title"], new)
|
|
|
+ for i, c in enumerate(new):
|
|
|
+ c["suggestion"] = suggestions[i] if i < len(suggestions) else FALLBACK_SUGGESTIONS[i % len(FALLBACK_SUGGESTIONS)]
|
|
|
+
|
|
|
+ changed.append({
|
|
|
+ "note": n,
|
|
|
+ "new_comments": new,
|
|
|
+ })
|
|
|
+ # 更新基线: 计数 + 已读时间(取已读评论的最大时间, 防止漏报)
|
|
|
+ max_create = max(c["create_ms"] for c in new)
|
|
|
+ notes_state[nid]["comment_count"] = n["comments_count"]
|
|
|
+ notes_state[nid]["last_check_time"] = max(now, max_create)
|
|
|
+ notes_state[nid]["title"] = n["title"]
|
|
|
+
|
|
|
+ _save_state(state)
|
|
|
+
|
|
|
+ if first_run:
|
|
|
+ print("[INIT] 首次运行: 已建立全部笔记基线, 不推送")
|
|
|
return
|
|
|
|
|
|
- date = time.strftime("%Y-%m-%d")
|
|
|
+ if not changed:
|
|
|
+ print("[SKIP] 无新增评论, 不推送")
|
|
|
+ return
|
|
|
|
|
|
- # 构建消息
|
|
|
- lines = []
|
|
|
- for n in noted_with_comments:
|
|
|
- suggestions = generate_reply_suggestions(n["title"], n["comment_count"])
|
|
|
- sug_text = "\n".join(f" · {s}" for s in suggestions)
|
|
|
- lines.append(
|
|
|
- f" · 《{n['title']}》({n['publish_date']})\n"
|
|
|
- f" 评论：{n['comment_count']} 条 | 阅读：{n['read_count']}\n"
|
|
|
- f" 💡 建议回复：\n{sug_text}"
|
|
|
- )
|
|
|
+ date = time.strftime("%Y-%m-%d %H:%M")
|
|
|
+ sections = []
|
|
|
+ for ch in changed:
|
|
|
+ note = ch["note"]
|
|
|
+ new_c = ch["new_comments"]
|
|
|
+ lines = [f"《{note['title']}》 评论 +{len(new_c)} (共 {note['comments_count']} 条)"]
|
|
|
+ if note.get("publish_time"):
|
|
|
+ lines[0] += f" | 发布 {note['publish_time']}"
|
|
|
+ for c in new_c:
|
|
|
+ kind = "回复" if c.get("is_sub") else "评论"
|
|
|
+ lines.append(
|
|
|
+ f"· [{kind}] @{c.get('nickname')} ({_fmt_ms(c.get('create_ms'))})\n"
|
|
|
+ f" {c.get('content') or '(图片)'}\n"
|
|
|
+ f" 💡 建议回复: {c.get('suggestion')}"
|
|
|
+ )
|
|
|
+ sections.append("\n".join(lines))
|
|
|
|
|
|
msg = (
|
|
|
- f"📬 小红书评论提醒（{date}）\n\n"
|
|
|
- + "\n\n".join(lines)
|
|
|
- + "\n\n💬 提示：以上为有评论的笔记，请及时回复以增加互动和粉丝粘性"
|
|
|
+ f"📬 小红书新增评论提醒（{date}）\n\n"
|
|
|
+ + "\n\n".join(sections)
|
|
|
+ + "\n\n💬 提示：以上为新增评论，请及时回复以增加互动和粉丝粘性"
|
|
|
)
|
|
|
print(msg)
|
|
|
|
|
|
- # 推送
|
|
|
- try:
|
|
|
- token = _get_token(cfg["feishu_app_id"], cfg["feishu_app_secret"])
|
|
|
- url = f"{FEISHU_MSG_URL}?receive_id_type=chat_id"
|
|
|
- r = requests.post(
|
|
|
- url,
|
|
|
- headers={"Authorization": f"Bearer {token}"},
|
|
|
- json={
|
|
|
- "receive_id": cfg["feishu_chat_id"],
|
|
|
- "msg_type": "text",
|
|
|
- "content": json.dumps({"text": msg}, ensure_ascii=False),
|
|
|
- },
|
|
|
- timeout=20,
|
|
|
- )
|
|
|
- d = r.json()
|
|
|
- if d.get("code") == 0:
|
|
|
- print(f"[OK] 评论通知已推送")
|
|
|
- else:
|
|
|
- print(f"[WARN] 推送失败: {d.get('msg')}")
|
|
|
- except Exception as e:
|
|
|
- print(f"推送失败: {e}")
|
|
|
+ if dry_run:
|
|
|
+ print("[DRY-RUN] 未实际推送")
|
|
|
+ return
|
|
|
+
|
|
|
+ push_feishu(cfg, msg)
|
|
|
+ print("[OK] 评论通知已推送")
|
|
|
|
|
|
except Exception as e:
|
|
|
print(f"检查评论失败: {e}")
|
|
|
@@ -234,4 +405,7 @@ def check_comments():
|
|
|
|
|
|
|
|
|
if __name__ == "__main__":
|
|
|
- check_comments()
|
|
|
+ parser = argparse.ArgumentParser(description="小红书新增评论检测")
|
|
|
+ parser.add_argument("--dry-run", action="store_true", help="只打印不推送")
|
|
|
+ args = parser.parse_args()
|
|
|
+ check_comments(dry_run=args.dry_run)
|
Sisyphus Agent