"""
91短剧 (91crdj.com / 91crdj.net) - TVBox 爬虫源
============================================================
接口：homeContent / homeVideoContent / categoryContent(分类+标签+排序)
/ detailContent / playerContent(视频/漫画/小说三态) / searchContent
站点特性（实勘确认 2026-09-16）：
1. 双线路同构：91crdj.com / 91crdj.net（页面请求依次尝试并缓存成功线路）
2. 分类列表：GET /api/list/fragment?scope={scope}&key={key}&sort={new|hot}&page={pg}
主分类 scope/key = category/duanju、category/dongman-sm(漫剧)、category/zhibo-huifang(真人剧)、
category/videos(成人视频)、comic/(漫画 空key)、novel/(小说 空key)；标签 scope=tag&key=标签slug
3. 列表页卡片：
封面在 img data-src（2026-09-16 实测图床迁址 pic.zdmhyg.cn → pic.tuafjz.cn，
泛 pic.* 域名直连返回 AES-128-CBC 密文，需本地代理解密；src 均为 data: 占位图）
4. 详情页 /{cate}/{id-slug}/：h1 标题、meta description 简介、JSON-LD(TVSeries)→年份/类型、
class="score">★ 评分、/biaoqian/ 链接标签、选集链接 href="/{cate}/{id-slug}/{n}/"
5. 播放页 /{cate}/{id-slug}/{n}/：内嵌 ', html)
if ld:
try:
parsed = json.loads(ld.group(1))
graph = parsed.get("@graph") or []
for node in graph:
if node.get("@type") == "TVSeries":
if node.get("datePublished"):
vod["vod_year"] = str(node["datePublished"])[:4]
g = node.get("genre")
if isinstance(g, list) and g:
vod["type_name"] = ",".join(g)
break
except Exception:
pass
sc = re.search(r'class="score"[^>]*>\u2605([\d.]+)', html)
if not sc:
sc = re.search(r'\u2605([\d.]+)', html)
if sc:
vod["vod_score"] = sc.group(1)
tags = []
tre = re.compile(r']+href="https://91crdj\.com/biaoqian/[^"]+"[^>]*>([^<]+)')
for tm in tre.finditer(html):
tag = tm.group(1).strip()
if tag and tag not in tags:
tags.append(tag)
if tags:
vod["vod_tag"] = ",".join(tags)
# 剧集列表
esc_id = re.sub(r'/', r'\\/', id_)
ere = re.compile(
r']+href="https://91crdj\.com/' + esc_id + r'/(\d+)/"[^>]*>([^<]*)'
)
eps = []
seen_ep = set()
for em in ere.finditer(html):
num = em.group(1)
nm = em.group(2).strip()
if not nm or nm in BTN_TEXTS or num in seen_ep:
continue
seen_ep.add(num)
eps.append([nm, num])
if not eps:
eps = [["第1集", "1"]]
urls = []
for nm, num in eps:
urls.append(f"{nm}${id_}/{num}")
vod["vod_play_url"] = "#".join(urls)
if eps:
vod["vod_remarks"] = f"共{len(eps)}集"
return {"list": [vod]}
# ==================== 播放 ====================
def playerContent(self, flag, id_, vipFlags):
parts = str(id_).split("/")
cate = parts[0] if parts else ""
video_id = ""
if len(parts) >= 2:
video_id = parts[1].split("-")[0]
ep = parts[2] if len(parts) >= 3 else "1"
if cate in COMIC_CATES:
return self._play_comic(id_)
if cate in NOVEL_CATES:
return self._play_novel(id_)
if cate not in VIDEO_CATES:
return {"parse": 0, "url": "", "header": {}}
url = ""
# 1) 播放页 playInitialData
html = self._get_text("/" + id_ + "/")
pm = re.search(r']+id="playInitialData"[^>]*>([\s\S]*?)', html or "")
if pm:
try:
data = json.loads(pm.group(1).strip())
if data and data.get("current") and data["current"].get("src"):
url = data["current"]["src"]
except Exception:
pass
# 2) playback API
if not url and video_id:
try:
api = self._get_text_with_headers(
f"/videos/{video_id}/episodes/{ep}/playback",
{"X-Requested-With": "fetch", "Referer": self.siteUrl + "/" + id_ + "/"},
TIMEOUT_API,
)
jd = json.loads(api or "{}")
if jd.get("data") and jd["data"].get("src"):
url = jd["data"]["src"]
except Exception:
pass
if not url:
return {"parse": 0, "url": "", "header": {}}
return {
"parse": 0,
"url": url,
"header": {"User-Agent": UA, "Referer": self.siteUrl + "/" + cate + "/"},
}
def _play_comic(self, id_):
"""漫画: pics:// + 本地代理（AES 解密在 localProxy）"""
html = self._get_text("/" + id_ + "/")
if not html:
return {"parse": 0, "url": "", "header": {}}
imgs = []
# 1) comic-page 的懒加载 data-src
r1 = re.compile(r'class="comic-page[^"]*"[^>]*data-src="([^"]+)"')
for m in r1.finditer(html):
u = self._norm_pic(m.group(1))
if u and u not in imgs:
imgs.append(u)
# 2) 任意 pic.* 的懒加载属性（data-src/data-original/data-lazy）
if not imgs:
r2 = re.compile(r'(?:data-src|data-original|data-lazy)="(https://pic\.[^"]+)"')
for m in r2.finditer(html):
u = self._norm_pic(m.group(1))
if u and u not in imgs:
imgs.append(u)
# 3) 任意 pic.* 的 src（兜底）
if not imgs:
r3 = re.compile(r'src="(https://pic\.[^"]+)"')
for m in r3.finditer(html):
u = self._norm_pic(m.group(1))
if u and u not in imgs:
imgs.append(u)
if not imgs:
return {"parse": 0, "url": "", "header": {}}
proxied = [PROXY_PREFIX + quote(u, safe='') for u in imgs]
return {"parse": 0, "url": "pics://" + "&&".join(proxied), "header": {}}
def _play_novel(self, id_):
"""小说: novel:// JSON({title, content})"""
html = self._get_text("/" + id_ + "/")
if not html:
return {"parse": 0, "url": "", "header": {}}
am = re.search(r']*>([\s\S]*?)', html)
if not am:
am2 = re.search(
r']*class="[^"]*(?:content|article|chapter|text|reader)[^"]*"[^>]*>([\s\S]*?)',
html,
)
if am2:
am = am2
if not am:
return {"parse": 0, "url": "", "header": {}}
raw = am.group(1)
raw = re.sub(r']*>[\s\S]*?', '', raw)
raw = re.sub(r']*>[\s\S]*?', '', raw)
raw = re.sub(r']*>[\s\S]*?', '', raw)
raw = re.sub(r'', '\n', raw)
raw = re.sub(r'', '\n\n', raw)
text = re.sub(r'<[^>]+>', '', raw)
text = (text.replace(' ', ' ').replace('<', '<')
.replace('>', '>').replace('&', '&')
.replace('"', '"').replace(''', "'"))
text = text.strip()
text = re.sub(r'\n{3,}', '\n\n', text)
ch_title = ""
cm = re.search(r']*class="novel-h"[^>]*>([\s\S]*?)', html)
if cm:
ch_title = re.sub(r'<[^>]+>', '', cm.group(1)).strip()
if not ch_title:
tm = re.search(r'([^<]+)', html)
ch_title = tm.group(1).strip() if tm else "正文"
payload = json.dumps({"title": ch_title, "content": text}, ensure_ascii=False)
return {"parse": 0, "url": "novel://" + payload, "header": {}}
# ==================== 本地代理（封面/漫画图 AES 解密） ====================
def localProxy(self, params):
"""壳子请求: 127.0.0.1:9978/proxy?do=py&action=proxy&type=cover&url={quote(原图URL)}
国色天香(PyramidStore 直推)协议: 返回 3 元组 [status, mime, 原始图片字节]。
流程: 解析 URL(明文/兼容旧 b64) -> 拉取二进制 -> 标准图直通 / AES-128-CBC 解密
-> 按字节前缀判 mime -> 返回 [200, mime, bytes]
"""
# 兼容壳子传参: Map 或 dict（值可能为 list）
def _first(v):
if isinstance(v, list):
return v[0].strip() if v and v[0] else ""
return (v or "").strip()
url = ""
if isinstance(params, dict):
url = _first(params.get("url")) or _first((params.get("params") or {}).get("url"))
elif isinstance(params, str):
url = params.strip()
# 若整串是 query 形式（如 "url=xxx&type=cover"），提取 url 值
if "url=" in url:
for seg in url.split("&"):
if seg.startswith("url="):
url = seg[4:]
break
url = (url or "").strip()
if not url:
return [404, "text/plain", b"no url"]
# 兼容: 明文 URL / urlencoded URL / 旧版 b64(url)
decoded = url
if not (url.startswith("http://") or url.startswith("https://")):
from urllib.parse import unquote as _unquote
dec_q = _unquote(url)
if dec_q.startswith("http://") or dec_q.startswith("https://"):
decoded = dec_q
else:
try:
cand = base64.urlsafe_b64decode(url.encode('utf-8')).decode('utf-8')
if cand.startswith("http://") or cand.startswith("https://"):
decoded = cand
except Exception:
pass
url = decoded
raw = None
try:
r = self.session.get(url, timeout=TIMEOUT_API,
headers={"User-Agent": UA,
"Referer": self.siteUrl + "/",
"Accept": "image/webp,image/apng,image/*,*/*;q=0.8"})
if r.status_code == 200 and r.content:
raw = r.content
except Exception:
try:
r = self.session.get(url, timeout=TIMEOUT_API)
if r.status_code == 200 and r.content:
raw = r.content
except Exception:
raw = None
if not raw:
return [404, "text/plain", b"req fail"]
# 标准图直通 / AES-128-CBC 解密（固定 key/iv，实测可解）
data = raw
if not (raw[:2] == b'\xff\xd8' or raw[:8] == b'\x89PNG\r\n\x1a\n'
or raw[:4] == b'RIFF' or raw[:4] == b'GIF8'):
dec = self._aes_decrypt_image(raw)
if dec and (dec[:2] == b'\xff\xd8' or dec[:8] == b'\x89PNG\r\n\x1a\n'
or dec[:4] == b'RIFF' or dec[:4] == b'GIF8'):
data = dec
mime = "image/webp"
if data[:2] == b'\xff\xd8':
mime = "image/jpeg"
elif data[:8] == b'\x89PNG\r\n\x1a\n':
mime = "image/png"
elif data[:4] == b'GIF8':
mime = "image/gif"
return [200, mime, data]
# ==================== 其他 ====================
def isVideoFormat(self, url):
if not url or not isinstance(url, str):
return False
if not url.startswith("http"):
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 destroy(self):
try:
self.session.close()
except Exception:
pass
def close(self):
self.destroy()
# ============================================================
# 自测入口（python 91短剧.py 直接运行）
# ============================================================
def _selftest():
print("=" * 64)
print("91短剧.py 自测（{}）".format(time.strftime('%Y-%m-%d %H:%M:%S')))
print("=" * 64)
sp = Spider()
sp.init()
# 1) home
home = sp.homeContent({})
print(f"[home] 分类数={len(home['class'])} "
f"主分类={[c['type_id'] for c in home['class'][:6]]}")
assert len(home["class"]) == len(ALL_CATES) == 6 + len(TAG_LIST)
# 2) homeVideo
hv = sp.homeVideoContent()
print(f"[homeVideo] {len(hv.get('list') or [])} 条, "
f"首条={((hv.get('list') or [{}])[0].get('vod_name'), (hv.get('list') or [{}])[0].get('vod_pic', '')[:50])}")
assert len(hv.get("list") or []) > 0
# 2.1) 海报断言：vod_pic 必须非空且走本地代理（修复项回归）
for it in (hv.get("list") or [])[:5]:
assert it.get("vod_pic", "").startswith(PROXY_PREFIX), f"海报缺失: {it.get('vod_name')}"
print("[poster] 首页海报 vod_pic 全部非空 ✓")
# 3) category（主分类 + 标签 + 排序）
cat = sp.categoryContent("duanju", 1, None, {"sort": "new"})
print(f"[category duanju] 返回={len(cat.get('list') or [])} "
f"page={cat.get('page')} pagecount={cat.get('pagecount')} "
f"total={cat.get('total')} 首条={((cat.get('list') or [{}])[0].get('vod_name'),)}")
assert len(cat.get("list") or []) > 0
cat_hot = sp.categoryContent("duanju", 1, None, {"sort": "hot"})
print(f"[category duanju hot] 返回={len(cat_hot.get('list') or [])}, "
f"首条={((cat_hot.get('list') or [{}])[0].get('vod_name'),)}")
cat_tag = sp.categoryContent("tag_xiaoyuanduanju", 1, None, None)
print(f"[category tag_xiaoyuanduanju] 返回={len(cat_tag.get('list') or [])}, "
f"首条={((cat_tag.get('list') or [{}])[0].get('vod_name'),)}")
assert len(cat_tag.get("list") or []) > 0 or cat_tag.get("total", 0) > 0
cat_manhua = sp.categoryContent("manhua", 1, None, None)
print(f"[category manhua] 返回={len(cat_manhua.get('list') or [])}, "
f"首条={((cat_manhua.get('list') or [{}])[0].get('vod_name'),)}")
cat_xiaoshuo = sp.categoryContent("xiaoshuo", 1, None, None)
print(f"[category xiaoshuo] 返回={len(cat_xiaoshuo.get('list') or [])}, "
f"首条={((cat_xiaoshuo.get('list') or [{}])[0].get('vod_name'),)}")
# 4) search
sr = sp.searchContent("校花", "", 1)
print(f"[search 校花] total={sr.get('total')} list={len(sr.get('list') or [])} "
f"pagecount={sr.get('pagecount')} 首条={((sr.get('list') or [{}])[0].get('vod_name'),)}")
assert len(sr.get("list") or []) > 0
# 5) detail + play（视频）
item = (cat.get("list") or [{}])[0]
vid = item.get("vod_id")
det = sp.detailContent([vid])["list"][0]
print(f"[detail {vid}] name={det.get('vod_name')} year={det.get('vod_year')} "
f"score={det.get('vod_score')} eps={det.get('vod_play_url', '')[:80]}")
assert det.get("vod_pic", "").startswith(PROXY_PREFIX), "详情海报缺失"
print(f"[poster] 详情海报={det.get('vod_pic', '')[:60]}...")
# 5.1) 海报端到端：localProxy 拉密文 → AES 解密 → 必须是真图字节
from urllib.parse import unquote as _unq
pic_url = _unq(det["vod_pic"][len(PROXY_PREFIX):])
rc = sp.localProxy({"url": pic_url})
assert isinstance(rc, list) and len(rc) == 3, f"localProxy 返回形状错: {type(rc)}"
assert rc[0] == 200, f"localProxy 状态码 {rc[0]}"
body = rc[2]
ok_magic = (body[:2] == b'\xff\xd8' or body[:8] == b'\x89PNG\r\n\x1a\n'
or body[:4] == b'RIFF' or body[:4] == b'GIF8')
assert ok_magic and len(body) > 1000, f"解密结果不是图片: {body[:16]!r} len={len(body)}"
print(f"[poster] localProxy 端到端 ✓ mime={rc[1]} 大小={len(body)}B")
play = sp.playerContent("", vid, {})
print(f"[play {vid}] url={play.get('url', '')[:90]} parse={play.get('parse')}")
sp.close()
if __name__ == "__main__":
_selftest()