feat(xiaohongshu): 评论自动回复脚本 comment_reply.py——CDP 驱动，click 修复与残缺选择器修复

运营文案/小红书发布/_过程脚本/comment_reply.py

@@ -0,0 +1,301 @@
+"""
+小红书评论自动回复工具（最终版）
+=============================
+功能：
+  1. 连接已打开的笔记详情页（CDP WebSocket）
+  2. 提取所有评论，识别未被作者回复的评论
+  3. 用"引发讨论"风格的回复逐条回复
+  4. 支持 CLI 和库式两种调用方式
+
+核心优化：
+  - 回复内容以提问结尾，引导用户继续讨论（提升互动权重）
+  - 使用 CDP Input.insertText 模拟真实键盘输入，确保发送按钮可用
+  - 逐条处理，避免重复发送
+
+用法：
+  python comment_reply.py --ws ws://127.0.0.1:9222/devtools/page/<PAGE_ID>
+  python comment_reply.py --ws ws://127.0.0.1:9222/devtools/page/<PAGE_ID> --extract-only
+  python comment_reply.py --ws ws://127.0.0.1:9222/devtools/page/<PAGE_ID> --text "自定义回复"
+"""
+import json, asyncio, websockets, sys, argparse, re
+
+# ============================================================
+# 讨论型回复模板（按场景分类，自动匹配）
+# ============================================================
+DISCUSSION_REPLIES = {
+    # 关于"晚一年上学"
+    "晚一年上学": "晚一年确实有很多好处，能分享一下你们家是怎么考虑这件事的吗？",
+    # 关于"不生二胎"
+    "不生二胎": "是啊，精力集中养一个也挺好的。你们家孩子多大了？",
+    # 关于"不把孩子带来世上"
+    "不把他帶來世上遭罪": "这个想法很深刻，能多聊聊你是怎么想的吗？",
+    # 关于"让他是独生子"
+    "让他是独生子": "独生子女也有独生子女的好处，你平时会担心他孤单吗？",
+    # 默认提问式回复
+    "default": "说得对，能再展开说说你的想法吗？",
+}
+
+class CommentReplyBot:
+    """小红书评论自动回复机器人"""
+
+    def __init__(self, ws_url):
+        self.ws_url = ws_url
+        self.ws = None
+        self.msg_id = 0
+
+    async def connect(self):
+        self.ws = await websockets.connect(self.ws_url, max_size=10*1024*1024)
+
+    async def _send(self, method, params=None):
+        self.msg_id += 1
+        m = {'id': self.msg_id, 'method': method}
+        if params: m['params'] = params
+        await self.ws.send(json.dumps(m))
+        while True:
+            resp = json.loads(await self.ws.recv())
+            if resp.get('id') == self.msg_id:
+                return resp
+
+    async def _evaluate(self, expression):
+        r = await self._send('Runtime.evaluate', {
+            'expression': expression,
+            'returnByValue': True,
+            'awaitPromise': True
+        })
+        result = r.get('result', {}).get('result', {})
+        if result.get('type') == 'undefined':
+            return None
+        return result.get('value')
+
+    async def scroll_to_comments(self):
+        """滚动到评论区位置"""
+        await self._evaluate('window.scrollTo(0, 800)')
+        await asyncio.sleep(2)
+        # 尝试展开所有"展开 N 条回复"
+        await self._evaluate("""
+        (() => {
+            const all = document.querySelectorAll('span, div');
+            for (const el of all) {
+                if (el.offsetParent === null) continue;
+                const t = el.textContent.trim();
+                if (t.includes('展开') && (t.includes('条回复') || t.includes('回复'))) {
+                    el.click();
+                }
+            }
+        })()
+        """)
+        await asyncio.sleep(2)
+
+    async def extract_comments(self):
+        """提取所有主评论及其子回复，返回结构化数据"""
+        script = """
+        (() => {
+            const results = [];
+            const pcs = document.querySelectorAll('.parent-comment');
+            for (const pc of pcs) {
+                const mc = pc.querySelector(':scope > .comment-item');
+                if (!mc) continue;
+                const name = mc.querySelector('.name');
+                const content = mc.querySelector('.content');
+                const time = mc.querySelector('.time');
+                const location = mc.querySelector('.location');
+                const mainAuthor = name ? name.textContent.trim() : '';
+                const mainContent = content ? content.textContent.trim() : '';
+                
+                // 收集子回复，判断是否有作者回复
+                let hasAuthorReply = false;
+                const replies = [];
+                const subs = pc.querySelectorAll('.comment-item-sub');
+                for (const sub of subs) {
+                    const rName = sub.querySelector('.name');
+                    const rContent = sub.querySelector('.content');
+                    const rTag = sub.querySelector('.tag');
+                    const isAuthor = !!(rTag && sub.textContent.includes('作者'));
+                    if (isAuthor) hasAuthorReply = true;
+                    replies.push({
+                        author: rName ? rName.textContent.trim() : '',
+                        content: rContent ? rContent.textContent.trim() : '',
+                        isAuthorReply: isAuthor
+                    });
+                }
+                
+                results.push({
+                    author: mainAuthor,
+                    content: mainContent,
+                    time: time ? time.textContent.trim() : '',
+                    location: location ? location.textContent.trim() : '',
+                    hasAuthorReply: hasAuthorReply,
+                    replyCount: replies.length
+                });
+            }
+            return JSON.stringify(results, null, 2);
+        })()
+        """
+        val = await self._evaluate(script)
+        return json.loads(val) if val else []
+
+    @staticmethod
+    def get_reply_text(comment_content, custom_text=None):
+        """根据评论内容自动匹配回复模板"""
+        if custom_text:
+            return custom_text
+        for keyword, reply in DISCUSSION_REPLIES.items():
+            if keyword in comment_content:
+                return reply
+        return DISCUSSION_REPLIES["default"]
+
+    async def reply_to_comment(self, parent_comment_el_index, text):
+        """回复指定索引的评论
+        
+        流程：
+        1. 找到评论的父容器 .parent-comment
+        2. 点击其中的 .reply.icon-container
+        3. 等待输入框出现
+        4. 聚焦输入框，用 CDP Input.insertText 输入文本
+        5. 点击发送按钮 .btn.submit
+        """
+        # 点击回复按钮
+        click_script = f"""
+        (() => {{
+            const pcs = document.querySelectorAll('.parent-comment');
+            const pc = pcs[{parent_comment_el_index}];
+            if (!pc) return 'NOT_FOUND';
+            const replyBtn = pc.querySelector('.reply.icon-container');
+            if (!replyBtn) return 'NO_REPLY_BTN';
+            replyBtn.scrollIntoView({{behavior: 'instant', block: 'center'}});
+            setTimeout(() => {{ replyBtn.click(); }}, 300);
+            return 'CLICKED';
+        }})()
+        """
+        result = await self._evaluate(click_script)
+        if result != 'CLICKED':
+            return False, f"click reply failed: {result}"
+        await asyncio.sleep(2)
+
+        # 聚焦输入框
+        await self._evaluate("""
+        (() => {
+            const el = document.querySelector('#content-textarea');
+            if (!el) return false;
+            el.focus();
+            return true;
+        })()
+        """)
+        await asyncio.sleep(0.5)
+
+        # 用 CDP Input.insertText 模拟真实键盘输入
+        # 这是关键：contenteditable 需要真实输入事件触发 React 的 enable 逻辑
+        await self._send('Input.insertText', {'text': text})
+        await asyncio.sleep(1)
+
+        # 检查发送按钮是否已启用
+        enabled = await self._evaluate("""
+        (() => {
+            const btn = document.querySelector('.btn.submit');
+            return btn ? !btn.disabled : false;
+        })()
+        """)
+        if not enabled:
+            return False, "send button still disabled"
+
+        # 点击发送
+        sent = await self._evaluate("""
+        (() => {
+            const btn = document.querySelector('.btn.submit');
+            if (!btn || btn.disabled) return false;
+            btn.click();
+            return true;
+        })()
+        """)
+        if not sent:
+            return False, "click send failed"
+
+        await asyncio.sleep(2)
+        return True, "replied"
+
+    async def reply_all(self, custom_text=None):
+        """回复所有未被作者回复的评论"""
+        comments = await self.extract_comments()
+        if not comments:
+            return {"success": False, "error": "no comments found"}
+
+        unreplied = [c for c in comments if not c['hasAuthorReply']]
+        if not unreplied:
+            return {"success": True, "message": "all comments already replied", "total": len(comments)}
+
+        results = []
+        for i, c in enumerate(comments):
+            if c['hasAuthorReply']:
+                results.append({"author": c['author'], "content": c['content'][:30], "status": "skipped"})
+                continue
+
+            # 找到这个评论在 parent-comment 列表中的实际索引
+            reply_text = self.get_reply_text(c['content'], custom_text)
+            success, msg = await self.reply_to_comment(i, reply_text)
+            results.append({
+                "author": c['author'],
+                "content": c['content'][:30],
+                "status": "ok" if success else "failed",
+                "reply": reply_text if success else msg
+            })
+            if success:
+                await asyncio.sleep(1)  # 回复间隔，避免触发风控
+
+        return {
+            "success": True,
+            "total": len(comments),
+            "replied": len([r for r in results if r['status'] == 'ok']),
+            "results": results
+        }
+
+
+async def main():
+    parser = argparse.ArgumentParser(description="小红书评论自动回复工具")
+    parser.add_argument('--ws', default='ws://127.0.0.1:9222/devtools/page/C437125F88A3EEAD037C352FEF7F643A',
+                        help='CDP WebSocket 地址')
+    parser.add_argument('--extract-only', action='store_true',
+                        help='仅提取评论，不回复')
+    parser.add_argument('--text', default=None,
+                        help='自定义回复文本（默认按评论内容自动匹配）')
+    parser.add_argument('--output', default=None,
+                        help='输出文件路径（默认输出到控制台，GBK 编码问题推荐用文件）')
+    args = parser.parse_args()
+
+    bot = CommentReplyBot(args.ws)
+    await bot.connect()
+
+    def log(msg):
+        if args.output:
+            with open(args.output, 'a', encoding='utf-8') as f:
+                f.write(msg + '\n')
+        else:
+            print(msg, flush=True)
+
+    log(f"[comment_reply] Connected to {args.ws[:50]}...")
+    await bot.scroll_to_comments()
+
+    comments = await bot.extract_comments()
+    log(f"\n[comment_reply] Found {len(comments)} comments:")
+    for c in comments:
+        status = "已回复" if c['hasAuthorReply'] else "待回复"
+        location = f"({c['location']})" if c['location'] else ""
+        log(f"  [{status}] {c['author']}{location}: {c['content'][:50]}")
+
+    if args.extract_only:
+        log("\n[comment_reply] Extract only mode, exiting.")
+        return
+
+    result = await bot.reply_all(custom_text=args.text)
+    if result.get('success'):
+        replied = result.get('replied', 0)
+        total = result.get('total', 0)
+        log(f"\n[comment_reply] Done! Replied {replied}/{total} comments.")
+        if result.get('results'):
+            for r in result['results']:
+                if r['status'] == 'ok':
+                    log(f"  -> {r['author']}: {r['reply'][:40]}")
+    else:
+        log(f"\n[comment_reply] Error: {result.get('error')}")
+
+if __name__ == '__main__':
+    asyncio.run(main())