更新

2026-07-17 09:24:47 +08:00
commit 530e7f839d
4353 changed files with 731879 additions and 0 deletions
@@ -0,0 +1,191 @@
+import hashlib
+import re
+import time
+import json
+import random
+import base64
+import urllib.parse
+from os import path
+import subprocess
+original_popen = subprocess.Popen
+def patched_popen(*args, **kwargs):
+    if kwargs.get('universal_newlines') or kwargs.get('text'):
+        if 'encoding' not in kwargs:
+            kwargs['encoding'] = 'utf-8'
+    return original_popen(*args, **kwargs)
+subprocess.Popen = patched_popen
+
+import execjs
+import requests
+
+basedir = path.dirname(__file__)
+static_dir = path.join(basedir, 'static')
+node_modules = path.join(static_dir, 'node_modules')
+
+# 全局唯一 User-Agent：a_bogus 签名、HTTP 请求头、protobuf body、webid 采集等
+# 必须全部使用同一个 UA，否则抖音服务端重算 a_bogus 时与请求头 UA 不一致 -> 7911。
+# 该值需与浏览器登录上下文（playwright new_context user_agent）保持一致。
+DEFAULT_USER_AGENT = (
+    "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
+    "(KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36"
+)
+
+# 动态编译 JS
+dy_path = path.join(static_dir, 'dy_ab.js')
+dy_js = execjs.compile(open(dy_path, 'r', encoding='utf-8').read(), cwd=node_modules)
+
+sign_path = path.join(static_dir, 'dy_live_sign.js')
+sign_js = execjs.compile(open(sign_path, 'r', encoding='utf-8').read(), cwd=node_modules)
+
+login_path = path.join(static_dir, 'login.js')
+login_js = execjs.compile(open(login_path, 'r', encoding='utf-8').read(), cwd=node_modules)
+
+
+def generateSecretPhoneNum(phone):
+    return login_js.call('generateSecretPhoneNum', phone)
+
+
+def generateSecretCode(phone, code):
+    return login_js.call('generateSecretCode', phone, code)
+
+
+def trans_cookies(cookies_str):
+    cookies = {}
+    for i in cookies_str.split("; "):
+        try:
+            parts = i.split('=')
+            key = parts[0].strip()
+            val = '='.join(parts[1:]).strip()
+            # 防御性清洗：过滤掉因为误粘贴 fetch 等包含非法字符或换行的 Cookie 键
+            if not key or any(c in key for c in "()[]{}'\"\n \t\\"):
+                continue
+            cookies[key] = val
+        except:
+            continue
+    return cookies
+
+
+def generate_req_sign(e, priK):
+    """私信传 obj，其他的拼接"""
+    return dy_js.call('get_req_sign', e, priK)
+
+
+def generate_a_bogus(query, data="", user_agent=None):
+    """query, data 都是拼接字符串。
+
+    user_agent 必须与实际发出请求所用的 User-Agent 完全一致（见 DEFAULT_USER_AGENT），
+    否则抖音服务端用请求头 UA 重算 a_bogus 会对不上，导致 7911 安全校验失败。
+    """
+    return dy_js.call('get_ab', query, data, user_agent or DEFAULT_USER_AGENT)
+
+
+def generate_signature(room_id, user_unique_id):
+    raw_string = f"live_id=1,aid=6383,version_code=180800,webcast_sdk_version=1.0.15,room_id={room_id},sub_room_id=,sub_channel_id=,did_rule=3,user_unique_id={user_unique_id},device_platform=web,device_type=,ac=,identity=audience"
+    x_ms_stub = hashlib.md5(raw_string.encode("utf-8")).hexdigest()
+    result = sign_js.call("get_signature", x_ms_stub)
+    return result.get("X-Bogus")
+
+
+def generate_ree_key(prik):
+    """传递私钥"""
+    return dy_js.call('get_ree_key', prik)
+
+
+def generate_bd_ticket_client_data(api, ticket, ts_sign, priK):
+    """传递 query, ticket, ts_sign, priK"""
+    timestamp = int(time.time())
+    res_sign = f"ticket={ticket}&path={api}&timestamp={timestamp}"
+    p = {
+        'ts_sign': ts_sign,
+        'req_content': 'ticket,path,timestamp',
+        'req_sign': generate_req_sign(res_sign, priK),
+        'timestamp': timestamp,
+    }
+    p = json.dumps(p, ensure_ascii=False, separators=(',', ':'))
+    return base64.urlsafe_b64encode(p.encode('utf-8')).decode('utf-8')
+
+
+def generate_msToken(randomlength=107):
+    random_str = ''
+    base_str = 'ABCDEFGHIGKLMNOPQRSTUVWXYZabcdefghigklmnopqrstuvwxyz0123456789='
+    length = len(base_str) - 1
+    for _ in range(randomlength):
+        random_str += base_str[random.randint(0, length)]
+    return random_str
+
+
+def generate_fake_webid(random_length=19):
+    random_str = ''
+    base_str = '0123456789'
+    length = len(base_str) - 1
+    for _ in range(random_length):
+        random_str += base_str[random.randint(0, length)]
+    return random_str
+
+
+def generate_webid(auth=None, url=""):
+    # 优先用已采集到的 web_id（避免每次发送都发起一次阻塞的 HTTP 请求，导致事件循环卡顿）
+    cached = getattr(auth, "web_id", None) if auth is not None else None
+    if cached:
+        return str(cached)
+    if url == "":
+        url = "https://www.douyin.com/discover?modal_id=7376449060384935209"
+    try:
+        from .auth import DouyinAuth
+        headers = {
+            "User-Agent": DEFAULT_USER_AGENT,
+            "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8",
+            "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8",
+            "upgrade-insecure-requests": "1"
+        }
+        if auth and auth.cookie_str:
+            headers['cookie'] = auth.cookie_str
+        try:
+            from rpa_engine.runtime_config import requests_proxies
+            proxies = requests_proxies()
+        except Exception:
+            proxies = None
+        response = requests.get(
+            url, headers=headers, verify=False, timeout=10, proxies=proxies
+        )
+        res_text = response.text
+        user_unique_id = re.findall(r'\\"user_unique_id\\":\\"(.*?)\\"', res_text)[0]
+        return user_unique_id
+    except Exception:
+        return generate_fake_webid()
+
+
+def generate_millisecond():
+    return int(round(time.time() * 1000))
+
+
+def normalize_client_cert(cert: str) -> str:
+    """统一 client_cert / sdk_cert 格式为「base64 证书体」。
+
+    web_protect.client_cert 与 frontier WS 的 sdk_cert 通常已是 base64(PEM)；
+    若误把 PEM 原文或 frontier 证书二次 base64，会导致 7911。
+    """
+    cert = (cert or "").strip()
+    if not cert:
+        return ""
+    if cert.startswith("-----BEGIN"):
+        return base64.b64encode(cert.encode("utf-8")).decode("utf-8")
+    return cert
+
+
+def resolve_proto_device_id(device_id: str = "", web_id: str = "", my_uid: int = 0) -> str:
+    """protobuf / frontier 更倾向使用数字 device_id（通常等于 my_uid）。"""
+    for candidate in (device_id, web_id, str(my_uid or "")):
+        c = str(candidate or "").strip()
+        if c.isdigit():
+            return c
+    return str(device_id or web_id or "0")
+
+
+def splice_url(params):
+    splice_url_str = ''
+    for key, value in params.items():
+        if value is None:
+            value = ''
+        splice_url_str += key + '=' + urllib.parse.quote(str(value)) + '&'
+    return splice_url_str[:-1]