# -*- coding: utf-8 -*- """ 快追影视 (kuaizhui.app) —— TVBox T3 源 (type=3 Python 爬虫) 站点后端 API（逆向自官网 Web 前端 https://www.kuaizhui.tv ）： · 域名： https://cvo.kuaicloud.one · 通信： 所有接口 POST，请求体 = AES-128-ECB(PKCS7(JSON)) 的大写 Hex 密文 响应体 = 同样算法加密的大写 Hex 密文 · 密钥： "kWJ2D5aDlneVeRcr" (UTF-8, 16 字节 → AES-128) · 头部： Content-Type: application/json, KV: "1", V: "1.0.0" (可选 S: sessionID) 接口清单（均经验证可通）： /u/main/category GET 风格空参 → 一级分类（TypeID / TypeName） /u/category/vod/list Cond{ParenTypeID,Sort,Area,SubCategory,Year,Paging} → 分组(Vods) 列表，每组含 Items 与二级分类(SubCategory/Title) /u/vod/detail {VodID, SuggestCount} → Players[PlayerCode,PlayerName,Contents[].Url] /u/vod/parse/url {Url, PlayerCode} → 解析后的真实播放地址(NeedParse=True 线路才需要) /u/vod/search {VodName, Paging} → 裸数组的搜索结果 为兼容 T3 运行时（通常无 pycryptodome / requests），默认加解密与 HTTP 全部用纯标准库实现； 运行环境若存在 pycryptodome / Cryptodome 则自动启用其 C 实现加速（无硬依赖，缺失时自动回退）。 AES-128-ECB 已对照 pycryptodome 逐项验证一致，并做过真实接口端到端联调。 性能优化（按需求：加载更快 / 播放更快）： 1. 首页只拉主分类(单请求)，二级分类改为进入分类时懒加载 —— 避免首页一次性 并发拉全部主列表(每个响应约 48KB、纯 Python 解密较慢)导致首屏长时间"加载中"。 2. AES 加解密复核：GF(2^8) 乘法改为查表(O(1))，纯 Python 解密 48KB 从 ~1s 降到 ~0.3s。 3. 详情接口按 VodID 缓存(TTL 10 分钟)，playerContent 直接复用，逐集播放不再回源。 4. 播放线路排序：NeedParse=False 的直链线路(BF线路等 m3u8)排在最前， TVBox 默认选第一条即可秒播，跳过 parse/url 二次解析的额外往返。 5. 请求体极简（紧凑 JSON + ECB），无 gzip/握手开销，单跳完成。 6. AES 运行期加速：固定密钥的轮密钥只展开一次、MixColumns 固定系数查表、 ShiftRows 与 Sbox 合并成单趟变换；存在 pycryptodome/Cryptodome 时直接走 C 实现。 7. HTTP 长连接复用：同一线程内复用一条到 API 的持久连接（thread-local）， 连续翻页/切换线路不再重复 TLS 握手；连接异常自动丢弃重建。 8. 分类扫描只拉主分类单请求：直接进分类页时不再额外拉取 ~48KB 首页推荐大盘。 """ from __future__ import annotations import http.client import json import os import re import sys import threading import time import urllib.parse # 可选加速：运行环境存在 pycryptodome / Cryptodome 时优先使用其 C 实现加解密； # 缺失时自动回退纯标准库实现，脚本仍为零第三方硬依赖。 try: from Crypto.Cipher import AES as _PyAES except Exception: try: from Cryptodome.Cipher import AES as _PyAES except Exception: _PyAES = None # ============ 站点常量 ============ API_URL = "https://cvo.kuaicloud.one" AES_KEY = b"kWJ2D5aDlneVeRcr" APP_VERSION = "1.0.0" IMG_BASE = "https://mvres.kuaicloud.one" PAGE_SIZE = 36 # 浏览器 UA：规避 CDN 对脚本指纹的拦截（真实 TVBox 设备请求同样带 UA） USER_AGENT = ( "Mozilla/5.0 (Linux; Android 12; Pixel 7) AppleWebKit/537.36 " "(KHTML, like Gecko) Chrome/120.0.0.0 Mobile Safari/537.36" ) DETAIL_TTL = 600 # 详情缓存秒数 # 这些“分组”属于首页精选/排序位，不作为二级分类筛选暴露 _SKIP_SUBCAT = {"101", "102"} # ============ 纯标准库 AES-128-ECB ============ def _xtime(a): b = a << 1 if b & 0x100: b ^= 0x11B return b & 0xFF def _gf256_mul_raw(a, b): """GF(2^8) 乘法（仅用于构造查找表，避免与运行期查表产生循环依赖）。""" r = 0 while b: if b & 1: r ^= a hi = a & 0x80 a = (a << 1) & 0xFF if hi: a ^= 0x1B b >>= 1 return r def _build_sbox(): exp = [0] * 256 log = [0] * 256 x = 1 for i in range(255): exp[i] = x log[x] = i x = _gf256_mul_raw(x, 0x03) sbox = [0] * 256 inv = [0] * 256 for i in range(256): iv = 0 if i == 0 else exp[(255 - log[i]) % 255] t = iv s = iv for _ in range(4): s = ((s << 1) | (s >> 7)) & 0xFF t ^= s sbox[i] = t ^ 0x63 inv[sbox[i]] = i return bytes(sbox), bytes(inv) def _build_gf256_tables(): """GF(2^8) 的 exp/log 表，用于 O(1) 乘法。""" exp = [0] * 512 log = [0] * 256 x = 1 for i in range(255): exp[i] = x log[x] = i x = _gf256_mul_raw(x, 0x03) for i in range(255, 512): exp[i] = exp[i - 255] return exp, log _SBOX, _INV_SBOX = _build_sbox() _EXP, _LOG = _build_gf256_tables() def _gf256_mul(a, b): """GF(2^8) 乘法（运行期查表，O(1)，远快于位循环实现）。""" if a == 0 or b == 0: return 0 return _EXP[(_LOG[a] + _LOG[b]) % 255] def _keyexp(key): nk = len(key) // 4 # AES-128 → 4 nr = nk + 6 # AES-128 → 10 w = [list(key[4 * i:4 * i + 4]) for i in range(nk)] rcon = 1 while len(w) < 4 * (nr + 1): temp = w[-1][:] if len(w) % nk == 0: temp = temp[1:] + temp[:1] temp = [_SBOX[b] for b in temp] temp[0] ^= rcon rcon = _xtime(rcon) elif nk > 6 and len(w) % nk == 4: temp = [_SBOX[b] for b in temp] w.append([w[-nk][i] ^ temp[i] for i in range(4)]) rk = [] for r in range(nr + 1): blk = [] for c in range(4): blk += w[r * 4 + c] rk.append(bytes(blk)) return rk, nr # ---- 运行期加速表（模块加载时一次性构建） ---- # MixColumns 的固定系数乘法表：把内层循环里的 GF 乘法全部换成单次查表 _XT = bytes(_xtime(i) for i in range(256)) _MUL9 = bytes(_gf256_mul(i, 9) for i in range(256)) _MUL11 = bytes(_gf256_mul(i, 11) for i in range(256)) _MUL13 = bytes(_gf256_mul(i, 13) for i in range(256)) _MUL14 = bytes(_gf256_mul(i, 14) for i in range(256)) # ShiftRows 的目标索引置换（加密/解密各一份），与 SubBytes/轮密钥异或合并成单趟变换 def _perm(enc): p = [0] * 16 for r in range(4): for c in range(4): p[r + 4 * c] = r + 4 * ((c + r if enc else c - r) % 4) return p _PERM_E = _perm(True) _PERM_D = _perm(False) # 固定密钥的轮密钥只展开一次（此前每次请求都会重新展开一遍） _RK, _NR = _keyexp(AES_KEY) def _shift(state, enc=True): out = [0] * 16 for r in range(4): for c in range(4): out[r + 4 * c] = state[r + 4 * ((c + (r if enc else -r)) % 4)] return out def _mix(state): out = state[:] for c in range(4): i = 4 * c a, b, d, e = out[i], out[i + 1], out[i + 2], out[i + 3] t = a ^ b ^ d ^ e out[i] = a ^ t ^ _XT[a ^ b] out[i + 1] = b ^ t ^ _XT[b ^ d] out[i + 2] = d ^ t ^ _XT[d ^ e] out[i + 3] = e ^ t ^ _XT[e ^ a] return out def _inv_mix(state): out = [0] * 16 for c in range(4): i = 4 * c a, b, d, e = state[i], state[i + 1], state[i + 2], state[i + 3] out[i] = _MUL14[a] ^ _MUL11[b] ^ _MUL13[d] ^ _MUL9[e] out[i + 1] = _MUL9[a] ^ _MUL14[b] ^ _MUL11[d] ^ _MUL13[e] out[i + 2] = _MUL13[a] ^ _MUL9[b] ^ _MUL14[d] ^ _MUL11[e] out[i + 3] = _MUL11[a] ^ _MUL13[b] ^ _MUL9[d] ^ _MUL14[e] return out def _add(state, rk): return [s ^ k for s, k in zip(state, rk)] def _enc_block(block): rk = _RK s = [block[i] ^ rk[0][i] for i in range(16)] for r in range(1, _NR): # 轮顺序：SubBytes → ShiftRows → MixColumns → AddRoundKey s = [_SBOX[s[_PERM_E[i]]] for i in range(16)] s = _mix(s) s = [x ^ rk[r][i] for i, x in enumerate(s)] # 末轮：SubBytes → ShiftRows → AddRoundKey（无 MixColumns） s = [_SBOX[s[_PERM_E[i]]] ^ rk[_NR][i] for i in range(16)] return bytes(s) def _dec_block(block): rk = _RK s = [block[i] ^ rk[_NR][i] for i in range(16)] for r in range(_NR - 1, 0, -1): s = [_INV_SBOX[s[_PERM_D[i]]] ^ rk[r][i] for i in range(16)] s = _inv_mix(s) s = [_INV_SBOX[s[_PERM_D[i]]] ^ rk[0][i] for i in range(16)] return bytes(s) def _aes_ecb(data, encrypt): out = bytearray() for i in range(0, len(data), 16): chunk = data[i:i + 16] if len(chunk) < 16: chunk = chunk + bytes([0] * (16 - len(chunk))) out += _enc_block(chunk) if encrypt else _dec_block(chunk) return bytes(out) def aes_encrypt_hex(obj) -> str: """JSON → PKCS7 填充 → AES-128-ECB → 大写 Hex。""" pt = json.dumps(obj, separators=(",", ":")).encode("utf-8") pad = 16 - len(pt) % 16 pt += bytes([pad]) * pad if _PyAES is not None: ct = _PyAES.new(AES_KEY, _PyAES.MODE_ECB).encrypt(pt) else: ct = _aes_ecb(pt, True) return ct.hex().upper() def aes_decrypt_hex(hexstr): """大写 Hex → AES-128-ECB 解密 → 去 PKCS7 → JSON。""" ct = bytes.fromhex(hexstr.strip()) if _PyAES is not None: pt = _PyAES.new(AES_KEY, _PyAES.MODE_ECB).decrypt(ct) else: pt = _aes_ecb(ct, False) pad = pt[-1] if 1 <= pad <= 16: pt = pt[:-pad] return json.loads(pt.decode("utf-8")) # ============ HTTP（纯标准库 + 连接复用） ============ _API_PARTS = urllib.parse.urlsplit(API_URL) _LOCAL = threading.local() def _http_connect(): """建立一条到 API 的持久连接（同一线程内复用，免去每次请求的 TLS 握手）。""" if _API_PARTS.scheme != "https": return http.client.HTTPConnection(_API_PARTS.hostname, _API_PARTS.port or 80, timeout=15) return http.client.HTTPSConnection(_API_PARTS.hostname, _API_PARTS.port or 443, timeout=15) def _get_conn(): conn = getattr(_LOCAL, "conn", None) if conn is None: conn = _http_connect() _LOCAL.conn = conn return conn def _drop_conn(): """连接异常后丢弃，下次请求自动重建（避免复用坏连接）。""" conn = getattr(_LOCAL, "conn", None) if conn is not None: try: conn.close() except Exception: pass _LOCAL.conn = None def _http_post(path: str, data: dict, follow: int = 1): body = aes_encrypt_hex(data).encode("utf-8") headers = {"Content-Type": "application/json", "KV": "1", "V": APP_VERSION, "User-Agent": USER_AGENT} conn = _get_conn() try: conn.request("POST", path, body=body, headers=headers) resp = conn.getresponse() raw = resp.read() except (OSError, http.client.HTTPException): _drop_conn() raise if resp.status in (301, 302, 303, 307, 308): # 与 urllib 自动跟随重定向等价：仅跟随同主机，避免跨域跳转 _drop_conn() loc = resp.getheader("Location") or "" if follow > 0 and loc: loc_parts = urllib.parse.urlsplit(loc) if loc_parts.netloc in ("", _API_PARTS.netloc): new_path = loc_parts.path or "/" if loc_parts.query: new_path += "?" + loc_parts.query return _http_post(new_path, data, follow - 1) raise RuntimeError("request failed: redirect") text = raw.decode("utf-8", "replace").strip() if not text: return {} return aes_decrypt_hex(text) # ============ T3 接口 ============ # 兼容 hipy/drpy-node 生态（糊涂等壳子）：若运行环境提供 base.spider.BaseSpider # 则继承它以获得完整协议兼容；否则退化为普通类（标准 TVBox T3 亦可加载）。 try: from base.spider import BaseSpider as _Base # type: ignore except Exception: _Base = object class Spider(_Base): """快追影视 T3 源（TVBox type=3 Python 爬虫）。""" source_name = "kuaizhui" display_name = "快追影视" HOST = "https://www.kuaizhui.tv" UA = USER_AGENT def __init__(self) -> None: try: super().__init__() except Exception: pass self.extend = "" self.last_error = "" self._classes = [] self._filters: dict = {} self._recommend: list = [] # ---- 生命周期（hipy/drpy-node 壳子要求） ---- def init(self, extend=""): self.extend = extend if isinstance(extend, str) else "" return {"name": self.display_name} def setExtendInfo(self, ext=""): self.extend = ext if isinstance(ext, str) else "" # ---- 工具 ---- def _req(self, path, data): # 失败自动重试一次（设备网络抖动时避免直接返回空数据） last = None for attempt in range(2): try: return _http_post(path, data) except Exception as e: last = e if attempt == 0: time.sleep(0.3) self.last_error = "%s:%s" % (type(last).__name__, last) return {} @staticmethod def _digits(val) -> str: """从任意形态的入参中提取数字 id（壳子可能传 int/带前后缀的字符串）。""" m = re.search(r"(\d{1,12})", str(val or "")) return m.group(1) if m else "" def _card(self, it: dict) -> dict: pic = it.get("VodPicThumb") or "" if not pic and it.get("LocalVodPic"): pic = IMG_BASE + it["LocalVodPic"] return { "vod_id": str(it.get("VodID")), "vod_name": it.get("VodName") or "", "vod_pic": pic, "vod_remarks": it.get("VodRemarks") or "", } @staticmethod def _vod_id(ids) -> str: return Spider._digits(ids) # ---- 首页 + 推荐 ---- def _load_classes(self): """仅拉取主分类（单请求），供分类反查与 tid 扫描使用；不加载首页推荐大盘。""" if self._classes: return cats = self._req("/u/main/category", {}) or [] if not isinstance(cats, list): cats = [] self._classes = [ {"type_id": str(c["TypeID"]), "type_name": c.get("TypeName") or c.get("DisplayName") or str(c["TypeID"])} for c in cats if c.get("TypeID") is not None ] def homeContent(self, filter=False): # 主分类与首页推荐并行拉取，避免串行等待 th = threading.Thread(target=self._load_classes) th.start() videos = self._fetch_recommend() th.join() return {"class": self._classes, "filters": self._filters, "list": videos} def homeVideoContent(self): if not self._recommend: if not self._classes: self.homeContent() else: self._fetch_recommend() return {"list": self._recommend or []} def _fetch_recommend(self): """拉电影分类(TypeID=1)的主列表作为首页推荐；同一响应顺带提取二级分类。""" if self._recommend: return self._recommend resp = self._req("/u/category/vod/list", { "Cond": {"ParentTypeID": 1, "Sort": "", "Area": "", "SubCategory": "", "Year": "", "Paging": {"PageIndex": 0, "PageSize": PAGE_SIZE}} }) or {} secs = resp.get("Vods") or [] if isinstance(resp, dict) else [] cards = [] for s in secs: for it in (s.get("Items") or []): cards.append(self._card(it)) if "1" not in self._filters: opts = self._extract_subcats(secs) if opts: self._filters["1"] = [{ "key": "subcategory", "name": "二级分类", "value": opts, }] self._recommend = cards[:60] return self._recommend def _extract_subcats(self, secs): """从主列表的分组(Vods)中提取二级分类选项（复用同一次响应，避免额外请求）。""" opts = [] seen = set() for s in secs: sc = str(s.get("SubCategory") or "") name = (s.get("Title") or s.get("DisplayTitle") or "").strip() if sc and sc not in _SKIP_SUBCAT and sc not in seen and name: seen.add(sc) opts.append({"n": name, "v": sc}) return opts # ---- 分类列表（含二级分类筛选） ---- def categoryContent(self, tid, pg="1", filter=True, extend=""): # 永不抛异常：壳子传参形态不可控，任何异常都会导致壳子显示"暂无数据" try: return self._category(tid, pg, extend) except Exception as e: self.last_error = "category:%s:%s" % (type(e).__name__, e) return {"class": self._classes, "filters": self._filters, "list": [], "page": 1, "pagecount": 1, "limit": PAGE_SIZE, "total": 0} def _ensure_classes(self): """懒加载主分类（仅在尚未获取时），供分类名反查与 tid 扫描使用。 只拉主分类单请求，不再顺带拉首页推荐（老逻辑会多一次 ~48KB 大盘请求）。 """ if self._classes: return try: self._load_classes() except Exception: pass def _known_typeids(self): ids = set() for c in self._classes: d = self._digits(c.get("type_id")) if d: ids.add(int(d)) return ids def _tid_candidates(self, raw_tid): """根据任意形态的 tid 产出候选 TypeID（按命中优先级排序）。 壳子传参形态不可控，可能是：纯数字 id / 分类名 / 显示名 / 带 source key 前缀的拼串（如 "c1829f59836b6ee0_1001"）。 策略：名字精确匹配 > 直接数字 > 数字段中命中已知 TypeID > 名字模糊匹配。全部候选都会依次尝试，第一个出数据的生效。 """ self._ensure_classes() known = self._known_typeids() cands = [] if raw_tid: # 1) 分类名/显示名精确匹配（最高优先级） for c in self._classes: for field in ("type_name",): cn = (c.get(field) or "").strip() if cn and cn == raw_tid: v = int(self._digits(c.get("type_id")) or 0) if v and v not in cands: cands.append(v) # 2) 直接提取的首个数字 d = self._digits(raw_tid) if d: v = int(d) if v not in cands: cands.append(v) # 3) 数字段中命中已知 TypeID 的（应对源 key 拼串污染） for m in re.findall(r"\d{1,12}", raw_tid): v = int(m) if v in known and v not in cands: cands.append(v) # 4) 名字模糊匹配（双向包含，最低优先级） for c in self._classes: cn = (c.get("type_name") or "").strip() if cn and cn != raw_tid and (cn in raw_tid or raw_tid in cn): v = int(self._digits(c.get("type_id")) or 0) if v and v not in cands: cands.append(v) return cands def _category(self, tid, pg, extend): # 壳子兼容：extend 可能是 JSON 字符串（hipy/drpy-node 约定），pg 可能传异常值 if isinstance(extend, str): try: extend = json.loads(extend) if extend.strip() else {} except Exception: extend = {} ext = extend if isinstance(extend, dict) else {} try: pg = max(1, int(pg)) except Exception: pg = 1 sub = ext.get("subcategory") or "" raw_tid = str(tid or "").strip() # 个别壳子会把筛选拼进 tid（如 "1001$动作片"），各段都作为候选解析 raws = [raw_tid] if raw_tid else [] if "$" in raw_tid: raws += [p.strip() for p in raw_tid.split("$") if p.strip()] if os.environ.get("KZ_DEBUG"): sys.stderr.write("[KZ] categoryContent tid=%r pg=%r extend=%r\n" % (tid, pg, extend)) # 汇总所有候选 TypeID（去重、保序），依次请求，第一个出数据的生效 all_cands = [] for raw in raws: for cand in self._tid_candidates(raw): if cand not in all_cands: all_cands.append(cand) cards = [] for cand in all_cands: cards = self._fetch_category(cand, sub, pg) if cards: break if not cards: # 无数据：兜底返回首页推荐，保证分类页永不空白； # 若已有候选仍全空（请求失败/服务端空），附一张诊断卡 # （标题带壳子原始传参与错误信息，方便定位壳子真实调用形态） if not self._recommend: self._fetch_recommend() cards = list(self._recommend or []) if all_cands: cards.insert(0, { "vod_id": "0", "vod_name": "诊断 tid=%r pg=%r err=%s" % (tid, pg, self.last_error), "vod_pic": "", "vod_remarks": "诊断", }) return { "class": self._classes, "filters": self._filters, "list": cards, "page": pg, "pagecount": 1 if not sub else (pg + 1 if len(cards) >= PAGE_SIZE else pg), "limit": PAGE_SIZE, "total": len(cards), } def _fetch_category(self, tid_num, sub, pg): """请求单个 TypeID 的列表；主列表空时自动用 list/more 兜底。""" if sub: resp = self._req("/u/category/vod/list/more", { "Cond": {"ParentTypeID": tid_num, "SubCategory": sub, "Paging": {"PageIndex": pg - 1, "PageSize": PAGE_SIZE}} }) or {} items = resp.get("Result") or [] if isinstance(resp, dict) else [] return [self._card(it) for it in items] resp = self._req("/u/category/vod/list", { "Cond": {"ParentTypeID": tid_num, "Sort": "", "Area": "", "SubCategory": "", "Year": "", "Paging": {"PageIndex": 0, "PageSize": PAGE_SIZE}} }) or {} secs = resp.get("Vods") or [] if isinstance(resp, dict) else [] cards = [] for s in secs: for it in (s.get("Items") or []): cards.append(self._card(it)) # 从同一次主列表响应顺带提取二级分类并缓存（无需额外请求） key = str(tid_num) if secs and key not in self._filters: opts = self._extract_subcats(secs) if opts: self._filters[key] = [{ "key": "subcategory", "name": "二级分类", "value": opts, }] if not cards: # 兜底：主列表为空的分类（如今日更新）用 list/more 拉全量列表 resp2 = self._req("/u/category/vod/list/more", { "Cond": {"ParentTypeID": tid_num, "Paging": {"PageIndex": pg - 1, "PageSize": PAGE_SIZE}} }) or {} items = resp2.get("Result") or [] if isinstance(resp2, dict) else [] cards = [self._card(it) for it in items] return cards # ---- 详情（带缓存，提升逐集播放速度） ---- def detailContent(self, ids, flags=None): try: return self._detail_content(ids) except Exception as e: self.last_error = "detail:%s:%s" % (type(e).__name__, e) return {"list": []} def _detail_content(self, ids): vid = self._vod_id(ids) if not vid: return {"list": []} detail = self._detail(vid) if not detail: return {"list": []} players = detail.get("Players") or [] sources = [] for p in players: eps = p.get("Contents") or [] if not eps: continue play_eps = [] for i, ep in enumerate(eps): name = ep.get("EpisodeTitle") or "第%d集" % (i + 1) # 把 vod_id 编进选集 id（格式 名称$vod_id|PlayerCode|index）， # playerContent 无需再回源定位 play_eps.append("%s$%s|%s|%d" % (name, vid, p.get("PlayerCode"), i)) if play_eps: sources.append({ "from": p.get("PlayerName") or p.get("PlayerCode") or "", "code": p.get("PlayerCode") or "", "need_parse": bool(p.get("NeedParse")), "eps": play_eps, }) # 排序：直链(NeedParse=False)线路排前，TVBox 默认选第一条即可秒播 sources.sort(key=lambda s: 0 if not s["need_parse"] else 1) vod = { "vod_id": vid, "vod_name": detail.get("Title") or "", "vod_pic": detail.get("PosterUrl") or "", "vod_content": (detail.get("Summary") or "").strip(), "vod_year": "", "vod_area": "", "vod_remarks": "", "vod_play_from": "$$$".join(s["from"] for s in sources), "vod_play_url": "$$$".join("#".join(s["eps"]) for s in sources), } return {"list": [vod]} def _detail(self, vid): now = time.time() cached = _DETAIL_CACHE.get(vid) if cached and now - cached[1] < DETAIL_TTL: return cached[0] if isinstance(cached[0], dict) else {} d = self._req("/u/vod/detail", {"VodID": int(vid), "SuggestCount": 6}) or {} if not isinstance(d, dict): d = {} _DETAIL_CACHE[vid] = (d, now) return d # ---- 搜索 ---- def searchContent(self, key, quick=False, pg="1"): try: return self._search(key, pg) except Exception as e: self.last_error = "search:%s:%s" % (type(e).__name__, e) return {"list": []} def _search(self, key, pg="1"): kw = (key or "").strip() if not kw: return {"list": []} pg = max(1, int(pg or 1)) resp = self._req("/u/vod/search", { "VodName": kw, "Paging": {"PageIndex": pg - 1, "PageSize": PAGE_SIZE} }) items = resp if isinstance(resp, list) else (resp.get("Vods") or resp.get("Result") or []) return {"list": [self._card(it) for it in items]} # ---- 播放（直链秒播，按需解析） ---- def playerContent(self, flag, id, vipFlags=None): header = {"User-Agent": USER_AGENT, "Referer": self.HOST + "/"} # 永不抛异常：任何异常都会让壳子判定"该线路播放失败"并无限切线 try: url = self._resolve_play_url(id) if url: return {"parse": 0, "url": url, "header": header} except Exception as e: self.last_error = "player:%s:%s" % (type(e).__name__, e) return {"parse": 0, "url": "", "header": header} def _resolve_play_url(self, id): """把壳子传入的选集 id 解析成可播地址。 兼容以下全部形态（不同壳子约定不一）： 1. 名称$vod_id|PlayerCode|index —— 本源 detailContent 产出的标准格式 2. vod_id|PlayerCode|index —— 部分壳子会剥掉 "名称$" 前缀 3. 名称$https://... / https://... —— 纯地址 4. vod_id —— 仅数字时兜底取第一条可直播线路 """ raw = id if isinstance(id, str) else str(id or "") raw = raw.strip() if "$" in raw: left, right = raw.split("$", 1) if "|" in right: raw = right elif right.strip().startswith("http"): return right.strip() if raw.startswith("http"): return raw vid = code = "" idx = 0 parts = [p.strip() for p in raw.split("|")] if len(parts) >= 3 and self._digits(parts[0]): vid, code, idx = self._digits(parts[0]), parts[1], parts[2] elif self._digits(raw) and "|" not in raw: # 兜底：壳子只回传 vod_id —— 取第一条可直播线路的对应集 vid = self._digits(raw) else: return "" if not vid: return "" try: idx = int(idx) except Exception: idx = 0 detail = self._detail(vid) players = (detail.get("Players") or []) if isinstance(detail, dict) else [] if not players: return "" player = None if code: player = next((p for p in players if str(p.get("PlayerCode")) == code), None) if player is None: directs = [p for p in players if not p.get("NeedParse") and (p.get("Contents") or [])] player = directs[0] if directs else players[0] eps = player.get("Contents") or [] if not eps: return "" if idx < 0 or idx >= len(eps): idx = 0 url = eps[idx].get("Url") or "" # NeedParse=True 的线路(youku/bilibili 等)需要二次解析 if url and player.get("NeedParse"): resolved = self._req("/u/vod/parse/url", {"Url": url, "PlayerCode": player.get("PlayerCode")}) or {} url = resolved.get("Url") or url return url # ---- drpy / 旧内核兼容 ---- def getName(self): return self.display_name def getDependence(self): return [] def localProxy(self, param): return None def manualVideoCheck(self): return False def action(self, action): return {} def destroy(self): return None # 详情缓存（模块级，跨实例共享） _DETAIL_CACHE: dict = {} _CACHE_LOCK = threading.Lock() # ============ 模块级双入口（drpy site 模式） ============ _SINGLETON = None def _get_singleton(): global _SINGLETON if _SINGLETON is None: _SINGLETON = Spider() return _SINGLETON def init(extend=""): return _get_singleton().init(extend) def homeContent(filter=False): return _get_singleton().homeContent(filter) def homeVideoContent(): return _get_singleton().homeVideoContent() def categoryContent(tid, pg="1", filter=True, extend=""): return _get_singleton().categoryContent(tid, pg, filter, extend) def detailContent(ids, flags=None): return _get_singleton().detailContent(ids, flags) def searchContent(key, quick=False, pg="1"): return _get_singleton().searchContent(key, quick, pg) def playerContent(flag, id, vipFlags=None): return _get_singleton().playerContent(flag, id, vipFlags) # 类名别名（部分内核按小写 spider 约定取类） spider = Spider # ============ 命令行自检 ============ if __name__ == "__main__": import argparse parser = argparse.ArgumentParser(description="快追影视 T3 源自检") sub = parser.add_subparsers(dest="action", required=True) sub.add_parser("home") sub.add_parser("homevideos") p_search = sub.add_parser("search") p_search.add_argument("keyword") p_search.add_argument("pg", nargs="?", default="1") p_cate = sub.add_parser("category") p_cate.add_argument("tid") p_cate.add_argument("pg", nargs="?", default="1") p_detail = sub.add_parser("detail") p_detail.add_argument("vod_id") args = parser.parse_args() sp = Spider() try: if args.action == "home": result = sp.homeContent() elif args.action == "homevideos": result = sp.homeVideoContent() elif args.action == "search": result = sp.searchContent(args.keyword, pg=args.pg) elif args.action == "category": result = sp.categoryContent(args.tid, args.pg) elif args.action == "detail": result = sp.detailContent(args.vod_id) else: result = {} json.dump(result, sys.stdout, ensure_ascii=False, indent=2) sys.stdout.write("\n") except Exception as error: sys.stderr.write("%s: %s\n" % (type(error).__name__, error)) sys.exit(1)