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

"""
小红书创作者中心日报采集脚本
=============================
每天早上 9:00（Windows 计划任务）运行：通过 CDP 连接已登录的 Chrome，
采集小红书创作者中心的账号总览 + 笔记列表数据及页面截图，
推送到飞书群。

技术栈：CDP（复用已在运行的 Chrome + 小红书登录态），requests，飞书 API。
复用自 C:\\code\\XiaohongshuSkills\\scripts\\cdp_publish.py 的 XiaohongshuPublisher。

用法：
    python xhs_daily_report.py        # 正常执行（连接 9222 端口 CDP）
    python xhs_daily_report.py --login   # 手动登录引导（保存 CDP 登录态）

先决条件：
    1. 本机 Chrome 已以 --remote-debugging-port=9222 启动且已登录小红书
    2. 运营文案/xhs_daily_config.json 已填入飞书 app_id/secret/群ID
"""
import argparse
import base64
import io
import json
import os
import re
import sys
import time

import requests

# 遵循项目规范：UTF-8 输出，避免 Windows GBK 乱码
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8', errors='replace')
sys.stderr = sys.stdout

# ---- 路径配置 ----
SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))                        # 小红书发布/_过程脚本/
XHS_DIR = os.path.dirname(SCRIPT_DIR)                                           # 小红书发布/
OPERATION_DIR = os.path.dirname(XHS_DIR)                                        # 运营文案/

CONFIG_PATH = os.path.join(OPERATION_DIR, "xhs_daily_config.json")
SCREENSHOT_DIR = os.path.join(XHS_DIR, "_过程文件", "截图")

# 2026-08-28 起新增脚本采用轻量记录，不依赖旧版 master_tracking.csv
RECORD_DIR = os.path.join(XHS_DIR, "_数据追踪")

# XiaohongshuSkills 库路径
XHS_SKILLS_PATH = r"C:\code\XiaohongshuSkills\scripts"
sys.path.insert(0, XHS_SKILLS_PATH)
if OPERATION_DIR not in sys.path:
    sys.path.insert(0, OPERATION_DIR)

CREATOR_HOME = "https://creator.xiaohongshu.com/new/home"
NOTES_MANAGE_URL = "https://creator.xiaohongshu.com/new/note-manager"


def _log(msg):
    print(f"[{time.strftime('%H:%M:%S')}] {msg}", flush=True)


def _load_config() -> dict:
    """读取飞书配置。"""
    if not os.path.exists(CONFIG_PATH):
        raise FileNotFoundError(
            f"配置文件不存在: {CONFIG_PATH}\n"
            f"请从 xhs_daily_config.json.example 复制并填入真实值"
        )
    with open(CONFIG_PATH, "r", encoding="utf-8") as f:
        cfg = json.load(f)
    missing = [k for k in ("feishu_app_id", "feishu_app_secret", "feishu_chat_id") if not cfg.get(k)]
    if missing:
        raise ValueError(f"配置缺少必填字段: {', '.join(missing)}")
    return cfg


def _to_int(text):
    """'1.2万' -> 12000, '1,234' -> 1234, '--'/'' -> 0。"""
    if not text:
        return 0
    t = str(text).strip().replace(",", "").replace(" ", "")
    if not t or t == "--":
        return 0
    m = re.match(r"^([\d.]+)(万|亿)?$", t)
    if not m:
        return 0
    val = float(m.group(1))
    unit = m.group(2)
    if unit == "万":
        val *= 10000
    elif unit == "亿":
        val *= 100000000
    return int(val)


def _clean_title(title):
    """清理标题：去掉换行与多余空白。"""
    if not title:
        return ""
    return " ".join(title.split())


