tihao-ai/tests/api-probe/probe_creators.py

"""
API 结构探针脚本 v2
目标：用最低接口成本测通 小红书(JustOne PGY) 和 抖音(TikHub 代理) 创作者搜索。

注意：TikHub 使用代理地址 https://server.fmode.cn/thapi 规避跨域问题。
每个测试只调用 1 次，count/pageSize=2，节省 API 成本。
"""

import urllib.request
import urllib.parse
import json
import sys
import time

# Windows GBK 终端强制 UTF-8 输出
if hasattr(sys.stdout, 'reconfigure'):
    sys.stdout.reconfigure(encoding='utf-8', errors='replace')
if hasattr(sys.stderr, 'reconfigure'):
    sys.stderr.reconfigure(encoding='utf-8', errors='replace')

# ─────────────── API Keys & 地址 ───────────────
JUSTONE_TOKEN = "wT0w78sxrwlrpJrQ"
TIKHUB_TOKEN  = "oL4nMZwrhXCro1+TQYye0BL8LASyaphtx0LUmGuRCfqnm1CeBFfbdmLAGQ=="

JUSTONE_BASE  = "https://api.justoneapi.com"
TIKHUB_PROXY  = "https://server.fmode.cn/thapi"   # TikHub 代理，规避跨域

KEYWORD = "护肤"


# ─────────────── 工具函数 ───────────────
def get(url: str, headers: dict | None = None, retry301: int = 3) -> dict:
    for attempt in range(1, retry301 + 2):
        req = urllib.request.Request(url, headers=headers or {})
        try:
            with urllib.request.urlopen(req, timeout=20) as resp:
                body = resp.read().decode("utf-8")
                data = json.loads(body)
                if data.get("code") == 301 and attempt <= retry301:
                    print(f"  [301] RETRY ({attempt}/{retry301})...")
                    time.sleep(3)
                    continue
                return data
        except urllib.error.HTTPError as e:
            body = e.read().decode("utf-8")
            print(f"  [HTTP {e.code}] {body[:300]}")
            return {}
        except Exception as ex:
            print(f"  [ERROR] {ex}")
            return {}
    return {}


def post(url: str, body: dict, headers: dict | None = None) -> dict:
    """POST JSON body 请求"""
    payload = json.dumps(body).encode("utf-8")
    h = dict(headers or {})
    h["Content-Type"] = "application/json"
    req = urllib.request.Request(url, data=payload, headers=h, method="POST")
    try:
        with urllib.request.urlopen(req, timeout=20) as resp:
            return json.loads(resp.read().decode("utf-8"))
    except urllib.error.HTTPError as e:
        body_resp = e.read().decode("utf-8")
        print(f"  [HTTP {e.code}] {body_resp[:300]}")
        return {}
    except Exception as ex:
        print(f"  [ERROR] {ex}")
        return {}


def print_section(title: str):
    print("\n" + "=" * 60)
    print(f"  {title}")
    print("=" * 60)


def describe_keys(obj: dict, indent: int = 0, max_depth: int = 3):
    """递归打印对象的键和值类型（不打印完整值，节省屏幕空间）"""
    prefix = "  " * indent
    for k, v in obj.items():
        if isinstance(v, dict):
            print(f"{prefix}[dict] {k}:")
            if indent < max_depth:
                describe_keys(v, indent + 1, max_depth)
        elif isinstance(v, list):
            print(f"{prefix}[list({len(v)})] {k}")
            if v and isinstance(v[0], dict) and indent < max_depth:
                print(f"{prefix}  └─ 第一个元素 keys:")
                describe_keys(v[0], indent + 2, max_depth)
        else:
            display = repr(v)[:80]
            print(f"{prefix}[{type(v).__name__}] {k} = {display}")


