内容: 小红书创作者中心每日日报采集与飞书推送模块

.gitignore

@@ -71,3 +71,6 @@ venv/
 
 # Skills temp directory
 .superpowers/sdd/
+
+# 小红书日报推送本地配置（含飞书密钥，禁止提交）
+运营文案/xhs_daily_config.json

运营文案/xhs_daily_config.json.example

@@ -0,0 +1,6 @@
+{
+  "feishu_app_id": "cli_xxxxxxxxxxxx",
+  "feishu_app_secret": "your_app_secret_here",
+  "feishu_chat_id": "oc_xxxxxxxxxxxx",
+  "receive_id_type": "chat_id"
+}

运营文案/xhs_feishu.py

@@ -0,0 +1,189 @@
+"""
+飞书企业机器人推送工具
+======================
+通过飞书开放平台（open.feishu.cn API）发送富文本卡片消息 + 图片附件。
+
+用法：
+    from xhs_feishu import send_daily_report
+    send_daily_report(config_path, date, overview, notes, screenshot_path)
+
+依赖：
+    requests（第三方库）
+    配置 JSON 文件（含 feishu_app_id / feishu_app_secret / feishu_chat_id）
+"""
+import io
+import json
+import os
+import sys
+import time
+
+# 遵循项目规范：UTF-8 输出，避免 Windows GBK 乱码
+sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8', errors='replace')
+
+import requests
+
+FEISHU_TOKEN_URL = "https://open.feishu.cn/open-apis/auth/v3/tenant_access_token/internal"
+FEISHU_MSG_URL = "https://open.feishu.cn/open-apis/im/v1/messages"
+FEISHU_IMG_URL = "https://open.feishu.cn/open-apis/im/v1/images"
+
+# 进程内 token 缓存：{key: (token, expire_ts)}
+_TOKEN_CACHE = {}
+
+
+def _get_token(app_id: str, app_secret: str) -> str:
+    """获取 tenant_access_token（带进程内缓存，过期的自动刷新）。
+
+    飞书 token 默认有效期 7200s，这里提前 60s 视为过期。
+    """
+    cache_key = f"{app_id}:{app_secret}"
+    now = time.time()
+    if cache_key in _TOKEN_CACHE:
+        token, expire_ts = _TOKEN_CACHE[cache_key]
+        if expire_ts > now:
+            return token
+    resp = requests.post(
+        FEISHU_TOKEN_URL,
+        json={"app_id": app_id, "app_secret": app_secret},
+        timeout=15,
+    )
+    resp.raise_for_status()
+    data = resp.json()
+    if data.get("code") != 0:
+        raise RuntimeError(f"获取飞书 token 失败: {data.get('msg')}")
+    token = data["tenant_access_token"]
+    expire = data.get("expire", 7200)
+    _TOKEN_CACHE[cache_key] = (token, now + expire - 60)
+    return token
+
+
+def _upload_image(token: str, image_path: str) -> str:
+    """上传图片到飞书，返回 image_key。"""
+    with open(image_path, "rb") as f:
+        resp = requests.post(
+            FEISHU_IMG_URL,
+            headers={"Authorization": f"Bearer {token}"},
+            files={"image": (os.path.basename(image_path), f, "image/jpeg")},
+            data={"image_type": "message"},
+            timeout=30,
+        )
+    resp.raise_for_status()
+    data = resp.json()
+    if data.get("code") != 0:
+        raise RuntimeError(f"上传图片失败: {data.get('msg')}")
+    return data["data"]["image_key"]
+
+
+def _build_card_json(date: str, overview: dict, notes: list) -> str:
+    """构建飞书富文本卡片 JSON 字符串。
+
+    overview 字段: fans_count, fans_growth_30d, total_read, total_interact, note_count
+    notes 元素字段: title, publish_date, read_count, like_count, collect_count,
+                    comment_count, share_count, unreplied_comment_count
+    """
+    ov = overview or {}
+    overview_lines = [
+        f"粉丝数：{ov.get('fans_count', 0):,}"
+        f"（近30天 {'+' if ov.get('fans_growth_30d', 0) >= 0 else ''}"
+        f"{ov.get('fans_growth_30d', 0):,}）",
+        f"总阅读量：{ov.get('total_read', 0):,}",
+        f"总互动量：{ov.get('total_interact', 0):,}",
+        f"笔记总数：{ov.get('note_count', 0)}",
+    ]
+    note_lines = []
+    for i, n in enumerate(notes, 1):
+        note_lines.append(
+            f"{i}. 《{n.get('title', '')}》\n"
+            f"   📅{n.get('publish_date', '')}  "
+            f"👀{n.get('read_count', 0):,}  "
+            f"❤️{n.get('like_count', 0):,}  "
+            f"⭐{n.get('collect_count', 0):,}  "
+            f"💬{n.get('comment_count', 0):,}  "
+            f"🔄{n.get('share_count', 0):,}\n"
+            f"   📮未回复评论：{n.get('unreplied_comment_count', 0)}"
+        )
+
+    content = "\n\n".join([
+        "**📈 账号概览**\n" + "\n".join(overview_lines),
+        f"**📝 笔记列表（{len(notes)}篇）**\n" + "\n\n".join(note_lines),
+        "🖼️ 附：创作者中心页面截图",
+    ])
+
+    card = {
+        "config": {"wide_screen_mode": True},
+        "header": {
+            "template": "blue",
+            "title": {"tag": "plain_text", "content": f"小红书创作者中心日报 | {date}"},
+        },
+        "elements": [
+            {"tag": "div", "text": {"tag": "lark_md", "content": content}},
+        ],
+    }
+    return json.dumps(card, ensure_ascii=False)
+
+
+def _send_message(token: str, cfg: dict, msg_type: str, content: str) -> dict:
+    """发送一条飞书消息，返回响应数据。"""
+    body = {
+        "receive_id_type": cfg.get("receive_id_type", "chat_id"),
+        "receive_id": cfg["feishu_chat_id"],
+        "msg_type": msg_type,
+        "content": content,
+    }
+    resp = requests.post(
+        FEISHU_MSG_URL,
+        headers={"Authorization": f"Bearer {token}"},
+        json=body,
+        timeout=30,
+    )
+    resp.raise_for_status()
+    data = resp.json()
+    if data.get("code") != 0:
+        raise RuntimeError(f"发送飞书{msg_type}消息失败: {data.get('msg')}")
+    return data
+
+
+def send_daily_report(
+    config_path: str,
+    date: str,
+    overview: dict,
+    notes: list,
+    screenshot_path: str = None,
+):
+    """发送小红书创作者中心日报到飞书群。
+
+    Args:
+        config_path: xhs_daily_config.json 路径
+        date: 报告日期 YYYY-MM-DD
+        overview: 账号总览 dict
+        notes: 笔记列表 list[dict]
+        screenshot_path: 截图文件路径（可选，存在则作为图片消息附发）
+    """
+    with open(config_path, "r", encoding="utf-8") as f:
+        cfg = json.load(f)
+    required = ["feishu_app_id", "feishu_app_secret", "feishu_chat_id"]
+    missing = [k for k in required if not cfg.get(k)]
+    if missing:
+        raise ValueError(f"配置缺少必填字段: {', '.join(missing)}")
+
+    token = _get_token(cfg["feishu_app_id"], cfg["feishu_app_secret"])
+    card_json = _build_card_json(date, overview, notes)
+    data = _send_message(token, cfg, "interactive", card_json)
+
+    if screenshot_path and os.path.exists(screenshot_path):
+        try:
+            image_key = _upload_image(token, screenshot_path)
+            _send_message(
+                token, cfg, "image",
+                json.dumps({"image_key": image_key}),
+            )
+        except Exception as e:
+            print(f"[WARN] 发送截图失败，但卡片已推送: {e}", flush=True)
+
+    print(f"[OK] 飞书日报已推送，message_id={data['data']['message_id']}", flush=True)
+    return data["data"]["message_id"]
+
+
+if __name__ == "__main__":
+    # 独立测试：本模块可作自检用
+    print("飞书推送工具模块加载成功")
+    print(f"token URL: {FEISHU_TOKEN_URL}")

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

