面向 AI 代理的工作者: 必需子技能:使用 superpowers:subagent-driven-development(推荐)或 superpowers:executing-plans 逐任务实现此计划。步骤使用复选框(
- [ ])语法来跟踪进度。
目标: 每天早上 9 点通过 CDP 连接小红书创作者中心(复用已登录 Chrome),采集账号总览 + 笔记列表数据及页面截图,推送到飞书群。
架构: Windows 任务计划程序每天 9:00 触发 xhs_daily_report.py → 复用 CDP(XiaohongshuPublisher 库,默认端口 9222)连接已打开的创作者中心页面 → 采集账号总览数据、笔记列表数据、未回复评论数 → 关键区域截图(用 CDP Page.captureScreenshot)→ 通过飞书企业机器人(app_id + app_secret)发送富文本卡片消息 + 截图附件。
技术栈: Python 3.12、requests、C:\code\XiaohongshuSkills\scripts\cdp_publish.py(XiaohongshuPublisher)、Windows 任务计划程序
项目位置: C:\code\cfc\运营文案\小红书发布\_过程脚本\xhs_daily_report.py(与现有 _过程脚本 风格一致)
关键约束(来自现有项目):
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8', errors='replace')(现有所有脚本的约定,避免 GBK 乱码)Page.captureScreenshot 截图(不是 Playwright),格式 jpeg, quality=60open.feishu.cn API(非 Webhook)xhs_daily_config.json(不提交)小红书发布/_过程文件/截图/xhs_daily_<YYYYMMDD>.jpgC:\code\cfc\运营文案\小红书发布\
├── _过程脚本\
│ └── xhs_daily_report.py # 新建:主脚本(采集+推送)
├── _过程文件\
│ └── 截图\
│ └── xhs_daily_20260828.jpg # 截图输出(运行时生成)
└── xhs_daily_config.json # 新建:飞书配置(加入 .gitignore)
文件:
C:\code\cfc\运营文案\xhs_daily_config.json.example.gitignore(项目根)追加 运营文案/xhs_daily_config.json创建:C:\code\cfc\运营文案\xhs_daily_config.json(占位,填入真实值)
[ ] 步骤 1:创建 xhs_daily_config.json.example
{
"feishu_app_id": "cli_xxxxxxxxxxxx",
"feishu_app_secret": "your_app_secret_here",
"feishu_chat_id": "oc_xxxxxxxxxxxx",
"receive_id_type": "chat_id"
}
[ ] 步骤 2:追加 .gitignore 忽略真实配置文件
在 C:\code\cfc\.gitignore 末尾追加一行:
运营文案/xhs_daily_config.json
确认方式:运行 git check-ignore -v 运营文案/xhs_daily_config.json 应输出匹配行。
[ ] 步骤 3:创建占位配置文件(用户稍后填入真实值)
{
"feishu_app_id": "",
"feishu_app_secret": "",
"feishu_chat_id": "",
"receive_id_type": "chat_id"
}
[ ] 步骤 4:验证配置读取
python -c "
import sys, io, json, os
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8', errors='replace')
cfg = json.load(open(r'C:\code\cfc\运营文案\xhs_daily_config.json', encoding='utf-8'))
assert 'feishu_app_id' in cfg, '缺少 feishu_app_id'
print('配置读取正常')
"
预期:输出 配置读取正常
[ ] 步骤 5:Commit
cd C:\code\cfc
git add 运营文案/xhs_daily_config.json.example 运营文案/xhs_daily_config.json
git commit -m "内容: 小红书日报推送配置模板"
文件:
创建:C:\code\cfc\运营文案\xhs_feishu.py(独立工具模块,可在任何脚本中 import)
[ ] 步骤 1:创建 xhs_feishu.py
"""
飞书企业机器人推送工具
======================
用法:
from xhs_feishu import send_daily_report
send_daily_report(config_path, overview, notes, screenshot_path)
"""
import json
import os
import time
import io
import sys
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_CACHE = {}
def _get_token(app_id: str, app_secret: str) -> str:
"""获取 tenant_access_token(带进程内缓存,有效期自动减去 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_lines = [
f"粉丝数:{overview.get('fans_count', 0):,}(近30天 {'+' if overview.get('fans_growth_30d', 0) >= 0 else ''}{overview.get('fans_growth_30d', 0):,})",
f"总阅读量:{overview.get('total_read', 0):,}",
f"总互动量:{overview.get('total_interact', 0):,}",
f"笔记总数:{overview.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_daily_report(config_path: str, date: str, overview: dict, notes: list, screenshot_path: str = None):
"""
发送日报到飞书群。
config_path: xhs_daily_config.json 路径
date: YYYY-MM-DD
overview: dict with keys fans_count, fans_growth_30d, total_read, total_interact, note_count
notes: list of dict, each with title, publish_date, read_count, like_count, collect_count, comment_count, share_count, unreplied_comment_count
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)
msg_body = {
"receive_id_type": cfg.get("receive_id_type", "chat_id"),
"receive_id": cfg["feishu_chat_id"],
"msg_type": "interactive",
"content": card_json,
}
resp = requests.post(
FEISHU_MSG_URL,
headers={"Authorization": f"Bearer {token}"},
json=msg_body,
timeout=30,
)
resp.raise_for_status()
data = resp.json()
if data.get("code") != 0:
raise RuntimeError(f"发送飞书消息失败: {data.get('msg')}")
image_key = None
if screenshot_path and os.path.exists(screenshot_path):
image_key = _upload_image(token, screenshot_path)
img_msg = {
"receive_id_type": cfg.get("receive_id_type", "chat_id"),
"receive_id": cfg["feishu_chat_id"],
"msg_type": "image",
"content": json.dumps({"image_key": image_key}),
}
resp2 = requests.post(
FEISHU_MSG_URL,
headers={"Authorization": f"Bearer {token}"},
json=img_msg,
timeout=30,
)
resp2.raise_for_status()
data2 = resp2.json()
if data2.get("code") != 0:
print(f"[WARN] 发送截图失败: {data2.get('msg')}", flush=True)
print(f"[OK] 飞书日报已推送", flush=True)
return data["data"]["message_id"]
[ ] 步骤 2:验证卡片构建(无网络依赖)
python -c "
import sys, io
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8', errors='replace')
sys.path.insert(0, r'C:\\code\\cfc\\运营文案')
from xhs_feishu import _build_card_json
overview = {'fans_count': 1234, 'fans_growth_30d': 56, 'total_read': 56789, 'total_interact': 3456, 'note_count': 89}
notes = [{'title': '测试笔记', 'publish_date': '2026-08-27', 'read_count': 123, 'like_count': 12, 'collect_count': 5, 'comment_count': 3, 'share_count': 1, 'unreplied_comment_count': 2}]
card = _build_card_json('2026-08-28', overview, notes)
import json
obj = json.loads(card)
assert '小红书创作者中心日报' in obj['header']['title']['content']
assert '1,234' in obj['elements'][0]['text']['content']
print('卡片构建通过')
"
预期:输出 卡片构建通过
[ ] 步骤 3:Commit
cd C:\code\cfc
git add 运营文案/xhs_feishu.py
git commit -m "内容: 飞书推送工具模块"
文件:
创建:C:\code\cfc\运营文案\小红书发布\_过程脚本\xhs_daily_report.py
[ ] 步骤 1:创建主脚本
"""
小红书创作者中心日报采集脚本
=============================
每天早上 9 点(Windows 计划任务)运行,采集账号总览 + 笔记列表数据及截图,
推送到飞书群。
用法:
python xhs_daily_report.py # 正常执行(自动连接 CDP)
python xhs_daily_report.py --headful # 有头模式(调试用)
python xhs_daily_report.py --login # 手动登录模式(引导 CDP 浏览器扫码)
"""
import argparse
import base64
import io
import json
import os
import re
import sys
import time
import requests
# 遵循项目规范:UTF-8 输出,避免 GBK 乱码
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8', errors='replace')
sys.stderr = sys.stdout
# 路径配置
PROJECT_ROOT = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))
DATA_TRACK_DIR = os.path.join(PROJECT_ROOT, "运营文案", "小红书发布", "_数据追踪")
SCREENSHOT_DIR = os.path.join(PROJECT_ROOT, "运营文案", "小红书发布", "_过程文件", "截图")
CONFIG_PATH = os.path.join(PROJECT_ROOT, "运营文案", "xhs_daily_config.json")
MASTER_CSV = os.path.join(DATA_TRACK_DIR, "master_tracking.csv")
# 插入 XiaohongshuSkills 到 path
XHS_SKILLS_PATH = r"C:\code\XiaohongshuSkills\scripts"
sys.path.insert(0, XHS_SKILLS_PATH)
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: str) -> int:
"""'1.2万' -> 12000, '1,234' -> 1234, '--' -> 0"""
if not text:
return 0
t = text.strip().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 _parse_date(text: str) -> str:
"""'08-28' / '2026-08-28' -> 'YYYY-MM-DD'"""
m = re.match(r"^(\d{4})-(\d{2})-(\d{2})", text)
if m:
return f"{m.group(1)}-{m.group(2)}-{m.group(3)}"
m = re.match(r"^(\d{2})-(\d{2})", text)
if m:
return f"{time.strftime('%Y')}-{m.group(1)}-{m.group(2)}"
return text
def collect_overview_text(page) -> dict:
"""从创作者中心首页/数据中心页面文本提取账号总览"""
page.wait_for_timeout(3000)
text = page.evaluate("document.body.innerText")
overview = {}
patterns = [
("粉丝数", r"粉丝数\s*[::]\s*([\d,]+万?)"),
("近30天新增粉丝", r"近30天.*?([\+-]?[\d,.]+万?)"),
("总阅读量", r"总阅读\s*[::]\s*([\d,]+万?)"),
("总互动量", r"总互动\s*[::]\s*([\d,]+万?)"),
("笔记总数", r"笔记总数\s*[::]\s*([\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_list(page, limit: int = 10) -> list:
"""从笔记管理页抓取笔记列表"""
page.wait_for_timeout(3000)
text = page.evaluate("document.body.innerText")
notes = []
# 用正则匹配每一行:标题 + 数字数据
# 小红书笔记列表格式(典型):
# 标题文本
# 发布时间 阅读数 点赞数 收藏数 评论数 分享数
lines = text.split("\n")
i = 0
while i < len(lines) and len(notes) < limit:
line = lines[i].strip()
if not line:
i += 1
continue
# 尝试匹配数据行(含多个数字)
nums = re.findall(r"([\d,]+)", line)
if len(nums) >= 3:
# 上一行通常是标题
title = lines[i - 1].strip() if i > 0 else ""
if not title:
title = line # 标题和数据在同一行
note = {"title": title}
# 按顺序匹配:阅读/点赞/收藏/评论/分享(从数据行取前5个数字)
for j, num in enumerate(nums[:5]):
key = ["read_count", "like_count", "collect_count", "comment_count", "share_count"][j]
note[key] = _to_int(num.replace(",", ""))
note.setdefault("publish_date", "")
note.setdefault("unreplied_comment_count", 0)
notes.append(note)
i += 1
continue
i += 1
# 备选:从 DOM 提取
if not notes:
notes = collect_notes_from_dom(page, limit)
return notes
def collect_notes_from_dom(page, limit: int = 10) -> list:
"""从 DOM 提取笔记列表(备选方案)"""
return page.evaluate(f"""
() => {{
const notes = [];
const rows = document.querySelectorAll('tr, .note-item, [class*="note-row"], .list-item');
let count = 0;
rows.forEach(row => {{
if (count >= {limit}) return;
const texts = Array.from(row.querySelectorAll('td, .stat, [class*="stat"]'))
.map(el => el.textContent.trim())
.filter(t => t);
const titleEl = row.querySelector('[class*="title"]');
const title = titleEl ? titleEl.textContent.trim() : texts[0] || '';
if (!title || title.length < 2) return;
const nums = texts.slice(1).map(t => parseInt(t.replace(/[,万,]/g, ''))).filter(n => !isNaN(n));
notes.push({{
title,
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,
publish_date: '',
unreplied_comment_count: 0,
}});
count++;
}});
return JSON.stringify(notes);
}}
""")
def collect_unreplied_comments_count(page) -> int:
"""从互动消息/评论管理页抓取未回复评论数"""
page.evaluate("window.location.href = 'https://creator.xiaohongshu.com/message/comment'")
page.wait_for_timeout(2000)
text = page.evaluate("document.body.innerText")
m = re.search(r"未回复\s*[::]\s*(\d+)", text) or re.search(r"未回复(\d+)", text)
if m:
return int(m.group(1))
return 0
def screenshot_page(page, output_path: str):
"""使用 CDP 截取当前页面(整页,JPEG)"""
os.makedirs(os.path.dirname(output_path), exist_ok=True)
# 使用 CDP Page.captureScreenshot(全页面)
resp = page.evaluate("""
() => {
return new Promise(resolve => {
// 滚动到顶部确保截图完整
window.scrollTo(0, 0);
// 用 jsPDF-like 方式:直接用 Page.captureScreenshot CDP 命令
// 这里通过 Runtime.evaluate 调用原生的 Page.captureScreenshot 不可行,
// 改用 CSS 方案:扩展页面高度后截图
document.body.style.zoom = '0.5';
const height = Math.max(
document.body.scrollHeight,
document.documentElement.scrollHeight,
document.body.offsetHeight,
document.documentElement.offsetHeight
);
document.body.style.zoom = '1';
resolve({ width: document.documentElement.clientWidth, height: height });
});
}
""")
# 通过 _send 调用 CDP 截图
result = page._send("Page.captureScreenshot", {"format": "jpeg", "quality": 60})
if "result" in result and "data" in result["result"]:
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_to_master_csv(date: str, overview: dict, notes: list):
"""追加当日数据到 master_tracking.csv"""
os.makedirs(DATA_TRACK_DIR, exist_ok=True)
if not os.path.exists(MASTER_CSV):
# 创建 CSV
with open(MASTER_CSV, "w", newline="", encoding="utf-8") as f:
writer = csv.writer(f)
writer.writerow(["日期", "粉丝数", "近30天新增粉丝", "总阅读量", "总互动量", "笔记总数", "笔记数"])
with open(MASTER_CSV, "a", newline="", encoding="utf-8") as f:
writer = csv.writer(f)
writer.writerow([
date,
overview.get("fans_count", 0),
overview.get("近30天新增粉丝", 0),
overview.get("总阅读量", 0),
overview.get("总互动量", 0),
overview.get("笔记总数", 0),
len(notes),
])
def main():
parser = argparse.ArgumentParser(description="小红书创作者中心日报采集")
parser.add_argument("--login", action="store_true", help="手动登录模式")
parser.add_argument("--headful", action="store_true", help="有头模式(调试)")
args = parser.parse_args()
cfg = _load_config()
date = time.strftime("%Y-%m-%d")
_log(f"开始采集日报 {date}")
# 导入 XiaohongshuPublisher
from cdp_publish import XiaohongshuPublisher
publisher = XiaohongshuPublisher()
if args.login:
_log("手动登录模式:请在弹出的浏览器中完成扫码登录,完成后回车")
publisher.connect(reuse_existing_tab=False)
input("登录完成后按回车保存登录态并退出...")
publisher.disconnect()
_log("登录态已保存(CDP 缓存)")
return
try:
publisher.connect(reuse_existing_tab=True)
if not publisher.check_login():
raise RuntimeError("未登录小红书,请先运行 --login 或手动打开浏览器完成扫码")
# 1. 采集账号总览
_log("采集账号总览数据...")
publisher._navigate("https://creator.xiaohongshu.com/new/home")
overview = collect_overview_text(publisher)
_log(f" 粉丝数: {overview.get('粉丝数', '-')}")
_log(f" 总阅读量: {overview.get('总阅读量', '-')}")
_log(f" 笔记总数: {overview.get('笔记总数', '-')}")
# 2. 采集笔记列表
_log("采集笔记列表...")
publisher._navigate("https://creator.xiaohongshu.com/publish/publish_manage")
notes = collect_notes_list(publisher)
_log(f" 采集到 {len(notes)} 篇笔记")
# 3. 采集未回复评论数
_log("采集未回复评论数...")
unreplied = collect_unreplied_comments_count(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_page(publisher, shot_path)
# 5. 追加 CSV
append_to_master_csv(date, overview, notes)
# 6. 推送飞书
_log("推送到飞书...")
from xhs_feishu import send_daily_report
overview_out = {
"fans_count": overview.get("粉丝数", 0),
"fans_growth_30d": overview.get("近30天新增粉丝", 0),
"total_read": overview.get("总阅读量", 0),
"total_interact": overview.get("总互动量", 0),
"note_count": overview.get("笔记总数", 0),
}
send_daily_report(CONFIG_PATH, date, overview_out, notes, screenshot_path=shot_path)
_log("✅ 日报完成")
except Exception as e:
_log(f"❌ 采集失败: {e}")
raise
finally:
try:
publisher.disconnect()
except Exception:
pass
if __name__ == "__main__":
main()
[ ] 步骤 2:语法检查(无运行时)
python -m py_compile scripts\xhs_daily_report.py 2>&1
# 或者在当前路径
python -m py_compile "C:\code\cfc\运营文案\小红书发布\_过程脚本\xhs_daily_report.py"
预期:无输出(成功),或提示语法错误(修复后重试)
[ ] 步骤 3:Commit
cd C:\code\cfc
git add 运营文案/小红书发布/_过程脚本/xhs_daily_report.py
git commit -m "内容: 小红书创作者中心日报采集脚本"
文件:
无(运行时操作)
[ ] 步骤 1:注册计划任务
schtasks /create /tn "XHS_Daily_Report" `
/tr "python `"C:\code\cfc\运营文案\小红书发布\_过程脚本\xhs_daily_report.py`"" `
/sc daily /st 09:00 /rl highest
预期:SUCCESS: 已成功创建计划任务 XHS_Daily_Report
[ ] 步骤 2:验证任务注册
schtasks /query /tn "XHS_Daily_Report" /fo LIST
预期:显示任务信息,下次运行时间 2026/8/29 9:00:00
| 风险 | 应对 |
|---|---|
| CDP 端口 9222 被占用或 Chrome 未启动 | 脚本开头检查 http://127.0.0.1:9222/json 是否可达,不可达时给出提示 |
| 创作者中心页面改版导致正则失效 | 保留 DOM 备选提取方案(collect_notes_from_dom) |
| 截图全页高度受限 | 当前用 Page.captureScreenshot 不带 clip,自动全页 |
| 飞书 token 过期 | _get_token 内置缓存 + 自动刷新 |
| 9 点机器睡眠/关机 | 任务计划程序「错过则立即运行」选项 |
| 未登录态(cookie 失效) | check_login() 会报错,提示重新 --login |
overview dict、notes list、CONFIG_PATH 在所有函数间命名一致 ✓