# ─────────────── 1. 小红书 JustOne PGY 创作者搜索 ───────────────
def test_xhs_justone():
    print_section("1. 小红书 — JustOne 蒲公英 Creator Search")

    params = urllib.parse.urlencode({
        "token":           JUSTONE_TOKEN,
        "keyword":         KEYWORD,
        "page":            1,
        "searchType":      "NOTE",
        "fansNumberLower": 20000,
        "fansNumberUpper": 500000,
        "gender":          "ALL",
    })
    url = f"{JUSTONE_BASE}/api/xiaohongshu-pgy/api/solar/cooperator/blogger/v2/v1?{params}"
    print(f"  请求 URL: {url.replace(JUSTONE_TOKEN, '***')}\n")

    data = get(url)
    if not data:
        print("  ⚠ 无响应")
        return

    print(f"  顶层 code  : {data.get('code')}")
    print(f"  顶层 message: {data.get('message')}")
    print(f"\n  完整响应结构：")
    describe_keys(data)

    # 定位列表
    d = data.get("data", {})
    if d:
        for list_key in ("kols", "bloggerList", "list", "items", "result", "authors"):
            lst = d.get(list_key)
            if lst and isinstance(lst, list):
                print(f"\n  [OK] 找到列表字段: data.data.{list_key}，共 {len(lst)} 条")
                if lst:
                    print(f"\n  第一条完整数据：")
                    print(json.dumps(lst[0], ensure_ascii=False, indent=2)[:2000])
                return
        print(f"\n  [WARN] 未在 data.data 中找到已知列表字段，data.data keys: {list(d.keys())}")


# ─────────────── 2. 抖音 JustOne 星图 创作者搜索（多接口尝试）───────────────
DOUYIN_JUSTONE_ENDPOINTS = [
    # 当前在用（已知返回 authors 但为空）
    ("/api/douyin-xingtu/gw/api/gsearch/search_for_author_square/v1",
     {"keyword": KEYWORD, "page": 1, "searchType": "CONTENT"},
     "Xingtu search_for_author_square (当前)"),
    # JustOne 文档 Douyin Xingtu KOL Keyword Search V1
    ("/api/douyin-xingtu/gw/api/gsearch/search_kol/v1",
     {"keyword": KEYWORD, "page": 1},
     "Xingtu search_kol v1"),
    # JustOne 文档 Douyin User Search V2
    ("/api/douyin/aweme/v1/search/user/",
     {"keyword": KEYWORD, "count": 2, "offset": 0},
     "Douyin user search v1 (通用)"),
    # 另一种可能路径
    ("/api/douyin-xingtu/gw/api/search/author/v1",
     {"keyword": KEYWORD, "page": 1},
     "Xingtu search author v1"),
]

def test_douyin_justone():
    print_section("2. 抖音 — JustOne 星图 KOL 搜索（逐一探测可用接口）")

    for endpoint, extra_params, label in DOUYIN_JUSTONE_ENDPOINTS:
        print(f"\n  ── {label} ──")
        params = urllib.parse.urlencode({
            "token": JUSTONE_TOKEN,
            **extra_params,
        })
        url = f"{JUSTONE_BASE}{endpoint}?{params}"
        print(f"  URL: {url.replace(JUSTONE_TOKEN, '***')}")

        data = get(url)
        if not data:
            print("  ⚠ 无响应，跳过")
            continue

        code = data.get("code")
        msg  = data.get("message", "")
        print(f"  code={code}  message={msg}")

        if code != 0:
            print(f"  [SKIP] 接口返回错误 code={code}")
            continue

        # 成功，打印完整结构
        print(f"\n  [OK] 接口 [{label}] 返回成功！")
        print(f"  完整响应结构：")
        describe_keys(data)

        # 定位列表
        d = data.get("data", {})
        for list_key in ("authors", "kols", "kol_list", "list", "items", "result", "userList"):
            lst = d.get(list_key) if isinstance(d, dict) else None
            if lst is None and isinstance(d, dict):
                d2 = d.get("data", {})
                lst = d2.get(list_key) if isinstance(d2, dict) else None
            if lst and isinstance(lst, list):
                print(f"\n  [LIST] 字段: {list_key}，共 {len(lst)} 条")
                if lst:
                    print(f"\n  第一条完整数据：")
                    print(json.dumps(lst[0], ensure_ascii=False, indent=2)[:2000])
                return

        print(f"  [WARN] 未找到列表字段，data keys = {list(data.get('data', {}).keys()) if isinstance(data.get('data'), dict) else data.keys()}")
        return

    print("\n  [FAIL] 所有 JustOne 抖音接口均无有效响应")