def collect_overview(publisher) -> dict:
    """
    从创作者中心首页 /new/home 获取账号总览数据。

    页面 innerText 结构（实测）：
      - 账号级字段: "数字\\n标签"（如 "20\\n粉丝数", "551\\n获赞与收藏"）
      - 周期数据: "标签\\n数字\\n环比xx%"（如 "观看数\\n209\\n环比-86%"）
      - 默认显示近7日，点击「近30日」tab 抓第二组

    返回 dict:
      fans_count / follow_count / interact_count（账号级）
      period_7d / period_30d: 各含 exposure/view/like/comment/collect/share/net_fans
    """
    overview = {}

    def _parse_period(text):
        """解析周期数据（标签\\n数字\\n环比%）。返回 dict[label] = (值, 原始字符串)"""
        p = {}
        # 匹配 "标签\n数字\n环比xx%" 或 "标签\n数字\n环比-"
        pattern = re.compile(
            r"^(曝光数|观看数|封面点击率|视频完播率|点赞数|评论数|收藏数|分享数|净涨粉|新增关注|取消关注|主页访客)"
            r"\n([\d.,万%]+)\n环比([+-]?\d*%?)", re.M)
        for m in pattern.finditer(text):
            p[m.group(1)] = m.group(2)
        return p

    # --- 账号级字段：数字在标签前 ---
    text = (publisher._evaluate("document.body.innerText") or "")
    for key, label in (("fans_count", "粉丝数"), ("follow_count", "关注数"),
                       ("interact_count", "获赞与收藏")):
        m = re.search(r"([\d.,万]+)\n" + re.escape(label), text)
        if m:
            overview[key] = _to_int(m.group(1))
        else:
            overview[key] = 0

    # --- 近7日（默认 tab） ---
    period_7d = _parse_period(text)

    # --- 近30日：点击 tab 抓第二组 ---
    period_30d = {}
    try:
        js = (
            "(() => {"
            "  const els = [...document.querySelectorAll('*')] ;"
            "  const t = els.find(e => e.children.length === 0 && e.textContent.trim() === '近30日');"
            "  if (t) { t.click(); return 'CLICKED'; }"
            "  return 'NOT_FOUND';"
            "})()"
        )
        r = publisher._evaluate(js)
        if r == "CLICKED":
            time.sleep(6)  # 等待数据刷新
            text2 = publisher._evaluate("document.body.innerText") or ""
            period_30d = _parse_period(text2)
            _log(f"  近30日: 观看={period_30d.get('观看数')} 净涨粉={period_30d.get('净涨粉')}")
            if not period_30d:
                _log("  [WARN] 近30日解析为空，重试一次")
                time.sleep(4)
                text2 = publisher._evaluate("document.body.innerText") or ""
                period_30d = _parse_period(text2)
                _log(f"  近30日(重试): 观看={period_30d.get('观看数')}")
        else:
            _log(f"  [WARN] 切换近30日 tab 失败: {r}")
    except Exception as e:
        _log(f"  [WARN] 抓近30日失败: {e}")

    overview["period_7d"] = period_7d
    overview["period_30d"] = period_30d
    return overview


def _count_unreplied_comments(publisher, note_id: str, xsec_token: str) -> int:
    """打开笔记详情页读取评论，统计未回复的父评论数。

    判定规则：父评论下所有楼中楼（subComments）中，
    若没有 userId 等于作者 userId 的子评论 → 计为未回复。
    """
    if not xsec_token or not note_id:
        return 0
    try:
        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,
        )
    except Exception as e:
        _log(f"  [WARN] 笔记 {note_id} 详情页读取失败: {e}")
        return 0
    detail = res.get("detail") or {}
    note_obj = detail.get("note")
    # note.user 可能是 dict（图文笔记）或 str（视频笔记的 userId 字段）
    user_obj = (note_obj or {}).get("user") if isinstance(note_obj, dict) else None
    if isinstance(user_obj, str):
        author_id = user_obj
    elif isinstance(user_obj, dict):
        author_id = str(user_obj.get("userId") or "")
    else:
        author_id = ""
    if not author_id:
        return 0
    comments = (detail.get("comments") or {}).get("list") or []
    unreplied = 0
    for c in comments:
        subs = c.get("subComments") or []
        replied = any(
            str((s.get("userInfo") or {}).get("userId") or "") == author_id
            for s in subs
        )
        if not replied:
            unreplied += 1
    return unreplied


