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 不得出现在对外结构里」这条不变量，以及权限未生效时空壳响应的处理。
242 lines
7.0 KiB
Python
242 lines
7.0 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""创作者后台客户端的解析与签名。
|
|
|
|
**字段名尚未亲眼验证过**：Phase 0 抓响应时账号的数据权限还没生效，列表接口返回的是
|
|
空壳（`data.result` 里只有 `{success, code, message}`）。所以这些解析写成多别名匹配，
|
|
而这份测试就是它的规格 —— 等真实响应到手，先跑这里看哪些假设破了。
|
|
"""
|
|
|
|
import pytest
|
|
|
|
from api.creator import signing
|
|
from api.creator.client import (
|
|
CreatorClient,
|
|
as_float,
|
|
as_int,
|
|
as_seconds,
|
|
find_note_list,
|
|
normalize_note,
|
|
trans_cookies,
|
|
)
|
|
from api.creator.models import CreatorAccount
|
|
from api.creator.service import _account_dict
|
|
|
|
|
|
# --- cookie 解析 -----------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"raw, expected",
|
|
[
|
|
("a1=abc; web_session=xyz", {"a1": "abc", "web_session": "xyz"}),
|
|
("a1=abc;web_session=xyz;", {"a1": "abc", "web_session": "xyz"}),
|
|
("a1=abc\nweb_session=xyz", {"a1": "abc", "web_session": "xyz"}),
|
|
(" a1 = abc ; ", {"a1": "abc"}),
|
|
("", {}),
|
|
(None, {}),
|
|
# 值里可以有等号，不能被截断
|
|
("a1=abc=def", {"a1": "abc=def"}),
|
|
],
|
|
)
|
|
def test_trans_cookies(raw, expected):
|
|
assert trans_cookies(raw) == expected
|
|
|
|
|
|
def test_client_without_a1_cannot_sign():
|
|
"""a1 参与签名，没有它连请求都发不出去 —— 要提前拦而不是发出去再猜。"""
|
|
assert CreatorClient("web_session=abc").looks_authenticated is False
|
|
assert CreatorClient("a1=abc").looks_authenticated is True
|
|
|
|
|
|
# --- 数值解析 --------------------------------------------------------------
|
|
#
|
|
# 后台返回的可能是数字，也可能是 "1.2万" / "12.3%" / "1分30秒" 这类展示值。
|
|
# 解析不出来一律 None —— 不是 0。0 是真实值，None 是"不知道"。
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"raw, expected",
|
|
[
|
|
(123, 123),
|
|
("123", 123),
|
|
("1,234", 1234),
|
|
("1.2万", 12000),
|
|
("3万", 30000),
|
|
("1.5w", 15000),
|
|
("1亿", 100000000),
|
|
(0, 0),
|
|
("0", 0),
|
|
# 这些必须是 None 而不是 0 —— 把"没给"当成"是零"会让报表说谎
|
|
(None, None),
|
|
("", None),
|
|
("-", None),
|
|
("暂无", None),
|
|
("abc", None),
|
|
(True, None),
|
|
],
|
|
)
|
|
def test_as_int(raw, expected):
|
|
assert as_int(raw) == expected
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"raw, expected",
|
|
[
|
|
(12.3, 12.3),
|
|
("12.3%", 12.3),
|
|
("12.3", 12.3),
|
|
(None, None),
|
|
("-", None),
|
|
("暂无数据", None),
|
|
],
|
|
)
|
|
def test_as_float(raw, expected):
|
|
assert as_float(raw) == expected
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"raw, expected",
|
|
[
|
|
(45, 45.0),
|
|
("45", 45.0),
|
|
("1分30秒", 90.0),
|
|
("2分", 120.0),
|
|
("30秒", 30.0),
|
|
("01:30", 90.0),
|
|
("1:00:00", 3600.0),
|
|
(None, None),
|
|
("-", None),
|
|
("abc", None),
|
|
],
|
|
)
|
|
def test_as_seconds(raw, expected):
|
|
assert as_seconds(raw) == expected
|
|
|
|
|
|
# --- 字段归一化 ------------------------------------------------------------
|
|
|
|
|
|
def test_normalize_note_maps_aliases():
|
|
"""不同来源的记录用不同字段名，别名表要能都接住。"""
|
|
note = normalize_note(
|
|
{
|
|
"note_id": "abc123",
|
|
"title": "标题",
|
|
"publish_time": 1700000000000,
|
|
"view_count": "1.2万",
|
|
"like_count": 34,
|
|
"collected_count": 5,
|
|
"share_count": 2,
|
|
"comment_count": 7,
|
|
"cover_click_rate": "12.5%",
|
|
"avg_watch_time": "1分30秒",
|
|
}
|
|
)
|
|
|
|
assert note["note_id"] == "abc123"
|
|
assert note["title"] == "标题"
|
|
assert note["views"] == 12000
|
|
assert note["likes"] == 34
|
|
assert note["favorites"] == 5
|
|
assert note["shares"] == 2
|
|
assert note["comments"] == 7
|
|
assert note["cover_ctr"] == 12.5
|
|
assert note["avg_watch_seconds"] == 90.0
|
|
|
|
|
|
def test_normalize_note_leaves_missing_fields_as_none():
|
|
note = normalize_note({"note_id": "abc123"})
|
|
|
|
assert note["note_id"] == "abc123"
|
|
assert note["views"] is None
|
|
assert note["likes"] is None
|
|
|
|
|
|
def test_find_note_list_digs_the_array_out_of_a_nested_payload():
|
|
"""接口的确切结构没见过，所以按"像是一批笔记记录"来找，不写死路径。"""
|
|
payload = {
|
|
"code": 0,
|
|
"data": {
|
|
"result": {
|
|
"success": True,
|
|
"notes": [
|
|
{"note_id": "n1", "views": 10, "likes": 1},
|
|
{"note_id": "n2", "views": 20, "likes": 2},
|
|
],
|
|
}
|
|
},
|
|
}
|
|
|
|
found = find_note_list(payload)
|
|
|
|
assert [item["note_id"] for item in found] == ["n1", "n2"]
|
|
|
|
|
|
def test_find_note_list_returns_empty_for_the_permission_gated_envelope():
|
|
"""权限未生效时接口返回的就是这个 —— 必须安静地给出空列表，不是报错。"""
|
|
payload = {
|
|
"code": 0,
|
|
"success": True,
|
|
"msg": "成功",
|
|
"data": {"result": {"success": True, "code": 0, "message": "success"}},
|
|
}
|
|
|
|
assert find_note_list(payload) == []
|
|
|
|
|
|
# --- 签名 ------------------------------------------------------------------
|
|
|
|
|
|
def test_signed_api_carries_the_url_prefix():
|
|
"""待签字符串必须带 `url=`。少了它网关返回 406，而 406 的响应体看不出错在哪。"""
|
|
assert signing.signed_api("/api/galaxy/user/info") == "url=/api/galaxy/user/info"
|
|
assert (
|
|
signing.signed_api("/api/x", "a=1&b=2")
|
|
== "url=/api/x?a=1&b=2"
|
|
)
|
|
|
|
|
|
def test_sign_returns_xs_and_xt():
|
|
headers = signing.sign_xyw("url=/api/galaxy/user/info", "some-a1")
|
|
|
|
assert set(headers) == {"x-s", "x-t"}
|
|
assert headers["x-s"].startswith("XYW_")
|
|
assert headers["x-t"].isdigit()
|
|
|
|
|
|
def test_signature_is_stable_for_a_fixed_timestamp():
|
|
"""同一输入同一时间戳必须得到同一签名 —— 否则说明有隐藏的随机源。"""
|
|
first = signing.sign_xyw("url=/api/x", "a1", timestamp_ms=1700000000000)
|
|
second = signing.sign_xyw("url=/api/x", "a1", timestamp_ms=1700000000000)
|
|
|
|
assert first == second
|
|
|
|
|
|
def test_signature_changes_with_the_signed_string():
|
|
"""签名必须真的绑定待签内容，否则改参数不会被发现 —— 那这个签名就没意义了。"""
|
|
base = signing.sign_xyw("url=/api/x?a=1", "a1", timestamp_ms=1700000000000)
|
|
other = signing.sign_xyw("url=/api/x?a=2", "a1", timestamp_ms=1700000000000)
|
|
|
|
assert base["x-s"] != other["x-s"]
|
|
|
|
|
|
# --- 凭证不外泄 ------------------------------------------------------------
|
|
|
|
|
|
def test_account_dict_never_carries_the_cookie():
|
|
"""cookie 等于登录态。对外结构里只该有 `has_cookie`。"""
|
|
account = CreatorAccount(
|
|
id=1,
|
|
nickname="测试",
|
|
user_id="u1",
|
|
cookie="a1=SECRET; web_session=SECRET",
|
|
created_at=0,
|
|
updated_at=0,
|
|
)
|
|
|
|
payload = _account_dict(account)
|
|
|
|
assert payload["has_cookie"] is True
|
|
assert "cookie" not in payload
|
|
assert "SECRET" not in str(payload)
|