#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ 夜社(yeshe.tv) - TVBox/OK影视/影视仓 标准 Python 源 (完整版 v4.0) 目标: 域名自愈 + 请求头轮换重试 + 全分类覆盖(含图集/漫画/小说) + 翻页修复 + 播放稳定 实勘: 2026-09-11 全站 1-100 分类 + /ntype/ + /aituoyi/ 三套路由全量枚举: - /type/ 路由 29 个分类全部有内容, 全部收录: 视频7: 2视频/13AI短剧/36擦边短剧/11国产视频/12日本AV/14欧美无码/35韩国BJ 动漫7: 1动漫/7同人作品/8动画卡通/10 3D动漫/9中文动漫/32里番/33泡面番 有声4: 3有声/15有声小说/16淫词艳曲/17激情骚麦 漫画4: 4漫画/18韩国H漫/19日本H漫/31 3D漫画(详情走 /detail/ 路由) 写真7: 5写真/20秀人系列/22网红COS/21机构套图/23内购私拍/34AI绘图/24各国套图 - /ntype/ 路由 7 个专题分类全部有内容, 全部收录(卡片走 /novel/ 路由): 6小说/25都市生活/26学生校园/27家庭乱伦/28玄幻武侠/29系统穿越/30同人改编 - /aituoyi/ 系列(AI换装/换脸/绘图/深度)为内容标签页, 无可解析卡片, 不收录 - 37-100 号段为噪音页(无卡片), 已排除 - 播放链路: 18 个视频分类详情页含 player_aaaa, m3u8 直链多前缀 (short/abevx/avtaix/misub/beegs/kanav/skrbj + 有声 asms3/hls), CDN 域名随站变动 - 写真/漫画/小说类详情页无播放器(纯图集/章节页), 列表可看封面, 点进不崩 v4.0 相比 v3.0(全分类版) 的核心改进: 1. 翻页修复: 站点后端不分页(/-N.html、?page=、/id/N、/page/N 全部返回同一批, 每次请求还有 1-2 张卡随机轮换, 这是"滑到底不出下一页""翻页跳"的根因)。 改为「客户端切片分页」: 抓全量列表 -> 进程内 TTL 缓存 -> 按页切片返回, pagecount 按真实条数计算, 每页内容不同且不重复。 2. 分类栏全量收录: 29(type) + 7(ntype) = 36 个有内容分类, 不再剔除图集/漫画/小说。 3. 卡片解析兼容三路由: /play/{vid}/1/1.html(视频/写真)、/detail/{vid}.html(漫画)、 /novel/{vid}.html(小说/专题), 统一按 module-poster-item 块级解析。 4. 详情路由自适应: vod_id 带路由前缀(d:/n:), 图集/漫画/小说详情返回标题+封面不崩。 5. 固定分类清单(不动态提取): 镜像间导航分类数不一致(24/15)会抖, 固定清单跨镜像稳定。 保留 v2.0/v3.0 全部稳定性能力: - 域名解析: extend配置 > 文件缓存 > 短链302 > 缓存兜底 > 硬编码兜底, 失败自愈 - 请求: 3 UA 轮换 + SSL 降级 + 200/206 白名单 + 解码有效性校验 - CDN 动态化: m3u8 前缀从 player_aaaa.url 实时提取, 不硬编码 """ import requests import json import re import sys import os import time import base64 from urllib.parse import quote, parse_qs # 兼容本地调试与 TVBox/FongMi/PyramidStore 环境 sys.path.append('..') try: from base.spider import Spider except ImportError: class Spider: def __init__(self): pass def init(self, extend=""): pass # ========== 短链与兜底域名 ========== # 短链 302 跳转当前镜像, Location 形如 {镜像}/type/13.html?chl=yeshetv _SHORT_URL = "https://ysurl.win/755WwN" # 兜底域名(短链与缓存都失效时的最后一层保底) # 2026-09-11: yeshe.tv 已 404, 当前镜像为 njrunyuan.net 系 / coffeemiaomiao.com 系轮换 _BACKUP_BASE_URLS = [ "https://xn--2ju9sgdp-bx9ot43dhs3cs66e.njrunyuan.net", "https://xn--pyubwdhg-do9mr75hub8b1p7d.njrunyuan.net", ] # 域名缓存文件名(跨启动复用, 与国色天香site.txt同思路; 写失败自动忽略) _CACHE_FILE = "夜社domain.txt" # 页面混淆提取正则: var a="base64串" (解码后才是真实 HTML) _OBFUSC_REGEX = re.compile(r'var\s+a\s*=\s*"([A-Za-z0-9+/=]+)"') # 请求头轮换池(目标域名拒绝/异常时切换 UA 重试) _UA_POOL = [ "Mozilla/5.0 (Windows NT 10.0; Win64; x64) " "AppleWebKit/537.36 (KHTML, like Gecko) " "Chrome/120.0.0.0 Safari/537.36", "Mozilla/5.0 (iPhone; CPU iPhone OS 17_0 like Mac OS X) " "AppleWebKit/605.1.15 (KHTML, like Gecko) " "Version/17.0 Mobile/15E148 Safari/604.1", "Mozilla/5.0 (Linux; Android 13; Pixel 7) " "AppleWebKit/537.36 (KHTML, like Gecko) " "Chrome/120.0.0.0 Mobile Safari/537.36", ] # /ntype/ 路由的分类 id(卡片走 /novel/ 路由; /type/{id}.html 对应页面为空/噪音) _NTYPE_IDS = {"6", "25", "26", "27", "28", "29", "30"} # /detail/ 路由的分类 id(漫画: 卡片走 /detail/{vid}.html; 详情页无播放器) _DETAIL_IDS = {"4", "18", "19", "31"} # 每页切片条数(TVBox 一屏常见 24; 站点单页全量 ~120 张卡 -> 约 5 页) _PAGE_SIZE = 24 # 分类列表进程内缓存有效期(秒): 翻页期间保持一致, 超时自动刷新 _LIST_TTL = 45 # 固定分类栏(36 个全部有内容; 顺序与站点导航一致: 视频/动漫/有声/漫画/写真/专题) # 站点新增分类时在此追加一条即可; 不动态提取导航(镜像间导航分类数不一致会抖) _DEFAULT_CLASSES = [ {"type_id": "2", "type_name": "视频"}, {"type_id": "13", "type_name": "AI短剧"}, {"type_id": "36", "type_name": "擦边短剧"}, {"type_id": "11", "type_name": "国产视频"}, {"type_id": "12", "type_name": "日本AV"}, {"type_id": "14", "type_name": "欧美无码"}, {"type_id": "35", "type_name": "韩国BJ"}, {"type_id": "1", "type_name": "动漫"}, {"type_id": "7", "type_name": "同人作品"}, {"type_id": "8", "type_name": "动画卡通"}, {"type_id": "10", "type_name": "3D动漫"}, {"type_id": "9", "type_name": "中文动漫"}, {"type_id": "32", "type_name": "里番"}, {"type_id": "33", "type_name": "泡面番"}, {"type_id": "3", "type_name": "有声"}, {"type_id": "15", "type_name": "有声小说"}, {"type_id": "16", "type_name": "淫词艳曲"}, {"type_id": "17", "type_name": "激情骚麦"}, {"type_id": "4", "type_name": "漫画"}, {"type_id": "18", "type_name": "韩国H漫"}, {"type_id": "19", "type_name": "日本H漫"}, {"type_id": "31", "type_name": "3D漫画"}, {"type_id": "5", "type_name": "写真"}, {"type_id": "20", "type_name": "秀人系列"}, {"type_id": "22", "type_name": "网红COS"}, {"type_id": "21", "type_name": "机构套图"}, {"type_id": "23", "type_name": "内购私拍"}, {"type_id": "34", "type_name": "AI绘图"}, {"type_id": "24", "type_name": "各国套图"}, {"type_id": "6", "type_name": "小说"}, {"type_id": "25", "type_name": "都市生活"}, {"type_id": "26", "type_name": "学生校园"}, {"type_id": "27", "type_name": "家庭乱伦"}, {"type_id": "28", "type_name": "玄幻武侠"}, {"type_id": "29", "type_name": "系统穿越"}, {"type_id": "30", "type_name": "同人改编"}, ] def _decode_page(text: str) -> str: """解码站点页面(var a="base64" 混淆)。 解码结果需同时满足"长度占比"与"站点特征"校验, 防止半截/错误页被当作 正常页面继续解析导致列表空白。 """ if not text: return "" m = _OBFUSC_REGEX.search(text) if m: try: dec = base64.b64decode(m.group(1)).decode('utf-8', errors='replace') if len(dec) > len(text) * 0.3: if '/play/' in dec or '/type/' in dec or '/novel/' in dec or ' str: """详情页 清洗: 去掉站点名后缀(名称_夜社 / 名称-夜社 / 名称 夜社)。""" if not raw: return "" name = raw.strip() name = re.sub(r'[_\-\s]*夜社.*$', '', name) name = re.sub(r'[_\-\s]*$', '', name) return name.strip() def _first_pic(html: str) -> str: """从详情页取第一张内容图(排除 js/css/logo/样式资源)。""" if not html: return "" skip = ('static', '.js', '.css', 'logo', 'icon', 'banner', 'bg', 'loading', 'default') for m in re.finditer(r'(?:src|data-src|pic)="(https?://[^"]+)"', html): u = m.group(1) if re.search(r'\.(webp|jpg|jpeg|png)$', u, re.I) and not any(x in u.lower() for x in skip): return u return "" class Spider(Spider): """夜社 TVBox 标准爬虫插件 (完整版 v4.0)""" def __init__(self): self.host = _BACKUP_BASE_URLS[0] self.userAgent = _UA_POOL[0] self.timeout = 15 self.session = requests.Session() self.session.headers.update({ "User-Agent": self.userAgent, "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8", "Accept-Language": "zh-CN,zh;q=0.9", "Accept-Encoding": "gzip, deflate", "Referer": self.host + "/", }) # 进程内域名缓存(会话内只解析一次短链, 大幅降低抖动窗口) self._domain = "" self.extend = "" # 分类列表进程内缓存: {tid: (时间戳, [卡片...])} —— 翻页切片的一致性来源 self._list_cache = {} # ==================== 通用请求 (UA 轮换 + SSL 降级) ==================== def _raw_get(self, url: str, allow_redirects: bool = False, timeout=None, **kw): """原始 GET: 依次轮换 UA 重试; SSL 失败降级不校验。 不约束状态码(允许 302 等跳转响应, 用于短链解析), 返回 Response 或 None。 """ timeout = timeout or self.timeout last_err = None for ua in _UA_POOL: headers = { "User-Agent": ua, "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8", "Accept-Language": "zh-CN,zh;q=0.9", "Referer": (self._domain or self.host) + "/", } try: return self.session.get(url, headers=headers, timeout=timeout, allow_redirects=allow_redirects, **kw) except requests.exceptions.SSLError: try: return self.session.get(url, headers=headers, timeout=timeout, allow_redirects=allow_redirects, verify=False, **kw) except Exception as e: last_err = e except Exception as e: last_err = e if last_err: print("[ERROR] _raw_get %s failed: %s" % (url, last_err)) return None def fetch(self, url: str, headers: dict = None, **kw): """统一请求: 依次轮换 UA 重试; SSL 握手失败时降级为不校验证书。 只接受 200/206, 其余状态码视为失败并换 UA 重试。 返回 requests.Response 或 None(全部 UA 均失败)。 """ if headers is None: headers = { "User-Agent": self.userAgent, "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8", "Accept-Language": "zh-CN,zh;q=0.9", # 注意: 不要在此调用 _get_domain(), 否则短链解析阶段会无限递归 "Referer": (self._domain or self.host) + "/", } timeout = kw.pop('timeout', None) or self.timeout last_err = None for ua in ([headers.get("User-Agent")] if headers else []) + _UA_POOL: if ua is None: continue merged = dict(headers) merged["User-Agent"] = ua try: resp = self.session.get(url, headers=merged, timeout=timeout, **kw) if resp.status_code in (200, 206): return resp last_err = "status=%s" % resp.status_code continue except requests.exceptions.SSLError: # SSL 校验失败 -> 降级不校验重试一次(适配无有效证书的运营商线路) try: resp = self.session.get(url, headers=merged, timeout=timeout, verify=False, **kw) if resp.status_code in (200, 206): return resp last_err = "status=%s" % resp.status_code except Exception as e: last_err = e continue except Exception as e: last_err = e continue print("[ERROR] fetch %s failed: %s" % (url, last_err)) return None # ==================== 动态域名解析 (短链 + 缓存 + 自愈) ==================== def _load_cached_domain(self) -> str: """读取文件缓存的镜像域名(跨进程复用; TVBox 只读环境自动忽略)。""" try: if os.path.exists(_CACHE_FILE): with open(_CACHE_FILE, "r", encoding="utf-8") as f: dm = f.read().strip() if re.match(r"^https?://[^/\s]+$", dm): return dm except Exception: pass return "" def _save_cached_domain(self, dm: str): try: with open(_CACHE_FILE, "w", encoding="utf-8") as f: f.write(dm) except Exception: pass def _resolve_domain(self, force: bool = False) -> str: """解析当前可用镜像域名。 优先级: extend 配置 -> 文件缓存(非 force) -> 短链 302 -> 文件缓存(force 兜底) -> 兜底域名 """ # 1) extend 指定镜像(用户配置优先) if self.extend and self.extend.startswith("http"): m = re.match(r"(https?://[^/]+)", self.extend) return m.group(1).rstrip("/") if m else self.extend.rstrip("/") # 2) 文件缓存(非强制刷新时直接复用) if not force: cached = self._load_cached_domain() if cached: print("[INFO] 从缓存读取镜像: %s" % cached) return cached # 3) 短链 302 -> Location 的 host loc = "" try: resp = self._raw_get(_SHORT_URL, allow_redirects=False, timeout=10) if resp is not None: loc = resp.headers.get("Location", "") or resp.headers.get("location", "") except Exception as e: print("[WARN] 短链解析异常: %s" % e) if loc: m = re.match(r"(https?://[^/]+)", loc) if m: final = m.group(1).rstrip("/") self._save_cached_domain(final) print("[INFO] 域名解析(短链): %s" % final) return final # 4) 短链失败 -> 旧缓存兜底(force 场景) -> 硬编码兜底域名 if force: cached = self._load_cached_domain() if cached: print("[WARN] 短链失败, 回退文件缓存: %s" % cached) return cached print("[WARN] 短链失败, 回退默认域名: %s" % self.host) return self.host def _get_domain(self) -> str: """进程内缓存 + 解析, 保证返回非空字符串。""" if not self._domain: self._domain = self._resolve_domain(force=False) return self._domain or self.host # ==================== 页面抓取 (解码 + 有效性校验 + 自愈重试) ==================== def _looks_valid(self, html: str) -> bool: """正常站内页应包含导航(/type/)或卡片(/play//detail//novel/)链接。""" if not html or not isinstance(html, str): return False if len(html) < 200: return False if "/play/" in html or "/type/" in html or "/novel/" in html: return True return "<div" in html and "<a" in html def _fetch_html(self, path: str, _depth: int = 0): """抓取并解码页面; 失败时自动刷新域名后重试一次(自愈)。 返回解码后的 HTML 字符串或 None。 """ url = path if path.startswith("http") else self._get_domain() + path resp = self.fetch(url) if resp is None: # 第一轮: 域名也可能失效 -> 强制刷新域名后重试一次(仅一层, 防死循环) if _depth == 0: print("[WARN] 首次请求失败, 强制刷新域名后重试") self._domain = "" self._domain = self._resolve_domain(force=True) return self._fetch_html(path, _depth=1) return None dec = _decode_page(resp.text) if not self._looks_valid(dec): if _depth == 0: print("[WARN] 页面解码无效, 强制刷新域名后重试") self._domain = "" self._domain = self._resolve_domain(force=True) return self._fetch_html(path, _depth=1) return None return dec # ==================== 数据格式化 ==================== def _wrap_pic(self, pic_url: str) -> str: """封面转成绝对 https 地址(兼容 //、/、相对路径)。""" if not pic_url: return "" pic_url = pic_url.strip().strip('"\'') if pic_url.startswith("//"): pic_url = "https:" + pic_url elif not pic_url.startswith(("http://", "https://")): if pic_url.startswith("/"): pic_url = self._get_domain() + pic_url else: pic_url = self._get_domain() + "/" + pic_url return pic_url def _parse_cards(self, html: str) -> list: """解析卡片列表, 兼容三种路由(2026-09-11 实勘确认站点卡片结构): /play/{vid}/1/1.html —— 视频/写真分类 /detail/{vid}.html —— 漫画分类 /novel/{vid}.html —— 小说/专题分类(ntype) 全部为 <a class="module-poster-item ..."> 块, 统一块级解析: 标题: a[title] 或 module-poster-item-title; 封面: img src 或 data-src(懒加载); 备注: time 角标(集数/话数/章数/P数); 分类标签: type 角标(并入备注, 便于区分图集/漫画/小说); 相同视频 id 去重; vod_id 按路由前缀编码(d:/n:), 供 detailContent 直连。 """ if not html: return [] vod_list = [] seen = set() # 块级匹配: 允许 href 与 class 属性任意顺序 pattern = re.compile( r'<a[^>]*href="/(play|detail|novel)/([^"/]+?)(?:/1/1)?\.html"[^>]*' r'class="module-poster-item[^"]*"[^>]*>(.*?)</a>', re.DOTALL) for m in pattern.finditer(html): route, vid, block = m.group(1), m.group(2), m.group(0) key = route + ":" + vid if key in seen: continue seen.add(key) name = "" title_m = re.search(r'title="([^"]*)"', block) if title_m: name = title_m.group(1).strip() if not name: t2 = re.search(r'module-poster-item-title">([^<]+)<', block) if t2: name = t2.group(1).strip() pic_url = "" img_m = re.search(r'<img[^>]+src="([^"]+)"', block) if not img_m: img_m = re.search(r'<img[^>]+data-src="([^"]+)"', block) if img_m: pic_url = self._wrap_pic(img_m.group(1)) remark = "" type_tag = "" tm = re.search(r'<div class="time">([^<]+)</div>', block) if tm: remark = tm.group(1).strip() if not remark: tm2 = re.search(r'<span class="time">([^<]+)</span>', block) if tm2: remark = tm2.group(1).strip() tm3 = re.search(r'<div class="type">([^<]+)</div>', block) if tm3: type_tag = tm3.group(1).strip() if type_tag and type_tag not in remark: remark = (remark + " " + type_tag).strip() # vod_id 路由前缀: play 系保持纯数字(兼容旧收藏), 其余编码 vod_id = vid if route == "detail": vod_id = "d:" + vid elif route == "novel": vod_id = "n:" + vid vod_list.append({ "vod_id": vod_id, "vod_name": name, "vod_pic": pic_url, "vod_remarks": remark, }) return vod_list # ==================== 分类列表 (客户端切片分页) ==================== def _category_path(self, tid: str) -> str: """分类 id -> 列表页路径: /ntype/ 系列走 ntype 路由, 其余走 /type/ 路由。""" if tid in _NTYPE_IDS: return "/ntype/%s.html" % tid return "/type/%s.html" % tid def _category_list(self, tid: str) -> list: """分类全量卡片列表(进程内 TTL 缓存)。 站点后端不分页(任意翻页 URL 均返回同一批 + 少量随机轮换), 因此在进程内缓存最近一次全量列表, 供 categoryContent 切片翻页, 保证同一实例内各页内容一致、不重复、不跳页。 """ now = time.time() cached = self._list_cache.get(tid) if cached and now - cached[0] < _LIST_TTL: return cached[1] html = self._fetch_html(self._category_path(tid)) lst = self._parse_cards(html) if html else [] self._list_cache[tid] = (now, lst) return lst # ==================== TVBox 标准接口 ==================== def init(self, extend=""): """插件初始化(框架回调): 解析 extend + 预解析域名。""" if isinstance(extend, list): self.extend = "" elif extend and isinstance(extend, str) and extend.startswith("http"): self.extend = extend else: self.extend = "" # 预解析域名: 一次短链请求, 后续所有请求走缓存 try: self._get_domain() except Exception: pass def getName(self): return "夜社短剧" def homeContent(self, filter): """分类栏: 固定 36 个全量分类(不做动态提取)。 2026-09-11 实勘结论(关键): - 站点存在多套镜像, /type/ 导航分类数镜像间不一致(24/15), 动态提取 会让分类栏随镜像轮换来回增减, 不稳定; - 全量枚举确认 36 个分类全部有内容: 29 个 /type/(视频/动漫/有声/漫画/写真 全收录, 图集类也收) + 7 个 /ntype/(小说/都市生活等专题); - 图集/漫画/小说详情无播放器, 但列表可完整浏览封面, 点进详情不崩。 站点新增分类时, 在 _DEFAULT_CLASSES 追加一条即可。 """ return {"class": _DEFAULT_CLASSES, "filters": {}} def homeVideoContent(self): """首页推荐: AI短剧(13)首屏; 失败依次回退 视频(2)。""" result = {"list": []} for tid in ("13", "2"): try: lst = self._category_list(tid)[:30] if lst: result["list"] = lst break except Exception: continue return result def categoryContent(self, tid, pg, filter, extend): """分类内容: 客户端切片分页(修复站点不分页导致的翻页跳/不出下一页)。 站点后端对任意翻页 URL 都返回同一批卡片(约120张, 还有1-2张随机轮换), 旧版直接转发 -N.html 导致每页内容重复 -> TVBox 判"没有新内容"。 本版改为: 抓全量 -> 按 (pg-1)*24 切片 -> 每页返回不同卡片, pagecount = ceil(总量/24), TVBox 滑到底能真正翻页且不重复。 """ result = {"list": [], "page": pg, "pagecount": 1, "limit": _PAGE_SIZE, "total": 0} try: pg = int(pg or 1) lst = self._category_list(str(tid)) total = len(lst) pagecount = max(1, (total + _PAGE_SIZE - 1) // _PAGE_SIZE) if total else 1 start = (pg - 1) * _PAGE_SIZE result.update({ "page": pg, "pagecount": pagecount, "limit": _PAGE_SIZE, "total": total, "list": lst[start:start + _PAGE_SIZE], }) except Exception as e: print("[ERROR] categoryContent %s/%s 异常: %s" % (tid, pg, e)) return result def detailContent(self, ids): """详情页, 三种处理: - play 路由(视频): player_aaaa.url 为 m3u8 直链, CDN 前缀动态提取, 多集拼接 - play 路由(写真图集): 无播放器, 返回标题+封面 - d: 前缀(漫画) / n: 前缀(小说/专题): /detail/ 或 /novel/ 详情, 无播放器, 返回标题+封面(图集浏览用) 任何异常返回空结构, 不影响 TVBox 其他分类。 """ result = {"list": []} if not ids: return result vod_id = ids[0] if isinstance(ids, list) else str(ids) # 路由解析: d:/n: 前缀直连; 其余走 /play/ vid = vod_id if vod_id.startswith("d:"): detail_path = "/detail/%s.html" % vod_id[2:] return self._build_gallery_detail(vid, detail_path) if vod_id.startswith("n:"): detail_path = "/novel/%s.html" % vod_id[2:] return self._build_gallery_detail(vid, detail_path) html = self._fetch_html("/play/%s/1/1.html" % vod_id) if not html: return result config = _extract_json(html, "player_aaaa") if not config: # 无播放器: 写真/套图类图集, 返回标题+封面(列表卡信息), 不崩 return self._build_gallery_detail(vod_id, None, html) vod_data = config.get("vod_data", {}) or {} vod_pic = vod_data.get("vod_pic", "") or "" vod_name = vod_data.get("vod_name", "") or "" vod_class = vod_data.get("vod_class", "") or "" # 集数: 优先详情页 vod_play_all, 其次集数链接数 ep_count = 1 ep_m = re.search(r"vod_play_all\s*=\s*(\d+)", html) if ep_m: ep_count = int(ep_m.group(1)) else: eps = set(re.findall(r"/play/%s/1/(\d+)\.html" % vod_id, html)) ep_count = len(eps) if eps else 1 if ep_count < 1: ep_count = 1 # 播放地址: 支持多前缀路径(short/vod、abevx/vod、avtaix/vod、misub/vod、 # beegs/vod、kanav/vod、skrbj/vod); 有声类为 hls 单集(直接透传) first_url = config.get("url", "") or "" nxt = config.get("url_next", "") or "" play_urls = [] base = "" if first_url: # 格式A: .../{前缀}/vod/{path}/{n}/play.m3u8 (url 本身带集数目录) m = re.search(r"^(https?://[^/]+(/[^/]+){0,3}/vod/[^/]+)/\d+/play\.m3u8$", first_url) if m: base = m.group(1) elif nxt: # 格式B: 只有第1集直链, 用 url_next 反推集数模板 m2 = re.search(r"^(https?://[^/]+(/[^/]+){0,3}/vod/[^/]+)/2/play\.m3u8$", nxt) if m2: base = m2.group(1) if base and ep_count > 1: for i in range(1, ep_count + 1): play_urls.append("第%d集$%s/%d/play.m3u8" % (i, base, i)) elif first_url: play_urls.append("第1集$%s" % first_url) if nxt: play_urls.append("第2集$%s" % nxt) else: for i in range(1, ep_count + 1): play_urls.append("第%d集$/play/%s/1/%d.html" % (i, vod_id, i)) vod = { "vod_id": str(vod_id), "vod_name": vod_name, "vod_pic": vod_pic, "type_name": vod_class, "vod_year": "", "vod_area": "", "vod_actor": "", "vod_director": "", "vod_content": "", "vod_remarks": "%d集" % ep_count, "vod_play_from": "夜社", "vod_play_url": "#".join(play_urls), } result["list"] = [vod] return result def _build_gallery_detail(self, vod_id: str, detail_path=None, html=None): """图集/漫画/小说详情: 无播放器, 尽力提取标题+封面返回(不崩)。 vod_id 为带前缀的原始 id(供列表回传一致性); 播放源留空, TVBox 端可浏览作品基本信息。 """ result = {"list": []} try: if html is None and detail_path: html = self._fetch_html(detail_path) if not html: return result name = "" title_m = re.search(r'<title>([^<]+)', html) if title_m: name = _clean_name(title_m.group(1)) pic = _first_pic(html) vod = { "vod_id": str(vod_id), "vod_name": name, "vod_pic": pic, "type_name": "", "vod_year": "", "vod_area": "", "vod_actor": "", "vod_director": "", "vod_content": "图集/漫画/小说作品(无在线播放源)", "vod_remarks": "图集", "vod_play_from": "夜社", "vod_play_url": "", } result["list"] = [vod] except Exception as e: print("[WARN] 图集详情构建失败 %s: %s" % (vod_id, e)) return result def searchContent(self, key, quick, pg=1): result = {"list": []} if isinstance(quick, dict): quick = quick.get("quick", False) pg = int(pg or 1) url = "/vod/search/wd/%s.html" % quote(key) if pg > 1: url = "/vod/search/wd/%s-%d.html" % (quote(key), pg) html = self._fetch_html(url) if html: result["list"] = self._parse_cards(html)[:30] return result def searchContentPage(self, key, quick, pg=1): return self.searchContent(key, quick, pg) def isVideoFormat(self, url): if not url or not isinstance(url, str): return False fmt = [".mp4", ".m3u8", ".ts", ".mkv", ".avi", ".webm", ".flv"] return any(f in url.lower() for f in fmt) def manualVideoCheck(self): return False def playerContent(self, flag, id, vipFlags): """播放: m3u8 直链直接透传; /play/ 页面地址二次解析 player_aaaa。""" result = {"parse": 0, "url": "", "header": {}} if not id: return result play_id = str(id) headers = { "User-Agent": self.userAgent, "Referer": self._get_domain() + "/", } if play_id.startswith("http") and ".m3u8" in play_id: result["url"] = play_id result["header"] = headers return result if play_id.startswith("/play/"): html = self._fetch_html(play_id) if html: config = _extract_json(html, "player_aaaa") if config and config.get("url"): result["url"] = config["url"] result["header"] = headers return result result["url"] = play_id result["header"] = headers return result def localProxy(self, param): """本地代理: 封面图直连转发(页面 no-referrer, 无需防盗链)。""" if isinstance(param, str): param_dict = parse_qs(param) else: param_dict = param do = param_dict.get("do", "") if isinstance(do, list): do = do[0] if do else "" if do != "img": return None url = param_dict.get("url", "") if isinstance(url, list): url = url[0] if url else "" if url: try: url = base64.urlsafe_b64decode(url).decode("utf-8") except Exception: pass if not url: return [404, "text/plain", "", {}] headers = { "User-Agent": self.userAgent, "Accept": "image/webp,image/apng,image/*,*/*;q=0.8", } try: r = requests.get(url, headers=headers, timeout=10) if r.status_code != 200 or not r.content: return [404, "text/plain", "cover not found", {}] ct = r.headers.get("Content-Type", "") if not ct or "image" not in ct: if ".png" in url: ct = "image/png" elif ".webp" in url: ct = "image/webp" elif ".gif" in url: ct = "image/gif" else: ct = "image/jpeg" return [200, ct, r.content, {}] except Exception as e: print("[ERROR] localProxy img 失败: %s" % e) return [500, "text/plain", str(e), {}] def destroy(self): try: self.session.close() except Exception: pass