def collect_notes(publisher, limit=10):
    """
    从笔记管理页采集笔记列表，并统计每篇未回复评论数。

    优先使用 posted API（含 xsec_token），失败时退回 innerText 解析。
    对每篇有评论的笔记打开详情页，统计作者尚未回复的父评论数。
    """
    # ── 优先：posted API（精确数据 + xsec_token） ──────────────────────
    raw_notes = []
    try:
        from xhs_unreplied_comments import capture_posted_notes
        raw_notes = capture_posted_notes(publisher, max_wait=20.0)
        _log(f"  [posted] 捕获 {len(raw_notes)} 篇笔记")
    except Exception as e:
        _log(f"  [WARN] posted API 捕获失败，退回 innerText 解析: {e}")
        raw_notes = []

    notes = []
    if raw_notes:
        for n in raw_notes[:limit]:
            entry = {
                "title": n.get("title", ""),
                "publish_date": (n.get("publish_time") or "")[:10],
                "read_count": n.get("view_count", 0),
                "comment_count": n.get("comments_count", 0),
                "like_count": n.get("likes", 0),
                "collect_count": n.get("collected_count", 0),
                "share_count": n.get("shared_count", 0),
                "unreplied_comment_count": 0,
                "_xsec_token": n.get("xsec_token", ""),
                "_note_id": n.get("id", ""),
            }
            notes.append(entry)
    else:
        # fallback：innerText 解析（原逻辑，无法获取 xsec_token，unreplied 固定为 0）
        text = publisher._evaluate("document.body.innerText") or ""
        lines = [ln.strip() for ln in text.split("\n")]
        dates = re.compile(r"^\d{4}-\d{2}-\d{2} \d{2}:\d{2}$")
        i = 0
        while i < len(lines) and len(notes) < limit:
            ln = lines[i]
            if dates.match(ln) and i + 5 < len(lines):
                title = _clean_title(lines[i - 1]) 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],
                        "unreplied_comment_count": 0,
                        "_xsec_token": "",
                        "_note_id": "",
                    })
                i = j if j > i + 1 else i + 1
            else:
                i += 1

    # ── 逐篇统计未回复评论 ─────────────────────────────────────────────
    for entry in notes:
        if entry["comment_count"] <= 0:
            continue
        unreplied = _count_unreplied_comments(
            publisher, entry["_note_id"], entry["_xsec_token"]
        )
        entry["unreplied_comment_count"] = unreplied
        _log(
            f"  [笔记] {entry['title'][:30]}  "
            f"评论={entry['comment_count']} "
            f"未回复={unreplied}"
        )

    # 清理内部字段（保持对外接口干净）
    for entry in notes:
        entry.pop("_xsec_token", None)
        entry.pop("_note_id", None)

    return notes[:limit]


def screenshot_current(publisher, output_path, full_page: bool = True):
    """用 CDP 截取当前页面（JPEG），默认整页截图。

    full_page=True 时用 captureBeyondViewport 截取整个可滚动页面。
    """
    os.makedirs(os.path.dirname(output_path), exist_ok=True)
    params = {"format": "jpeg", "quality": 60}
    if full_page:
        params["captureBeyondViewport"] = True
    result = publisher._send("Page.captureScreenshot", params)

    data = result.get("data") or (result.get("result") or {}).get("data")
    if data:
        img_data = base64.b64decode(data)
        with open(output_path, "wb") as f:
            f.write(img_data)
        _log(f"截图已保存: {output_path} ({len(img_data)} bytes)")
    else:
        raise RuntimeError(f"截图失败: {result}")


# 账号概览页 URL（创作者中心数据看板）
ACCOUNT_OVERVIEW_URL = "https://creator.xiaohongshu.com/statistics/account/v2"


def capture_three_screenshots(publisher, shot_dir, prefix="xhs_bot"):
    """依次截取三张整页截图：首页 / 笔记管理 / 账号概览。

    Returns:
        list[str]: 三张截图的文件路径（按顺序：首页, 笔记管理, 账号概览）
    """
    os.makedirs(shot_dir, exist_ok=True)
    ts = int(time.time())
    shots = []

    targets = [
        (CREATOR_HOME, f"{prefix}_{ts}_home.jpg"),
        (NOTES_MANAGE_URL, f"{prefix}_{ts}_notes.jpg"),
        (ACCOUNT_OVERVIEW_URL, f"{prefix}_{ts}_overview.jpg"),
    ]
    for url, fname in targets:
        try:
            publisher._navigate(url)
            time.sleep(5)
            path = os.path.join(shot_dir, fname)
            screenshot_current(publisher, path, full_page=True)
            shots.append(path)
        except Exception as e:
            _log(f"[WARN] 截图失败 {url}: {e}")

    return shots


