cfc/运营文案/xhs_unreplied_comments.py

"""小红书评论检测脚本 - 记录每篇笔记评论数，发现新增评论时读取具体评论并推送给回复建议

运行逻辑:
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
import re
import sys
import time

import requests

if not sys.stdout.isatty():
    try:
        sys.stdout.reconfigure(encoding='utf-8', errors='replace')
    except Exception:
        pass

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)
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():
    with open(CONFIG_PATH, "r", encoding="utf-8") as f:
        return json.load(f)


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": {}}


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)


def _fmt_ms(ms: int) -> str:
    try:
        return datetime.datetime.fromtimestamp(ms / 1000).strftime("%m-%d %H:%M")
    except Exception:
        return ""


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),
            "likes": int(n.get("likes") or 0),
            "collected_count": int(n.get("collected_count") or 0),
            "shared_count": int(n.get("shared_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

        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

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

    try:
        if not publisher.check_login():
            raise RuntimeError("未登录小红书")

        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

        if not changed:
            print("[SKIP] 无新增评论, 不推送")
            return

        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(sections)
            + "\n\n💬 提示：以上为新增评论，请及时回复以增加互动和粉丝粘性"
        )
        print(msg)

        if dry_run:
            print("[DRY-RUN] 未实际推送")
            return

        push_feishu(cfg, msg)
        print("[OK] 评论通知已推送")

    except Exception as e:
        print(f"检查评论失败: {e}")
        import traceback
        traceback.print_exc()
    finally:
        try:
            publisher.disconnect()
        except Exception:
            pass


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="小红书新增评论检测")
    parser.add_argument("--dry-run", action="store_true", help="只打印不推送")
    args = parser.parse_args()
    check_comments(dry_run=args.dry_run)