# ─────────────── 3. 抖音 — TikHub 星图 KOL 搜索（代理地址）───────────────
def test_douyin_tikhub_xingtu():
    """
    通过代理 https://server.fmode.cn/thapi 调用 TikHub 星图 KOL 搜索。
    代理规避浏览器跨域，Python 直连同样可用。
    只请求 count=2，节省成本。
    """
    print_section("3. 抖音 — TikHub 星图 KOL 搜索（代理）")

    headers = {"Authorization": f"Bearer {TIKHUB_TOKEN}"}

    # 按优先级逐一尝试，成功则停止
    # 每个条目: (url, body_dict, label, use_post)
    candidates = [
        (
            f"{TIKHUB_PROXY}/api/v1/douyin/xingtu/fetch_kol_search_v1",
            {"keyword": KEYWORD, "page": 1, "count": 2},
            "Xingtu KOL 搜索 V1 [GET]",
            False
        ),
        (
            f"{TIKHUB_PROXY}/api/v1/douyin/xingtu/fetch_kol_search_v1",
            {"keyword": KEYWORD, "page": 1, "count": 2},
            "Xingtu KOL 搜索 V1 [POST]",
            True
        ),
        (
            f"{TIKHUB_PROXY}/api/v1/douyin/xingtu/fetch_kol_advanced_search_v2",
            {"keyword": KEYWORD, "page": 1, "count": 2},
            "Xingtu KOL 高级搜索 V2 [POST]",
            True
        ),
        (
            f"{TIKHUB_PROXY}/api/v1/douyin/xingtu_v2/get_author_market_fields",
            {},
            "Xingtu V2 达人广场筛选字段 [GET]",
            False
        ),
    ]

    for url_base, body, label, use_post in candidates:
        print(f"\n  ── {label} ──")
        if use_post:
            print(f"  POST {url_base}  body={body}")
            data = post(url_base, body, headers)
        else:
            params = urllib.parse.urlencode(body) if body else ""
            url = f"{url_base}?{params}" if params else url_base
            print(f"  GET {url}")
            data = get(url, headers)

        if not data:
            print("  [SKIP] 无响应")
            continue

        code = data.get("code")
        msg  = data.get("message", "")
        print(f"  code={code}  message={str(msg)[:100]}")

        if code not in (200, 0):
            print(f"  [SKIP] 非成功状态")
            continue

        print(f"\n  [OK] {label} 通过！")
        print("  响应结构：")
        describe_keys(data)
        _find_and_print_list(data, label)
        return

    print("\n  [FAIL] TikHub 星图接口均失败，改用 Douyin 用户搜索")
    test_douyin_tikhub_user_search()


def test_douyin_tikhub_user_search():
    """
    备用：通过代理调用 TikHub Douyin 用户搜索。
    docs 中 UserSearchRequest / UserSearchRequestV2 是 POST body schema，
    故改为 POST + JSON body（之前用 GET 是错误的）。
    """
    print_section("3b. 抖音备用 — TikHub Douyin 用户搜索（POST，代理）")

    headers = {"Authorization": f"Bearer {TIKHUB_TOKEN}"}

    # (url, body, label, use_post)
    candidates = [
        (
            f"{TIKHUB_PROXY}/api/v1/douyin/search/fetch_user_search_v2",
            {"keyword": KEYWORD, "offset": 0, "count": 2},
            "Douyin Search 用户搜索 V2 [POST]",
            True
        ),
        (
            f"{TIKHUB_PROXY}/api/v1/douyin/search/fetch_user_search",
            {"keyword": KEYWORD, "offset": 0, "count": 2},
            "Douyin Search 用户搜索 V1 [POST]",
            True
        ),
        (
            f"{TIKHUB_PROXY}/api/v1/douyin/web/fetch_user_search",
            {"keyword": KEYWORD, "offset": 0, "count": 2},
            "Douyin Web 用户搜索 [GET]",
            False
        ),
        (
            f"{TIKHUB_PROXY}/api/v1/douyin/web/fetch_user_search",
            {"keyword": KEYWORD, "offset": 0, "count": 2},
            "Douyin Web 用户搜索 [POST]",
            True
        ),
    ]

    for url_base, body, label, use_post in candidates:
        print(f"\n  ── {label} ──")
        if use_post:
            print(f"  POST {url_base}  body={body}")
            data = post(url_base, body, headers)
        else:
            params = urllib.parse.urlencode(body)
            url = f"{url_base}?{params}"
            print(f"  GET {url}")
            data = get(url, headers)

        if not data:
            print("  [SKIP] 无响应")
            continue

        code = data.get("code")
        msg  = data.get("message", "")
        print(f"  code={code}  message={str(msg)[:100]}")

        if code not in (200, 0):
            print(f"  [SKIP] 非成功状态")
            continue

        print(f"\n  [OK] {label} 通过！")
        print("  响应结构：")
        describe_keys(data)
        _find_and_print_list(data, label)
        return

    print("  [FAIL] 所有 TikHub 抖音用户搜索接口均失败")