def append_daily_record(date, overview, notes):
    """追加当日快照到本地 JSON 记录（轻量，不依赖旧 CSV 结构）。"""
    os.makedirs(RECORD_DIR, exist_ok=True)
    rec_path = os.path.join(RECORD_DIR, "daily_reports.jsonl")
    rec = {
        "date": date,
        "overview": overview,
        "notes": notes,
        "recorded_at": time.strftime("%Y-%m-%d %H:%M:%S"),
    }
    with open(rec_path, "a", encoding="utf-8") as f:
        f.write(json.dumps(rec, ensure_ascii=False) + "\n")


def main():
    parser = argparse.ArgumentParser(description="小红书创作者中心日报采集与飞书推送")
    parser.add_argument("--login", action="store_true", help="手动登录态（需人工扫码）")
    parser.add_argument("--limit", type=int, default=10, help="推送笔记条数上限")
    args = parser.parse_args()

    cfg = _load_config()

    from cdp_publish import XiaohongshuPublisher
    publisher = XiaohongshuPublisher()

    if args.login:
        try:
            publisher.connect(reuse_existing_tab=True)
        except Exception as e:
            _log(f"连接 CDP 失败: {e}")
            _log("请先用下列命令启动带调试端口的 Chrome 并登录小红书：")
            _log('chrome.exe --remote-debugging-port=9222 --user-data-dir="%LOCALAPPDATA%\\Google\\Chrome\\User Data"')
            sys.exit(1)
        publisher.check_login()
        publisher.disconnect()
        _log("登录态已确认/缓存。之后直接运行 xhs_daily_report.py 即可。")
        return

    # ---- 常规日报流程 ----
    date = time.strftime("%Y-%m-%d")
    overview = {"fans_count": 0, "fans_growth_30d": 0, "total_read": 0,
                "total_interact": 0, "note_count": 0}
    try:
        _log("连接 Chrome CDP (9222)...")
        publisher.connect(reuse_existing_tab=True)
        if not publisher.check_login():
            raise RuntimeError("未登录小红书，请先运行 --login 或手动打开浏览器完成扫码")

        # 1. 账号总览
        _log("采集账号总览...")
        publisher._navigate(CREATOR_HOME)
        time.sleep(4)
        overview = collect_overview(publisher)
        _log(f"  粉丝数: {overview.get('fans_count')}, 总阅读: {overview.get('total_read')}, "
             f"总互动: {overview.get('total_interact')}, 笔记数: {overview.get('note_count')}")

        _log(f"采集笔记列表（最多 {args.limit} 条）...")
        publisher._navigate(NOTES_MANAGE_URL)
        time.sleep(4)
        notes = collect_notes(publisher, args.limit)
        _log(f"  采集到 {len(notes)} 篇笔记")


        _log("截图（首页/笔记管理/账号概览 三张整页）...")
        shot_paths = capture_three_screenshots(publisher, SCREENSHOT_DIR, prefix=f"xhs_daily_{date}")
        _log(f"  截图 {len(shot_paths)} 张")


        append_daily_record(date, overview, notes)


        _log("推送到飞书...")
        sys.path.insert(0, OPERATION_DIR)
        from xhs_feishu import send_daily_report
        send_daily_report(CONFIG_PATH, date, overview, notes, screenshot_paths=shot_paths)
        _log("✅ 日报流程完成")

    except Exception as e:
        _log(f"❌ 采集失败: {e}")
        import traceback
        traceback.print_exc(file=sys.stderr)
        sys.exit(1)
    finally:
        try:
            publisher.disconnect()
        except Exception:
            pass


if __name__ == "__main__":
    main()