butubb
c2b310c7bf
侧边栏在「监控」右边加了「运营」：账号列表 → 点进二级详情看该账号的数据。 【为什么是独立模块而不是监控的子视图】两者形状不同：监控是公开数据（点赞/收藏/评论/分享）的每轮快照+差分；运营是创作者后台按日期给出的曝光/观看/完播率/涨粉。凭据不同、采集方式也不同 —— 那边要浏览器登录态，这边是纯请求。硬塞进同一个模型会同时污染两边。 【扫码登录的关键差异】监控的扫码把登录态写进浏览器默认 profile（爬虫要复用）。运营要的是 cookie 字符串（纯请求够用），所以每次登录开一个**临时上下文**，扫完取出 cookie 就丢弃 —— 登第二个账号不会把第一个顶掉，也不影响监控那个登录态，十个账号互不干扰。 【决策依据】tools/probe_creator_api.py 的 Phase 0 实测：签名可自造（XYW_：MD5 → base64 → AES-128-CBC，与 xhshow 内置实现常量逐字节一致）；主站 cookie 即可认证创作者后台；接口与参数已与真实页面对齐。 后端： - api/creator/models.py: creator_account / creator_note_stat。**复用 MonitorBase**，这样 create_all 与上一轮改成元数据驱动的 _ensure_columns 会自动覆盖新表 - api/creator/signing.py: XYW_ 签名，带三条实测结论（url= 前缀、appId=ugc、401 与 406 的区别） - api/creator/client.py: 纯 httpx 客户端。字段名尚未亲眼验证过，所以写成多别名匹配；解析不出来存 None 而非 0 - api/creator/service.py: 账号 CRUD 与同步。cookie 绝不进入对外结构，只给 has_cookie - api/creator/login.py: 临时上下文的扫码登录 - api/routers/creator.py: 8 条路由，全部带鉴权 前端： - 侧边栏「运营」+ OperationView（账号列表 → 二级详情）+ AddAccountDialog - 权限状态显眼呈现：pending 时照抄后台原话「已为您申请数据权限，次日可查看」，并说明此时同步返回 0 条是正常的，不是采集失败 测试：tests/test_creator_client.py 新增 48 例，含「cookie 不得出现在对外结构里」这条不变量，以及权限未生效时空壳响应的处理。
346 lines
13 KiB
Python
346 lines
13 KiB
Python
# -*- coding: utf-8 -*-
|
||
# Copyright (c) 2025 [email protected]
|
||
#
|
||
# This file is part of MediaCrawler project.
|
||
# Repository: https://github.com/NanmiCoder/MediaCrawler/blob/main/api/creator/client.py
|
||
# GitHub: https://github.com/NanmiCoder
|
||
# Licensed under NON-COMMERCIAL LEARNING LICENSE 1.1
|
||
#
|
||
# 声明：本代码仅供学习和研究目的使用。使用者应遵守以下原则：
|
||
# 1. 不得用于任何商业用途。
|
||
# 2. 使用时应遵守目标平台的使用条款和robots.txt规则。
|
||
# 3. 不得进行大规模爬取或对平台造成运营干扰。
|
||
# 4. 应合理控制请求频率，避免给目标平台带来不必要的负担。
|
||
# 5. 不得用于任何非法或不当的用途。
|
||
#
|
||
# 详细许可条款请参阅项目根目录下的LICENSE文件。
|
||
# 使用本代码即表示您同意遵守上述原则和LICENSE中的所有条款。
|
||
|
||
"""小红书创作者后台的纯请求客户端。
|
||
|
||
无浏览器：签名在本地算（见 signing.py），请求走 httpx。
|
||
|
||
**关于字段名的谨慎**：Phase 0 抓到的响应里，列表接口因为账号权限未生效而返回空壳
|
||
（`data.result` 只有 `{success, code, message}` 没有数据），所以**真实字段名尚未亲眼
|
||
见过**。因此每个指标都写成**多别名匹配**，并且解析不出来时存 `None` 而不是 0 ——
|
||
0 是真实值，None 是"不知道"，两者混淆会让报表说谎。
|
||
"""
|
||
|
||
import re
|
||
from typing import Any, Dict, List, Optional
|
||
|
||
import httpx
|
||
|
||
from .signing import sign_xyw, signed_api
|
||
|
||
CREATOR_ORIGIN = "https://creator.xiaohongshu.com"
|
||
DATA_ANALYSIS_PAGE = f"{CREATOR_ORIGIN}/statistics/data-analysis"
|
||
|
||
USER_INFO_PATH = "/api/galaxy/user/info"
|
||
PERMISSION_PATH = "/api/galaxy/creator/datacenter/permission/query"
|
||
NOTE_LIST_PATH = "/api/galaxy/creator/datacenter/note/analyze/list"
|
||
|
||
USER_AGENT = (
|
||
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
|
||
"(KHTML, like Gecko) Chrome/138.0.0.0 Safari/537.36"
|
||
)
|
||
|
||
# 签名被网关拒绝时返回的响应体。区分它和普通业务错误很重要：406 说明签名写错了，
|
||
# 而应用层的 code/-100 说明签名没问题、只是没有登录态。
|
||
SIGNATURE_REJECTED_MARKERS = ("code", -1)
|
||
|
||
|
||
class CreatorApiError(RuntimeError):
|
||
"""调用创作者后台失败。``status`` 用于区分是网关拒绝还是业务错误。"""
|
||
|
||
def __init__(self, message: str, status: int = 0, payload: Any = None):
|
||
super().__init__(message)
|
||
self.status = status
|
||
self.payload = payload
|
||
|
||
|
||
def trans_cookies(cookie_str: str) -> Dict[str, str]:
|
||
"""把 cookie 字符串解析成字典。容忍末尾分号、换行和零散空格。"""
|
||
jar: Dict[str, str] = {}
|
||
for chunk in re.split(r"[;\n]", cookie_str or ""):
|
||
chunk = chunk.strip()
|
||
if not chunk or "=" not in chunk:
|
||
continue
|
||
name, value = chunk.split("=", 1)
|
||
name = name.strip()
|
||
if name:
|
||
jar[name] = value.strip()
|
||
return jar
|
||
|
||
|
||
def _pick(item: Dict[str, Any], *names: str) -> Any:
|
||
"""按别名顺序取第一个存在的键。
|
||
|
||
字段名来自二手资料，未亲眼验证，所以不赌单一命名。
|
||
"""
|
||
for name in names:
|
||
if name in item and item[name] is not None:
|
||
return item[name]
|
||
return None
|
||
|
||
|
||
_COUNT_UNITS = {"万": 10_000, "w": 10_000, "W": 10_000, "亿": 100_000_000, "k": 1_000, "K": 1_000}
|
||
|
||
|
||
def as_int(value: Any) -> Optional[int]:
|
||
"""解析计数。处理 "1.2万"、"1,234"、"123" 与已经是数字的情况。"""
|
||
if value is None or isinstance(value, bool):
|
||
return None
|
||
if isinstance(value, (int, float)):
|
||
return int(value)
|
||
|
||
text = str(value).strip().replace(",", "").replace(" ", "")
|
||
if not text or text in ("-", "--", "暂无"):
|
||
return None
|
||
|
||
unit = 1
|
||
suffix = text[-1]
|
||
if suffix in _COUNT_UNITS:
|
||
unit = _COUNT_UNITS[suffix]
|
||
text = text[:-1]
|
||
try:
|
||
return int(float(text) * unit)
|
||
except ValueError:
|
||
return None
|
||
|
||
|
||
def as_float(value: Any) -> Optional[float]:
|
||
"""解析比率。``"12.3%"`` -> 12.3；``"0.123"`` 原样返回数字。"""
|
||
if value is None or isinstance(value, bool):
|
||
return None
|
||
if isinstance(value, (int, float)):
|
||
return float(value)
|
||
|
||
text = str(value).strip().replace("%", "")
|
||
try:
|
||
return float(text)
|
||
except ValueError:
|
||
return None
|
||
|
||
|
||
_DURATION_RE = re.compile(r"(?:(\d+)\s*分)?\s*(?:(\d+)\s*秒)?")
|
||
|
||
|
||
def as_seconds(value: Any) -> Optional[float]:
|
||
"""解析时长。处理 ``"1分30秒"``、``"01:30"``、``"45"``（秒）。"""
|
||
if value is None or isinstance(value, bool):
|
||
return None
|
||
if isinstance(value, (int, float)):
|
||
return float(value)
|
||
|
||
text = str(value).strip()
|
||
if not text or text in ("-", "--"):
|
||
return None
|
||
|
||
if ":" in text:
|
||
parts = text.split(":")
|
||
try:
|
||
total = 0.0
|
||
for part in parts:
|
||
total = total * 60 + float(part)
|
||
return total
|
||
except ValueError:
|
||
return None
|
||
|
||
if "分" in text or "秒" in text:
|
||
minutes = re.search(r"(\d+)\s*分", text)
|
||
seconds = re.search(r"(\d+)\s*秒", text)
|
||
if not minutes and not seconds:
|
||
return None
|
||
return float(minutes.group(1) if minutes else 0) * 60 + float(
|
||
seconds.group(1) if seconds else 0
|
||
)
|
||
|
||
try:
|
||
return float(text)
|
||
except ValueError:
|
||
return None
|
||
|
||
|
||
# 指标 -> 候选原始字段名。别名来自公开资料，权威与否只能等真实响应来验证。
|
||
_FIELD_ALIASES: Dict[str, tuple] = {
|
||
"exposure": ("exposure", "exposure_count", "imp", "impression", "impression_count"),
|
||
"views": ("views", "view", "view_count", "watch", "watch_count", "read_count"),
|
||
"likes": ("likes", "like", "like_count", "liked_count"),
|
||
"comments": ("comments", "comment", "comment_count", "comments_count"),
|
||
"favorites": ("favorites", "favorite", "favorite_count", "collect", "collect_count", "collected_count"),
|
||
"shares": ("shares", "share", "share_count", "shared_count"),
|
||
"new_followers": ("new_followers", "fans_growth", "follower_growth", "increase_fans"),
|
||
"danmaku": ("danmaku", "danmaku_count", "barrage"),
|
||
"cover_ctr": ("cover_ctr", "cover_click_rate", "cover_click_ratio", "ctr"),
|
||
"avg_watch_seconds": ("avg_watch_seconds", "avg_watch_time", "average_watch_time"),
|
||
"two_second_exit_rate": ("two_second_exit_rate", "2s_exit_rate", "exit_rate_2s"),
|
||
"completion_rate": ("completion_rate", "finish_rate", "complete_rate"),
|
||
}
|
||
|
||
_COUNT_FIELDS = {
|
||
"exposure", "views", "likes", "comments", "favorites", "shares", "new_followers", "danmaku",
|
||
}
|
||
_DURATION_FIELDS = {"avg_watch_seconds"}
|
||
_RATE_FIELDS = {"two_second_exit_rate", "completion_rate", "cover_ctr"}
|
||
|
||
NOTE_ID_ALIASES = ("note_id", "noteId", "id", "content_id", "item_id")
|
||
TITLE_ALIASES = ("title", "content", "note_title", "display_title")
|
||
PUBLISH_TIME_ALIASES = ("publish_time", "publishTime", "post_time", "create_time", "time")
|
||
|
||
|
||
def _normalize_metric(name: str, raw: Any) -> Any:
|
||
if name in _COUNT_FIELDS:
|
||
return as_int(raw)
|
||
if name in _DURATION_FIELDS:
|
||
return as_seconds(raw)
|
||
if name in _RATE_FIELDS:
|
||
return as_float(raw)
|
||
return raw
|
||
|
||
|
||
def normalize_note(item: Dict[str, Any]) -> Dict[str, Any]:
|
||
"""把一条原始记录规范化成落库用的字段。"""
|
||
note: Dict[str, Any] = {
|
||
"note_id": str(_pick(item, *NOTE_ID_ALIASES) or ""),
|
||
"title": str(_pick(item, *TITLE_ALIASES) or ""),
|
||
"publish_time": as_int(_pick(item, *PUBLISH_TIME_ALIASES)),
|
||
}
|
||
for name, aliases in _FIELD_ALIASES.items():
|
||
note[name] = _normalize_metric(name, _pick(item, *aliases))
|
||
return note
|
||
|
||
|
||
def find_note_list(payload: Any) -> List[Dict[str, Any]]:
|
||
"""在响应里找出笔记数组。
|
||
|
||
接口的**确切结构还没亲眼见过**（权限未生效时 `data.result` 里没有数据），
|
||
所以不写死路径：遍历 JSON，挑出"看起来像一批笔记记录"的那个列表 ——
|
||
元素是 dict，且至少带一个指标字段。找不到就返回空列表，让上层如实报"没数据"。
|
||
"""
|
||
best: List[Dict[str, Any]] = []
|
||
metric_keys = {alias for aliases in _FIELD_ALIASES.values() for alias in aliases}
|
||
|
||
def walk(value: Any) -> None:
|
||
nonlocal best
|
||
if isinstance(value, dict):
|
||
for child in value.values():
|
||
walk(child)
|
||
elif isinstance(value, list):
|
||
if value and isinstance(value[0], dict):
|
||
keys = set(value[0].keys())
|
||
if keys & metric_keys and len(value) > len(best):
|
||
best = value
|
||
for child in value:
|
||
walk(child)
|
||
|
||
walk(payload)
|
||
return best
|
||
|
||
|
||
class CreatorClient:
|
||
"""一个账号的客户端。``cookie`` 就是它的全部身份。"""
|
||
|
||
def __init__(self, cookie: str, timeout: float = 25.0):
|
||
self.cookies = trans_cookies(cookie)
|
||
self.a1 = self.cookies.get("a1", "")
|
||
self._timeout = timeout
|
||
|
||
@property
|
||
def looks_authenticated(self) -> bool:
|
||
"""签名需要 a1；没有它连请求都签不出来。"""
|
||
return bool(self.a1)
|
||
|
||
def _headers(self, api: str, body: dict | None = None) -> Dict[str, str]:
|
||
cookie_header = "; ".join(f"{k}={v}" for k, v in self.cookies.items())
|
||
return {
|
||
"user-agent": USER_AGENT,
|
||
"accept": "application/json, text/plain, */*",
|
||
"accept-language": "zh-CN,zh;q=0.9",
|
||
"origin": CREATOR_ORIGIN,
|
||
"referer": DATA_ANALYSIS_PAGE,
|
||
"cookie": cookie_header,
|
||
**sign_xyw(api, self.a1, body=body),
|
||
}
|
||
|
||
async def _get(self, path: str, params: Dict[str, Any]) -> Dict[str, Any]:
|
||
if not self.looks_authenticated:
|
||
raise CreatorApiError("cookie 里没有 a1，无法完成签名", status=0)
|
||
|
||
query = "&".join(f"{k}={v}" for k, v in params.items())
|
||
api = signed_api(path, query)
|
||
url = f"{CREATOR_ORIGIN}{path}?{query}" if query else f"{CREATOR_ORIGIN}{path}"
|
||
|
||
async with httpx.AsyncClient(timeout=self._timeout, follow_redirects=False) as client:
|
||
response = await client.get(url, headers=self._headers(api))
|
||
return self._unwrap(response)
|
||
|
||
@staticmethod
|
||
def _unwrap(response: httpx.Response) -> Dict[str, Any]:
|
||
if response.status_code == 406:
|
||
raise CreatorApiError(
|
||
"签名被网关拒绝（406）—— 待签字符串的拼法不对", status=406
|
||
)
|
||
try:
|
||
payload = response.json()
|
||
except Exception as exc: # noqa: BLE001
|
||
raise CreatorApiError(
|
||
f"响应不是 JSON（HTTP {response.status_code}）", status=response.status_code
|
||
) from exc
|
||
|
||
if response.status_code == 401 or payload.get("code") == -100:
|
||
raise CreatorApiError("登录态无效或已过期", status=401, payload=payload)
|
||
if not payload.get("success", True):
|
||
raise CreatorApiError(
|
||
str(payload.get("msg") or "接口返回失败"),
|
||
status=response.status_code,
|
||
payload=payload,
|
||
)
|
||
return payload
|
||
|
||
async def fetch_user_info(self) -> Dict[str, Any]:
|
||
"""当前 cookie 属于哪个账号。登录后用它取名与去重。"""
|
||
payload = await self._get(USER_INFO_PATH, {})
|
||
data = payload.get("data") or {}
|
||
return {
|
||
"user_id": str(data.get("userId") or ""),
|
||
"nickname": str(data.get("userName") or ""),
|
||
"avatar": str(data.get("userAvatar") or ""),
|
||
"red_id": str(data.get("redId") or ""),
|
||
"role": str(data.get("role") or ""),
|
||
"permissions": list(data.get("permissions") or []),
|
||
}
|
||
|
||
async def fetch_permission(self) -> Dict[str, Any]:
|
||
"""数据权限状态。
|
||
|
||
``tip_msg`` 是后台原话（实测是"已为您申请数据权限，次日可查看"），照抄不改写 ——
|
||
这条信息必须原样交给用户，它解释了"为什么没有数据"。
|
||
"""
|
||
payload = await self._get(PERMISSION_PATH, {})
|
||
data = payload.get("data") or {}
|
||
return {
|
||
"display": data.get("display"),
|
||
"status": data.get("status"),
|
||
"tip": str(data.get("tip_msg") or ""),
|
||
}
|
||
|
||
async def fetch_note_list(
|
||
self, start_ms: int, end_ms: int, page_num: int = 1, page_size: int = 10, note_type: int = 0
|
||
) -> List[Dict[str, Any]]:
|
||
"""按发布时间区间取笔记列表。
|
||
|
||
参数与顺序**照抄真实页面的请求**（见 Phase 0 抓包），不要凭感觉改。
|
||
"""
|
||
payload = await self._get(
|
||
NOTE_LIST_PATH,
|
||
{
|
||
"post_begin_time": start_ms,
|
||
"post_end_time": end_ms,
|
||
"type": note_type,
|
||
"page_size": page_size,
|
||
"page_num": page_num,
|
||
},
|
||
)
|
||
return [normalize_note(item) for item in find_note_list(payload)]
|