""" 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)