import sys import os import re import json import time import struct import base64 import random import hashlib import binascii import socket import threading import zlib from pathlib import Path from urllib.parse import (quote, urlencode, parse_qs, urlparse, urlsplit, parse_qsl, unquote) from datetime import datetime, timezone, timedelta from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from concurrent.futures import ThreadPoolExecutor, as_completed from typing import Optional, Dict, Any, List, Tuple sys.path.append('..') try: import requests except ImportError: requests = None try: from Crypto.Cipher import AES from Crypto.Util import Counter except ImportError: AES = None Counter = None try: from base.spider import Spider except ImportError: class Spider: pass try: import warnings import urllib3 urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) warnings.filterwarnings('ignore', message='Unverified HTTPS request') except Exception: pass CONFIG_DEVICE_ID = '829902912863503360' CONFIG_INSTALL_ID = '250404579274111008' CONFIG_PLATFORM = 'android' CONFIG_CACHE_SECONDS = 3600 _CURRENT_DOMAIN = 'http://127.0.0.1:9877' _EMBEDDED_SERVER = None _EMBEDDED_PORT = 0 # ------------------------------------------------------------------ # 官网域名自动获取（2026-09 修复：站点改为纯前端 SPA，SSR 全局回退， # 页面里不再有 _ROUTER_DATA，必须走 Modern.js 的 __loader 接口。 # 同时支持站点更换/新增域名时自动探测可用域名。） # ------------------------------------------------------------------ # 候选域名列表：第一个为默认主域名，其余为备用/疑似镜像。 # init(extend) 时若传入了自定义域名，会被自动追加到候选列表最前面。 _CANDIDATE_DOMAINS = [ 'https://hongguoduanju.com', 'https://www.hongguoduanju.com', 'https://hongguo.tv', 'https://www.hongguo.tv', 'https://hongguoduanju.tv', ] # 全局 HTTP 连接池（复用 TCP/TLS 连接，避免每次新建连接耗时） _http_session = None def _get_http_session(): global _http_session if _http_session is None: _http_session = requests.Session() _http_session.trust_env = False return _http_session # 已验证可用的域名缓存（仅内存态） _ACTIVE_SITE = '' _SITE_LOCK = threading.Lock() # 域名健康追踪（避免持续请求已挂的域名） _DOMAIN_HEALTH = {} # {domain: {'fail_count': int, 'last_fail': timestamp, 'next_probe': timestamp}} _HEALTH_LOCK = threading.Lock() # 第三方搜索聚合接口（官网接口挂了时的兜底） _THIRD_SEARCH_API = 'https://api-v2.cenguigui.cn/api/duanju/api.php' _THIRD_SEARCH_TIMEOUT = 8 # 本地搜索缓存（内存态，进程重启失效） _SEARCH_CACHE = {} _SEARCH_CACHE_LOCK = threading.Lock() _SEARCH_CACHE_TTL = 300 # 缓存5分钟 def _normalize_domain(value: str) -> str: """把任意输入规整成 scheme://host 形式（不带尾部斜杠）。""" value = str(value or '').strip() if not value: return '' if not value.startswith('http://') and not value.startswith('https://'): value = 'https://' + value try: parsed = urlparse(value) if not parsed.netloc: return '' scheme = parsed.scheme or 'https' return '%s://%s' % (scheme, parsed.netloc) except Exception: return '' def _register_domain(domain: str) -> None: """登记一个可用域名（去重、置顶候选列表）。""" global _ACTIVE_SITE domain = _normalize_domain(domain) if not domain: return with _SITE_LOCK: if domain not in _CANDIDATE_DOMAINS: _CANDIDATE_DOMAINS.insert(0, domain) if _ACTIVE_SITE != domain: _ACTIVE_SITE = domain print('[红果果] 启用官网域名: %s' % domain) def get_active_site() -> str: with _SITE_LOCK: if _ACTIVE_SITE: return _ACTIVE_SITE return _CANDIDATE_DOMAINS[0] if _CANDIDATE_DOMAINS else 'https://hongguoduanju.com' def _site_probe_ok(domain: str) -> bool: """探测域名是否存活：首页可达且是红果短剧的页面。""" try: resp = _get_http_session().get( domain + '/', headers={'User-Agent': 'Mozilla/5.0 (Linux; Android 12; TV) ' 'AppleWebKit/537.36 (KHTML, like Gecko) ' 'Chrome/126.0 Safari/537.36'}, timeout=2, verify=False, ) if resp.status_code != 200: return False text = resp.text[:20000] return ('hongguo' in text.lower()) or ('短剧' in text) except Exception: return False def _probe_domain(domain: str) -> Tuple[bool, str]: """并发探测域名，返回 (是否成功, 域名)。用于 ThreadPoolExecutor。""" ok = _site_probe_ok(domain) return (ok, domain) def probe_site_domains(extra_domains=None) -> str: """并发探测候选域名，返回第一个可用域名并登记；全部失败返回空串。 优化：使用 ThreadPoolExecutor 并发探测，不再逐个等超时。""" candidates = [] for d in (extra_domains or []): nd = _normalize_domain(d) if nd and nd not in candidates: candidates.append(nd) with _SITE_LOCK: candidates.extend([d for d in _CANDIDATE_DOMAINS if d not in candidates]) # 并发探测所有域名，谁先活下来就用谁 with ThreadPoolExecutor(max_workers=len(candidates)) as executor: futures = {executor.submit(_probe_domain, d): d for d in candidates} for future in as_completed(futures): try: ok, domain = future.result(timeout=5) if ok: _register_domain(domain) print('[红果果] 探测到可用域名: %s' % domain) return domain except Exception: continue return '' # 域名探测结果缓存（避免每次init都重新探测） _PROBE_CACHE = {} _PROBE_CACHE_LOCK = threading.Lock() _PROBE_CACHE_TTL = 300 # 缓存5分钟 def _site_probe_ok_cached(domain: str) -> bool: """带缓存的域名探测，避免重复网络请求。""" now = time.time() with _PROBE_CACHE_LOCK: entry = _PROBE_CACHE.get(domain) if entry and now - entry['time'] < _PROBE_CACHE_TTL: return entry['ok'] # 缓存未命中，执行实际探测 ok = _site_probe_ok(domain) with _PROBE_CACHE_LOCK: _PROBE_CACHE[domain] = {'ok': ok, 'time': now} return ok def _get_domain_health(domain: str) -> dict: """获取域名健康状态。""" with _HEALTH_LOCK: return _DOMAIN_HEALTH.get(domain, {'fail_count': 0, 'last_fail': 0, 'next_probe': 0}) def _record_domain_failure(domain: str) -> None: """记录域名失败，标记短暂冷却。""" with _HEALTH_LOCK: h = _DOMAIN_HEALTH.get(domain, {'fail_count': 0, 'last_fail': 0, 'next_probe': 0}) h['fail_count'] = h.get('fail_count', 0) + 1 h['last_fail'] = time.time() h['next_probe'] = time.time() + 60 # 1分钟后重试 _DOMAIN_HEALTH[domain] = h def _record_domain_success(domain: str) -> None: """记录域名成功，清除失败计数。""" with _HEALTH_LOCK: h = _DOMAIN_HEALTH.get(domain, {'fail_count': 0, 'last_fail': 0, 'next_probe': 0}) h['fail_count'] = 0 h['last_fail'] = 0 h['next_probe'] = 0 _DOMAIN_HEALTH[domain] = h def extract_domains_from_extend(extend) -> list: """从 init(extend) 参数中提取用户自定义域名（字符串/列表/JSON 均可）。""" domains = [] if not extend: return domains try: if isinstance(extend, str): text = extend.strip() if text.startswith('{') or text.startswith('['): obj = json.loads(text) else: obj = text else: obj = extend if isinstance(obj, dict): extend = obj.get('extend', obj) for key in ('hongguo_site', 'site', 'domain', 'api', 'url', 'host'): v = obj.get(key) if isinstance(obj, dict) else None if v: domains.append(v) obj = extend if isinstance(obj, str): for part in re.split(r'[\s,;#\n]+', obj): nd = _normalize_domain(part) if nd: domains.append(nd) elif isinstance(obj, list): for part in obj: nd = _normalize_domain(part) if nd: domains.append(nd) except Exception: pass return domains def _pkcs7_pad(data, block_size=16): pad_len = block_size - (len(data) % block_size) return data + bytes([pad_len] * pad_len) def _sm3_rotl(x, n, bits=32): n = n % bits return ((x << n) | (x >> (bits - n))) & ((1 << bits) - 1) def _sm3_p0(x): return x ^ _sm3_rotl(x, 9) ^ _sm3_rotl(x, 17) def _sm3_p1(x): return x ^ _sm3_rotl(x, 15) ^ _sm3_rotl(x, 23) def _sm3_ff(x, y, z, j): if j < 16: return x ^ y ^ z return (x & y) | (x & z) | (y & z) def _sm3_gg(x, y, z, j): if j < 16: return x ^ y ^ z return (x & y) | ((~x) & z) def _sm3_hash(msg_list): IV = [0x7380166f, 0x4914b2b9, 0x172442d7, 0xda8a0600, 0xa96f30bc, 0x163138aa, 0xe38dee4d, 0xb0fb0e4e] msg = bytes(msg_list) length = len(msg) msg += b'\x80' while len(msg) % 64 != 56: msg += b'\x00' msg += (length * 8).to_bytes(8, 'big') V = list(IV) for i in range(0, len(msg), 64): block = msg[i:i+64] W = [int.from_bytes(block[j*4:j*4+4], 'big') for j in range(16)] for j in range(16, 68): w = _sm3_p1(W[j-16] ^ W[j-9] ^ _sm3_rotl(W[j-3], 15)) ^ _sm3_rotl(W[j-13], 7) ^ W[j-6] W.append(w) W_prime = [W[j] ^ W[j+4] for j in range(64)] A, B, C, D, E, F, G, H = V for j in range(64): t_j = 0x79cc4519 if j < 16 else 0x7a879d8a SS1 = _sm3_rotl((_sm3_rotl(A, 12) + E + _sm3_rotl(t_j, j % 32)) & 0xFFFFFFFF, 7) SS2 = SS1 ^ _sm3_rotl(A, 12) TT1 = (_sm3_ff(A, B, C, j) + D + SS2 + W_prime[j]) & 0xFFFFFFFF TT2 = (_sm3_gg(E, F, G, j) + H + SS1 + W[j]) & 0xFFFFFFFF D = C C = _sm3_rotl(B, 9) B = A A = TT1 H = G G = _sm3_rotl(F, 19) F = E E = _sm3_p0(TT2) V = [(V[k] ^ [A, B, C, D, E, F, G, H][k]) & 0xFFFFFFFF for k in range(8)] return ''.join('%08x' % x for x in V) class SM3: def __init__(self, data=b''): if isinstance(data, str): data = data.encode("utf-8") self._data = bytearray(data) def update(self, data): self._data.extend(data) def digest(self): h = _sm3_hash([b for b in self._data]) return bytes.fromhex(h) def _enc_varint(value): buf = bytearray() while value > 0x7F: buf.append((value & 0x7F) | 0x80) value >>= 7 buf.append(value & 0x7F) return bytes(buf) def _zigzag32(n): return ((n << 1) ^ (n >> 31)) & 0xFFFFFFFF def _zigzag64(n): return ((n << 1) ^ (n >> 63)) & 0xFFFFFFFFFFFFFFFF def _enc_sint32(field, value): if value == 0: return b'' return _enc_varint(field << 3) + _enc_varint(_zigzag32(value)) def _enc_sint64(field, value): if value == 0: return b'' return _enc_varint(field << 3) + _enc_varint(_zigzag64(value)) def _enc_string(field, value): if not value: return b'' data = value.encode('utf-8') return _enc_varint((field << 3) | 2) + _enc_varint(len(data)) + data def _enc_bytes(field, value): if not value: return b'' data = bytes(value) return _enc_varint((field << 3) | 2) + _enc_varint(len(data)) + data def _enc_float(field, value): if value == 0.0: return b'' return _enc_varint((field << 3) | 5) + struct.pack('> (32 - shift))) & 0xFFFFFFFF def rl8(x: int, k: int) -> int: n = 8 s = k & (n - 1) return ((x << s) | (x >> (n - s))) & 0xff def ror32(value, count): count %= 32 low = value << (32 - count) value >>= count value |= low value &= 0xFFFFFFFF return value def ror(value, count): count %= 64 low = value << (64 - count) value >>= count value |= low value &= 0xFFFFFFFFFFFFFFFF return value def get_key_hash(key, rand): to_hash = bytearray(68) to_hash[:32] = key to_hash[32:36] = rand.to_bytes(4, byteorder='little') to_hash[36:] = key d1 = (rand >> 16) & 0x000000ff d2 = (d1 << 11) | (rand >> 24) d2 ^= (d1 >> 5) ^ d1 d2 = ~d2 & 0xffffffff return SM3(to_hash).digest(), d2.to_bytes(4, "little") def split_blocks(message, block_size=16, require_padding=True): assert len(message) % block_size == 0 or not require_padding return [message[i:i + 16] for i in range(0, len(message), block_size)] def add_round_key(s, k): for i in range(4): for j in range(4): s[i][j] ^= k[i][j] def add_round_key_con(s, k, con): for i in range(4): for j in range(4): s[i][j] ^= k[i][con[j]] def bytes2matrix(text): return [list(text[i:i + 4]) for i in range(0, len(text), 4)] def xor_bytes(a, b): return bytearray(i ^ j for i, j in zip(a, b)) def matrix2bytes(matrix): return bytearray(sum(matrix, [])) def mix_single_column(a, i): t = a[0][i] ^ a[1][i] ^ a[2][i] ^ a[3][i] u = a[0][i] a[0][i] ^= t ^ xtime(a[0][i] ^ a[1][i]) a[1][i] ^= t ^ xtime(a[1][i] ^ a[2][i]) a[2][i] ^= t ^ xtime(a[2][i] ^ a[3][i]) a[3][i] ^= t ^ xtime(a[3][i] ^ u) def mix_columns(s): for i in range(4): mix_single_column(s, i) def inv_mix_columns(s): for i in range(0, 4): inv_mix_single_column(s, i) def inv_mix_single_column(a, i): u = xtime(xtime(a[0][i] ^ a[2][i])) v = xtime(xtime(a[1][i] ^ a[3][i])) a[0][i] ^= u a[1][i] ^= v a[2][i] ^= u a[3][i] ^= v mix_single_column(a, i) s_box = b''.join([ b'\xFA\x7D\x08\x6B\x9C\x59\xB3\x4B\x04\x5F\x39\xD0\x38\x4A\x91\x99', b'\x00\x67\xA6\x20\x9F\xF5\x4D\x82\x73\x26\xEE\xDF\x18\x66\x83\x33', b'\x80\x03\x19\xFB\xD9\xFE\xAE\xAA\xA9\xB0\x52\xC6\x0B\xF3\x79\x25', b'\x4E\x78\xB4\x36\xAC\x5D\x1A\x27\x9E\x88\xDB\xBD\x3C\x63\xEC\x49', b'\x15\xC1\x30\x1F\xDC\xB8\x56\xD4\x6C\xCD\xCA\x09\x43\xC8\x35\xA3', b'\xEF\x1E\xF4\x96\xD2\xFC\x0E\x72\x7B\x94\x84\xD1\xEA\x45\x5A\x62', b'\x02\x3F\xD3\x12\x81\x34\x2B\xDD\x7E\xE6\x28\xF2\xA5\x46\x13\x01', b'\x3B\x21\xF6\x61\x37\x29\x2A\x0D\xED\x8C\xAF\xBF\x9D\x5C\xBB\x24', b'\x76\x0F\x75\xE4\x53\x89\xE1\x98\x8D\xB1\x9A\x65\x70\x4F\x54\x4C', b'\x58\xAB\x6E\x6F\x8B\x23\xC4\x07\x11\x0C\xBA\xCF\xA0\xA4\x8E\xD8', b'\x05\x3D\x14\xB2\xDA\x74\xC3\xD7\xE7\xBE\xD6\x7F\xDE\x48\x16\x3E', b'\x85\x90\xA1\x55\xB7\x77\x42\x22\xC9\x86\x50\x2E\x17\xF9\x64\x31', b'\x2C\x9B\xF1\x6D\x1C\x44\x68\xE3\xE9\xA8\x93\x97\xCB\x32\x57\xEB', b'\xE5\x71\x6A\xAD\xC0\xCC\xC7\xC5\xFD\x60\x1D\xA2\x2D\x47\xA7\xE2', b'\x51\x69\x5E\x7A\xCE\x0A\x41\xB6\x95\x8F\xF7\xB9\x87\xE0\x3A\x06', b'\x10\x8A\xB5\xF8\x5B\xD5\xF0\xBC\x92\xFF\x7C\x2F\xC2\xE8\x1B\x40', b'\xEC\x1B\xDA\xBD\xBA\x98\x91\x0C\xB2\x2B\x83\x41\x34\x67\xFB\x0A', b'\xD8\x76\xB5\x46\x05\x59\x61\x23\x75\x90\x87\x2A\xE3\x50\x15\x4C', b'\xAC\xB1\x79\xEB\xAE\xE5\x95\x47\x04\x68\xF0\x86\x3D\x51\x8B\x0F', b'\xCA\x8E\xE4\xB9\x4E\xF2\x12\x82\xBC\x0E\xD5\xF7\xEF\x28\x25\xCF', b'\x5B\x5D\xE9\x6A\x55\x02\xE1\x33\xBE\x93\xE7\xF5\xAD\x9D\x3E\x39', b'\x24\xA8\xE2\xFA\x17\x57\xD0\x7A\x0D\x08\x30\xD6\xB8\xA3\x8D\xFD', b'\x07\x9A\xC4\x1E\x6E\x22\x64\x97\xD2\x1D\xB0\xBF\x45\x66\x3F\x6C', b'\xDD\xDB\x27\x80\xA7\x11\xDC\xA6\xC5\x52\xF8\xC0\xB6\xC8\x5C\x00', b'\x73\x60\x7B\xA0\x19\x13\xAA\xC9\x35\x48\x4B\xD3\xA4\xCD\x9F\x99', b'\xF3\x10\x44\x40\x54\x7E\x29\xF4\x06\x1F\xA2\xAB\xA1\x2F\x3C\xF6', b'\xAF\x85\x62\x36\x21\x7F\x5E\xDF\x20\x1A\xB3\xB4\xE6\xFF\x72\x84', b'\x8F\x65\x26\x94\x5A\x77\xEA\x43\x78\xC7\x4A\xCC\x2C\x14\x6B\xC6', b'\xE8\x74\x53\xFC\xD4\x1C\xCE\x31\x70\x03\x18\x8C\x96\x38\x32\x89', b'\xF1\x3A\x5F\xD7\xF9\xA9\x69\xB7\x63\x37\x58\xC2\x3B\xC3\x71\xCB', b'\x9E\x92\x01\x8A\x0B\x4D\x88\x9B\xBB\x4F\x6D\x6F\xE0\xFE\xA5\x49', b'\xDE\x56\x16\x09\xED\x9C\xC1\x2D\xEE\x81\x7D\xD9\x7C\xD1\x2E\x42', b'\x5B\x4D\xC1\xA6\x5D\xEA\x44\xFD\x45\x4E\x1B\xA1\x3F\xD1\x89\xE1', b'\x7D\x2F\xAA\xDB\xAB\xAD\x59\xCB\xB1\xCE\x9A\x28\xC9\xE0\xF6\x70', b'\x39\x4A\xD7\xFF\x30\xF5\xDD\xBC\x57\x3B\x11\x8D\xB2\xEE\x00\xB6', b'\xE6\x1A\x5A\x7C\xF9\xDE\xC4\xCD\x2E\x80\xBB\xB9\x4C\xA5\x9F\x84', b'\x08\xC6\x6F\x42\x6C\xF0\x27\xE7\x8B\x3A\x9C\x51\xFB\x67\x21\x75', b'\x41\x31\xA7\xCA\x20\x43\x2A\xB7\xBF\xD9\x7A\xF2\xB5\xF8\x8C\x2C', b'\x23\x83\x4F\x8F\x60\xA0\x04\x13\x37\x14\xE3\x01\xC5\x63\x66\x5C', b'\x74\x81\xDF\x58\xBD\x68\x90\x3D\xD2\xB3\x34\xF4\x19\x93\x32\x29', b'\xD6\x49\xAE\x0D\x4B\xD8\x07\x9E\xAC\x1E\x2D\x0B\x40\xB8\x72\xBA', b'\x76\x10\x71\xA8\xE4\x56\x1D\x48\xFE\xE5\xC2\x47\x91\xDA\x87\x26', b'\x9D\x1F\x88\x6B\xC0\x98\xBE\x25\x09\x97\x33\xA3\x85\x16\x5E\x7F', b'\xDC\x6E\x54\xE9\xF7\xA9\xC8\xE8\xC3\x77\xD0\x82\x2B\xEC\x02\x62', b'\x8A\x92\x0E\x3E\xB0\x0F\x05\xF3\xF1\x96\x78\x38\x86\x36\x18\x3C', b'\x24\xCF\x0A\xB4\x53\xCC\x61\x65\xA4\xC7\x94\xD5\x15\x7E\x6D\xEF', b'\x79\x22\x35\x12\x6A\x8E\x52\x06\x55\x7B\x46\x64\x50\x95\xE2\x0C', b'\xED\xD3\x17\x03\xA2\x9B\x99\xEB\x1C\xFC\xAF\xD4\x73\x69\xFA\x5F', b'\xF7\x2C\x1E\xBF\xC8\xE1\xF3\x9F\x76\x80\x71\x48\xAA\x94\xAD\x64', b'\xFB\x89\xC6\x60\xC3\x32\xB3\x4D\xD2\xE0\x44\xDD\x5F\xA8\xB1\xC7', b'\x68\x23\x34\xC9\x6D\x12\x7F\xB7\xEB\x15\xBE\xA9\xD1\x78\x93\xA0', b'\x0C\x92\xA4\xD7\x47\xE3\x8A\xC2\x70\xAB\x26\x41\x9A\x79\xA7\xD8', b'\x14\x85\x8F\xC0\x6F\x56\xD0\x8C\x11\xB9\x2E\x3C\xE2\x9D\xCF\x0E', b'\xDE\x03\x5D\x46\x3E\xCD\x38\x43\x0F\x33\x5A\xD9\x1A\x65\x6C\x22', b'\x3B\xFC\x30\xA6\x88\xEA\x37\xA2\xB4\x8D\x8E\x51\x9C\xD6\x40\xEE', b'\xF9\xF8\x84\xF4\xAE\x97\xE9\xCA\x0A\x45\x67\x57\x04\x2F\x83\x5C', b'\xD5\xC5\xC4\x82\xB6\xA3\x91\x98\x1F\x4A\xAC\x96\x81\x6E\xCB\x1B', b'\x09\x08\xAF\x18\x95\x49\x7D\x54\xED\xFA\x16\x31\x3A\xDA\xB8\x66', b'\xF5\xA5\xF1\xFE\x10\x01\x06\x74\xCC\x63\xDF\x7C\x28\x25\xF6\xCE', b'\xB2\x4F\x8B\xE5\xBC\x87\x69\xBB\x86\x21\x07\x00\x36\xE7\x0B\x50', b'\x59\x9B\x1C\xE8\x62\x58\x19\x61\xF2\xBD\x27\x5E\xBA\x1D\xE6\x99', b'\x42\x3D\x0D\x2A\xB5\xDC\x5B\x29\xF0\x2D\x4C\x53\x7B\x6A\x73\x4E', b'\x3F\x75\xFF\x4B\xA1\x35\x17\x55\x72\x39\x20\xD3\xB0\xFD\xEF\x02', b'\xEC\x77\x7E\xE4\x2B\xDB\x90\xC1\x05\x9E\x7A\xD4\x52\x6B\x24\x13']) inv_s_box = bytearray(1024) for i in range(4): idx = i * 256 for j in range(256): inv_s_box[idx + s_box[idx + j]] = j class AES_V3(): r_con = [[1, 0, 2, 3], [1, 3, 0, 2], [0, 1, 3, 2], [1, 0, 2, 3]] r_con2 = [[1, 0, 2, 3], [2, 0, 3, 1], [0, 1, 3, 2], [1, 0, 2, 3]] r_orders = [[0, 9, 14, 11, 4, 13, 2, 7, 8, 1, 6, 15, 12, 5, 10, 3], [0, 9, 14, 15, 4, 13, 2, 7, 8, 1, 6, 3, 12, 5, 10, 11], [0, 9, 14, 7, 4, 13, 2, 11, 8, 1, 6, 3, 12, 5, 10, 15], [0, 9, 14, 11, 4, 13, 2, 7, 8, 1, 6, 15, 12, 5, 10, 3]] def __init__(self, aes_key, khronos): self.word_size = khronos & 3 self.aes_key = aes_key self.s_box = s_box[self.word_size << 8:] self.inv_s_box = inv_s_box[self.word_size << 8:] self.master_key = self._expand_key() self._key_matrices = bytes2matrix(self.master_key) self.con = self.r_con[self.word_size] self.con2 = self.r_con2[self.word_size] self.order = self.r_orders[self.word_size] def _expand_key(self): init_values = [0xca025ddc, 0x823dc546, 0xc9420583, 0xc298225f] init_value = init_values[self.word_size] mk = bytearray(init_value.to_bytes(4, "little")) mk = mk * 4 mk = xor_bytes(mk, self.aes_key) mk += bytearray(32) rounds = 8 for i in range(4, 12): idx = 4 * (i - 1) k0, k1, k2, k3 = mk[idx], mk[idx + 1], mk[idx + 2], mk[idx + 3] if i & 3 == 0: k00 = (init_value >> (rounds & 24)) ^ self.s_box[k1] k1 = self.s_box[k2] k2 = self.s_box[k3] k3 = self.s_box[k0] k0 = k00 & 0xff rounds += 2 mk[idx + 4] = k0 ^ mk[idx - 12] mk[idx + 5] = k1 ^ mk[idx - 11] mk[idx + 6] = k2 ^ mk[idx - 10] mk[idx + 7] = k3 ^ mk[idx - 9] return mk @staticmethod def sum_data(data): key = bytearray(32) for i in range(31): idx = i * 8 n0 = (data[idx] >> 4) & 2 n1 = n0 | data[idx + 1] & 64 n2 = n1 | (data[idx + 2] >> 2) & 1 n3 = n2 | (data[idx + 3] << 3) & -128 n4 = n3 | (data[idx + 4] >> 1) & 4 n5 = n4 | (data[idx + 5] << 3) & 16 n6 = n5 | (data[idx + 6] << 5) & 32 n7 = n6 | (data[idx + 7] >> 4) & 8 key[i] = (n7 & 0xff) key[31] = 1 return key @staticmethod def mix_columns(data, key): data = bytearray(data) for i in range(31): kk = key[i] idx = i * 8 data[idx + 0] = (data[idx + 0] & -33) | ((kk << 4) & 0xff & 32) data[idx + 1] = (data[idx + 1] & -65) | (kk & 64) data[idx + 2] = (data[idx + 2] & -5) | ((kk * 4) & 4) data[idx + 3] = (data[idx + 3] & -17) | ((kk >> 3) & 16) data[idx + 4] = (data[idx + 4] & -9) | ((kk + kk) & 8) data[idx + 5] = (data[idx + 5] & -3) | ((kk >> 3) & 2) data[idx + 6] = (data[idx + 6] & -2) | ((kk >> 5) & 1) data[idx + 7] = (data[idx + 7] & 127) | ((kk << 4) & 0xff & -128) return data def encrypt(self, data, iv): plaintext = self.sum_data(data) blocks = [] previous = iv for plaintext_block in split_blocks(plaintext): x = xor_bytes(plaintext_block, previous) block = self.encrypt_block(x) blocks.append(block) previous = block key = b''.join(blocks) data = self.mix_columns(data, key) return key[-1:] + data def encrypt_block(self, plaintext): plain_state = bytes2matrix(plaintext) add_round_key_con(plain_state, self._key_matrices[0:4], self.con2) for i in range(1, 3): self.sub_bytes(plain_state) self.shift_rows(plain_state) if i == 1: self.shift_rows_con(plain_state, self.con2) mix_columns(plain_state) add_round_key_con(plain_state, self._key_matrices[i * 4:], self.con2) add_round_key(plain_state, self._key_matrices[4:]) return matrix2bytes(plain_state) def decrypt(self, ciphertext, iv, data): assert len(iv) == 16 blocks = [] previous = iv for ciphertext_block in split_blocks(ciphertext): dc = xor_bytes(previous, self.decrypt_block(ciphertext_block)) blocks.append(dc) previous = ciphertext_block key = b''.join(blocks) data = self.mix_columns(data, key) return data def decrypt_block(self, ciphertext): assert len(ciphertext) == 16 cipher_state = bytes2matrix(ciphertext) add_round_key(cipher_state, self._key_matrices[4:]) for i in range(2, 0, -1): add_round_key_con(cipher_state, self._key_matrices[i * 4:], self.con2) if i == 1: inv_mix_columns(cipher_state) self.shift_rows_con(cipher_state, self.con) self.inv_shift_rows(cipher_state) self.inv_sub_bytes(cipher_state) add_round_key_con(cipher_state, self._key_matrices[0:4], self.con2) return matrix2bytes(cipher_state) def shift_rows_con(self, s, c): for i in range(4): s[i][0], s[i][1], s[i][2], s[i][3] = s[i][c[0]], s[i][c[1]], s[i][c[2]], s[i][c[3]] def shift_rows(self, s): bs = matrix2bytes(s) for i in range(4): for j in range(4): s[i][j] = bs[self.order[i * 4 + j]] def inv_shift_rows(self, s): order = bytearray(16) for i in range(16): order[self.order[i]] = i bs = matrix2bytes(s) for i in range(4): for j in range(4): s[i][j] = bs[order[i * 4 + j]] def sub_bytes(self, s): for i in range(4): for j in range(4): s[i][j] = self.s_box[s[i][j]] s[0], s[1], s[2], s[3] = s[self.con2[0]], s[self.con2[1]], s[self.con2[2]], s[self.con2[3]] def inv_sub_bytes(self, s): for i in range(4): for j in range(4): s[i][j] = self.inv_s_box[s[i][j]] s[0], s[1], s[2], s[3] = s[self.con[0]], s[self.con[1]], s[self.con[2]], s[self.con[3]] SV = [0xa7aefe20, 0x7149f1d6, 0x47e4ca07, 0xe9b58f67, 0x93b924de, 0xc614d0f5, 0x38afe0ef, 0xb2bbad73, 0xe24444c3, 0x9d3aec9b, 0xdf7b37e4, 0xd8b16d40, 0xf8ac31b8, 0x76b9a90b, 0x31d833ee, 0x953fce64, 0x6a6b2b48, 0x8c138276, 0x6d24a010, 0x18da124b, 0xbbeb82ee, 0x39f7ea56, 0x149f4fe1, 0x229946bc, 0x327f309, 0xf4f50e66, 0x62d569aa, 0x78419f92, 0x4db088a7, 0x7430a018, 0xf31d88f4, 0x2f4ed651, 0x9b13fe59, 0x45c5910d, 0x374bf0ec, 0xd5a5b69e, 0xefaceb16, 0x714f4811, 0x4c98233b, 0xc9e6e7b1, 0xd0623b3f, 0xdf5e4f6f, 0xccb31244, 0xaddab856, 0x6faf9c9e, 0x62804e10, 0x6fa7e29d, 0x7e429d51, 0xcdb31e01, 0xb339e419, 0xe497a8a3, 0xebeca6bc, 0x1746cd57, 0xfbf4f54a, 0xd5ea2a8b, 0xb4f02ed2, 0x512b192, 0xefad4d29, 0xd59bec05, 0x2a1436e2, 0xfaea73c1, 0x1ebdbae8, 0x8ff257bd, 0x2d59351d] def leftCircularShift(k, bits): bits = bits % 32 k = k % (2 ** 32) upper = (k << bits) % (2 ** 32) result = upper | (k >> (32 - (bits))) return (result) def blockDivide(block, chunks): result = [] size = len(block) // chunks for i in range(0, chunks): result.append(int.from_bytes(block[i * size:(i + 1) * size], byteorder="little")) return (result) def F(X, Y, Z): return ((X & Y) | ((~X) & Z)) def G(X, Y, Z): return ((X & Z) | (Y & (~Z))) def H(X, Y, Z): return (X ^ Y ^ Z) def I(X, Y, Z): return (Y ^ (X | (~Z))) & 0xffffffff def FF(a, b, c, d, M, s, t): result = b + leftCircularShift((a + F(b, c, d) + M + t), s) return (result) def GG(a, b, c, d, M, s, t): result = b + leftCircularShift((a + G(b, c, d) + M + t), s) return (result) def HH(a, b, c, d, M, s, t): result = b + leftCircularShift((a + H(b, c, d) + M + t), s) return (result) def II(a, b, c, d, M, s, t): result = b + leftCircularShift((a + I(b, c, d) + M + t), s) return (result) def md5sum(msg, n=0): A = 0x79e0f2fb B = 0xc8b52570 C = 0xebc2f8cd D = 0x7c104d93 a = A b = B c = C d = D block = msg[:64] M = blockDivide(block, 16) a = FF(a, b, c, d, M[0], 7, SV[0]) d = FF(d, a, b, c, M[13], 12, SV[1]) c = FF(c, d, a, b, M[14], 17, SV[2]) b = FF(b, c, d, a, M[12], 22, SV[3]) a = FF(a, b, c, d, M[11], 7, SV[4]) d = FF(d, a, b, c, M[10], 12, SV[5]) c = FF(c, d, a, b, M[9], 17, SV[6]) b = FF(b, c, d, a, M[7], 22, SV[7]) a = FF(a, b, c, d, M[8], 7, SV[8]) d = FF(d, a, b, c, M[6], 12, SV[9]) c = FF(c, d, a, b, M[5], 17, SV[10]) b = FF(b, c, d, a, M[4], 22, SV[11]) a = FF(a, b, c, d, M[3], 7, SV[12]) d = FF(d, a, b, c, M[2], 12, SV[13]) c = FF(c, d, a, b, M[1], 17, SV[14]) b = FF(b, c, d, a, M[15], 22, SV[15]) a = GG(a, b, c, d, M[0], 5, SV[16]) d = GG(d, a, b, c, M[5], 9, SV[17]) c = GG(c, d, a, b, M[10], 14, SV[18]) b = GG(b, c, d, a, M[1], 20, SV[19]) a = GG(a, b, c, d, M[12], 5, SV[20]) d = GG(d, a, b, c, M[11], 9, SV[21]) c = GG(c, d, a, b, M[15], 14, SV[22]) b = GG(b, c, d, a, M[2], 20, SV[23]) a = GG(a, b, c, d, M[9], 5, SV[24]) d = GG(d, a, b, c, M[13], 9, SV[25]) c = GG(c, d, a, b, M[3], 14, SV[26]) b = GG(b, c, d, a, M[8], 20, SV[27]) a = GG(a, b, c, d, M[6], 5, SV[28]) d = GG(d, a, b, c, M[7], 9, SV[29]) c = GG(c, d, a, b, M[4], 14, SV[30]) b = GG(b, c, d, a, M[14], 20, SV[31]) a = HH(a, b, c, d, M[8], 4, SV[32]) d = HH(d, a, b, c, M[5], 11, SV[33]) c = HH(c, d, a, b, M[12], 16, SV[34]) b = HH(b, c, d, a, M[13], 23, SV[35]) a = HH(a, b, c, d, M[1], 4, SV[36]) d = HH(d, a, b, c, M[12], 11, SV[37]) c = HH(c, d, a, b, M[7], 16, SV[38]) b = HH(b, c, d, a, M[14], 23, SV[39]) a = HH(a, b, c, d, M[10], 4, SV[40]) d = HH(d, a, b, c, M[0], 11, SV[41]) c = HH(c, d, a, b, M[2], 16, SV[42]) b = HH(b, c, d, a, M[6], 23, SV[43]) a = HH(a, b, c, d, M[4], 4, SV[44]) d = HH(d, a, b, c, M[11], 11, SV[45]) c = HH(c, d, a, b, M[15], 16, SV[46]) b = HH(b, c, d, a, M[3], 23, SV[47]) a = II(a, b, c, d, M[8], 6, SV[48]) d = II(d, a, b, c, M[6], 10, SV[49]) c = II(c, d, a, b, M[15], 15, SV[50]) b = II(b, c, d, a, M[5], 21, SV[51]) a = II(a, b, c, d, M[14], 6, SV[52]) d = II(d, a, b, c, M[9], 10, SV[53]) c = II(c, d, a, b, M[10], 15, SV[54]) b = II(b, c, d, a, M[2], 21, SV[55]) a = II(a, b, c, d, M[2], 6, SV[56]) d = II(d, a, b, c, M[13], 10, SV[57]) c = II(c, d, a, b, M[7], 15, SV[58]) b = II(b, c, d, a, M[12], 21, SV[59]) a = II(a, b, c, d, M[4], 6, SV[60]) d = II(d, a, b, c, M[1], 10, SV[61]) c = II(c, d, a, b, M[11], 15, SV[62]) b = II(b, c, d, a, M[3], 21, SV[63]) A = (A + a) % (2 ** 32) ^ 0x19be4866 B = (B + b) % (2 ** 32) ^ 0xe85986b4 C = (C + c) % (2 ** 32) ^ 0xe19b326e D = (D + d) % (2 ** 32) ^ 0x71d1d7d4 result = bytearray( A.to_bytes(4, "little") + B.to_bytes(4, "little") + C.to_bytes(4, "little") + D.to_bytes(4, "little")) result += sum_md5(result).to_bytes(4, "little") return result def sum_md5(data): check_sum = 0x20220420 for i in range(12): if i % 2 == 0: temp = (check_sum >> 3) ^ check_sum check_sum = data[i] ^ (check_sum << 7) else: temp = (check_sum >> 5) ^ check_sum check_sum = data[i] | (check_sum << 11) check_sum ^= 0xffffffff check_sum ^= temp check_sum &= 0xffffffff check_sum |= 4 check_sum ^= 0x1000000 return check_sum SV2 = [0xa7aefe20, 0x7149f1d6, 0x47e4ca07, 0xe9b58f67, 0x93b924de, 0xc614d0f5, 0x38afe0ef, 0xb2bbad73, 0xe24444c3, 0x9d3aec9b, 0xdf7b37e4, 0xd8b16d40, 0xf8ac31b8, 0x76b9a90b, 0x31d833ee, 0x953fce64, 0x353595a4, 0x4609c13b, 0x36925008, 0x8c6d0925, 0x5df5c177, 0x1cfbf52b, 0x8a4fa7f0, 0x114ca35e, 0x8193f984, 0x7a7a8733, 0x316ab4d5, 0x3c20cfc9, 0xa6d84453, 0x3a18500c, 0x798ec47a, 0x97a76b28, 0x66c4ff96, 0x51716443, 0xdd2fc3b, 0xb5696da7, 0xbbeb3ac5, 0x5c53d204, 0xd32608ce, 0x7279b9ec, 0xf4188ecf, 0xf7d793db, 0x332cc491, 0xab76ae15, 0x9bebe727, 0x18a01384, 0x5be9f8a7, 0x5f90a754, 0x39b663c0, 0x36673c83, 0x7c92f514, 0x9d7d94d7, 0xe2e8d9aa, 0x5f7e9ea9, 0x7abd4551, 0x569e05da, 0x40a25632, 0x3df5a9a5, 0xbab37d80, 0x454286dc, 0x3f5d4e78, 0x3d7b75d, 0xb1fe4af7, 0xa5ab26a3] def md5sum_v3(msg, count_v2, orders, count_v1, n=0): count = count_v2 & 0xff sv = [0] * 64 for i in range(64): sv[i] = ror32(SV2[i], count_v1) start = [ ror32(0x79e0f2fb, count), ror32(0xc8b52570, count), ror32(0xebc2f8cd, count), ror32(0x7c104d93, count) ] count = (count_v2 + 6) & 0xff end = [ ror32(0x19be4866, count), ror32(0xe85986b4, count), ror32(0xe19b326e, count), ror32(0x71d1d7d4, count) ] A = start[0] B = start[1] C = start[2] D = start[3] a = A b = B c = C d = D block = msg[:64] M = blockDivide(block, 16) order1 = orders[:16] order2 = orders[16:32] order3 = orders[32:48] order4 = orders[48:] a = FF(a, b, c, d, M[order1[0]], 7, sv[0]) d = FF(d, a, b, c, M[order1[1]], 12, sv[1]) c = FF(c, d, a, b, M[order1[2]], 17, sv[2]) b = FF(b, c, d, a, M[order1[3]], 22, sv[3]) a = FF(a, b, c, d, M[order1[4]], 7, sv[4]) d = FF(d, a, b, c, M[order1[5]], 12, sv[5]) c = FF(c, d, a, b, M[order1[6]], 17, sv[6]) b = FF(b, c, d, a, M[order1[7]], 22, sv[7]) a = FF(a, b, c, d, M[order1[8]], 7, sv[8]) d = FF(d, a, b, c, M[order1[9]], 12, sv[9]) c = FF(c, d, a, b, M[order1[10]], 17, sv[10]) b = FF(b, c, d, a, M[order1[11]], 22, sv[11]) a = FF(a, b, c, d, M[order1[12]], 7, sv[12]) d = FF(d, a, b, c, M[order1[13]], 12, sv[13]) c = FF(c, d, a, b, M[order1[14]], 17, sv[14]) b = FF(b, c, d, a, M[order1[15]], 22, sv[15]) a = GG(a, b, c, d, M[order2[0]], 5, sv[16]) d = GG(d, a, b, c, M[order2[1]], 9, sv[17]) c = GG(c, d, a, b, M[order2[2]], 14, sv[18]) b = GG(b, c, d, a, M[order2[3]], 20, sv[19]) a = GG(a, b, c, d, M[order2[4]], 5, sv[20]) d = GG(d, a, b, c, M[order2[5]], 9, sv[21]) c = GG(c, d, a, b, M[order2[6]], 14, sv[22]) b = GG(b, c, d, a, M[order2[7]], 20, sv[23]) a = GG(a, b, c, d, M[order2[8]], 5, sv[24]) d = GG(d, a, b, c, M[order2[9]], 9, sv[25]) c = GG(c, d, a, b, M[order2[10]], 14, sv[26]) b = GG(b, c, d, a, M[order2[11]], 20, sv[27]) a = GG(a, b, c, d, M[order2[12]], 5, sv[28]) d = GG(d, a, b, c, M[order2[13]], 9, sv[29]) c = GG(c, d, a, b, M[order2[14]], 14, sv[30]) b = GG(b, c, d, a, M[order2[15]], 20, sv[31]) a = HH(a, b, c, d, M[order3[0]], 4, sv[32]) d = HH(d, a, b, c, M[order3[1]], 11, sv[33]) c = HH(c, d, a, b, M[order3[2]], 16, sv[34]) b = HH(b, c, d, a, M[order3[3]], 23, sv[35]) a = HH(a, b, c, d, M[order3[4]], 4, sv[36]) d = HH(d, a, b, c, M[order3[5]], 11, sv[37]) c = HH(c, d, a, b, M[order3[6]], 16, sv[38]) b = HH(b, c, d, a, M[order3[7]], 23, sv[39]) a = HH(a, b, c, d, M[order3[8]], 4, sv[40]) d = HH(d, a, b, c, M[order3[9]], 11, sv[41]) c = HH(c, d, a, b, M[order3[10]], 16, sv[42]) b = HH(b, c, d, a, M[order3[11]], 23, sv[43]) a = HH(a, b, c, d, M[order3[12]], 4, sv[44]) d = HH(d, a, b, c, M[order3[13]], 11, sv[45]) c = HH(c, d, a, b, M[order3[14]], 16, sv[46]) b = HH(b, c, d, a, M[order3[15]], 23, sv[47]) a = II(a, b, c, d, M[order4[0]], 6, sv[48]) d = II(d, a, b, c, M[order4[1]], 10, sv[49]) c = II(c, d, a, b, M[order4[2]], 15, sv[50]) b = II(b, c, d, a, M[order4[3]], 21, sv[51]) a = II(a, b, c, d, M[order4[4]], 6, sv[52]) d = II(d, a, b, c, M[order4[5]], 10, sv[53]) c = II(c, d, a, b, M[order4[6]], 15, sv[54]) b = II(b, c, d, a, M[order4[7]], 21, sv[55]) a = II(a, b, c, d, M[order4[8]], 6, sv[56]) d = II(d, a, b, c, M[order4[9]], 10, sv[57]) c = II(c, d, a, b, M[order4[10]], 15, sv[58]) b = II(b, c, d, a, M[order4[11]], 21, sv[59]) a = II(a, b, c, d, M[order4[12]], 6, sv[60]) d = II(d, a, b, c, M[order4[13]], 10, sv[61]) c = II(c, d, a, b, M[order4[14]], 15, sv[62]) b = II(b, c, d, a, M[order4[15]], 21, sv[63]) A = (A + a) % (2 ** 32) ^ end[0] B = (B + b) % (2 ** 32) ^ end[1] C = (C + c) % (2 ** 32) ^ end[2] D = (D + d) % (2 ** 32) ^ end[3] result = bytearray( A.to_bytes(4, "little") + B.to_bytes(4, "little") + C.to_bytes(4, "little") + D.to_bytes(4, "little")) result += sum_md5(result).to_bytes(4, "little") return result def bxor(b1, b2): b3 = bytearray(len(b1)) for i in range(len(b1)): b3[i] = b1[i] ^ b2[i] return b3 def get_iv(iv, data): for i in range(len(data)): if i & 1 == 0: iv = (iv >> 4) ^ iv ^ (iv << 6) ^ data[i] else: iv = ~((iv >> 7) ^ iv ^ (data[i] | iv << 12)) iv = iv & 0xffffffff return iv def hash_f13(query_sm3, body_md5_bytes, ts_bytes, khronos): iv = get_iv(0x20230928, query_sm3) iv = get_iv(iv, body_md5_bytes) iv = get_iv(iv, ts_bytes) iv_v0 = ((iv & 15) * 171) >> 9 branch = (iv & 15) - ((iv_v0 * 3) & 0xff) if branch == 0: return branch_0(iv_v0, khronos, query_sm3, body_md5_bytes, ts_bytes) elif branch == 1: return branch_1(iv_v0, khronos, query_sm3, body_md5_bytes, ts_bytes) elif branch == 2: return branch_2(iv, khronos, query_sm3, body_md5_bytes, ts_bytes) else: raise Exception("no branch: " + str(branch)) def branch_0(iv_v0, khronos, query_sm3, body_md5_bytes, ts_bytes): tt01 = [0xc4a78580, 0xb3c0fd39, 0xc58c5686, 0xc9aa3ba7, 0xf5a7adf2, 0x963c2ed1] iv_v1 = tt01[iv_v0] count_v1 = (iv_v1 + khronos + 1) & 0xff count_v2 = (iv_v1 + khronos) & 0xffffffff tt02 = [0xebb64faf, 0x7aadcc2, 0xcf3187bf, 0xe01138ff, 0x6d0bfcff, 0x5a30a3be, 0xb41ad638, 0x34180eb8, 0xf233eb6f, 0xb1a584cc, 0xccc30dc7, 0x47d1db51, 0xd55653de, 0x70a84fa1, 0x57473c12, 0xf76f0288, 0x2c077f0a, 0xda0dcad0, 0xfbb86f6c, 0xfdc4cf00, 0x688a020d, 0xe676c6a6, 0x8cd6338b, 0x1a3c8d0e, 0xcce8b06b, 0x6ad0ed0b, 0xa0522717, 0xdc71ac83, 0x2285db71, 0xd5b4dda6, 0x736f8650, 0x6560306c, 0x617ce2a6, 0xe423417e, 0xa40e143, 0x544e4032, 0x88dffb2a, 0x716c1ae0, 0x4c467a88, 0x5b23bb3, 0xe1d0b866, 0xbaa3dcb8, 0xae3374d3, 0xc3381a50, 0x1702f75b, 0xfe6da368, 0xf0b4cf48, 0x4e0ffbb8, 0x72aad10d, 0x26c53a3d, 0xf2bce0f6, 0xb4557581, 0x4a257fdd, 0x8c3182a2, 0xab0b3b86, 0x3d5dfb14, 0x4f103634, 0xd37b52d7, 0x444eff16, 0xeb0a33d1, 0x6ca86f6e, 0x284ba7, 0x8387cfa, 0x5fb37586] tt03 = [0] * 64 for i in range(0, 64): tt03[i] = ror32(tt02[i], count_v1) & 0xffffffff n0 = (count_v2 + 2) & 7 pad = bytearray(4) seed = bytes([0xfa, 0x45, 0x61, 0xd7]) for i in range(4): v = int.from_bytes(bytes([seed[i], seed[i]]), 'little') v = v >> n0 pad[i] = v & 0xff count_v2 = count_v2 & 0xff init_value = [ ror32(0x7aba4fc8, count_v2), ror32(0x67166507, count_v2), ror32(0x6403fa00, count_v2), ror32(0x340f512f, count_v2), ror32(984304912, count_v2), ror32(3005047866, count_v2), ror32(2874125293, count_v2), ror32(2152413264, count_v2) ] data = query_sm3 + body_md5_bytes + ts_bytes + pad + bytes.fromhex(' 00 00 00 00 00 00 01 a0') di = [0] * (len(data) // 4) for i in range(len(data) // 4): di[i] = int.from_bytes(data[i * 4:i * 4 + 4], "big") di0 = di[0] for i in range(112): di1, di14 = di[i + 1], di[i + 14] r_di1 = rol(di1, 14) ^ rol(di1, 25) ^ (di1 >> 3) r_di2 = rol(di14, 13) ^ rol(di14, 15) ^ (di14 >> 10) di0 = di0 + di[i + 9] + r_di1 + r_di2 di.append(di0 & 0xffffffff) di0 = di1 if iv_v0 == 5: v_j = branch0_xor(init_value, iv_v1, di, tt03, 100, 2, 0, 3, 5, 4, 6, 7, 2, 1, 5) elif iv_v0 == 4: v_j = branch0_xor(init_value, iv_v1, di, tt03, 96, 0, 5, 6, 7, 3, 1, 2, 5, 4, 4) elif iv_v0 == 3: v_j = branch0_xor(init_value, iv_v1, di, tt03, 99, 3, 6, 2, 4, 5, 1, 0, 0, 7, 6) elif iv_v0 == 2: v_j = branch0_xor(init_value, iv_v1, di, tt03, 96, 7, 6, 2, 1, 4, 0, 5, 4, 3, 5) elif iv_v0 == 1: v_j = branch0_xor(init_value, iv_v1, di, tt03, 96, 0, 6, 7, 5, 3, 2, 1, 5, 4, 4) elif iv_v0 == 0: v_j = branch0_xor(init_value, iv_v1, di, tt03, 101, 5, 7, 6, 3, 2, 1, 0, 5, 4, 3) ret = bytearray(32) for i in range(8): ret[i * 4:i * 4 + 4] = ((v_j[i] + init_value[i]) & 0xffffffff).to_bytes(4, 'big') ret = bxor(ret[:16], ret[16:]) sum = sum_md5(ret) ret += sum.to_bytes(4, "little") return ret def swap_v0(src, src_xor, tt2, table_f, order1, typ=None): da0 = [0] * 8 ha0 = src_xor[:] for i in range(8): da0[i] = src[order1[i]] ^ src_xor[i] for round in range(10): rr_0 = r00(ha0[0], ha0[1], ha0[2], ha0[3], ha0[4], ha0[5], ha0[6], ha0[7], 0, table_f) rr_1 = r00(ha0[1], ha0[2], ha0[3], ha0[4], ha0[5], ha0[6], ha0[7], ha0[0], 0, table_f) rr_2 = r00(ha0[2], ha0[3], ha0[4], ha0[5], ha0[6], ha0[7], ha0[0], ha0[1], 0, table_f) rr_3 = r00(ha0[3], ha0[4], ha0[5], ha0[6], ha0[7], ha0[0], ha0[1], ha0[2], 0, table_f) rr_4 = r00(ha0[4], ha0[5], ha0[6], ha0[7], ha0[0], ha0[1], ha0[2], ha0[3], 0, table_f) rr_5 = r00(ha0[5], ha0[6], ha0[7], ha0[0], ha0[1], ha0[2], ha0[3], ha0[4], 0, table_f) rr_6 = r00(ha0[6], ha0[7], ha0[0], ha0[1], ha0[2], ha0[3], ha0[4], ha0[5], 0, table_f) rr_7 = r00(ha0[7], ha0[0], ha0[1], ha0[2], ha0[3], ha0[4], ha0[5], ha0[6], 0, table_f) rr_7 = rr_7 ^ tt2[round + 1] d0, d1, d2, d3, d4, d5, d6, d7 = da0[0], da0[1], da0[2], da0[3], da0[4], da0[5], da0[6], da0[7] da0[0] = r00(d0, d1, d2, d3, d4, d5, d6, d7, rr_0, table_f) da0[1] = r00(d1, d2, d3, d4, d5, d6, d7, d0, rr_1, table_f) da0[2] = r00(d2, d3, d4, d5, d6, d7, d0, d1, rr_2, table_f) da0[3] = r00(d3, d4, d5, d6, d7, d0, d1, d2, rr_3, table_f) da0[4] = r00(d4, d5, d6, d7, d0, d1, d2, d3, rr_4, table_f) da0[5] = r00(d5, d6, d7, d0, d1, d2, d3, d4, rr_5, table_f) da0[6] = r00(d6, d7, d0, d1, d2, d3, d4, d5, rr_6, table_f) da0[7] = r00(d7, d0, d1, d2, d3, d4, d5, d6, rr_7, table_f) ha0[0], ha0[1], ha0[2], ha0[3], ha0[4], ha0[5], ha0[6], ha0[7] = rr_0, rr_1, rr_2, rr_3, rr_4, rr_5, rr_6, rr_7 if typ is None: src[0] = da0[0] ^ src_xor[0] ^ src[0] src[1] = da0[7] ^ src_xor[1] ^ src[1] src[2] = da0[6] ^ src_xor[2] ^ src[2] src[3] = da0[5] ^ src_xor[3] ^ src[3] src[4] = da0[4] ^ src_xor[4] ^ src[4] src[5] = da0[3] ^ src_xor[5] ^ src[5] src[6] = da0[2] ^ src_xor[6] ^ src[6] src[7] = da0[1] ^ src_xor[7] ^ src[7] else: src[0] = da0[7] ^ typ[1] src[1] = da0[6] ^ typ[2] src[2] = da0[5] ^ typ[3] src[3] = da0[4] ^ typ[4] src[4] = da0[3] ^ typ[5] src[5] = da0[2] ^ typ[6] src[6] = da0[1] ^ typ[7] ^ src[7] src[7] = da0[0] ^ typ[0] return src def branch_1(iv_v0, khronos, query_sm3, body_md5_bytes, ts_bytes): tt1 = [0x808a9c79, 0xf079807e, 0xbadf79c5, 0xa785d3ff, 0x82d8438c] iv_v1 = tt1[iv_v0] c_v1 = (iv_v1 + khronos) & 0xff c_v1 = ror(khronos, c_v1) orders = bytes.fromhex('''05 07 01 02 04 00 06 03 00 05 02 04 01 03 07 06 05 07 02 04 01 06 03 00 03 00 02 04 06 07 01 05 04 05 00 03 06 02 01 07 00 00 00 00 00 00 00 00''') order1 = orders[iv_v0 * 8:iv_v0 * 8 + 8] iv_v1 = (iv_v1 + khronos + 1) & 63 tt1 = [0x87aeea5dab37cd6b, 0x7ff48becb4f54087, 0xb0724c06706bbd5d, 0x1fe5dfb1143e328d, 0x1a2331d00af4f1f2, 0xcaff7131bb1e71ba, 0x33385e1042752218, 0xff01ed65d4a441fb, 0xadb1ec8828c80e8, 0x62475d12f4e06fe7, 0xbd0b238da4fe72] tt2 = [0] * 11 for i in range(0, 11): tt2[i] = ror(tt1[i], iv_v1) to_sign = bytearray(query_sm3 + body_md5_bytes + ts_bytes + bytes([0x80, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0])) src = [0] * (len(to_sign) // 8) for i in range(len(to_sign) // 8): src[i] = int.from_bytes(to_sign[i * 8:i * 8 + 8], 'big') src_xor = [c_v1, c_v1, c_v1, c_v1, c_v1, c_v1, c_v1, c_v1] table_f = get_branch_1_table_f(iv_v0) swap = swap_v0(src, src_xor, tt2, table_f, order1) data = [0] * 8 data[7] = 416 src_xor = [swap[4], swap[3], swap[2], swap[1], swap[0], swap[7], swap[6], swap[5]] swap = swap_v0(data, src_xor, tt2, table_f, order1, swap) ret = bytearray(32) n = 0 for i in range(7, -1, -1): pp = swap[i].to_bytes(8, "little") ret[n + 0], ret[n + 1], ret[n + 2], ret[n + 3] = pp[1], pp[3], pp[5], pp[6] n += 4 for i in range(16): ret[i] ^= ret[i + 16] ret = ret[:16] sum = sum_md5(ret) ret += sum.to_bytes(4, "little") return ret def r00(r0, r1, r2, r3, r4, r5, r6, r7, tt, table_f, v=0): x = table_f(8 * (r0 >> 56)) r1 = (r1 >> 45) & 2040 if tt > 0: x = x ^ tt x = x ^ table_f(r1 + 2048) r2 = (r2 >> 37) & 2040 x = x ^ table_f(r2 + 4096) r3 = (r3 >> 29) & 2040 x = x ^ table_f(r3 + 6144) r4 = (r4 >> 21) & 2040 x = x ^ table_f(r4 + 8192) r5 = (r5 >> 13) & 2040 x = x ^ table_f(r5 + 10240) r6 = (r6 >> 8) & 255 x = x ^ table_f(8 * r6 + 12288) r7 = r7 & 255 x = x ^ table_f(8 * r7 + 14336) return x branch_2_orders = bytes.fromhex(''' 0f 07 04 00 09 08 03 0a 06 0b 05 0d 0e 01 0c 02 0f 05 08 0c 00 09 02 01 03 07 0e 06 0b 0a 0d 04 06 05 00 07 0c 00 0a 04 08 0f 01 0b 0d 09 02 0e 06 0b 02 05 04 03 08 01 01 07 0a 00 0d 0c 09 0e 0d 07 0e 0f 0b 02 08 03 0c 05 09 01 00 04 06 0a 0d 09 02 06 0f 0b 0a 04 08 07 00 0c 05 03 01 0e 0c 09 0f 07 06 0f 03 0e 02 0d 04 05 01 0b 0a 00 0c 05 0a 09 0e 08 02 04 04 07 03 0f 01 06 0b 00 0b 0f 04 08 02 0a 07 00 09 0d 06 01 0e 03 05 0c 0b 06 0a 05 08 02 0c 03 07 0f 0e 09 0d 00 01 04 09 06 08 0f 05 08 00 04 0a 0b 03 0d 01 02 0c 0e 09 0d 0c 06 04 07 0a 03 03 0f 00 08 01 05 02 0e 01 00 0d 0f 09 0a 0b 0e 04 02 08 07 03 06 0c 05 01 08 0a 0c 0f 09 05 06 0b 00 03 04 02 0e 07 0d 04 08 0f 00 0c 0f 0e 0d 0a 01 06 02 07 09 05 03 04 02 05 08 0d 0b 0a 06 06 00 0e 0f 07 0c 09 03 0a 08 04 0f 00 0b 01 06 0d 0c 07 09 03 0e 05 02 0a 07 0b 05 0f 00 02 0e 01 08 03 0d 0c 06 09 04 0d 07 0f 08 05 0f 06 04 0b 0a 0e 0c 09 00 02 03 0d 0c 02 07 04 01 0b 0e 0e 08 06 0f 09 05 00 03 ''') def branch_2(iv, khronos, query_sm3, body_md5_bytes, ts_bytes): n0 = (iv & 15 - 2) * 86 iv_v0 = (n0 >> 15 & 0xff) + (n0 >> 8 & 0xff) t001 = [0x8980f29b, 0xeb549c7f, 0xb08726db, 0xd40cb5e6, 0xe8f559e4] n1 = t001[iv_v0] count_v1 = (n1 + khronos + 1) & 0xff count_v2 = (n1 + khronos) & 0xffffffff n0 = (count_v2 + 5) & 7 pad = bytearray(8) seed = bytes([0x84, 0x96, 0x77, 0x9d, 0xd4, 0x15, 0x0b, 0xf8]) for i in range(8): v = int.from_bytes(bytes([seed[i], seed[i]]), 'little') v = v >> n0 pad[i] = v & 0xff data = query_sm3 + body_md5_bytes + ts_bytes + pad + bytes.fromhex('a0 01 00 00') idx = iv_v0 << 6 ret = md5sum_v3(data, count_v2, branch_2_orders[idx:idx + 64], count_v1) return ret def branch0_xor(data, base, di, tt03, round, x1, x2, x3, x4, x5, x6, x7, x8, x9, x10): d = data[:] for i in range(round): offset = base + i n0 = di[offset & 127] n1 = ((d[x3] ^ d[x4]) & d[x1]) ^ d[x3] n2 = rol(d[x1], 26) ^ rol(d[x1], 21) ^ rol(d[x1], 7) offset = offset & 63 n3 = tt03[offset] n4 = (n0 + n1 + n2 + n3 + d[x5]) & 0xffffffff n5 = rol(d[x2], 30) ^ rol(d[x2], 19) ^ rol(d[x2], 10) n6 = (d[x2] & d[x6]) | ((d[x2] | d[x6]) & d[x7]) n7 = n5 + n6 o = d[x9] d[0], d[1], d[2], d[3], d[4], d[5], d[6], d[7] = d[7], d[0], d[1], d[2], d[3], d[4], d[5], d[6] d[x10] = (n7 + n4) & 0xffffffff d[x8] = (o + n4) & 0xffffffff return d _BR1_RAW = None def get_branch_1_table_f(iv_v0): global _BR1_RAW if _BR1_RAW is None: _BR1_RAW = bytes([item for sublist in _get_branch_one_table() for item in sublist]) table = _BR1_RAW[iv_v0 << 14:] def table_f(x): return int.from_bytes(table[x:x + 8], 'little') return table_f import hashlib def rc4_xg(data, key): S = list(range(256)) j = 0 for i in range(256): j = (j + S[i] + key[i % len(key)]) % 256 S[i] = S[j] i = j = 0 result = bytearray(len(data)) for k, v in enumerate(data): i += 1 x = S[i] j += x y = S[j & 0xff] S[i] = y result[k] = v ^ S[(y + y) & 0xff] return result def reverse_bits(num): bin_num = bin(num)[2:].zfill(8) rev_bin_num = bin_num[::-1] rev_num = int(rev_bin_num, 2) return rev_num gorgon_84 = bytes([0x4a, 0x16, 0x47, 0x6c, 0x84, 0x04]) def encrypt_gorgon(body, query, khronos, xg_rand, dataType): xg_seed = 320 body_md5 = xssstub_hash_md5_hex(data=body, dataType=dataType).lower() if body else "" data = hashlib.md5(query.encode()).digest()[:4] if len(body_md5) > 0: body_md5 = bytes.fromhex(body_md5) data += body_md5[:4] else: data += bytearray([0, 0, 0, 0]) data += bytearray([0, 0, 0, 0]) mssdkVersionInt = 67503104 data += mssdkVersionInt.to_bytes(4, "little") data += khronos.to_bytes(4, "big") key = bytes([ gorgon_84[0], 320 & 0xff, gorgon_84[1], (xg_rand >> 8) & 0xff, gorgon_84[2], gorgon_84[3], (320 >> 8) & 0xff, xg_rand & 0xff, ]) out = rc4_xg(data, key) for i in range(len(out)): a = out[i] out[i] = (a >> 4 | (a << 4)) & 0xFF a = out[0] if i + 1 < len(out): a = out[i + 1] a ^= out[i] a = reverse_bits(a) out[i] = (~(a ^ 20)) & 0xFF ret = gorgon_84[-2:] ret += xg_rand.to_bytes(2, "little") ret += xg_seed.to_bytes(2, "little") ret += out return ret.hex() import base64 import hashlib import random def ror(value, count): count %= 64 low = value << (64 - count) value >>= count value |= low value &= 0xFFFFFFFFFFFFFFFF return value HexTable = b"0123456789abcdef" def _helios_bytes2matrix(text): return [int.from_bytes(list(text[i:i + 8]), 'little') for i in range(0, len(text), 8)] def encrypt_helios_input(hash_table, in_data): data0 = int.from_bytes(in_data[:8], 'little') data1 = int.from_bytes(in_data[8:], 'little') for i in range(0, 0x22): hash1 = hash_table[i] data1 = hash1 ^ (data0 + ror(data1, 8)) data1 &= 0xFFFFFFFFFFFFFFFF data0 = data1 ^ ror(data0, 61) data0 &= 0xFFFFFFFFFFFFFFFF return data0.to_bytes(8, 'little') + data1.to_bytes(8, 'little') def encrypt_helios(khronos, rand=0): rand = rand or int(random.randint(0, 0xFFFFFFFF)) data = rand.to_bytes(4, "little") data += str(8662).encode() key_sum = hashlib.md5(data).digest() keys = bytearray(32) for i in range(16): v1 = key_sum[i] keys[2 * i] = HexTable[v1 >> 4] keys[2 * i + 1] = HexTable[v1 & 15] hash_table = [] hash_table.append(int.from_bytes(keys[:8], "little")) keys = _helios_bytes2matrix(keys) buffer_b0 = keys[0] buffer_b8 = keys[1] keys.pop(0) keys.pop(0) for i in range(0, 0x22): x9 = buffer_b0 x8 = buffer_b8 x8 = ror(x8, 8) x8 = x8 + x9 x8 = (x8 ^ i) & 0xFFFFFFFFFFFFFFFF keys.append(x8) x8 = x8 ^ ror(x9, 61) x8 &= 0xFFFFFFFFFFFFFFFF hash_table.append(x8) buffer_b0 = x8 buffer_b8 = keys[0] keys.pop(0) in_data = _pkcs7_pad(bytes(f"{khronos}-1588093228-8662", 'utf-8'), 16) out = bytearray() for i in range(len(in_data) // 16): out += encrypt_helios_input(hash_table, in_data[i * 16:i * 16 + 16]) out = data[:4] + out return base64.b64encode(out).decode() import base64 import random import time def xmxor(data, key): data = xmxor_two(data, key) last_flag = data[-1] ^ data[-2] data0 = data[0] data[0] = (~last_flag + data[0]) & 0xff data[1] = ((data[0] ^ data[-1] ^ 254) + data[1]) & 0xff data[2] = (data[2] + ((last_flag - data0) ^ rl8(data[1], 3) ^ 2)) & 0xff for i in range(len(data) - 4): temp = (rl8(data[i + 2], 3)) ^ data[i + 1] ^ (i + 3) data[i + 3] = (~temp + data[i + 3]) & 0xff data[-1] ^= data[-2] sum = 0 for i in range(len(data) - 1): sum += int(data[i + 1]) data[0] = ((data[0] ^ data[1]) + sum) & 0xff return data def xmxor_two(data, key): enc_data = bytearray(len(data)) for i in range(len(data)): index = (i * 4) & 28 d0 = key[index] d1 = key[index + 1] d2 = (rl8(data[i], 4) + d0) ^ d1 d2 = ~d2 d2 = (rl8(d2 & 0xff, 3)) & 0xff d2 += d1 d2 = (d2 & 0xff) ^ d0 enc_data[-i - 1] = ~d2 & 0xff return enc_data def gen_medusa(url, url_params, devices, data:dict, khronos:int=None, lanusk=None, hash_rand=0, xm_rand=0, dataType=None): config = { 'fix': bytes([0x35]), 'aes_key': b'\xf1Y3vvn\xa9\x8d4\xf3\x1b\x05z\x9d[\xe4', 'aes_iv': b'\x1f\xe1\t\xa4\x12R\x83\xf4\x18\xde\x9e\x05\x1a\x96\x9e\x12', 'signKey': b'\x8e\xbd\xfa8\x06\xec\xc5\xce\xe7\x94#\xe6\x02\x9e\xd8%@\xbc"\x18\xbb~\xae\xf7\x1c\xb6\x91\xf7\xaa\x8a\xa2\xf5', } data, query_sm3, body_sm3 = gen_medusa_proto(url, url_params, devices, data, khronos, lanusk, dataType) hash_rand = hash_rand or int(random.randint(0, 0xFFFFFFFF)) xm_rand = xm_rand or int(random.randint(0, 0xFFFFFFFF)) key, seed = get_key_hash(config['signKey'], hash_rand) data = xmxor(data, key) xg_seed = 320 data = xg_seed.to_bytes(8, "little") + data data = bytearray(data[::-1]) for i in range(0, len(data)): data[i] = data[i] ^ seed[~i & 3] hash_rand_bytes = hash_rand.to_bytes(4, "little") check_bit = (query_sm3[0] & 63) << 14 check_bit |= 0x18000001 check_bit |= (body_sm3[0] & 63) << 8 data = config['fix'] + xm_rand.to_bytes(4, 'little') + check_bit.to_bytes(4, 'little') + data + hash_rand_bytes[2:] data = AES_V3(config['aes_key'], khronos).encrypt(data, config['aes_iv']) version_or = bytes() version = bytes.fromhex("03 00 00 00 f7 e8 5f fa d7 d7 dc 3b d6 2a c8 70 57 cf 61 18 ") for i in range(0, 20, 4): d = int.from_bytes(version[i:i + 4], "little") d ^= khronos version_or += d.to_bytes(4, "little") data = version_or + hash_rand_bytes[:2] + int(256).to_bytes(2, "little") + data return base64.b64encode(data).decode() def gen_medusa_proto(url, url_params, devices, data:dict, khronos:int=None, lanusk=None, dataType=None): xg_seed = 320 body_md5 = xssstub_hash_md5_hex(data=data, dataType=dataType).lower() if data else "" if body_md5 != '': body_md5_bytes = bytes.fromhex(body_md5) else: body_md5_bytes = bytes(16) ts_bytes = khronos.to_bytes(4, "little") query_sm3 = SM3(url.split('?')[1]).digest() if len(lanusk) > 0: lanusk_sm3 = SM3(bytes.fromhex(lanusk) + ts_bytes).digest() lanusk_hash = md5sum( lanusk_sm3 + body_md5_bytes + bytes.fromhex("84 96 77 9d db 6d bc b6 d4 15 0b f8 80 01 00 00")) lanusk_hash = lanusk_hash psk_version = "1" query_body_hash_sm3 = SM3(url.split('?')[1].encode() + body_md5_bytes + b"1").digest() else: lanusk_hash = b'' psk_version = 'none' query_body_hash_sm3 = SM3(url.split('?')[1].encode() + body_md5_bytes + b"none").digest() proto_data = Medusa( magic=bytearray(b'\xf7\xe8_\xfa\xd7\xd7\xdc;\xd6*\xc8pW\xcfa\x18'), version=3, rand=int(random.randint(0, 0xFFFFFFFF)), ms_app_id='8662', device_id=str(devices.get("device_id", url_params.get("device_id", url_params.get("did", "")))), license_id='1588093228', app_version=devices.get("version_name", url_params.get("version_name")), sdk_version_str='v04.06.04-ml-android', sdk_version=67503104, xg_seed_bytes=xg_seed.to_bytes(8, "little"), time=khronos, query_body_ts_hash=hash_f13(query_sm3, body_md5_bytes, ts_bytes, khronos), query_sm3=bytearray(query_sm3[:6]), request=MedushaAlgorithmCount(sign_count=111, report_count=10, setting_count=694367, unknown4=0, unknown5=586952199), sec_device_token='AXYQOS6n2m60x1fVZHIrH3iol', time2=khronos, lanusk_hash=lanusk_hash, query_body_hash_sm3=query_body_hash_sm3, psk_version=psk_version, call_type=312, env=Env( launch_time=random.randint(100, 120), unknown2=146331399, unknown3=146331396, unknown5=7, version='v04.06.04.03-bugfix', pid=random.randint(10001, 12000), device=Device( d1=1, collect_stat=2, aid='8662', device_id=str(devices.get("device_id", url_params.get("device_id", url_params.get("did", "")))), sec_device_token='Ai6svO3PyrwDOUSmO6ZcResxu', app_version='!noperm!', battery=-888888, battery2=-888888, battery_health=3, battery_changed=-888888, network='!notset!', tz='Asia/Shanghai,8', lan='zh_CN', cpu=4, sdcard=255.24993896484375, sdcard_used=35.58599090576172, memory=3.467449188232422, memory2=3.467449188232422, data=255.1754913330078, data_used=42.17544174194336, os_version=devices.get("os_version", url_params.get("os_version")), brightness=41, volume=36, ts=1728388016635, ts2=1728388016635, ts3=1728388016635, ts4=1728388016637, usb=-1, hw_version=devices.get("device_model", url_params.get("device_type")), brand=devices.get("device_brand", url_params.get("device_brand")), board=devices.get("device_model", url_params.get("device_type")), product_name=devices.get("device_model", url_params.get("device_type")), product_device=devices.get("device_manufacturer", devices.get("device_brand", url_params.get("device_brand"))), product_manufacturer=devices.get("device_brand", url_params.get("device_brand")), hardware=devices.get("device_brand", url_params.get("device_brand")), unknown38=31 ), report=Report( time=devices.get("report_time", int(time.time())), state=-2, code=200, times=0, unknown6=0 ), app_version=devices.get("version_name", url_params.get("version_name")) ), unknown24='{"cmr":16777216,"cmr2":16777216,"un_h":1879194040,"vpn":0,"kd":0,"fkd":3672518972,"pd":-1872573247,"dyn":"","do":0,"tk":true}' ) return bytes(proto_data), query_sm3, proto_data.query_body_ts_hash import base64 import json import random import time def lowerHeader(header: dict): new_header = {} for key, value in header.items(): new_header[str(key).lower()] = value header.clear() header.update(new_header) return new_header def core_sixgod(surl, params, devices, data:dict={}, common=None, header=None, lanusk="", log=False, rticket_override=None, ts_override=None, khronos_override=None): if devices: for k, v in devices.items(): if k in data: data[k] = v dataType = header.get("content-type", header.get("Content-Type")) xg_rand = int(random.randint(0, 0xFFFF)) url, url_params, x_common = get_params_encrypturl(surl, params=params, devices=devices, common=common, rticket_override=rticket_override, ts_override=ts_override) khronos = khronos_override if khronos_override is not None else int(time.time()) params = EecryptParams() params.khronos = str(khronos) params.ladon = base64.b64encode(khronos.to_bytes(4, 'big')).decode() params.argus = base64.b64encode(khronos.to_bytes(4, 'little')).decode() params.gorgon = encrypt_gorgon(data, url_encode(url_params), khronos, xg_rand, dataType) params.helios = encrypt_helios(khronos, rand=0) params.medusa = gen_medusa( url, url_params, devices, data, khronos, lanusk, dataType=dataType ) six = result(params, data, url, common=x_common, dataType=dataType, log=log) headers = header.copy() lowerHeader(header=headers) headers.update(six.get("sign_header")) if devices: headers["x-tt-dt"] = devices.get("x_tt_dt", "") headers["user-agent"] = devices.get("ua", "") return headers, six.get("sign_url") def result(params: EecryptParams, data, eurl, common=None, sell=False, dataType=None, log=False): six = { 'x-ladon': params.ladon, 'x-khronos': params.khronos, 'x-argus': params.argus, 'x-gorgon': params.gorgon, 'x-helios': params.helios, 'x-medusa': params.medusa } if data: six["x-ss-stub"] = xssstub_hash_md5_hex(data=data, dataType=dataType) if sell: datas: dict = dict() datas["api"] = eurl datas["sign"] = six if common: datas["sign"]["x-common-params-v2"] = common return datas datas: dict = dict() datas["sign_url"] = eurl datas["sign_header"] = six if common: datas["sign_header"]["x-common-params-v2"] = common if log: print(json.dumps(datas, indent=4, ensure_ascii=False)) return datas USER_AGENT = ( "com.phoenix.read/71332 (Linux; U; Android 16; zh_CN; 25053RT47C; " "Build/BP2A.250605.031.A3; Cronet/TTNetVersion:04657795 2026-01-23 " "QuicVersion:c67e9834 2025-09-08)" ) VIDEO_MODEL_URL_TEMPLATE = ( "https://api5-normal-sinfonlineb.fqnovel.com/novel/player/multi_video_model/v1/" "?iid={install_id}&device_id={device_id}&ac=wifi&channel=update_64&aid=8662" "&app_name=novelread&version_code=71332&version_name=7.1.3.32" "&device_platform=android&os=android&ssmix=a&device_type=25053RT47C" "&device_brand=Redmi&language=zh&os_api=36&os_version=16" "&manifest_version_code=71332&resolution=1280*2772&dpi=520" "&update_version_code=71332&host_abi=arm64-v8a&dragon_device_type=phone" "&pv_player=71332&compliance_status=0&need_personal_recommend=1" "&player_so_load=1&is_android_pad_screen=0" ) def load_local_config() -> Dict[str, Any]: return {"device_id": CONFIG_DEVICE_ID, "install_id": CONFIG_INSTALL_ID, "platform": CONFIG_PLATFORM, "cache_seconds": CONFIG_CACHE_SECONDS} def get_device_keys() -> Dict[str, str]: config = load_local_config() device_id = str(config.get("device_id") or "").strip() install_id = str(config.get("install_id") or "").strip() platform = str(config.get("platform") or "android").strip() or "android" if not device_id or not install_id: raise RuntimeError( "Missing device configuration. Set DUANJU_DEVICE_ID / " "DUANJU_INSTALL_ID, or open the web UI and save local config." ) return { "device_id": device_id, "install_id": install_id, "platform": platform, } def build_liushen_device(device_keys: Dict[str, str]) -> Dict[str, str]: return { "device_id": device_keys.get("device_id", ""), "iid": device_keys.get("install_id", ""), "install_id": device_keys.get("install_id", ""), "device_brand": "Redmi", "device_model": "25053RT47C", "device_type": "25053RT47C", "device_manufacturer": "Xiaomi", "os_version": "16", "version_name": "7.1.3.32", "ua": USER_AGENT, } def _compute_branch(query_string, body_bytes, khronos): query_sm3 = SM3(query_string).digest() body_md5 = hashlib.md5(body_bytes).digest() if body_bytes else bytes(16) ts_bytes = khronos.to_bytes(4, "little") iv = get_iv(0x20230928, query_sm3) iv = get_iv(iv, body_md5) iv = get_iv(iv, ts_bytes) low = iv & 15 iv_v0 = (low * 171) >> 9 return low - (iv_v0 * 3) def sign_json_request_with_liushen( url: str, body_obj: Dict[str, Any], device_keys: Dict[str, str], ) -> Tuple[str, Dict[str, str], bytes]: body_text = json.dumps(body_obj, ensure_ascii=False, separators=(",", ":")) body_data = json.loads(body_text) body_bytes = body_text.encode("utf-8") ts = str(int(time.time() * 1000)) base_headers = { "User-Agent": USER_AGENT, "Accept": "application/json; charset=utf-8,application/x-protobuf", "Content-Type": "application/json; charset=UTF-8", "x-xs-from-web": "0", "x-ss-req-ticket": ts, "x-tt-request-tag": "t=0;n=0", "sdk-version": "2", "passport-sdk-version": "50561", "x-vc-bdturing-sdk-version": "3.7.2.cn", } url_parts = urlsplit(url) base_url = f"{url_parts.scheme}://{url_parts.netloc}{url_parts.path}" params = dict(parse_qsl(url_parts.query, keep_blank_values=True)) devices = build_liushen_device(device_keys) khronos = int(time.time()) base_rticket = int(time.time() * 1000) safe_rticket = base_rticket safe_khronos = khronos for offset in range(32): rticket = base_rticket + offset eurl, _, _ = get_params_encrypturl( base_url, params=params, devices=devices, rticket_override=rticket, ts_override=khronos, ) query_string = eurl.split("?", 1)[1] if "?" in eurl else "" branch = _compute_branch(query_string, body_bytes, khronos) if branch != 1: safe_rticket = rticket safe_khronos = khronos break sign_headers, sign_url = core_sixgod( surl=base_url, params=params, data=body_data, devices=devices, header=base_headers, log=False, rticket_override=safe_rticket, ts_override=safe_khronos, khronos_override=safe_khronos, ) return sign_url, sign_headers, body_bytes def _quality_number(value: str) -> int: text = str(value or "") m = re.match(r".*?(2160|1440|1080|720|576|540|480|360)", text) if m: return int(m.group(1)) mapping = {"1920": 1080, "1280": 720, "1024": 576, "854": 480, "640": 360} if text in mapping: return mapping[text] if re.match(r"^\d{3,4}$", text): return int(text) return 0 def _quality_label(value: str) -> str: n = _quality_number(value) if n: return "%dP" % n labels = { "low": "\u6d41\u7545", "smooth": "\u6d41\u7545", "medium": "\u6807\u6e05", "normal": "\u9ad8\u6e05", "high": "\u8d85\u6e05", "original": "\u539f\u753b", "uhd": "\u539f\u753b", "super_high": "\u539f\u753b", } return labels.get(str(value or "").lower(), str(value or "\u81ea\u52a8")) def _quality_rows(video_list: Dict[str, Any]) -> List[Dict[str, Any]]: rows: List[Dict[str, Any]] = [] if not video_list or not isinstance(video_list, dict): return rows for key, value in video_list.items(): item = value if isinstance(value, dict) else {} quality = str( item.get("quality_desc") or item.get("height") or item.get("vheight") or item.get("quality") or key ) if item.get("main_url") and not any( r["key"] == key for r in rows ): rows.append({"key": key, "quality": quality, "item": item}) rows.sort( key=lambda r: _quality_number(r["quality"] or r["key"]), reverse=True, ) return rows def fetch_quality_rows(vid: str) -> List[Dict[str, Any]]: cache_key = str(vid) cached = _quality_cache.get(cache_key) if cached and time.time() - cached[0] < 300: return cached[1] rows: List[Dict[str, Any]] = [] try: device_keys = get_device_keys() target_url = build_video_model_url( device_keys["device_id"], device_keys["install_id"] ) post_payload = { "biz_param": { "detail_page_version": 0, "device_level": 3, "disable_digg_stat": False, "need_all_video_definition": True, "need_mp4_align": False, "use_os_player": False, "use_server_dns": False, "video_platform": 1024, }, "mixed_video_id_map": {"1004": [str(vid)]}, } signed_url, headers, post_body = sign_json_request_with_liushen( target_url, post_payload, device_keys ) resp = curl_request(signed_url, headers, post_body, 20) data = json.loads(resp) fallback_api, video_model = extract_fallback_api(data, str(vid)) resp2 = curl_request(fallback_api, {"User-Agent": USER_AGENT}, None, 20) outer = json.loads(resp2) video_data = outer.get("video_info", {}).get("data", {}) if not isinstance(video_data, dict): video_data = {} video_list = video_data.get("video_list", {}) if isinstance(video_list, dict): rows = _quality_rows(video_list) except Exception: pass _quality_cache[cache_key] = (time.time(), rows) return rows _quality_cache: Dict[str, Tuple[float, List]] = {} def replace_failed_device(device_id: str, platform: str) -> None: pass def get_current_domain(request=None) -> str: return _CURRENT_DOMAIN _RUNTIME_BASE_DIR = Path(os.path.dirname(os.path.abspath(__file__))) if "__file__" in dir() else Path.cwd() def get_runtime_base_dir() -> Path: return _RUNTIME_BASE_DIR DEFAULT_TIMEOUT = 15 VIDEO_WORKER_POOL = ThreadPoolExecutor(max_workers=4) FFMPEG_BIN = "ffmpeg" VIDEO_TTL_SECONDS = 300 def schedule_video_cleanup(filepath: Path, delay_seconds: int = VIDEO_TTL_SECONDS) -> None: def _delete_file() -> None: try: filepath.unlink(missing_ok=True) print(f"[cleanup] deleted_expired_video={filepath.name}") except Exception as exc: print(f"[cleanup] delete_failed file={filepath.name} error={exc}") timer = threading.Timer(delay_seconds, _delete_file) timer.daemon = True timer.start() def curl_request( url: str, headers: Dict[str, str], post_body: Optional[bytes] = None, timeout: int = DEFAULT_TIMEOUT, ) -> bytes: session = requests.Session() session.trust_env = False if post_body is not None: resp = session.post(url, headers=headers, data=post_body, timeout=timeout) else: resp = session.get(url, headers=headers, timeout=timeout) resp.raise_for_status() return resp.content def handle_video_request( video_id: str, request=None, max_retries: int = 3, stream_mode: bool = False, quality_key: Optional[str] = None, ) -> Dict[str, Any]: return resolve_video_url( video_id, request, max_retries, stream_mode=stream_mode, quality_key=quality_key, ) def resolve_video_url( video_id: str, request=None, max_retries: int = 3, stream_mode: bool = False, quality_key: Optional[str] = None, ) -> Dict[str, Any]: last_err: Optional[Exception] = None for attempt in range(max_retries): device_keys = get_device_keys() target_url = build_video_model_url( device_keys["device_id"], device_keys["install_id"] ) post_payload = { "biz_param": { "detail_page_version": 0, "device_level": 3, "disable_digg_stat": False, "need_all_video_definition": True, "need_mp4_align": False, "use_os_player": False, "use_server_dns": False, "video_platform": 1024, }, "mixed_video_id_map": { "1004": [video_id], }, } signed_url, headers, post_body = sign_json_request_with_liushen( target_url, post_payload, device_keys ) try: resp = curl_request(signed_url, headers, post_body, 30) except Exception as exc: last_err = Exception(f"video_model request failed: {exc}") time.sleep(0.1) continue try: data = json.loads(resp) except Exception as exc: last_err = Exception(f"video_model JSON parse failed: {exc}") continue if not isinstance(data, dict) or "data" not in data: print("video_model raw response:") print(json.dumps(data, ensure_ascii=False, indent=2)) try: fallback_api, video_model = extract_fallback_api(data, video_id) except Exception as exc: last_err = exc continue try: result = download_and_decrypt_video( request, fallback_api, video_model, device_keys, video_id, max_retries=3, stream_mode=stream_mode, quality_key=quality_key, ) return result except Exception as exc: last_err = exc time.sleep(0.1) continue raise Exception(f"Video request failed, retried {max_retries} times: {last_err}") def build_video_model_url(device_id: str, install_id: str) -> str: from urllib.parse import quote return VIDEO_MODEL_URL_TEMPLATE.format( install_id=quote(install_id, safe=""), device_id=quote(device_id, safe=""), ) def extract_fallback_api( data: Dict[str, Any], video_id: str ) -> Tuple[str, Dict[str, Any]]: data_map = data.get("data") if not isinstance(data_map, dict): raise ValueError("Response missing data field") video_entry: Optional[Dict[str, Any]] = None if video_id in data_map and isinstance(data_map[video_id], dict): video_entry = data_map[video_id] if video_entry is None: for v in data_map.values(): if isinstance(v, dict): video_entry = v break if isinstance(v, list) and len(v) > 0 and isinstance(v[0], dict): video_entry = v[0] break if video_entry is None: keys = list(data_map.keys()) raise ValueError( f"video entry not found (looked for {video_id}, available keys: {keys})" ) video_model: Optional[Dict[str, Any]] = None vm = video_entry.get("video_model") if isinstance(vm, str): video_model = json.loads(vm) elif isinstance(vm, dict): video_model = vm else: raise ValueError("video_model is empty or unknown format") fallback_raw = video_model.get("fallback_api") fallback_str = parse_fallback_api(fallback_raw) if not fallback_str: raise ValueError(f"fallback_api cannot be parsed: {type(fallback_raw)} => {fallback_raw}") return fallback_str, video_model def parse_fallback_api(raw: Any) -> str: if isinstance(raw, str): if raw.startswith("{"): try: decoded = json.loads(raw) if isinstance(decoded, dict) and "fallback_api" in decoded: return str(decoded["fallback_api"]) except (json.JSONDecodeError, TypeError): pass if len(raw) > 10: return raw elif isinstance(raw, list) and len(raw) > 0: if isinstance(raw[0], str): return raw[0] elif isinstance(raw, dict): if "fallback_api" in raw: return str(raw["fallback_api"]) return "" def download_and_decrypt_video( request, fallback_api: str, video_model: Dict[str, Any], device_keys: Dict[str, str], video_id: str, max_retries: int = 3, stream_mode: bool = False, quality_key: Optional[str] = None, ) -> Dict[str, Any]: current_url = fallback_api current_device_keys = device_keys last_err: Optional[Exception] = None for attempt in range(max_retries): headers = {"User-Agent": USER_AGENT} try: resp = curl_request(current_url, headers, None, 30) except Exception as exc: last_err = Exception(f"fallback_api request failed: {exc}") else: try: data = json.loads(resp) except Exception as exc: last_err = Exception(f"fallback_api JSON parse failed: {exc}") else: video_info = data.get("video_info", {}) if not isinstance(video_info, dict): video_info = {} video_data = video_info.get("data", {}) if not isinstance(video_data, dict): video_data = {} if not video_data: last_err = Exception("fallback_api response structure abnormal") else: key_seed_b64 = video_data.get("key_seed", "") key_seed_raw = b64_decode_padded(key_seed_b64) video_list = video_data.get("video_list", {}) if isinstance(video_list, dict): best_key, best_item = select_best_quality(video_list, preferred_key=quality_key) if best_key and best_item: spade_a = best_item.get("spade_a", "") content_key = None if spade_a: try: content_key = derive_content_key(spade_a) except Exception: pass raw_main_url = best_item.get("main_url", "") if raw_main_url: real_main_url = raw_main_url if key_seed_raw and len(raw_main_url) > 10: try: dec = decrypt_spade_url(raw_main_url, key_seed_raw) if dec: real_main_url = dec except Exception as exc: print(f"[debug] spade_url_decrypt_failed={exc}") backup_urls = [] for bk_key in ("backup_url_1", "backup_url_2", "url", "play_addr"): bk_raw = best_item.get(bk_key, "") if not bk_raw or not isinstance(bk_raw, str): continue bk_real = bk_raw if key_seed_raw and len(bk_raw) > 10: try: bk_dec = decrypt_spade_url(bk_raw, key_seed_raw) if bk_dec: bk_real = bk_dec except Exception: pass if bk_real and bk_real != real_main_url: backup_urls.append(bk_real) if stream_mode: stream_url = build_stream_url( request, real_main_url, content_key, backup_urls) if stream_url: best_item["main_url"] = stream_url else: local_url = download_decrypt_and_serve( request, real_main_url, content_key, backup_urls) if local_url: best_item["main_url"] = local_url video_data["video_list"] = {best_key: best_item} video_info["data"] = video_data data["video_info"] = video_info return build_response_payload(video_id, video_model, video_data, best_item) if attempt < max_retries - 1 and video_id: if current_device_keys: replace_failed_device( current_device_keys.get("device_id", ""), current_device_keys.get("platform", ""), ) try: new_url, new_keys = refresh_fallback_url(video_id) if new_url: current_url = new_url current_device_keys = new_keys except Exception: pass raise Exception( f"fallback_api request failed, retried {max_retries} times: {last_err}" ) def download_decrypt_and_serve( request, video_url: str, content_key: Optional[bytes], backup_urls=None, ) -> Optional[str]: pipeline_start = time.perf_counter() encrypted_data = download_video_bytes(video_url, backup_urls) decrypted_data = decrypt_video_bytes(encrypted_data, content_key) local_url = save_video_bytes(request, decrypted_data) pipeline_seconds = time.perf_counter() - pipeline_start print(f"[timing] video_pipeline_seconds={pipeline_seconds:.3f}") return local_url def download_video_bytes(video_url: str, backup_urls=None) -> bytes: download_start = time.perf_counter() session = requests.Session() session.trust_env = False dl_headers = { "User-Agent": "com.phoenix.read/71332", "Referer": "https://novel.snssdk.com/", } candidate_urls = [video_url] + (backup_urls or []) last_exc = None for url in candidate_urls: try: resp = session.get(url, headers=dl_headers, timeout=(8, 120)) resp.raise_for_status() download_seconds = time.perf_counter() - download_start print(f"[timing] download_seconds={download_seconds:.3f}") return resp.content except Exception as exc: print(f"[download] cdn_failed url={url[:80]} error={exc}") last_exc = exc continue raise Exception(f"Failed to download video: all CDN nodes failed: {last_exc}") def decrypt_video_bytes(encrypted_data: bytes, content_key: Optional[bytes]) -> bytes: if content_key is None: print("[timing] decrypt_seconds=0.000 content_key_missing=true") return encrypted_data decrypt_start = time.perf_counter() try: decrypted_data = decrypt_mp4_cenc(encrypted_data, content_key) decrypt_seconds = time.perf_counter() - decrypt_start print(f"[timing] decrypt_seconds={decrypt_seconds:.3f}") return decrypted_data except Exception as exc: raise Exception(f"Failed to decrypt video: {exc}") def save_video_bytes(request, decrypted_data: bytes) -> str: save_start = time.perf_counter() src_dir = get_runtime_base_dir() / "src" src_dir.mkdir(parents=True, exist_ok=True) filename = f"video_{time.time_ns()}.mp4" filepath = src_dir / filename filepath.write_bytes(decrypted_data) schedule_video_cleanup(filepath) current_domain = get_current_domain(request) save_seconds = time.perf_counter() - save_start print(f"[timing] save_seconds={save_seconds:.3f}") return f"{current_domain}/src/{filename}" def build_stream_url(request, encrypted_url: str, content_key: Optional[bytes], backup_urls=None) -> str: current_domain = get_current_domain(request) key_b64 = base64.b64encode(content_key).decode('ascii') if content_key else '' params_list = [('url', encrypted_url), ('key', key_b64)] for bk in (backup_urls or []): params_list.append(('bk', bk)) params = urlencode(params_list) return f"{current_domain}/stream?{params}" def stream_cenc_decrypt_chunk(chunk: bytearray, chunk_start: int, decrypt_map: Dict[int, Tuple[int, bytes]], content_key: bytes) -> bytes: chunk_end = chunk_start + len(chunk) for sample_off in list(decrypt_map.keys()): sample_sz, iv = decrypt_map[sample_off] sample_end = sample_off + sample_sz if sample_off >= chunk_end or sample_end <= chunk_start: continue overlap_start = max(sample_off, chunk_start) overlap_end = min(sample_end, chunk_end) overlap_size = overlap_end - overlap_start offset_in_sample = overlap_start - sample_off start_block = offset_in_sample // 16 ctr_int = int.from_bytes(bytes(iv) + b'\x00' * 8, 'big') + start_block ctr_bytes = ctr_int.to_bytes(16, 'big') cipher = _new_aes('ctr', content_key, initial_value=ctr_bytes) block_offset = offset_in_sample % 16 if block_offset > 0: cipher.decrypt(b'\x00' * block_offset) start_in_chunk = overlap_start - chunk_start decrypted = cipher.decrypt(bytes(chunk[start_in_chunk:start_in_chunk + overlap_size])) chunk[start_in_chunk:start_in_chunk + overlap_size] = decrypted return bytes(chunk) def refresh_fallback_url(video_id: str) -> Tuple[str, Dict[str, str]]: device_keys = get_device_keys() target_url = build_video_model_url( device_keys["device_id"], device_keys["install_id"] ) post_payload = { "biz_param": { "detail_page_version": 0, "device_level": 3, "disable_digg_stat": False, "need_all_video_definition": True, "need_mp4_align": False, "use_os_player": False, "use_server_dns": False, "video_platform": 1024, }, "mixed_video_id_map": {"1004": [video_id]}, } signed_url, headers, post_body = sign_json_request_with_liushen( target_url, post_payload, device_keys ) resp = curl_request(signed_url, headers, post_body, 30) data = json.loads(resp) fallback_api, _ = extract_fallback_api(data, video_id) return fallback_api, device_keys def derive_content_key(spade_b64: str) -> bytes: s = spade_b64.strip() m = 4 - len(s) % 4 if m != 4: s += "=" * m raw = base64.b64decode(s) if len(raw) < 3: raise ValueError(f"spade_a too short: {len(raw)} bytes") v6 = raw[0] ^ raw[1] ^ raw[2] v8 = len(raw) - v6 + 47 if v8 <= 0 or v8 > len(raw) * 2: raise ValueError(f"spade_a: computed v8={v8} out of range") if 1 + v8 > len(raw): v8 = len(raw) - 1 if v8 < 33: raise ValueError(f"spade_a: v8={v8} too small (need >=33)") v13 = bytearray(raw[1 : 1 + v8]) vA, vB = 85, 246 for i in range(v8): popcnt = bin(i).count("1") if i & 1: v24 = vA vA = v13[i] else: v24 = vB vB = v13[i] v25 = v24 ^ v13[i] v26 = -21 - popcnt v13[i] = (v26 + v25) & 0xFF hex_str = bytes(v13[1:33]).decode("ascii") key = binascii.unhexlify(hex_str) return key def decrypt_mp4_cenc(data: bytes, content_key: bytes) -> bytes: data = bytearray(data) ftyp_end = struct.unpack(">I", data[0:4])[0] if ftyp_end + 8 >= len(data): raise ValueError("invalid MP4: ftyp too large") moov_size = struct.unpack(">I", data[ftyp_end : ftyp_end + 4])[0] if ftyp_end + 8 + moov_size > len(data): raise ValueError("invalid MP4: moov out of range") moov = data[ftyp_end + 8 : ftyp_end + moov_size] t1_off, t1_sz = find_box(moov, "trak", 0) t2_off, _ = find_box(moov, "trak", t1_off + t1_sz) for t_off in (t1_off, t2_off): if t_off < 0: continue result = parse_track(moov, t_off) if result is None: continue sizes, offsets, cns, aux_off, aux_sz, ns = result if ns == 0: continue if aux_off + aux_sz > len(data): continue aux = data[aux_off : aux_off + aux_sz] si, ap = 0, 0 for ci, off in enumerate(offsets): for k in range(cns[ci]): if si >= ns: break sz = sizes[si] if off + sz > len(data): break iv = bytearray(8) if ap + 8 <= len(aux): iv[:] = aux[ap : ap + 8] ctr_bytes = bytes(iv) + b"\x00" * 8 cipher = _new_aes('ctr', content_key, initial_value=ctr_bytes) decrypted = cipher.decrypt(bytes(data[off : off + sz])) data[off : off + sz] = decrypted off += sz si += 1 ap += 8 for old, new in ((b"encv", b"hvc1"), (b"enca", b"mp4a")): _replace_fourcc(data, old, new) _replace_sinf(data) return bytes(data) def find_box(data: bytearray, fourcc: str, start: int) -> Tuple[int, int]: b = fourcc.encode("ascii") for i in range(start, len(data) - 8): if data[i : i + 4] == b and i >= 4: sz = struct.unpack(">I", data[i - 4 : i])[0] if 0 < sz < 5000000: return i - 4, sz return -1, 0 def get_box(data: bytearray, fourcc: str, stbl_off: int): o, sz = find_box(data, fourcc, stbl_off) if o >= 0: return data[o + 8 : o + sz] return None def parse_track( moov: bytearray, t_off: int ) -> Optional[Tuple[List[int], List[int], List[int], int, int, int]]: stbl_off, _ = find_box(moov, "stbl", t_off + 8) stsz = get_box(moov, "stsz", stbl_off) if stsz is None: return None ds = struct.unpack(">I", stsz[4:8])[0] ns = struct.unpack(">I", stsz[8:12])[0] sizes: List[int] = [] if ds == 0: for i in range(ns): sizes.append(struct.unpack(">I", stsz[12 + i * 4 : 16 + i * 4])[0]) else: sizes = [ds] * ns stco = get_box(moov, "stco", stbl_off) if stco is None: return None nc = struct.unpack(">I", stco[4:8])[0] offsets = [] for i in range(nc): offsets.append(struct.unpack(">I", stco[8 + i * 4 : 12 + i * 4])[0]) stsc = get_box(moov, "stsc", stbl_off) if stsc is None: return None nsc = struct.unpack(">I", stsc[4:8])[0] entries = [] for i in range(nsc): entries.append(( struct.unpack(">I", stsc[8 + i * 12 : 12 + i * 12])[0], struct.unpack(">I", stsc[12 + i * 12 : 16 + i * 12])[0], struct.unpack(">I", stsc[16 + i * 12 : 20 + i * 12])[0], )) cns = [0] * nc for i in range(nsc): fc = entries[i][0] spc = entries[i][1] end = nc if i + 1 < nsc: end = entries[i + 1][0] - 1 for c in range(fc - 1, min(end, nc)): cns[c] = spc saiz = get_box(moov, "saiz", stbl_off) if saiz is None: return None da = saiz[4] na = struct.unpack(">I", saiz[5:9])[0] saio = get_box(moov, "saio", stbl_off) if saio is None: return None aux_off = struct.unpack(">I", saio[8:12])[0] aux_sz = na * max(da, 8) return sizes, offsets, cns, aux_off, aux_sz, ns def select_best_quality( video_list: Dict[str, Any], preferred_key: Optional[str] = None ) -> Tuple[str, Dict[str, Any]]: if preferred_key and preferred_key in video_list and isinstance( video_list[preferred_key], dict ): return preferred_key, video_list[preferred_key] best_key = "" best_item: Dict[str, Any] = {} best_height = 0 for k, item in video_list.items(): if not isinstance(item, dict): continue h = int(item.get("vheight", 0)) if h > best_height: best_height = h best_key = k best_item = item elif h == best_height and best_item: cur_br = int(item.get("bitrate", 0)) best_br = int(best_item.get("bitrate", 0)) if cur_br > best_br: best_key = k best_item = item return best_key, best_item def build_response_payload( video_id: str, video_model: Dict[str, Any], video_data: Dict[str, Any], best_item: Dict[str, Any], ) -> Dict[str, Any]: pic = first_non_empty( best_item.get("cover"), best_item.get("poster"), video_model.get("origin_cover"), video_model.get("cover_url"), video_model.get("dynamic_cover"), video_model.get("cover"), video_data.get("cover"), video_data.get("poster"), ) url = first_non_empty( best_item.get("main_url"), best_item.get("play_addr"), best_item.get("backup_url_1"), best_item.get("url"), ) height = stringify_int(first_non_empty(best_item.get("vheight"), best_item.get("height"))) width = stringify_int(first_non_empty(best_item.get("vwidth"), best_item.get("width"))) return { "vid": video_id, "pic": normalize_media_url(pic), "url": normalize_media_url(url), "quality": format_quality(best_item, height), "duration": format_duration(first_non_empty(video_model.get("duration"), video_data.get("duration"))), "size": format_size(first_non_empty(best_item.get("size"), best_item.get("data_size"), best_item.get("file_size"))), "height": height, "width": width, "create_time": format_create_time( first_non_empty( video_model.get("create_time"), video_model.get("publish_time"), video_data.get("create_time"), video_data.get("publish_time"), ) ), } def first_non_empty(*values: Any) -> str: for value in values: normalized = unwrap_media_value(value) if normalized: return normalized return "" def unwrap_media_value(value: Any) -> str: if value is None: return "" if isinstance(value, str): return value.strip() if isinstance(value, (int, float)): return str(value) if isinstance(value, list): for item in value: normalized = unwrap_media_value(item) if normalized: return normalized return "" if isinstance(value, dict): for key in ("url", "uri", "src", "download_url"): normalized = unwrap_media_value(value.get(key)) if normalized: return normalized for key in ("url_list", "urls"): normalized = unwrap_media_value(value.get(key)) if normalized: return normalized return "" def normalize_media_url(value: str) -> str: if value.startswith("//"): return f"https:{value}" return value def stringify_int(value: Any) -> str: if value in (None, ""): return "" try: return str(int(float(value))) except (TypeError, ValueError): return str(value) def format_quality(best_item: Dict[str, Any], height: str) -> str: label = first_non_empty( best_item.get("quality"), best_item.get("definition"), best_item.get("gear_name"), best_item.get("quality_desc"), ) if label: return label if height: return f"{height}p" return "" def format_duration(value: Any) -> str: if value in (None, ""): return "" try: total_seconds = int(float(value)) except (TypeError, ValueError): return str(value) minutes, seconds = divmod(total_seconds, 60) hours, minutes = divmod(minutes, 60) if hours > 0: return f"{hours}小时{minutes}分钟{seconds}秒" return f"{minutes}分钟{seconds}秒" def format_size(value: Any) -> str: if value in (None, ""): return "" try: size_bytes = float(value) except (TypeError, ValueError): return str(value) units = ["B", "KB", "MB", "GB", "TB"] unit_index = 0 while size_bytes >= 1024 and unit_index < len(units) - 1: size_bytes /= 1024 unit_index += 1 return f"{size_bytes:.2f}{units[unit_index]}" def format_create_time(value: Any) -> str: if value in (None, ""): return "" if isinstance(value, str): text = value.strip() if not text: return "" if text.endswith("Z"): return datetime.fromisoformat(text.replace("Z", "+00:00")).astimezone( timezone(timedelta(hours=8)) ).isoformat() try: numeric = float(text) except ValueError: return text value = numeric if isinstance(value, (int, float)): timestamp = float(value) if timestamp > 1e12: timestamp /= 1000 dt = datetime.fromtimestamp(timestamp, tz=timezone(timedelta(hours=8))) return dt.isoformat() return str(value) class _SpadeCbcDecryptor: """AES-CBC 解密适配器：cryptography 库无 .new()/.decrypt(bytes) 一体接口。""" def __init__(self, cipher): self._dec = cipher.decryptor() def decrypt(self, data): out = self._dec.update(bytes(data)) + self._dec.finalize() return out class _AesCtrAdapter: """AES-CTR 适配器：模拟 PyCryptodome 的 AES.new(key, MODE_CTR, nonce=b'', initial_value=n)。 cryptography 的 CTR 用 16 字节全宽大端计数器，语义一致。""" def __init__(self, key, initial_value): from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes if isinstance(initial_value, int): iv16 = initial_value.to_bytes(16, 'big') else: iv16 = bytes(initial_value) self._dec = Cipher(algorithms.AES(bytes(key)), modes.CTR(iv16)).decryptor() def decrypt(self, data): return self._dec.update(bytes(data)) + self._dec.finalize() def _new_aes(mode, key, **kw): """统一 AES 后端：优先 PyCryptodome，缺库时回退 cryptography。""" if AES is not None: if mode == 'ctr': return AES.new(key, AES.MODE_CTR, nonce=b"", initial_value=kw['initial_value']) return AES.new(key, AES.MODE_CBC, iv=kw['iv']) from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes if mode == 'ctr': return _AesCtrAdapter(key, kw['initial_value']) return _SpadeCbcDecryptor(Cipher(algorithms.AES(bytes(key)), modes.CBC(kw['iv']))) def decrypt_spade_url(b64_str: str, key_seed: bytes) -> str: if not b64_str: return "" raw = b64_decode_padded(b64_str) if len(raw) < 5: raise ValueError("Ciphertext too short") if raw[0] != 0xA8 or raw[2] != 0x01 or raw[3] != 0x00: raise ValueError("Ciphertext header format error") cipher_data = raw[4:] cipher_len = (len(cipher_data) // 16) * 16 cipher_data = cipher_data[:cipher_len] constants = bytes([ 0x4D, 0xD4, 0xC2, 0xE6, 0xB8, 0x31, 0x62, 0x09, 0x0E, 0x52, 0xB3, 0xC7, 0xA6, 0x73, 0x3B, 0xA4, 0x1C, 0xB2, 0x46, 0x2B, 0x82, 0x9A, 0xB5, 0x8A, 0x19, 0x6B, 0x39, 0xDB, 0x57, 0x17, 0x75, 0x24, 0xF4, 0x9B, 0xAF, 0x7F, 0x08, 0xE8, 0xD6, 0x8D, 0x26, 0xA7, 0x2E, 0x37, 0xC1, 0xA9, 0x5A, 0x2F, 0x1F, 0x05, 0xA5, 0x18, 0x92, 0xAE, 0xF2, 0x94, 0x97, 0x32, 0xB6, 0x2A, 0x38, 0xAA, 0xDD, 0x58, ]) h1 = hashlib.sha512(key_seed).digest() h2 = hashlib.sha512(h1 + constants).digest() aes_key = h2[:16] iv = h2[16:32] cipher = _new_aes('cbc', aes_key, iv=iv) plaintext = cipher.decrypt(cipher_data) if plaintext: pad = plaintext[-1] if 1 <= pad <= 16 and pad <= len(plaintext): plaintext = plaintext[:-pad] return plaintext.rstrip(b"\x00").decode("utf-8", errors="replace") def b64_decode_padded(s: str) -> bytes: s = s.strip() pad = len(s) % 4 if pad: s += "=" * (4 - pad) try: return base64.b64decode(s) except Exception: return base64.urlsafe_b64decode(s) def _replace_fourcc(data: bytearray, old: bytes, new: bytes) -> None: old_len = len(old) for i in range(len(data) - old_len): if data[i : i + old_len] == old: data[i : i + len(new)] = new def _replace_sinf(data: bytearray) -> None: i = 0 while i < len(data) - 4: if data[i : i + 4] == b"sinf": if i >= 4: sz = struct.unpack(">I", data[i - 4 : i])[0] if 0 < sz < 50000: data[i : i + 4] = b"free" end = min(i - 4 + sz, len(data)) for j in range(i + 4, end): data[j] = 0 i = end continue i += 1 class _PlayHandler(BaseHTTPRequestHandler): parser = None src_dir = None base_url = '' _resolve_cache = {} _vid_locks = {} _guard = threading.Lock() _download_locks = {} _download_guard = threading.Lock() _RESOLVE_TTL = 300 def do_GET(self): parsed = urlparse(self.path) path = parsed.path params = parse_qs(parsed.query) if path == '/play': self._handle_play(params) elif path == '/stream': self._handle_stream(params) elif path.startswith('/src/'): self._handle_src(path[5:]) elif path == '/health': self._handle_health() else: self.send_error(404) def do_HEAD(self): parsed = urlparse(self.path) path = parsed.path if path == '/stream': params = parse_qs(parsed.query) self._handle_stream(params, head_only=True) elif path.startswith('/src/'): self._handle_src(path[5:], head_only=True) else: self.send_error(404) def _handle_play(self, params): vid = params.get('vid', [''])[0].strip() if not vid.isdigit(): self._json(400, {'error': 'vid invalid'}) return try: result = self._resolve_cached(vid) media_url = str(result.get('url') or '') if not media_url: self._json(502, {'error': 'no media url'}) return self.send_response(302) self.send_header('Location', media_url) self.send_header('Access-Control-Allow-Origin', '*') self.end_headers() except Exception as exc: self._json(502, {'error': str(exc)}) def _resolve_cached(self, vid): with self._guard: if vid not in self._vid_locks: self._vid_locks[vid] = threading.Lock() if len(self._resolve_cache) > 1000: now = time.time() for k in [k for k, v in self._resolve_cache.items() if now - v[0] > self._RESOLVE_TTL]: self._resolve_cache.pop(k, None) lock = self._vid_locks[vid] with lock: hit = self._resolve_cache.get(vid) if hit and time.time() - hit[0] < self._RESOLVE_TTL: return hit[1] result = handle_video_request(vid, None, max_retries=3, stream_mode=True) self._resolve_cache[vid] = (time.time(), result) return result def _handle_src(self, filename, head_only=False): if not self.src_dir: self.send_error(503, 'src_dir not configured') return safe_name = Path(filename).name filepath = self.src_dir / safe_name if not filepath.exists() or not filepath.is_file(): self.send_error(404) return file_size = filepath.stat().st_size range_header = self.headers.get('Range', '') start = 0 end = file_size - 1 is_partial = False if range_header: m = re.match(r'bytes=(\d+)-(\d*)', range_header) if m: start = int(m.group(1)) if m.group(2): end = int(m.group(2)) is_partial = True if start > end or start >= file_size: self.send_error(416) return content_length = end - start + 1 if is_partial: self.send_response(206) self.send_header('Content-Range', 'bytes %d-%d/%d' % (start, end, file_size)) else: self.send_response(200) self.send_header('Content-Type', 'video/mp4') self.send_header('Content-Length', str(content_length)) self.send_header('Accept-Ranges', 'bytes') self.send_header('Access-Control-Allow-Origin', '*') self.send_header('Access-Control-Expose-Headers', 'Content-Length, Content-Range, Accept-Ranges') self.end_headers() if head_only: return with open(filepath, 'rb') as f: f.seek(start) remaining = content_length while remaining > 0: chunk = f.read(min(65536, remaining)) if not chunk: break self.wfile.write(chunk) remaining -= len(chunk) def _handle_health(self): self._json(200, { 'ok': True, 'service': 'hongguo-embedded', 'parser': self.parser is not None, 'src_dir': str(self.src_dir) if self.src_dir else '', }) def _handle_stream(self, params, head_only=False): raw_url = params.get('url', [''])[0] key_b64 = params.get('key', [''])[0] if not raw_url: self.send_error(400, 'missing url param') return backup_urls = [u for u in params.get('bk', []) if u] candidate_urls = [raw_url] + backup_urls try: content_key = base64.b64decode(key_b64) if key_b64 else None except Exception: content_key = None if content_key is not None: url_hash = hashlib.md5(raw_url.encode()).hexdigest()[:16] filename = 'video_%s.mp4' % url_hash if not self.src_dir: self.send_error(503, 'src_dir not configured') return filepath = self.src_dir / filename if filepath.exists() and filepath.is_file(): self._handle_src(filename, head_only) return range_header = self.headers.get('Range', '') is_full = (not range_header or re.match(r'bytes=0-\s*$', range_header) is not None) if is_full and not head_only: try: with self._download_guard: if filename not in self._download_locks: self._download_locks[filename] = threading.Lock() s_lock = self._download_locks[filename] if s_lock.acquire(blocking=False): try: if filepath.exists() and filepath.is_file(): self._handle_src(filename, head_only) return self._stream_cenc_direct( raw_url, backup_urls, content_key, total_size_hint=0, filename=filename, filepath=filepath) return finally: s_lock.release() else: with s_lock: pass if filepath.exists() and filepath.is_file(): self._handle_src(filename, head_only) return print('[stream] streaming peer failed, fallback to full download') except Exception as exc: print('[stream] streaming failed, fallback to full download: %s' % exc) try: with self._download_guard: if filename not in self._download_locks: self._download_locks[filename] = threading.Lock() dl_lock = self._download_locks[filename] with dl_lock: if filepath.exists() and filepath.is_file(): self._handle_src(filename, head_only) return print('[stream] full download + decrypt...') encrypted_data = download_video_bytes(raw_url, backup_urls) decrypted_data = decrypt_video_bytes(encrypted_data, content_key) filepath.write_bytes(decrypted_data) schedule_video_cleanup(filepath) print('[stream] cached %s (%d bytes)' % (filename, len(decrypted_data))) self._handle_src(filename, head_only) except Exception as exc: print('[stream] download_decrypt error: %s' % exc) try: self.send_error(502, 'stream failed: %s' % exc) except Exception: pass return session = requests.Session() session.trust_env = False dl_headers = { 'User-Agent': 'com.phoenix.read/71332', 'Referer': 'https://novel.snssdk.com/', } total_size = 0 active_url = candidate_urls[0] for cand in candidate_urls: try: head_resp = session.head(cand, headers=dl_headers, timeout=(8, 10)) total_size = int(head_resp.headers.get('Content-Length', 0)) active_url = cand break except Exception: continue if total_size == 0: for cand in candidate_urls: try: probe = session.get(cand, headers=dict(dl_headers, Range='bytes=0-0'), timeout=(8, 15), stream=True) cr = probe.headers.get('Content-Range', '') if '/' in cr: total_size = int(cr.rsplit('/', 1)[1]) probe.close() active_url = cand break except Exception: continue range_header = self.headers.get('Range', '') start = 0 end = total_size - 1 if total_size > 0 else 0 is_partial = False if range_header and total_size > 0: m = re.match(r'bytes=(\d+)-(\d*)', range_header) if m: start = int(m.group(1)) if m.group(2): end = int(m.group(2)) is_partial = True if content_key is None: try: range_headers = dict(dl_headers) if is_partial: range_headers['Range'] = 'bytes=%d-%d' % (start, end) resp = None last_exc = None for cand in candidate_urls: try: resp = session.get(cand, headers=range_headers, stream=True, timeout=(8, 90)) resp.raise_for_status() break except Exception as exc: last_exc = exc resp = None continue if resp is None: raise last_exc or RuntimeError('all cdn nodes failed') content_length = int(resp.headers.get('Content-Length', 0)) if is_partial: self.send_response(206) self.send_header('Content-Range', resp.headers.get('Content-Range', 'bytes %d-%d/%d' % (start, end, total_size))) else: self.send_response(200) self.send_header('Content-Type', 'video/mp4') if content_length: self.send_header('Content-Length', str(content_length)) self.send_header('Accept-Ranges', 'bytes') self.send_header('Access-Control-Allow-Origin', '*') self.send_header('Access-Control-Expose-Headers', 'Content-Length, Content-Range, Accept-Ranges') self.end_headers() if head_only: return for chunk in resp.iter_content(65536): self.wfile.write(chunk) except Exception as exc: try: self.send_error(502, 'proxy failed: %s' % exc) except Exception: pass return def _stream_cenc_direct(self, raw_url, backup_urls, content_key, total_size_hint=0, filename='', filepath=None): session = requests.Session() session.trust_env = False dl_headers = { 'User-Agent': 'com.phoenix.read/71332', 'Referer': 'https://novel.snssdk.com/', } candidate_urls = [raw_url] + (backup_urls or []) resp = None last_exc = None for cand in candidate_urls: try: resp = session.get(cand, headers=dl_headers, stream=True, timeout=(8, 90)) resp.raise_for_status() break except Exception as exc: last_exc = exc resp = None continue if resp is None: raise last_exc or RuntimeError('all cdn nodes failed') total_size = int(resp.headers.get('Content-Length', 0)) or total_size_hint chunk_iter = resp.iter_content(256 * 1024) buf = bytearray() ftyp_size = 0 moov_end = 0 for chunk in chunk_iter: buf.extend(chunk) if len(buf) >= 8 and ftyp_size == 0: ftyp_size = struct.unpack('>I', buf[0:4])[0] if ftyp_size > 0 and len(buf) >= ftyp_size + 8 and moov_end == 0: moov_size = struct.unpack('>I', buf[ftyp_size:ftyp_size + 4])[0] moov_end = ftyp_size + moov_size if moov_end > 0 and len(buf) >= moov_end: break if moov_end == 0 or len(buf) < moov_end: raise RuntimeError('moov not found in first chunk') moov_data = bytearray(buf[ftyp_size + 8:moov_end]) for old, new in ((b'encv', b'hvc1'), (b'enca', b'mp4a')): _replace_fourcc(moov_data, old, new) _replace_sinf(moov_data) decrypt_map = {} t1_off, t1_sz = find_box(moov_data, 'trak', 0) t2_off, _ = find_box(moov_data, 'trak', t1_off + t1_sz) for t_off in (t1_off, t2_off): if t_off < 0: continue result = parse_track(moov_data, t_off) if result is None: continue sizes, offsets, cns, aux_off, aux_sz, ns = result if ns == 0: continue aux_data = None if aux_off + aux_sz <= len(buf): aux_data = bytes(buf[aux_off:aux_off + aux_sz]) else: try: aux_resp = session.get(cand, headers={ **dl_headers, 'Range': 'bytes=%d-%d' % (aux_off, aux_off + aux_sz - 1), }, timeout=10) aux_data = aux_resp.content except Exception: aux_data = None if not aux_data: continue ivs = [] for i in range(0, len(aux_data), 8): if i + 8 <= len(aux_data): ivs.append(aux_data[i:i + 8]) si = 0 for ci, chunk_off in enumerate(offsets): off = chunk_off for _ in range(cns[ci]): if si >= ns: break sz = sizes[si] iv = ivs[si] if si < len(ivs) else b'\x00' * 8 decrypt_map[off] = (sz, iv) off += sz si += 1 self.send_response(200) if total_size > 0: self.send_header('Content-Length', str(total_size)) self.send_header('Content-Type', 'video/mp4') self.send_header('Accept-Ranges', 'bytes') self.send_header('Access-Control-Allow-Origin', '*') self.send_header('Access-Control-Expose-Headers', 'Content-Length, Content-Range, Accept-Ranges') self.end_headers() header_bytes = bytes(buf[0:ftyp_size]) + buf[ftyp_size:ftyp_size + 8] + bytes(moov_data) part_path = filepath.with_suffix('.part') if filepath else None file_handle = None try: if part_path: file_handle = open(part_path, 'wb') except Exception: file_handle = None try: self.wfile.write(header_bytes) if file_handle: file_handle.write(header_bytes) current_pos = moov_end if len(buf) > moov_end: remaining = bytearray(buf[moov_end:]) decrypted = stream_cenc_decrypt_chunk( remaining, current_pos, decrypt_map, content_key) self.wfile.write(decrypted) self.wfile.flush() if file_handle: file_handle.write(decrypted) current_pos += len(remaining) buf = None for chunk in chunk_iter: chunk = bytearray(chunk) decrypted = stream_cenc_decrypt_chunk( chunk, current_pos, decrypt_map, content_key) self.wfile.write(bytes(decrypted)) self.wfile.flush() if file_handle: file_handle.write(decrypted) current_pos += len(chunk) if file_handle: file_handle.close() file_handle = None if part_path and part_path.exists() and part_path.stat().st_size > 0: part_path.replace(filepath) schedule_video_cleanup(filepath) print('[stream] streaming complete, cached %s' % filename) except BrokenPipeError: print('[stream] player disconnected mid-stream') except Exception as exc: print('[stream] write error: %s' % exc) finally: if file_handle: file_handle.close() if part_path and part_path.exists(): try: part_path.unlink() except Exception: pass def _stream_cenc(self, session, candidate_urls, dl_headers, content_key, total_size, req_start, req_end, is_partial, head_only): resp = None last_exc = None for cand in candidate_urls: try: resp = session.get(cand, headers=dl_headers, stream=True, timeout=(8, 90)) resp.raise_for_status() break except Exception as exc: last_exc = exc resp = None continue if resp is None: raise last_exc or RuntimeError('all cdn nodes failed') chunk_iter = resp.iter_content(256 * 1024) buf = bytearray() ftyp_size = 0 moov_end = 0 for chunk in chunk_iter: buf.extend(chunk) if len(buf) >= 8 and ftyp_size == 0: ftyp_size = struct.unpack('>I', buf[0:4])[0] if ftyp_size > 0 and len(buf) >= ftyp_size + 8 and moov_end == 0: moov_size = struct.unpack('>I', buf[ftyp_size:ftyp_size + 4])[0] moov_end = ftyp_size + moov_size if moov_end > 0 and len(buf) >= moov_end: break if moov_end == 0 or len(buf) < moov_end: print('[stream] moov parse failed, falling back to full download') for chunk in chunk_iter: buf.extend(chunk) decrypted = decrypt_mp4_cenc(buf, content_key) self._send_cenc_response(decrypted, total_size, req_start, req_end, is_partial, head_only) return moov_data = bytearray(buf[ftyp_size + 8:moov_end]) for old, new in ((b'encv', b'hvc1'), (b'enca', b'mp4a')): _replace_fourcc(moov_data, old, new) _replace_sinf(moov_data) decrypt_map = {} t1_off, t1_sz = find_box(moov_data, 'trak', 0) t2_off, _ = find_box(moov_data, 'trak', t1_off + t1_sz) for t_off in (t1_off, t2_off): if t_off < 0: continue result = parse_track(moov_data, t_off) if result is None: continue sizes, offsets, cns, aux_off, aux_sz, ns = result if ns == 0: continue aux_data = None if aux_off + aux_sz <= len(buf): aux_data = bytes(buf[aux_off:aux_off + aux_sz]) else: try: aux_resp = session.get(cand, headers={ **dl_headers, 'Range': 'bytes=%d-%d' % (aux_off, aux_off + aux_sz - 1), }, timeout=10) aux_data = aux_resp.content except Exception as exc: print('[stream] aux fetch failed: %s' % exc) aux_data = None if not aux_data: continue ivs = [] for i in range(0, len(aux_data), 8): if i + 8 <= len(aux_data): ivs.append(aux_data[i:i + 8]) si = 0 for ci, chunk_off in enumerate(offsets): off = chunk_off for _ in range(cns[ci]): if si >= ns: break sz = sizes[si] iv = ivs[si] if si < len(ivs) else b'\x00' * 8 decrypt_map[off] = (sz, iv) off += sz si += 1 if total_size > 0: content_length = req_end - req_start + 1 if is_partial: self.send_response(206) self.send_header('Content-Range', 'bytes %d-%d/%d' % (req_start, req_end, total_size)) else: self.send_response(200) self.send_header('Content-Length', str(content_length)) else: self.send_response(200) self.send_header('Content-Type', 'video/mp4') self.send_header('Accept-Ranges', 'bytes') self.send_header('Access-Control-Allow-Origin', '*') self.send_header('Access-Control-Expose-Headers', 'Content-Length, Content-Range, Accept-Ranges') self.end_headers() if head_only: return header_bytes = bytes(buf[0:ftyp_size]) + buf[ftyp_size:ftyp_size + 8] + bytes(moov_data) if req_start < moov_end: write_end = min(req_end + 1, moov_end) self.wfile.write(header_bytes[req_start:write_end]) current_pos = moov_end if len(buf) > moov_end: remaining = bytearray(buf[moov_end:]) decrypted_remaining = stream_cenc_decrypt_chunk( remaining, current_pos, decrypt_map, content_key) self._write_range_data(decrypted_remaining, current_pos, req_start, req_end) current_pos += len(remaining) buf = None for chunk in chunk_iter: chunk = bytearray(chunk) decrypted = stream_cenc_decrypt_chunk( chunk, current_pos, decrypt_map, content_key) if not self._write_range_data(decrypted, current_pos, req_start, req_end): break current_pos += len(chunk) def _write_range_data(self, data, data_start, req_start, req_end): data_end = data_start + len(data) if data_end <= req_start: return True if data_start > req_end: return False write_start = max(0, req_start - data_start) write_end = min(len(data), req_end - data_start + 1) if write_start < write_end: self.wfile.write(data[write_start:write_end]) return True def _send_cenc_response(self, data, total_size, req_start, req_end, is_partial, head_only): content_length = len(data) if is_partial: self.send_response(206) self.send_header('Content-Range', 'bytes %d-%d/%d' % (req_start, req_end, total_size or content_length)) self.send_header('Content-Length', str(req_end - req_start + 1)) else: self.send_response(200) self.send_header('Content-Length', str(content_length)) self.send_header('Content-Type', 'video/mp4') self.send_header('Accept-Ranges', 'bytes') self.send_header('Access-Control-Allow-Origin', '*') self.send_header('Access-Control-Expose-Headers', 'Content-Length, Content-Range, Accept-Ranges') self.end_headers() if head_only: return if is_partial: self.wfile.write(data[req_start:req_end + 1]) else: self.wfile.write(data) def _json(self, code, data): body = json.dumps(data, ensure_ascii=False).encode('utf-8') self.send_response(code) self.send_header('Content-Type', 'application/json; charset=utf-8') self.send_header('Content-Length', str(len(body))) self.send_header('Access-Control-Allow-Origin', '*') self.end_headers() self.wfile.write(body) def log_message(self, *args): pass def _find_free_port(preferred): for port in range(preferred, preferred + 50): try: with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: s.bind(('0.0.0.0', port)) return port except OSError: continue return 0 class Spider(Spider): # SITE 改为动态获取：get_site() 返回当前可用域名（支持自动探测/自定义） UA = ('Mozilla/5.0 (Linux; Android 12; TV) AppleWebKit/537.36 ' '(KHTML, like Gecko) Chrome/126.0 Safari/537.36') HEADERS = { 'User-Agent': UA, 'Accept-Language': 'zh-CN,zh;q=0.9', } DEFAULT_PORT = 9877 @staticmethod def SITE() -> str: return get_active_site() CATEGORY_CONFIG = { 'rank_hot': {'type_name': '红果热播榜', 'kind': 'rank', 'route': 'hot-drama'}, 'rank_human': {'type_name': '真人剧热播榜', 'kind': 'rank', 'route': 'hot-real-drama'}, 'rank_comic': {'type_name': '漫剧热播榜', 'kind': 'rank', 'route': 'hot-comic-drama'}, 'rank_ai': {'type_name': 'AI剧热播榜', 'kind': 'rank', 'route': 'hot-ai-drama'}, 'short': {'type_name': '短剧', 'kind': 'category', 'query': 'tab=1&sort_type=1'}, } CATEGORIES = [ {'type_id': k, 'type_name': v['type_name']} for k, v in CATEGORY_CONFIG.items() ] RANK_ROUTES = {'hot-drama', 'hot-real-drama', 'hot-comic-drama', 'hot-ai-drama'} SELECTOR_GROUPS = [ { 'key': 'background', 'name': '\u5168\u90e8\u80cc\u666f', 'items': [ ['\u73b0\u4ee3', 'cate_757'], ['\u90fd\u5e02', 'cate_1'], ['\u53e4\u4ee3', 'cate_758'], ['\u4e61\u6751', 'cate_11'], ['\u5e74\u4ee3', 'cate_79'], ['\u67b6\u7a7a', 'cate_452'], ['\u804c\u573a', 'cate_127'], ['\u6c11\u56fd', 'cate_390'], ['\u6821\u56ed', 'cate_4'], ['\u5bab\u5ef7', 'cate_1153'], ['\u8352\u5c9b', 'cate_1162'], ], }, { 'key': 'topic', 'name': '\u5168\u90e8\u4e3b\u9898', 'items': [ ['\u73b0\u8a00', 'cate_1021'], ['\u5973\u6027\u6210\u957f', 'cate_1048'], ['\u8111\u6d1e', 'cate_262'], ['\u5947\u5e7b', 'cate_1020'], ['\u7384\u5e7b', 'cate_1019'], ['\u53e4\u8a00', 'cate_439'], ['\u6218\u795e', 'cate_1038'], ['\u5bab\u6597', 'cate_246'], ['\u4ed9\u4fa0', 'cate_1013'], ['\u6743\u8c0b', 'cate_1047'], ['\u79cd\u7530', 'cate_1180'], ['\u5e74\u4ee3\u7231\u60c5', 'cate_1022'], ['\u559c\u5267', 'cate_303'], ['\u60ac\u7591', 'cate_165'], ['\u9752\u6625', 'cate_297'], ['\u5fd7\u602a', 'cate_1027'], ['\u6c11\u56fd\u7231\u60c5', 'cate_1025'], ['\u7075\u5f02', 'cate_751'], ['\u5bb6\u56fd\u60c5\u6000', 'cate_1235'], ['\u6cd5\u5f8b', 'cate_1136'], ['\u5211\u4fa6', 'cate_1148'], ['\u6297\u6218', 'cate_504'], ['\u6b66\u4fa0', 'cate_1172'], ['\u6c11\u56fd\u4f20\u5947', 'cate_1240'], ['\u52a8\u4f5c', 'cate_302'], ['\u6c42\u751f', 'cate_1168'], ['\u79d1\u5e7b', 'cate_1092'], ['\u6050\u6016', 'cate_1219'], ['\u5546\u6218', 'cate_1225'], ], }, { 'key': 'setting', 'name': '\u5168\u90e8\u8bbe\u5b9a', 'items': [ ['\u6253\u8138\u8650\u6e23', 'cate_1051'], ['\u5927\u7537\u4e3b', 'cate_1207'], ['\u5927\u5973\u4e3b', 'cate_760'], ['\u9a6c\u7532', 'cate_266'], ['\u91cd\u751f', 'cate_36'], ['\u7a7f\u8d8a', 'cate_37'], ['\u7cfb\u7edf', 'cate_19'], ['\u5148\u5a5a\u540e\u7231', 'cate_265'], ['\u5bb6\u957f\u91cc\u77ed', 'cate_862'], ['\u5c0f\u4eba\u7269', 'cate_1010'], ['\u7834\u955c\u91cd\u5706', 'cate_475'], ['\u795e\u8c6a', 'cate_20'], ['\u8c6a\u95e8', 'cate_936'], ['\u5f3a\u8005\u56de\u5f52', 'cate_1045'], ['\u5f02\u80fd', 'cate_598'], ['\u4f20\u627f\u89c9\u9192', 'cate_1007'], ['\u8650\u604b', 'cate_1008'], ['\u533b\u751f', 'cate_487'], ['\u5f3a\u5f3a\u8054\u5408', 'cate_1049'], ['\u8d58\u5a7f\u9006\u88ad', 'cate_1044'], ['\u751c\u5ba0', 'cate_96'], ['\u5a31\u4e50\u5708', 'cate_43'], ['\u795e\u533b', 'cate_26'], ['\u9752\u6885\u7af9\u9a6c', 'cate_387'], ['\u59d0\u5f1f\u604b', 'cate_762'], ['\u7384\u5b66', 'cate_929'], ['\u8ffd\u59bb\u706b\u846c\u573a', 'cate_616'], ['\u4e1a\u754c\u7cbe\u82f1', 'cate_1293'], ['\u4e00\u89c1\u949f\u60c5', 'cate_477'], ['\u798f\u5b9d', 'cate_1291'], ['\u635e\u504f\u95e8', 'cate_1287'], ['\u53cd\u6d3e\u4e3b\u89d2', 'cate_1042'], ['\u840c\u5ba0', 'cate_428'], ['\u65b9\u8a00', 'cate_1255'], ['\u53cc\u5411\u6551\u8d4e', 'cate_1200'], ['\u767d\u6708\u5149', 'cate_615'], ['\u7075\u9b42\u4e92\u6362', 'cate_831'], ['\u75c5\u5a07', 'cate_380'], ['\u66b4\u5bcc', 'cate_1191'], ['\u9ed1\u9053', 'cate_826'], ['\u4e27\u5c38', 'cate_582'], ['\u7279\u79cd\u5175', 'cate_375'], ], }, { 'key': 'gender', 'name': '\u5168\u90e8\u53d7\u4f17', 'items': [['\u7537\u9891', '1'], ['\u5973\u9891', '0']], }, { 'key': 'time', 'name': '\u5168\u90e8\u65f6\u95f4', 'items': [ ['7\u5929\u5185\u4e0a\u65b0', '1'], ['14\u5929\u5185\u4e0a\u65b0', '2'], ['30\u5929\u5185\u4e0a\u65b0', '3'], ['90\u5929\u5185\u4e0a\u65b0', '4'], ], }, { 'key': 'sort_type', 'name': '\u5168\u90e8\u63a8\u8350', 'items': [['\u6700\u65b0', '2'], ['\u6700\u70ed', '1']], }, ] PAGE_SIZE = 24 def __init__(self): self._cache = {} self._server = None self._server_port = self.DEFAULT_PORT self._server_started = False self._parser = None def getName(self): return '\u7ea2\u679c\u679c[\u77ed]' def init(self, extend=''): # 极速启动：不做任何网络请求，不启动内嵌服务 # 域名和播放服务改为懒加载（真正需要时才请求/启动） try: extra = extract_domains_from_extend(extend) if extra: _register_domain(extra[0]) # 不再做域名探测，直接使用候选列表中的默认域名 # 如果默认域名不可用，后续API请求会自然失败并触发重试机制 # 异步预连接服务器（建立TCP/TLS连接池，后续请求复用） threading.Thread(target=self._preconnect, daemon=True).start() print('[红果果] 极速模式启动，域名懒探测') except Exception as exc: print('[红果果] init异常:', exc) # 内嵌播放服务不在init时启动，改为playerContent时懒启动 # 这样进入接口时瞬间完成，不会阻塞10+秒 def _preconnect(self): """异步预连接到服务器，建立TCP/TLS连接池""" try: session = _get_http_session() domain = get_active_site() if domain: session.get(domain + '/', headers=self.HEADERS, timeout=3, verify=False) print('[红果果] 预连接成功: %s' % domain) except Exception as exc: print('[红果果] 预连接失败: %s' % exc) def _port_file(self): return Path(os.path.dirname(os.path.abspath(__file__))) / '.hongguo_embedded_port' def _read_registered_port(self): try: return int(self._port_file().read_text().strip()) except Exception: return 0 def _ensure_server(self): if self._server_started: try: with socket.create_connection(('127.0.0.1', self._server_port), timeout=1): return except OSError: self._server_started = False self._server = None reg_port = self._read_registered_port() if reg_port: try: with socket.create_connection(('127.0.0.1', reg_port), timeout=1): self._server_port = reg_port self._server_started = True global _CURRENT_DOMAIN _CURRENT_DOMAIN = 'http://127.0.0.1:%d' % reg_port return except OSError: pass try: self._start_embedded_server(self.DEFAULT_PORT) except Exception as exc: print('[红果果] 内嵌服务重启失败: %s' % exc) def _start_embedded_server(self, port): global _CURRENT_DOMAIN, _EMBEDDED_SERVER, _EMBEDDED_PORT if _EMBEDDED_SERVER is not None: try: with socket.create_connection(('127.0.0.1', _EMBEDDED_PORT), timeout=1): self._server = _EMBEDDED_SERVER self._server_port = _EMBEDDED_PORT self._server_started = True _CURRENT_DOMAIN = 'http://127.0.0.1:%d' % _EMBEDDED_PORT return except OSError: pass reg_port = self._read_registered_port() if reg_port and reg_port != port: try: with socket.create_connection(('127.0.0.1', reg_port), timeout=1): self._server_port = reg_port self._server_started = True _CURRENT_DOMAIN = 'http://127.0.0.1:%d' % reg_port return except OSError: pass actual_port = _find_free_port(port) if actual_port == 0: raise RuntimeError('no free port near %d' % port) public_url = 'http://127.0.0.1:%d' % actual_port _CURRENT_DOMAIN = public_url src_dir = Path(os.path.dirname(os.path.abspath(__file__))) / 'src' src_dir.mkdir(parents=True, exist_ok=True) _PlayHandler.src_dir = src_dir _PlayHandler.base_url = public_url os.environ['APP_PORT'] = str(actual_port) server = ThreadingHTTPServer(('0.0.0.0', actual_port), _PlayHandler) server_thread = threading.Thread(target=server.serve_forever, daemon=True) server_thread.start() _EMBEDDED_SERVER = server _EMBEDDED_PORT = actual_port try: self._port_file().write_text(str(actual_port)) except Exception: pass self._server = server self._server_port = actual_port self._server_started = True print('[\u7ea2\u679c\u679c] \u5185\u5d4c\u670d\u52a1\u5df2\u542f\u52a8: %s' % public_url) def isVideoFormat(self, url): return False def manualVideoCheck(self): return False def destroy(self): return def _get(self, url): try: resp = _get_http_session().get(url, headers=self.HEADERS, timeout=25, verify=False) resp.encoding = 'utf-8' return resp.text except Exception: return '' # ---- Modern.js loader 接口（2026-09 官网改版后唯一的稳定数据源） ---- # 页面路由 id： # 分类 /category -> category_page # 详情 /detail?series_id=xx -> detail_page # 搜索 /search/<关键词> -> search_(keyword)/page # 榜单 /rank/ -> rank_/page # 播放 /player// -> player_(series_id)/(vid)/page LOADER_TIMEOUT = 5 @staticmethod def _loader_url(path: str, route_id: str, params: dict = None) -> str: base = get_active_site() + path query = {'__loader': route_id, '__ssrDirect': 'true'} if params: query.update(params) return base + '?' + urlencode(query) def _loader_data(self, path: str, route_id: str, params: dict = None, allow_empty: bool = False): """请求 Modern.js loader 接口，返回 JSON 数据（失败返回 None 或抛异常）。""" url = self._loader_url(path, route_id, params) cached = self._cache.get(url) if cached and time.time() - cached[0] < 180: return cached[1] try: resp = _get_http_session().get( url, headers=dict(self.HEADERS, Referer=get_active_site() + '/'), timeout=self.LOADER_TIMEOUT, verify=False, ) resp.encoding = 'utf-8' if resp.status_code != 200: raise RuntimeError('loader HTTP %s: %s' % (resp.status_code, url)) text = resp.text # 多数接口是纯 JSON；榜单接口是 JSON + SSE(data:{...}) 延迟块： # 首段 JSON 里 content 为 "__deferred_promise:content"，真实数据在 data: 块里。 stripped = text.lstrip() if stripped.startswith('{'): try: data = json.loads(text) except Exception: data = json.JSONDecoder().raw_decode(stripped)[0] chunks = re.findall(r'^data:(\{.*)$', text, re.M) if chunks and isinstance(data, dict): merged = dict(data) for chunk in chunks: try: cd = json.loads(chunk) except Exception: continue content = cd.get('content') if isinstance(content, dict): merged['content'] = content for key in ('rankList', 'isSuccess'): if key in cd and key not in merged: merged[key] = cd[key] data = merged else: chunks = re.findall(r'^data:(\{.*)$', text, re.M) if not chunks: raise RuntimeError('loader 返回格式未知: ' + url) data = json.loads(chunks[0]) content = data.get('content') if isinstance(data.get('content'), dict) else data data = content except Exception as exc: if allow_empty: return None raise if isinstance(data, dict) and data.get('message') and not data.get('isSuccess'): if allow_empty: return None raise RuntimeError('loader 错误: %s (%s)' % (data.get('message'), url)) self._cache[url] = (time.time(), data) return data def _router_data(self, url): """兼容保留：官网已不再输出 _ROUTER_DATA，这里自动降级到 loader 接口。""" try: parsed = urlparse(url) query = parse_qs(parsed.query) path = parsed.path or '/' if path.startswith('/category'): params = {k: v[0] for k, v in query.items() if k not in ('__loader', '__ssrDirect', 'page')} page = query.get('page', ['1'])[0] if page and page != '1': params['page_num'] = page return self._loader_data('/category', 'category_page', params) if path.startswith('/detail'): sid = (query.get('series_id', [''])[0] or '') data = self._loader_data('/detail', 'detail_page', {'series_id': sid} if sid else None) # loader 返回的是平铺结构，包一层兼容旧的 loaderData 取法 return {'loaderData': {'detail_page': data or {}}} if path.startswith('/search/'): keyword = path[len('/search/'):] page = query.get('page', ['1'])[0] params = {'page': page} if page and page != '1' else None data = self._loader_data('/search/' + keyword, 'search_(keyword)/page', params, allow_empty=True) return {'loaderData': {'search_(keyword)/page': data or {}}} if path.startswith('/rank/'): route = path[len('/rank/'):] page = query.get('page', ['1'])[0] params = {'page': page} if page and page != '1' else None data = self._loader_data('/rank/' + route, 'rank_%s/page' % route, params, allow_empty=True) return {'rankList': (data or {}).get('rankList') or []} except Exception: pass raise RuntimeError('页面获取失败(官网已改版，无 _ROUTER_DATA): ' + url) @staticmethod def _clean_url(value): if not value: return '' return str(value).replace('\\/', '/').replace('\\u0026', '&').replace('&', '&') @staticmethod def _decode_html(value): if not value: return '' return (str(value) .replace('"', '"') .replace('/', '/').replace('/', '/') .replace(''', "'").replace(''', "'") .replace('<', '<').replace('>', '>') .replace('&', '&')) def _vod(self, item): source = item.get('video_data') if isinstance(item.get('video_data'), dict) else item tags = source.get('tags') or [] if isinstance(tags, list): tags = ' / '.join(str(t) for t in tags[:5] if t) count = source.get('episode_cnt') or source.get('series_episode_info', {}).get('episode_cnt') or 0 remark = str(source.get('episode_right_text') or '') if not remark and count: remark = '\u5168%s\u96c6' % count return { 'vod_id': str(source.get('series_id') or ''), 'vod_name': str(source.get('series_name') or source.get('series_title') or item.get('name') or item.get('title') or ''), 'vod_pic': self._clean_url(source.get('series_cover') or source.get('cover')), 'vod_remarks': remark, 'vod_tag': str(tags or ''), 'vod_content': str(source.get('series_intro') or source.get('description') or ''), } def _category_items(self, query, page=1): # 拆分 query：命名筛选参数直接透传；分页参数用 page（loader 接口忽略 page_num） params = {} for k, v in parse_qs(query).items(): if v and v[0] and k not in ('page', 'page_num'): params[k] = v[0] if page > 1: params['page'] = str(page) # 优化：对首页列表做短时缓存（非分页请求缓存10秒） cache_key = 'cat_%s_%d' % (query, page) if page == 1: cached = self._cache.get(cache_key) if cached and time.time() - cached[0] < 10: return cached[1] # 重试2次（减少等待） for attempt in range(2): try: data = self._loader_data('/category', 'category_page', params) items = (data or {}).get('recommendList') or [] seen = set() result = [] for item in items: sid = str(item.get('series_id') or '') if sid and sid not in seen: seen.add(sid) result.append(item) if page == 1: self._cache[cache_key] = (time.time(), result) return result except Exception as exc: print('[红果果] 分类列表重试(%d/2): %s' % (attempt + 1, exc)) if attempt < 1: time.sleep(0.5) else: raise return [] def _rank_items(self, route, page=1): params = {'page': str(page)} if page > 1 else None data = self._loader_data('/rank/' + route, 'rank_%s/page' % route, params, allow_empty=False) payload = data if isinstance(data, dict) else {} if isinstance(payload.get('content'), dict): payload = payload['content'] raw_list = payload.get('rankList') or [] if not payload.get('isSuccess', True) or not isinstance(raw_list, list): raise RuntimeError('榜单数据格式错误') result = [] for item in raw_list: episode_vids = item.get('episodeVids') or item.get('episode_vids') or [] vid_first = '' if isinstance(episode_vids, list) and episode_vids: vid_first = str(episode_vids[0]) vod = self._vod({ 'series_id': item.get('seriesId') or item.get('series_id') or item.get('id'), 'series_name': item.get('title') or item.get('seriesTitle') or item.get('series_name'), 'series_cover': item.get('cover') or item.get('seriesCover') or item.get('series_cover'), 'series_intro': item.get('description') or item.get('intro') or item.get('seriesIntro'), 'episode_cnt': item.get('episodeCount') or item.get('episode_cnt'), 'tags': item.get('tags'), }) if vod['vod_id'] and vod['vod_name']: if vid_first: vod['vod_id'] = vod['vod_id'] + '@' + vid_first result.append(vod) return result def _category_type(self, value): raw = str(value).replace('category?', '').replace('type_id=', '') raw = raw.split('@')[0] if raw in self.CATEGORY_CONFIG: return raw parsed = parse_qs(raw) route = parsed.get('rank', [''])[0] or parsed.get('route', [''])[0] if route in self.RANK_ROUTES: for k, v in self.CATEGORY_CONFIG.items(): if v.get('route') == route and v['kind'] == 'rank': return k return 'short' @staticmethod def _filter_values(extend): if not extend or not isinstance(extend, dict): return {} result = {} for k, v in extend.items(): if k == 'filters': continue if v is not None and str(v) and str(v) != '-1': result[k] = str(v) return result def _build_filters(self): filters = {} filter_groups = [] for group in self.SELECTOR_GROUPS: filter_groups.append({ 'key': group['key'], 'name': group['name'], 'value': [{'n': '\u5168\u90e8', 'v': ''}] + [ {'n': n, 'v': v} for n, v in group['items'] ], }) for type_id, cfg in self.CATEGORY_CONFIG.items(): if cfg['kind'] == 'category': filters[type_id] = [dict(g) for g in filter_groups] else: filters[type_id] = [] return filters def homeContent(self, filter): result = {'class': [dict(c) for c in self.CATEGORIES]} if filter: result['filters'] = self._build_filters() # 极速返回：先尝试缓存，未命中则异步加载不阻塞UI cached = self._cache.get('_home_video') if cached and time.time() - cached[0] < 30: result['list'] = cached[1].get('list', []) else: result['list'] = [] # 异步预加载首页视频（不阻塞返回） threading.Thread(target=self._preload_home_video, daemon=True).start() return result def homeVideoContent(self): # 优化：首页数据缓存30秒，避免每次请求都拉取 now = time.time() cached = self._cache.get('_home_video') if cached and now - cached[0] < 30: return cached[1] try: items = self._category_items('tab=1&sort_type=1')[:12] result = {'list': [self._vod(x) for x in items]} self._cache['_home_video'] = (now, result) return result except Exception as exc: print('[红果果] 首页读取失败:', exc) # 不做域名探测重试，直接返回空列表 # 域名探测会阻塞10+秒，用户会感觉卡死 # 如果域名真的挂了，用户切换源即可 return {'list': []} def _preload_home_video(self): """异步预加载首页视频数据（不阻塞UI）""" try: items = self._category_items('tab=1&sort_type=1')[:12] result = {'list': [self._vod(x) for x in items]} self._cache['_home_video'] = (time.time(), result) print('[红果果] 首页视频异步加载完成: %d条' % len(items)) except Exception as exc: print('[红果果] 首页视频异步加载失败: %s' % exc) def categoryContent(self, tid, pg, filter, extend): page = max(1, int(pg or 1)) raw_id = str(tid or 'short') type_id = self._category_type(raw_id) config = self.CATEGORY_CONFIG[type_id] requested = self._filter_values(extend or {}) try: if config['kind'] == 'rank': items = self._rank_items(config['route'], page) return {'list': items, 'page': page, 'pagecount': 1, 'limit': len(items), 'total': len(items)} query = config['query'] if '=' in raw_id and raw_id not in self.CATEGORY_CONFIG: parsed = parse_qs(raw_id.replace('category?', '')) for k, v_list in parsed.items(): query = self._set_param(query, k, v_list[0]) for k, v in requested.items(): query = self._set_param(query, k, v) if page > 1: query = self._set_param(query, 'page', str(page)) items = self._category_items(query, page) vods = [self._vod(x) for x in items] page_count = max(1, page + 1) if len(vods) >= self.PAGE_SIZE else page return {'list': vods, 'page': page, 'pagecount': page_count, 'limit': self.PAGE_SIZE, 'total': page * self.PAGE_SIZE + len(vods)} except Exception as exc: print('[红果果] 分类读取失败:', exc) return {'list': [], 'page': page, 'pagecount': 1, 'limit': 0, 'total': 0} @staticmethod def _set_param(query, key, value): params = parse_qs(query) params[key] = [value] return urlencode({k: v[0] for k, v in params.items()}) def detailContent(self, ids): try: # 兼容榜单 id 格式 "sid@vid"、自定义 tid "category?..." 等，取第一段作为 series_id series_id = str(ids[0]).split('@')[0].split('#')[0] if series_id.startswith('category?'): series_id = parse_qs(series_id.replace('category?', '')).get('type_id', [series_id])[0] series = {} vids = [] # 策略1: 尝试旧loader接口 try: data = self._loader_data('/detail', 'detail_page', {'series_id': series_id}) series = (data or {}).get('seriesDetail') or {} vids = series.get('vid_list') or [] vids = [str(v) for v in vids if str(v)] except Exception as exc: print('[红果果] 详情loader失败: %s' % exc) # 策略2: 用搜索API通过series_id查找 if not vids: try: search_items = self._search_api(series_id, 1) for item in search_items: vd = item.get('video_data') or {} if str(vd.get('series_id', '')) == str(series_id): series = vd vids = vd.get('vid_list') or [] vids = [str(v) for v in vids if str(v)] if not series.get('series_name') and item.get('name'): series['series_name'] = item.get('name') break if not vids and search_items: series = search_items[0].get('video_data') or {} vids = series.get('vid_list') or [] vids = [str(v) for v in vids if str(v)] except Exception as exc: print('[红果果] 详情搜索API失败: %s' % exc) # 策略3: 从分类列表查找 if not vids: try: cat_items = self._category_items('tab=1&sort_type=1') for item in cat_items: if str(item.get('series_id', '')) == str(series_id): series = item vids = item.get('vid_list') or [] vids = [str(v) for v in vids if str(v)] break except Exception as exc: print('[红果果] 详情分类回退失败: %s' % exc) if not vids: return {'list': []} try: accessible = int(series.get('accessible_episode_cnt') or 0) except (TypeError, ValueError): accessible = 0 if accessible <= 0: accessible = len(vids) rows = [] try: rows = fetch_quality_rows(vids[0]) except Exception as exc: print('[红果果] 清晰度列表获取失败:', exc) if not rows: rows = [{'key': 'auto', 'quality': 'auto'}] sources = [] for row in rows: qkey = row.get('key', 'auto') qname = _quality_label(row.get('quality', qkey)) episodes = [] for idx, vid in enumerate(vids): ep = idx + 1 label = '第%d集' % ep episodes.append('%s$%s|%s' % (label, vid, qkey)) sources.append({'name': qname, 'episodes': '#'.join(episodes)}) tags = series.get('tags') or [] if isinstance(tags, list): tags = ' / '.join(str(t) for t in tags[:5] if t) count = series.get('episode_cnt') or len(vids) remark = str(series.get('episode_right_text') or '') if not remark and count: remark = '全%s集' % count vod = { 'vod_id': series_id, 'vod_name': str(series.get('series_name') or series.get('series_title') or '红果短剧'), 'vod_pic': self._clean_url(series.get('series_cover')), 'type_name': str(tags or ''), 'vod_remarks': remark, 'vod_content': str(series.get('series_intro') or ''), 'vod_play_from': '$$$'.join(s['name'] for s in sources), 'vod_play_url': '$$$'.join(s['episodes'] for s in sources), } return {'list': [vod]} except Exception as exc: print('[红果果] 详情读取失败:', exc) return {'list': []} def _search_api(self, keyword, page=1): """使用新API搜索短剧（2026-10修复：官网搜索页改为SPA，旧loader接口返回HTML失效，改用/api/search/result） 增强版：支持多域名自动切换 + 第三方接口兜底 + 本地缓存""" # 先查本地缓存 cache_key = '%s_%d' % (keyword, page) with _SEARCH_CACHE_LOCK: cached = _SEARCH_CACHE.get(cache_key) if cached and time.time() - cached[0] < _SEARCH_CACHE_TTL: print('[红果果] 搜索缓存命中: %s' % keyword) return cached[1] # 策略1: 尝试所有候选域名（优先试当前活跃域名，快速命中） all_domains = list(_CANDIDATE_DOMAINS) # 把当前活跃域名移到最前面，减少首次连接等待 active = get_active_site() if active in all_domains: all_domains.remove(active) all_domains.insert(0, active) for domain in all_domains: try: url = domain + '/api/search/result' params = {'query': keyword} if page > 1: params['page'] = str(page) # 第一个域名（活跃域名）用较短超时快速失败 if domain == active: req_timeout = min(5, self.LOADER_TIMEOUT) else: req_timeout = self.LOADER_TIMEOUT resp = _get_http_session().get( url, params=params, headers=dict(self.HEADERS, Referer=domain + '/'), timeout=req_timeout, verify=False, ) resp.encoding = 'utf-8' if resp.status_code != 200: print('[红果果] 搜索API HTTP错误(%s): %s' % (domain, resp.status_code)) continue data = resp.json() search_list = data.get('searchList') or [] print('[红果果] 搜索API[%s]返回 %d 条结果' % (domain, len(search_list))) # 缓存成功结果 with _SEARCH_CACHE_LOCK: _SEARCH_CACHE[cache_key] = (time.time(), search_list) return search_list except Exception as exc: print('[红果果] 搜索API[%s]异常: %s' % (domain, exc)) continue # 策略2: 尝试第三方聚合接口（官网全部不可用时兜底） try: params = {'name': keyword, 'page': str(page)} resp = _get_http_session().get( _THIRD_SEARCH_API, params=params, timeout=_THIRD_SEARCH_TIMEOUT, verify=False, ) resp.encoding = 'utf-8' if resp.status_code == 200: data = resp.json() if data.get('code') == 200: third_party_items = data.get('data', []) print('[红果果] 第三方搜索返回 %d 条结果' % len(third_party_items)) # 转换第三方格式为官网格式 converted = [] for item in third_party_items: converted.append({ 'name': item.get('title', ''), 'video_data': { 'series_id': item.get('book_id', ''), 'series_name': item.get('title', ''), 'series_cover': item.get('cover', ''), 'episode_cnt': item.get('episode_cnt', 0), }, }) with _SEARCH_CACHE_LOCK: _SEARCH_CACHE[cache_key] = (time.time(), converted) return converted except Exception as exc: print('[红果果] 第三方搜索异常: %s' % exc) # 全部失败 print('[红果果] 所有搜索接口均失败') return [] def searchContent(self, key, quick, pg=1): page = max(1, int(pg or 1)) word = str(key).strip() if not word: return {'list': [], 'page': page} # 策略1: 使用增强版搜索API（多域名重试 + 第三方兜底 + 缓存） try: items = self._search_api(word, page) result = [] for item in items: # 新API返回的item: {name, keyword(series_id), video_data{...}} vod = self._vod(item.get('video_data') or item) if not vod['vod_name']: vod['vod_name'] = item.get('name', '') if vod['vod_id'] and vod['vod_name']: result.append(vod) if result: print('[红果果] 搜索成功: %s -> %d条' % (word, len(result))) return {'list': result, 'page': page} except Exception as exc: print('[红果果] 搜索API失败:', exc) # 策略2: 从分类列表本地搜索（多域名重试） try: all_items = [] # 尝试多次获取分类列表 for attempt in range(2): try: items = self._category_items('tab=1&sort_type=1') if items: all_items.extend(items) break except Exception as exc: print('[红果果] 分类列表获取失败(尝试%d): %s' % (attempt + 1, exc)) time.sleep(0.5) if all_items: keyword = word.lower() matches = [ x for x in all_items if keyword in str(x.get('series_name') or '').lower() or keyword in str(x.get('series_intro') or '').lower() ] per_page = 20 start = (page - 1) * per_page print('[红果果] 本地搜索: %s -> %d条匹配' % (word, len(matches))) return { 'list': [self._vod(x) for x in matches[start:start + per_page]], 'page': page, } except Exception as exc: print('[红果果] 搜索本地回退失败:', exc) # 全部失败 print('[红果果] 搜索无结果: %s' % word) return {'list': [], 'page': page} def searchContentPage(self, key, quick, pg=1): return self.searchContent(key, quick, pg) def playerContent(self, flag, pid, vipFlags): raw = str(pid).split('#')[0] quality_key = 'auto' if '|' in raw: vid_part, quality_key = raw.split('|', 1) quality_key = quality_key or 'auto' else: vid_part = raw # 详情集格式: "第N集$vid|qkey"；榜单集格式: "第N集$sid@vid" vid, sid = vid_part, '' if '@' in vid_part: sid_part, vid = vid_part.split('@', 1) sid = sid_part.split('|')[0] if '|' in sid_part else sid_part if '|' in vid_part: quality_key = vid_part.split('|', 1)[1] or 'auto' vid = vid.split('|', 1)[0] elif '_' in vid_part: vid, sid = vid_part.split('_', 1) self._ensure_server() result = { 'parse': 0, 'playUrl': '', 'url': '', 'header': { 'User-Agent': self.UA, 'Referer': get_active_site() + '/', }, } if sid: try: data = self._loader_data( '/player/%s/%s' % (sid, vid), 'player_(series_id)/(vid)/page', None, allow_empty=True, ) vpi = (data or {}).get('video_player_info') or {} play_url = str(vpi.get('main_url') or '') if play_url: play_url = self._clean_url(play_url) result['url'] = play_url result['header'] = {'User-Agent': self.UA} return result except Exception: pass if self._server_started: try: resolved = handle_video_request(str(vid), None, max_retries=3, stream_mode=True, quality_key=quality_key) stream_url = str(resolved.get('url') or '') if stream_url and stream_url.startswith('http'): result['url'] = stream_url result['header'] = {'User-Agent': self.UA} return result except Exception as exc: print('[playerContent] resolve_direct_failed: %s' % exc) play_url = 'http://127.0.0.1:%d/play?%s' % ( self._server_port, urlencode({'vid': vid})) result['url'] = play_url return result def localProxy(self, params): return None