def test_douyin_tikhub_billboard():
    """
    测试 TikHub Douyin-Billboard-API 账号搜索。
    搜索用户名或抖音号 — 可能不需要高级订阅。
    docs 中 Billboard 端点带有 _post 后缀，故先试 POST 再试 GET。
    """
    print_section("3c. 抖音 — TikHub Billboard 账号搜索")

    headers = {"Authorization": f"Bearer {TIKHUB_TOKEN}"}
    url = f"{TIKHUB_PROXY}/api/v1/douyin/billboard/fetch_account_search_list"
    body = {"keyword": KEYWORD, "count": 2}

    for label, use_post in [("[POST]", True), ("[GET]", False)]:
        print(f"\n  ── Billboard 账号搜索 {label} ──")
        if use_post:
            print(f"  POST {url}  body={body}")
            data = post(url, body, headers)
        else:
            params = urllib.parse.urlencode(body)
            print(f"  GET {url}?{params}")
            data = get(f"{url}?{params}", headers)

        if not data:
            print("  [SKIP] 无响应")
            continue

        code = data.get("code")
        msg  = data.get("message", "")
        print(f"  code={code}  message={str(msg)[:100]}")

        if code not in (200, 0):
            continue

        print(f"\n  [OK] Billboard 账号搜索 {label} 通过！")
        describe_keys(data)
        _find_and_print_list(data, "Billboard")
        return

    print("  [FAIL] Billboard 账号搜索均失败")


def _find_and_print_list(data: dict, label: str):
    """在任意层级自动定位列表字段并打印第一条"""
    LIST_KEYS = ("kol_list", "user_list", "authors", "kols", "list",
                 "items", "result", "data", "user_info")

    def search(obj, depth=0):
        if depth > 4 or not isinstance(obj, dict):
            return False
        for k, v in obj.items():
            if isinstance(v, list) and v and k in LIST_KEYS:
                print(f"\n  [LIST] 路径 .{k}，共 {len(v)} 条")
                print(f"  第一条完整数据：")
                print(json.dumps(v[0], ensure_ascii=False, indent=2)[:2500])
                return True
        for k, v in obj.items():
            if isinstance(v, dict):
                if search(v, depth + 1):
                    return True
        return False

    if not search(data):
        print(f"  [WARN] 未在响应中找到已知列表字段，完整 keys: {list(data.keys())}")


# ─────────────── 4. 字段解析验证（纯 Python 模拟 TS 解析逻辑）───────────────
def parse_xhs_creator(item: dict) -> dict:
    """模拟 justone.service.ts 中 parseXhsCreator 的解析逻辑"""
    raw_content = item.get("contentTags", [])
    content_tags = []
    for t in raw_content:
        if isinstance(t, str):
            content_tags.append(t)
        elif isinstance(t, dict):
            tag = t.get("taxonomy1Tag")
            if isinstance(tag, str):
                content_tags.append(tag)

    feature_tags = [v for v in item.get("featureTags", []) if isinstance(v, str)]
    fans = item.get("fansNum") or item.get("fansCount") or 0
    image_price = item.get("picturePrice") or item.get("imagePrice") or 0
    video_price = item.get("videoPrice") or 0
    min_price   = item.get("lowerPrice") or image_price or 0
    cooperate   = "active" if item.get("cooperateState") == 1 else ""
    gender_map  = {"女": "female", "男": "male", "FEMALE": "female", "MALE": "male"}
    gender      = gender_map.get(str(item.get("gender", "")), str(item.get("gender", "")))

    return {
        "userId":        item.get("userId") or item.get("user_id") or "",
        "nickname":      item.get("name") or item.get("nickname") or "",
        "redId":         item.get("redId") or "",
        "location":      item.get("location") or item.get("city") or "",
        "fansCount":     int(fans),
        "imagePrice":    float(image_price),
        "videoPrice":    float(video_price),
        "minPrice":      float(min_price),
        "cooperationStatus": cooperate,
        "personalTags":  feature_tags,
        "contentTags":   content_tags,
        "gender":        gender,
    }