@@ -0,0 +1,336 @@
+"""
+小红书创作者中心日报采集脚本
+=============================
+每天早上 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
+
+# ---- 路径配置 ----
+OPERATION_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))   # 运营文案/
+SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))                        # 小红书发布/_过程脚本/
+XHS_DIR = os.path.dirname(SCRIPT_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)
+
+CREATOR_HOME = "https://creator.xiaohongshu.com/new/home"
+NOTES_MANAGE_URL = "https://creator.xiaohongshu.com/publish/publish_manage"
+COMMENTS_URL = "https://creator.xiaohongshu.com/message/comment"
+
+
+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:
+    """从创作者中心首页获取账号总览数据。"""
+    overview = {}
+    text = (publisher._evaluate("document.body.innerText") or "")
+
+    # 粉丝数 / 总阅读 / 总互动 / 笔记总数（宽泛匹配，兼容页面 word 差异）
+    patterns = [
+        ("fans_count", r"粉丝数\s*[：:\n]\s*([\d.,万]+)"),
+        ("total_read", r"总阅读[量]?\s*[：:\n]\s*([\d.,万]+)"),
+        ("total_interact", r"总互动[量]?\s*[：:\n]\s*([\d.,万]+)"),
+        ("note_count", r"笔记总数\s*[：:\n]\s*([\d.,]+)"),
+        ("fans_growth_30d", r"(?:近30天|近3[0-9]天)[^\d\n]*?([+-]?[\d.,万]+)"),
+    ]
+    for key, pat in patterns:
+        m = re.search(pat, text)
+        if m:
+            overview[key] = _to_int(m.group(1))
+    return overview
+
+
+def collect_notes_from_dom(publisher, limit=10):
+    """从笔记管理页 DOM 提取笔记列表（完整标题 + 数据）。"""
+    expr = f"""() => {{
+        const notes = [];
+        // 常见容器选择器：数据表格行 / 笔记卡片
+        const rows = document.querySelectorAll(
+            'table tbody tr, .note-item, [class*="note-row"], [class*="note-item"], [class*="data-card"], li'
+        );
+        let count = 0;
+        for (const row of rows) {{
+            if (count >= {int(limit)}) break;
+            const titleEl = row.querySelector(
+                '[class*="title"], [class*="Title"], [class*="name"], [class*="content"], a[href*="/explore/"]'
+            );
+            const titleFull = (titleEl ? titleEl.textContent : row.childNodes[0] ? row.childNodes[0].textContent : row.textContent)
+                .trim();
+            if (!titleFull || titleFull.length < 2 || /(阅读|点赞|评论|收藏|发布)/.test(titleFull)) continue;
+            // 收集该行所有数字
+            const nums = Array.from(row.querySelectorAll('td, [class*="stat"], [class*="value"], [class*="count"]'))
+                .map(el => el.textContent.replace(/[, ]/g, ''))
+                .filter(t => /^\\d+$/.test(t))
+                .map(Number);
+            // 常见顺序: 阅读 / 点赞? / 收藏 / 评论，或不含单位。按存在数量尽力填充。
+            const note = {{
+                title: titleFull,
+                publish_date: '',
+                read_count: nums[0] || 0,
+                like_count: nums[1] || 0,
+                collect_count: nums[2] || 0,
+                comment_count: nums[3] || 0,
+                share_count: nums[4] || 0,
+                unreplied_comment_count: 0,
+            }};
+            // 找发布日期
+            const m = row.textContent.match(/(\\d{{4}}[-/]\\d{{1,2}}[-/]\\d{{1,2}})|(\\d{{1,2}}[-/]\\d{{1,2}})/);
+            if (m) note.publish_date = m[0];
+            notes.push(note);
+            count++;
+        }}
+        return JSON.stringify(notes);
+    }}"""
+    try:
+        raw = publisher._evaluate(expr)
+        if isinstance(raw, str):
+            parsed = json.loads(raw) if raw.strip().startswith("[") else []
+            for n in parsed:
+                n["title"] = _clean_title(n.get("title", ""))
+            return parsed
+    except Exception as e:
+        _log(f"[warn] DOM 提取笔记失败: {e}")
+    return []
+
+
+def collect_notes_fallback(publisher) -> list:
+    """兜底：从页面纯文本按行解析笔记数据。"""
+    text = publisher._evaluate("document.body.innerText") or ""
+    notes = []
+    lines = [ln.strip() for ln in text.split("\n") if ln.strip()]
+    i = 0
+    while i < len(lines) and len(notes) < 10:
+        line = lines[i]
+        # 数据行特征：含 2+ 个数字
+        nums = re.findall(r"[\d.,万]+", line)
+        if len(nums) >= 2 and not re.match(r'^\d+$', line):
+            # 标题 = 上一行若非数字行
+            title = lines[i - 1] if i > 0 and not re.fullmatch(r"[\d,\.]+", lines[i - 1]) else line
+            values = [_to_int(nm) for nm in nums]
+            note = {
+                "title": _clean_title(title),
+                "publish_date": "",
+                "read_count": values[0],
+                "like_count": values[1] if len(values) > 1 else 0,
+                "collect_count": values[2] if len(values) > 2 else 0,
+                "comment_count": values[3] if len(values) > 3 else 0,
+                "share_count": values[4] if len(values) > 4 else 0,
+                "unreplied_comment_count": 0,
+            }
+            notes.append(note)
+        i += 1
+    return notes
+
+
+def collect_notes(publisher, limit=10):
+    """采集笔记列表：先 DOM，失败则兜底文本。"""
+    notes = collect_notes_from_dom(publisher, limit)
+    if not notes:
+        notes = collect_notes_fallback(publisher)
+    return notes[:limit]
+
+
+def collect_unreplied_comments(publisher) -> int:
+    """采集未回复评论数（跳转评论消息页）。"""
+    publisher._navigate(COMMENTS_URL)
+    time.sleep(2)
+    try:
+        text = publisher._evaluate("document.body.innerText") or ""
+    except Exception:
+        return 0
+    m = re.search(r"未回复[^\\d]*(\\d+)", text)
+    if m:
+        return int(m.group(1))
+    # 备选：查未读提醒徽标 / 未回复 tab 计数
+    m2 = re.search(r"未回复评论[^\\d]*(\\d+)", text)
+    return int(m2.group(1)) if m2 else 0
+
+
+def screenshot_current(publisher, output_path):
+    """用 CDP 截取当前页面（JPEG）。"""
+    os.makedirs(os.path.dirname(output_path), exist_ok=True)
+    result = publisher._send("Page.captureScreenshot", {"format": "jpeg", "quality": 60})
+    if result.get("result", {}).get("data"):
+        img_data = base64.b64decode(result["result"]["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}")
+
+
+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')}")
+
+        # 2. 笔记列表
+        _log(f"采集笔记列表（最多 {args.limit} 条）...")
+        publisher._navigate(NOTES_MANAGE_URL)
+        time.sleep(3)
+        notes = collect_notes(publisher, args.limit)
+        _log(f"  采集到 {len(notes)} 篇笔记")
+
+        # 3. 未回复评论
+        _log("采集未回复评论数...")
+        unreplied = collect_unreplied_comments(publisher)
+        for n in notes:
+            n["unreplied_comment_count"] = unreplied
+        _log(f"  未回复评论: {unreplied}")
+
+        # 4. 截图（回到创作者首页，截账号概览看板）
+        _log("截图...")
+        shot_path = os.path.join(SCREENSHOT_DIR, f"xhs_daily_{date}.jpg")
+        screenshot_current(publisher, shot_path)
+
+        # 5. 本地记录
+        append_daily_record(date, overview, notes)
+
+        # 6. 飞书推送
+        _log("推送到飞书...")
+        sys.path.insert(0, OPERATION_DIR)
+        from xhs_feishu import send_daily_report
+        send_daily_report(CONFIG_PATH, date, overview, notes, screenshot_path=shot_path)
+        _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()