"""App 端接口客户端 —— 无签名、无 Cookie、无登录。 实测链路（每一步都验过）： 1. POST /service/2/device_register/ 引导调用，不需要任何签名 -> device_id / install_id 2. GET /aweme/v1/aweme/detail/ 带上真实 device_id/iid 即可 -> 完整 aweme_detail，含 video.play_addr.uri 3. GET /aweme/v1/play/?video_id=URI 免鉴权，302 到 douyinvod CDN -> mp4 对比 web 端：web 的 /aweme/v1/web/aweme/detail/ 对匿名用户返回 {"name":"强制登录"}，且要算 a_bogus。App 端两样都不需要。 早先测 App 端拿到空 body，是因为 device_id/iid 是伪造的；换成注册得来的 真实凭据就正常了 —— 空 body 不代表要签名。 """ from __future__ import annotations import json import random import time import uuid from pathlib import Path from typing import Any, Dict, Optional from urllib.parse import urlencode import requests from .client import Aweme, DouyinClient, DouyinError REGISTER_HOSTS = ( "https://log.snssdk.com/service/2/device_register/", "https://api.amemv.com/service/2/device_register/", "https://log-api.amemv.com/service/2/device_register/", "https://toblog.ctobsnssdk.com/service/2/device_register/", ) API_HOSTS = ( "https://aweme.snssdk.com", "https://api.amemv.com", ) APP_VERSION = "9.9.9" VERSION_CODE = "990" UPDATE_VERSION_CODE = "99013" APP_ID = "1128" CHANNEL = "douyin_official" # 机型池。openudid/cdid 本来就每次随机，但 UA、机型、分辨率这些以前是写死的， # 意味着我们所有注册请求在这些维度上完全一致。 # # 注册受限时到底是按 IP 还是按"UA+机型"这类特征分桶，我们没验证过——本机没有 # 第二个出口 IP，做不了对照实验。把机型随机化是成本极低的两用手段： # 如果是按机型特征分桶，它能缓解；如果是纯 IP，它无害。 # # 每项是 (device_type, device_brand, os_version, os_api, resolution, dpi)。 DEVICE_POOL = ( ("SM-G9750", "samsung", "12", "31", "1080*2340", "420"), ("SM-S9080", "samsung", "13", "33", "1440*3088", "560"), ("PDRM00", "OPPO", "12", "31", "1080*2400", "480"), ("V2118A", "vivo", "12", "31", "1080*2376", "480"), ("M2012K11AC", "Xiaomi", "13", "33", "1080*2400", "440"), ("ELS-AN00", "HUAWEI", "12", "31", "1176*2400", "480"), ("21091116AC", "Redmi", "12", "31", "1080*2400", "440"), ("NOH-AN01", "HONOR", "12", "31", "1344*2772", "480"), ) # 兼容旧引用：默认取池子里第一项 DEVICE_TYPE, DEVICE_BRAND, OS_VERSION, OS_API = DEVICE_POOL[0][:4] class DeviceRegisterError(DouyinError): pass class NotAVideo(DouyinError): """作品存在，但不是视频（图文/图集）。""" def _hex(n: int) -> str: return "".join(random.choice("0123456789abcdef") for _ in range(n)) class Device: """一台虚拟设备的身份。注册一次可长期复用，所以落盘缓存。""" def __init__(self, device_id: str, install_id: str, openudid: str, cdid: str, clientudid: str, profile: Optional[tuple] = None) -> None: self.device_id = device_id self.install_id = install_id self.openudid = openudid self.cdid = cdid self.clientudid = clientudid # 机型档案必须跟着设备走：注册时报的是小米、后续 detail 请求的 UA # 却是三星，这种自相矛盾本身就是个可疑特征。 self.profile = tuple(profile) if profile else DEVICE_POOL[0] @property def device_type(self) -> str: return self.profile[0] @property def device_brand(self) -> str: return self.profile[1] @property def os_version(self) -> str: return self.profile[2] @property def os_api(self) -> str: return self.profile[3] @property def resolution(self) -> str: return self.profile[4] @property def dpi(self) -> str: return self.profile[5] @property def user_agent(self) -> str: return (f"com.ss.android.ugc.aweme/{UPDATE_VERSION_CODE} " f"(Linux; U; Android {self.os_version}; zh_CN; {self.device_type}; " f"Build/QP1A.190711.020; Cronet/TTNetVersion:6c7b9b1a 2022-11-29)") def to_dict(self) -> Dict[str, Any]: return { "device_id": self.device_id, "install_id": self.install_id, "openudid": self.openudid, "cdid": self.cdid, "clientudid": self.clientudid, "profile": list(self.profile), } @classmethod def from_dict(cls, d: Dict[str, Any]) -> "Device": # profile 是后加的，旧缓存没有这个字段，退回池子第一项 return cls(d["device_id"], d["install_id"], d["openudid"], d["cdid"], d["clientudid"], d.get("profile")) # ------------------------------------------------------------------ # @classmethod def register(cls, session: Optional[requests.Session] = None, timeout: float = 20.0, retries: int = 5, backoff: float = 30.0) -> "Device": """向 ByteDance 注册一台新设备。这一步不需要签名。 短时间内注册太多次，所有 host 都会返回 device_id='0'。实测： 指纹（openudid/cdid）每次都是新的照样被拒，4 个 host 同时被拒， 约 10 分钟后自行恢复 —— 所以是临时限流，且不是按设备指纹或按 单个 host 分桶的。 至于分桶的键到底是 IP 还是 UA/机型这类特征，我们没验证过：本机 没有第二个出口 IP，做不了对照实验。DEVICE_POOL 的随机化对两种 情况都无害，权当两用。 这里做指数退避重试而不是直接抛错，否则调用方会把后续所有条目 瞬间判为失败。 """ last: Optional[Exception] = None for attempt in range(retries): try: return cls._register_once(session, timeout) except DeviceRegisterError as exc: last = exc if attempt < retries - 1: wait = backoff * (2 ** attempt) print(f" [注册受限] 等待 {wait:.0f}s 后重试 " f"({attempt + 1}/{retries})", flush=True) time.sleep(wait) raise last # type: ignore[misc] @classmethod def _register_once(cls, session: Optional[requests.Session] = None, timeout: float = 20.0) -> "Device": session = session or requests.Session() openudid = _hex(16) cdid = str(uuid.UUID(int=random.getrandbits(128))) clientudid = str(uuid.UUID(int=random.getrandbits(128))) profile = random.choice(DEVICE_POOL) dev_type, dev_brand, os_ver, os_api, resolution, dpi = profile ua = (f"com.ss.android.ugc.aweme/{UPDATE_VERSION_CODE} " f"(Linux; U; Android {os_ver}; zh_CN; {dev_type}; " f"Build/QP1A.190711.020; Cronet/TTNetVersion:6c7b9b1a 2022-11-29)") query = { "ac": "wifi", "channel": CHANNEL, "aid": APP_ID, "app_name": "aweme", "version_code": VERSION_CODE, "version_name": APP_VERSION, "device_platform": "android", "os": "android", "ssmix": "a", "device_type": dev_type, "device_brand": dev_brand, "language": "zh", "os_api": os_api, "os_version": os_ver, "manifest_version_code": VERSION_CODE, "resolution": resolution, "dpi": dpi, "update_version_code": UPDATE_VERSION_CODE, "_rticket": str(int(time.time() * 1000)), "cdid": cdid, "openudid": openudid, } payload = { "magic_tag": "ss_app_log", "header": { "display_name": "抖音", "update_version_code": int(UPDATE_VERSION_CODE), "manifest_version_code": int(VERSION_CODE), "aid": int(APP_ID), "channel": CHANNEL, "package": "com.ss.android.ugc.aweme", "app_version": APP_VERSION, "version_code": int(VERSION_CODE), "sdk_version": "2.13.5", "os": "Android", "os_version": os_ver, "os_api": int(os_api), "device_model": dev_type, "device_brand": dev_brand, "device_manufacturer": dev_brand, "cpu_abi": "arm64-v8a", "density_dpi": int(dpi), "display_density": "mdpi", "resolution": resolution.replace("*", "x"), "language": "zh", "timezone": 8, "access": "wifi", "not_request_sender": 0, "region": "CN", "tz_name": "Asia/Shanghai", "tz_offset": 28800, "sim_region": "cn", "openudid": openudid, "clientudid": clientudid, "cdid": cdid, "rom": "QP1A.190711.020", "rom_version": f"{dev_brand}/{os_ver}", "device_id": 0, "install_id": 0, }, "_gen_time": int(time.time() * 1000), } headers = { "User-Agent": ua, "Content-Type": "application/json; charset=utf-8", "Accept-Encoding": "gzip", "X-SS-REQ-TICKET": str(int(time.time() * 1000)), } errors = [] for host in REGISTER_HOSTS: try: r = session.post(host + "?" + urlencode(query), data=json.dumps(payload), headers=headers, timeout=timeout) data = r.json() except (requests.RequestException, ValueError) as exc: errors.append(f"{host}: {type(exc).__name__}") continue did = str(data.get("device_id_str") or data.get("device_id") or "") iid = str(data.get("install_id_str") or data.get("install_id") or "") if did and did != "0": return cls(did, iid, openudid, cdid, clientudid, profile) errors.append(f"{host}: device_id={did!r}") raise DeviceRegisterError("设备注册全部失败: " + "; ".join(errors)) @classmethod def load_or_register(cls, cache: str | Path, session: Optional[requests.Session] = None) -> "Device": cache = Path(cache) if cache.exists(): try: return cls.from_dict(json.loads(cache.read_text(encoding="utf-8"))) except (ValueError, KeyError, OSError): pass dev = cls.register(session) cache.parent.mkdir(parents=True, exist_ok=True) cache.write_text(json.dumps(dev.to_dict(), indent=2), encoding="utf-8") return dev class AppClient: """按 aweme_id 取详情。不需要签名、Cookie 或登录态。""" def __init__(self, device: Optional[Device] = None, cache: str | Path = ".device.json", timeout: float = 25.0) -> None: self.session = requests.Session() self.device = device or Device.load_or_register(cache, self.session) self.timeout = timeout self.session.headers.update({ "User-Agent": self.device.user_agent, "Accept-Encoding": "gzip", "Connection": "keep-alive", "sdk-version": "2", }) def _params(self) -> Dict[str, str]: return { "device_platform": "android", "os": "android", "ssmix": "a", "_rticket": str(int(time.time() * 1000)), "cdid": self.device.cdid, "channel": CHANNEL, "aid": APP_ID, "app_name": "aweme", "version_code": VERSION_CODE, "version_name": APP_VERSION, "manifest_version_code": VERSION_CODE, "update_version_code": UPDATE_VERSION_CODE, "resolution": self.device.resolution, "dpi": self.device.dpi, "device_type": self.device.device_type, "device_brand": self.device.device_brand, "language": "zh", "os_api": self.device.os_api, "os_version": self.device.os_version, "openudid": self.device.openudid, "device_id": self.device.device_id, "iid": self.device.install_id, "ac": "wifi", "ts": str(int(time.time())), } def fetch_detail(self, aweme_id: str, retries: int = 3) -> Aweme: last = "" for attempt in range(retries): host = API_HOSTS[attempt % len(API_HOSTS)] params = self._params() params["aweme_id"] = aweme_id url = f"{host}/aweme/v1/aweme/detail/?" + urlencode(params) try: r = self.session.get(url, timeout=self.timeout) except requests.RequestException as exc: last = f"{type(exc).__name__}: {exc}" continue if not r.content: last = f"HTTP {r.status_code} 空 body" time.sleep(1 + attempt) continue try: data: Dict[str, Any] = r.json() except ValueError: last = f"非 JSON: {r.text[:120]!r}" continue detail = data.get("aweme_detail") if not detail: # 服务端会在 filter_detail 里说明为什么不给正片， # 例如图文作品 filter_reason=images_base f = data.get("filter_detail") or {} reason = f.get("filter_reason") or "" if reason == "images_base": raise NotAVideo(f"{aweme_id} 是图文作品（图集），不是视频") if reason: raise DouyinError( f"{aweme_id} 被服务端过滤：{reason}" + (f"（{f.get('notice')}）" if f.get("notice") else "") ) raise DouyinError( f"无 aweme_detail，status_code={data.get('status_code')}，" "作品可能已删除或为私密" ) return DouyinClient.parse(detail) raise DouyinError(f"App detail 连续 {retries} 次失败（{last}）")