def parse_douyin_xingtu_creator(item: dict) -> dict:
    """模拟 justone.service.ts 中 parseDouyinXingtuCreator 的解析逻辑"""
    attr = item.get("attribute_datas") or {}
    fans = int(attr.get("follower") or attr.get("fans_count") or 0)
    gender_map = {"1": "male", "2": "female", "0": ""}
    gender = gender_map.get(str(attr.get("gender", "0")), "")

    # 从 last_10_items 提取视频标题中的 hashtag 作为 contentTags
    content_tags = []
    try:
        last_items = json.loads(attr.get("last_10_items") or "[]")
        for it in last_items[:5]:
            title = it.get("item_title", "")
            tags = [t.lstrip("#") for t in title.split() if t.startswith("#")]
            content_tags.extend(tags)
        content_tags = list(dict.fromkeys(content_tags))[:10]   # 去重限10个
    except Exception:
        pass

    # keyword_association 作为 personalTags
    personal_tags = []
    try:
        kw_map = json.loads(attr.get("author_thin_mid_word_association_index") or "{}")
        personal_tags = list(kw_map.keys())
    except Exception:
        pass

    return {
        "userId":        str(item.get("star_id") or attr.get("id") or attr.get("core_user_id") or ""),
        "nickname":      str(attr.get("nickname") or attr.get("name") or ""),
        "location":      str(attr.get("city") or attr.get("province") or ""),
        "fansCount":     fans,
        "imagePrice":    0,
        "videoPrice":    0,
        "minPrice":      0,
        "cooperationStatus": "active" if attr.get("author_status") == "1" else "",
        "personalTags":  personal_tags,
        "contentTags":   content_tags,
        "gender":        gender,
    }


def test_parse_validation():
    print_section("4. 字段解析验证（Python 模拟解析，不调用接口）")

    # ── XHS 验证 ──
    print("\n  [XHS] 模拟解析第一条 kol 数据...")
    xhs_sample = {
        "userId": "5f869dc2000000000101db7c",
        "name": "imtender.",
        "redId": "1011627787",
        "location": "山东 潍坊 潍城区",
        "fansCount": 0,        # 始终为 0，正确字段是 fansNum
        "fansNum": 20022,
        "picturePrice": 0.0,
        "videoPrice": 4000.0,
        "lowerPrice": 4000.0,
        "cooperateState": 1,
        "gender": "女",
        "contentTags": [
            {"taxonomy1Tag": "护肤", "taxonomy2Tags": ["护肤合集"]},
            {"taxonomy1Tag": "个人护理", "taxonomy2Tags": ["身体护理", "头发产品"]},
        ],
        "featureTags": ["沉浸式", "vlog", "氛围感", "抗老", "控油", "开箱"],
    }
    parsed_xhs = parse_xhs_creator(xhs_sample)
    print(f"  nickname      = {parsed_xhs['nickname']!r}   (期望: 'imtender.')")
    print(f"  fansCount     = {parsed_xhs['fansCount']}     (期望: 20022)")
    print(f"  imagePrice    = {parsed_xhs['imagePrice']}    (期望: 0.0)")
    print(f"  videoPrice    = {parsed_xhs['videoPrice']}  (期望: 4000.0)")
    print(f"  minPrice      = {parsed_xhs['minPrice']}   (期望: 4000.0)")
    print(f"  contentTags   = {parsed_xhs['contentTags']}  (期望: ['护肤', '个人护理'])")
    print(f"  personalTags  = {parsed_xhs['personalTags'][:3]}  (期望: ['沉浸式', 'vlog', '氛围感']...)")
    print(f"  gender        = {parsed_xhs['gender']!r}   (期望: 'female')")
    print(f"  cooperationStatus = {parsed_xhs['cooperationStatus']!r}  (期望: 'active')")

    xhs_ok = (
        parsed_xhs["nickname"] == "imtender."
        and parsed_xhs["fansCount"] == 20022
        and parsed_xhs["videoPrice"] == 4000.0
        and "护肤" in parsed_xhs["contentTags"]
        and parsed_xhs["gender"] == "female"
        and parsed_xhs["cooperationStatus"] == "active"
    )
    print(f"\n  XHS 解析{'[PASS]' if xhs_ok else '[FAIL]'}")

    # ── Douyin 验证 ──
    print("\n  [Douyin] 模拟解析第一条 author 数据...")
    dy_sample = {
        "star_id": "7174756942059929657",
        "attribute_datas": {
            "follower": "2161358",
            "gender": "1",
            "city": "湖州",
            "author_status": "1",
            "author_thin_mid_word_association_index": '{"服帖":0.33,"清爽":0.15}',
            "last_10_items": json.dumps([
                {"item_title": "愿我们都能在梦幻的世界里被童话治愈 #艾莎 #沉浸式护肤 #解压", "like_cnt": "2158"},
                {"item_title": "给男友挑战黄色护肤 #沉浸式护肤 #情侣日常", "like_cnt": "3837"},
            ]),
        },
    }
    parsed_dy = parse_douyin_xingtu_creator(dy_sample)
    print(f"  userId        = {parsed_dy['userId']!r}  (期望: '7174756942059929657')")
    print(f"  fansCount     = {parsed_dy['fansCount']}  (期望: 2161358)")
    print(f"  gender        = {parsed_dy['gender']!r}   (期望: 'male')")
    print(f"  location      = {parsed_dy['location']!r}  (期望: '湖州')")
    print(f"  contentTags   = {parsed_dy['contentTags']}  (期望含: '沉浸式护肤')")
    print(f"  personalTags  = {parsed_dy['personalTags']}  (期望: ['服帖', '清爽'])")
    print(f"  cooperationStatus = {parsed_dy['cooperationStatus']!r}  (期望: 'active')")

    dy_ok = (
        parsed_dy["userId"] == "7174756942059929657"
        and parsed_dy["fansCount"] == 2161358
        and parsed_dy["gender"] == "male"
        and any("护肤" in t for t in parsed_dy["contentTags"])
        and "服帖" in parsed_dy["personalTags"]
        and parsed_dy["cooperationStatus"] == "active"
    )
    print(f"\n  Douyin 解析{'[PASS]' if dy_ok else '[FAIL]'}")

    if xhs_ok and dy_ok:
        print("\n  [PASS] 所有字段解析验证通过，可以更新 justone.service.ts")
    else:
        print("\n  [FAIL] 解析有误，需先修复 Python 逻辑再对应修改 TS 服务")


# ─────────────── 主程序 ───────────────
if __name__ == "__main__":
    print("=" * 60)
    print("  API 结构探针  —  关键词:", KEYWORD)
    print("  每平台只调用 1 次，节省成本")
    print("=" * 60)

    test_xhs_justone()             # 小红书：JustOne 蒲公英
    test_douyin_justone()          # 抖音：JustOne 星图（301→重试→成功）
    test_douyin_tikhub_xingtu()    # 抖音：TikHub 星图（GET+POST，修正方法）
    test_douyin_tikhub_billboard() # 抖音：TikHub Billboard 账号搜索（新增）
    test_parse_validation()        # 纯 Python 字段解析验证（不调接口）

    print("\n" + "=" * 60)
    print("  探针完成。根据结果决定抖音数据源：")
    print("    - TikHub Xingtu POST 通：使用 TikHub（有定价），成本低")
    print("    - TikHub Billboard 通：可搜索用户，但无定价")
    print("    - 均失败：继续用 JustOne 星图（需 301 重试，有定价）")
    print("=" * 60)