yeyin/tvbox_collect
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Sync all projects

Browse Source
This commit is contained in:
github-actions[bot]
2026-07-30 17:32:45 +00:00
parent 3f366cb15e
commit fb3649e275
31 changed files with 26017 additions and 20790 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tvbox/py/ddmf.py
+229
View File
@@ -0,0 +1,229 @@
#!/usr/bin/python
# -*- coding: utf-8 -*-
import re, json, requests
from urllib.parse import quote
try:
from lxml import etree
except Exception:
etree = None
from base.spider import Spider
PAGEFMT = ["%s?page=%s", "%s/page/%s"]
class Spider(Spider):
def getName(self): return "蛋蛋魔法影视"
def init(self, extend=""):
self.host = "https://www.ddmf.net"
try: ext = json.loads(extend) if str(extend).strip().startswith("{") else {}
except Exception: ext = {}
if ext.get("host"): self.host = ext["host"].rstrip("/")
self.pageFmt = ext.get("pageFmt", "")
ua = ext.get("ua", "Mozilla/5.0 (Linux; Android 13; V2219A) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/131.0.0.0 Mobile Safari/537.36")
self.headers = {
"User-Agent": ua,
"Referer": self.host + "/",
"Origin": self.host,
"X-Turbo-Charged-By": "LiteSpeed",
}
if ext.get("cookie"): self.headers["Cookie"] = ext["cookie"]
self.relay = ext.get("relay", "").rstrip("/")
self.categories = [{"type_id": "1", "type_name": "电影"}, {"type_id": "2", "type_name": "电视剧"}, {"type_id": "3", "type_name": "综艺"}, {"type_id": "4", "type_name": "动漫"}]
def _fix(self, u):
if not u: return ""
if u.startswith("//"): return "https:" + u
if u.startswith("/"): return self.host + u
return u
def _gated(self, html):
marks = ("Just a moment", "cf-chl", "__cf_chl", "Enable JavaScript and cookies", "cf_clearance", "Checking your browser")
return bool(html) and any(m in html for m in marks)
def _garbled(self, html):
if not html: return False
sample = html[:200]
bad = sum(1 for c in sample if ord(c) < 32 and c not in "\t\n\r" or ord(c) > 0xfffd or (0xd800 <= ord(c) <= 0xdfff))
return len(sample) > 0 and bad / len(sample) > 0.15
def _direct(self, url):
try:
r = requests.get(url, headers=self.headers, timeout=15); r.encoding = "utf-8"
return r.status_code, (r.text or "")
except requests.exceptions.Timeout: print("[ERROR] 请求超时: %s" % url); return None, ""
except requests.exceptions.ConnectionError: print("[ERROR] 连接错误: %s" % url); return None, ""
except Exception as e: print("[ERROR] 请求失败: %s, %s" % (url, str(e))); return None, ""
def _via_relay(self, url):
if not self.relay: return None
try:
r = requests.get(self.relay + "/?url=" + quote(url, safe=""), headers={"User-Agent": self.headers["User-Agent"]}, timeout=45)
r.encoding = "utf-8"
return r.text or ""
except Exception as e:
print("[ERROR] relay失败: %s" % str(e)); return None
def _get(self, path):
url = path if path.startswith("http") else self.host + path
status, body = self._direct(url)
garbled = self._garbled(body)
blocked = status is None or status >= 400 or self._gated(body) or garbled
if blocked:
if garbled:
reason = "响应体疑似未正确解压(乱码，非拦截页文字)"
elif self._gated(body):
reason = "疑似JS挑战/机器人识别页"
else:
reason = "状态码%s" % status
print("[WARN] 直连异常(%s) url=%s len=%d 片段=%r" % (reason, url, len(body), body[:200]))
if garbled:
print("[WARN] 响应体是压缩后未解压的二进制，不是拦截页内容，无法据此判断是否被拦截，请检查Accept-Encoding与本环境的解压库支持")
return None
if self.relay:
print("[INFO] 尝试通过relay取页: %s" % url)
relayed = self._via_relay(url)
if relayed and not self._gated(relayed) and not self._garbled(relayed):
return relayed
print("[WARN] relay取页仍失败或仍被拦截")
else:
print("[WARN] 浏览器可正常访问但脚本被拦，可能是请求头/TLS指纹识别，也可能是其他原因，需拿到真实响应体内容才能判断。"
"建议：1) 确认上面的浏览器级请求头是否已生效；2) 配置 ext.relay 走真实浏览器内核中转")
return None
return body
def _tree(self, html, tag="页面"):
if not html: return None
if etree is None: print("[WARN] lxml 不可用"); return None
tree = etree.HTML(html)
if tree is None: print("[WARN] %s etree解析为空，长度=%d 片段=%r" % (tag, len(html), html[:80]))
return tree
def _regex_list(self, html):
out, seen = [], set()
for vid, title in re.findall(r'href="[^"]*?/voddetail/(\d+)\.html"[^>]*title="([^"]*)"', html):
if vid in seen: continue
seen.add(vid); out.append({"vod_id": vid, "vod_name": title, "vod_pic": ""})
return out
def _parse_list(self, html):
if not html: return []
tree = self._tree(html, "列表") if etree else None
if tree is None: return self._regex_list(html)
results, seen = [], set()
for a in tree.xpath('//a[contains(@href,"/voddetail/")]'):
m = re.search(r'/voddetail/(\d+)\.html', a.get("href", ""))
if not m or m.group(1) in seen: continue
name = (a.get("title") or "".join(a.xpath('.//text()'))).strip()
if not name: continue
seen.add(m.group(1))
pic = ""
for at in ("data-original", "data-src", "src"):
v = a.xpath('.//img/@%s' % at)
if v and "load.gif" not in v[0]: pic = v[0]; break
note = "".join(a.xpath('.//span//text()')).strip()
results.append({"vod_id": m.group(1), "vod_name": name, "vod_pic": self._fix(pic), "vod_remarks": note})
return results
def _first(self, lst): return lst[0]["vod_id"] if lst else ""
def _paged(self, base, pg):
if pg == "1": return self._get(base)
if self.pageFmt: return self._get(self.pageFmt % (base, pg))
first = self._first(self._parse_list(self._get(base)))
for f in PAGEFMT:
html = self._get(f % (base, pg))
got = self._parse_list(html)
if got and self._first(got) != first:
self.pageFmt = f
print("[INFO] 分页格式确定: %s" % (f % ("{base}", "{pg}")))
return html
print("[WARN] 未能确定分页格式(vodshow页码位置未验证)，仅返回首屏")
return self._get(base)
def _safe_home_list(self):
# 首页 / 混杂了福利视频分区条目，不作为列表源；改为仅拼接白名单分类(电影/电视剧/综艺/动漫)的
# vodtype 落地页，从源头避免福利视频内容混入
out, seen = [], set()
for c in self.categories:
for v in self._parse_list(self._get("/vodtype/%s.html" % c["type_id"])):
if v["vod_id"] in seen: continue
seen.add(v["vod_id"]); out.append(v)
if len(out) >= 60: return out
return out
def homeContent(self, filter):
fl = {c["type_id"]: [{"key": "year", "name": "年份", "value": [{"n": "全部", "v": ""}] + [{"n": str(y), "v": str(y)} for y in range(2026, 2014, -1)]}] for c in self.categories}
return {"class": self.categories, "list": self._safe_home_list(), "filters": fl}
def homeVideoContent(self): return {"list": self._safe_home_list()}
def categoryContent(self, tid, pg, filter, extend):
pg = str(pg or "1")
year = (extend or {}).get("year", "")
base = "/vodshow/%s-----------%s.html" % (tid, year) if year else "/vodshow/%s-----------.html" % tid
lst = self._parse_list(self._paged(base, pg))
return {"page": int(pg), "pagecount": int(pg) + 1 if lst else int(pg), "limit": 32, "total": 999999, "list": lst}
def searchContent(self, key, quick, pg="1"):
pg = str(pg or "1")
lst = self._parse_list(self._get("/vodsearch/%s-------------.html" % quote(key)))
return {"list": lst, "page": int(pg)}
def _field(self, text, key):
m = re.search(r'%s\s*[:：]\s*([^\n]{1,200})' % key, text)
return m.group(1).strip(" \u3000|/") if m else ""
def detailContent(self, ids):
vid = re.sub(r'\D', '', str(ids[0]))
html = self._get("/voddetail/%s.html" % vid)
tree = self._tree(html, "详情页")
if tree is None: return {"list": []}
text = "\n".join(x.strip() for x in tree.xpath('//text()') if x.strip())
pic = ""
for v in tree.xpath('//img/@data-original | //img/@src'):
if v and "load.gif" not in v: pic = v; break
vod = {"vod_id": vid,
"vod_name": "".join(tree.xpath('//h1//text()')).strip(),
"vod_pic": self._fix(pic),
"vod_year": self._field(text, "年份") or "".join(tree.xpath('//a[contains(@href,"/vodshow/")]/text()')[:1]).strip(),
"vod_area": "".join(tree.xpath('//a[contains(@href,"/vodshow/")]/text()')[1:2]).strip(),
"type_name": "".join(tree.xpath('//a[contains(@href,"/vodshow/")]/text()')[2:3]).strip(),
"vod_director": self._field(text, "导演"), "vod_actor": self._field(text, "主演"),
"vod_remarks": self._field(text, "连载") or self._field(text, "更新"),
"vod_content": self._field(text, "剧情") or self._field(text, "简介")}
groups, seen = {}, set()
for a in tree.xpath('//a[contains(@href,"/vodplay/")]'):
lk = a.get("href", "")
if lk in seen: continue
m = re.search(r'/vodplay/%s-(\d+)-(\d+)\.html' % vid, lk)
if not m: continue
seen.add(lk)
src, ep = m.group(1), int(m.group(2))
nm = (a.get("title") or "".join(a.xpath('.//text()'))).strip()
nm = re.sub(r'^播放.*?第|集$', '', nm) or ("第%d集" % ep)
groups.setdefault(src, []).append((ep, ("第%s集" % nm if nm.isdigit() else nm), lk))
froms, urls = [], []
for src in sorted(groups, key=lambda s: (len(s), s)):
eps = sorted(groups[src], key=lambda x: x[0])
froms.append("线路%s" % src)
urls.append("#".join(nm.replace("$", "").replace("#", "") + "$" + self._fix(lk) for _, nm, lk in eps))
vod["vod_play_from"] = "$$$".join(froms) if froms else "蛋蛋魔法影视"
vod["vod_play_url"] = "$$$".join(urls) if urls else ("正片$%s/voddetail/%s.html" % (self.host, vid))
return {"list": [vod]}
def playerContent(self, flag, id, vipFlags):
pid = id if id.startswith("http") else self._fix(id)
html = self._get(pid) or ""
url = ""
for p in [r'var\s+player_\w*\s*=\s*(\{.*?\})\s*[<;]', r'"url"\s*:\s*"([^"]+)"', r'var\s+now\s*=\s*["\']([^"\']+)["\']', r'url:\s*["\']([^"\']+\.(?:m3u8|mp4)[^"\']*)["\']', r'(https?://[^\s"\'\\<>]+\.(?:m3u8|mp4)[^\s"\'\\<>]*)']:
m = re.search(p, html.replace("\\/", "/"), re.S)
if not m: continue
val = m.group(1)
if val.startswith("{"):
try: val = json.loads(val).get("url", "")
except Exception:
m2 = re.search(r'"url"\s*:\s*"([^"]+)"', val); val = m2.group(1).replace("\\/", "/") if m2 else ""
if val: url = self._fix(val); break
if not url: return {"parse": 1, "url": pid, "header": self.headers}
return {"parse": 0, "url": url, "header": {"User-Agent": self.headers["User-Agent"], "Referer": self.host + "/"}}
tvbox/py/ddys.py
-161
View File
@@ -1,161 +0,0 @@
#!/usr/bin/python
# -*- coding: utf-8 -*-
import re, json, requests
from urllib.parse import quote
try:
from lxml import etree
except Exception:
etree = None
from base.spider import Spider
BADIMG = ("logo", "loading", "nopic", "no-pic", "static/")
class Spider(Spider):
def getName(self): return "低端影视"
def init(self, extend=""):
self.host = "https://www.ddys.run"
try: ext = json.loads(extend) if str(extend).strip().startswith("{") else {}
except Exception: ext = {}
if ext.get("host"): self.host = ext["host"].rstrip("/")
self.headers = {"User-Agent": ext.get("ua", "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/131.0.0.0 Safari/537.36"), "Referer": self.host + "/", "Accept-Language": "zh-CN,zh;q=0.9", "Cookie": ext.get("cookie", "pageLogin=ok")}
self.listFmt = ext.get("listFmt", "")
self.categories = [{"type_id": "dianying", "type_name": "电影"}, {"type_id": "juji", "type_name": "剧集"}, {"type_id": "dongman", "type_name": "动漫"}]
def _fix(self, u):
if not u: return ""
if u.startswith("//"): return "https:" + u
if u.startswith("/"): return self.host + u
return u
def _get(self, path):
url = path if path.startswith("http") else self.host + path
try:
r = requests.get(url, headers=self.headers, timeout=15); r.encoding = "utf-8"
if r.status_code >= 400: print("[WARN] status=%s url=%s" % (r.status_code, url)); return None
return r.text
except requests.exceptions.Timeout: print("[ERROR] 请求超时: %s" % url)
except requests.exceptions.ConnectionError: print("[ERROR] 连接错误: %s" % url)
except Exception as e: print("[ERROR] 请求失败: %s, %s" % (url, str(e)))
return None
def _post(self, path, data):
try:
r = requests.post(self.host + path, data=data, headers=self.headers, timeout=15); r.encoding = "utf-8"; return r.text
except Exception as e: print("[ERROR] POST失败: %s, %s" % (path, str(e))); return None
def _img(self, node):
for at in ("data-original", "data-src", "data-echo", "src"):
for v in node.xpath('.//@%s' % at):
if v and re.search(r'\.(jpg|jpeg|png|webp|avif)', v) and not any(b in v for b in BADIMG): return v
return ""
def _parse_list(self, html):
if not html: return []
if etree is None:
print("[WARN] lxml 不可用，降级为正则解析")
out, seen = [], set()
for slug, title in re.findall(r'href="/video/([^"]+)\.html"[^>]*title="([^"]*)"', html):
if slug in seen: continue
seen.add(slug); out.append({"vod_id": slug, "vod_name": title, "vod_pic": ""})
return out
tree = etree.HTML(html); results, seen = [], set()
items = tree.xpath('//a[contains(@class,"stui-vodlist__thumb") and contains(@href,"/video/")]') or tree.xpath('//div[contains(@class,"stui-vodlist__box")]//a[contains(@href,"/video/")]') or tree.xpath('//a[contains(@href,"/video/") and @title]')
for it in items:
try:
m = re.search(r'/video/([^/.]+)\.html', it.get("href", ""))
if not m or m.group(1) in seen: continue
name = (it.get("title") or "".join(it.xpath('.//img/@alt')[:1])).strip()
if not name: continue
seen.add(m.group(1))
note = "".join(it.xpath('.//span[contains(@class,"pic-text")]//text()')).strip()
results.append({"vod_id": m.group(1), "vod_name": name, "vod_pic": self._fix(self._img(it)), "vod_remarks": note})
except Exception: continue
return results
def _dcount(self, sep):
try: return len(re.findall(r'/list/[a-z0-9]+(%s*)\.html' % re.escape(sep), self._home)[0])
except Exception: return 11
def homeContent(self, filter):
self._home = self._get("/") or ""
fl = {}
return {"class": self.categories, "list": self._parse_list(self._home), "filters": fl}
def homeVideoContent(self): return {"list": self._parse_list(self._get("/"))}
def categoryContent(self, tid, pg, filter, extend):
pg = str(pg or "1")
cands = ["/list/%s-%s----------.html" % (tid, pg), "/list/%s----------%s-.html" % (tid, pg), "/list/%s-----------.html?page=%s" % (tid, pg)]
if self.listFmt: cands = [self.listFmt % (tid, pg)]
lst = []
for c in cands:
lst = self._parse_list(self._get(c))
if lst:
self.listFmt = c.replace(tid, "%s", 1).replace("-%s-" % pg, "-%s-", 1).replace("%s-" % pg + "-", "%s--", 1) if not self.listFmt else self.listFmt
self.listFmt = c.split(tid, 1)[0] + "%s" + c.split(tid, 1)[1].replace(pg, "%s", 1)
break
return {"page": int(pg), "pagecount": int(pg) + 1 if lst else int(pg), "limit": 48, "total": 999999, "list": lst}
def searchContent(self, key, quick, pg="1"):
pg = str(pg or "1")
lst = self._parse_list(self._get("/search/-------------.html?wd=%s&page=%s" % (quote(key), pg)))
if not lst:
lst = self._parse_list(self._post("/search/-------------.html", {"wd": key}))
if not lst:
lst = self._parse_list(self._get("/vodsearch/-------------.html?wd=%s" % quote(key)))
return {"list": lst, "page": int(pg)}
def detailContent(self, ids):
slug = str(ids[0])
html = self._get("/video/%s.html" % slug)
if not html or etree is None: return {"list": []}
tree = etree.HTML(html)
text = "\n".join(x.strip() for x in tree.xpath('//text()') if x.strip())
pic = self._img(tree) or "".join(tree.xpath('//img[contains(@class,"lazyload")]/@data-original')[:1])
vod = {"vod_id": slug,
"vod_name": ("".join(tree.xpath('//h1//text()')).strip() or "".join(tree.xpath('//div[contains(@class,"stui-content__detail")]//h1/text()')).strip()),
"vod_pic": self._fix(pic),
"vod_year": self._field(text, "年份"), "vod_area": self._field(text, "地区"),
"type_name": self._field(text, "分类") or self._field(text, "类型"),
"vod_lang": self._field(text, "语言"),
"vod_actor": self._field(text, "主演"), "vod_director": self._field(text, "导演"),
"vod_remarks": self._field(text, "备注") or self._field(text, "状态"),
"vod_content": re.sub(r'\s+', ' ', "".join(tree.xpath('//span[contains(@class,"detail-content")]//text() | //div[contains(@class,"detail")]//span[contains(@class,"content")]//text()'))).strip()}
froms, urls = [], []
heads = tree.xpath('//div[contains(@class,"stui-pannel__head") or contains(@class,"playlist")]//h3/text() | //div[contains(@class,"stui-vodlist__head")]//h3/text()')
lists = tree.xpath('//ul[contains(@class,"stui-content__playlist") or contains(@class,"stui-content__list") or contains(@class,"content__playlist")]')
for i, ul in enumerate(lists):
eps = []
for a in ul.xpath('.//a'):
nm = ("".join(a.xpath('.//text()')).strip() or a.get("title", "")).strip()
lk = a.get("href", "")
if not nm or not lk: continue
eps.append(nm.replace("$", "").replace("#", "") + "$" + self._fix(lk))
if eps:
froms.append(heads[i].strip() if i < len(heads) else "线路%d" % (i + 1))
urls.append("#".join(eps))
vod["vod_play_from"] = "$$$".join(froms) if froms else "低端影视"
vod["vod_play_url"] = "$$$".join(urls) if urls else ("正片$%s/video/%s.html" % (self.host, slug))
return {"list": [vod]}
def _field(self, text, key):
m = re.search(r'%s\s*[:：]\s*([^\n]{1,200})' % key, text)
return m.group(1).strip(" \u3000|/") if m else ""
def playerContent(self, flag, id, vipFlags):
pid = id if id.startswith("http") else self._fix(id)
html = self._get(pid) or ""
url = ""
for p in [r'var\s+player_\w*\s*=\s*(\{.*?\})\s*[<;]', r'"url"\s*:\s*"([^"]+)"', r'var\s+now\s*=\s*["\']([^"\']+)["\']', r'url:\s*["\']([^"\']+\.(?:m3u8|mp4)[^"\']*)["\']', r'(https?://[^\s"\'\\<>]+\.(?:m3u8|mp4)[^\s"\'\\<>]*)']:
m = re.search(p, html.replace("\\/", "/"), re.S)
if not m: continue
val = m.group(1)
if val.startswith("{"):
try: val = json.loads(val).get("url", "")
except Exception:
m2 = re.search(r'"url"\s*:\s*"([^"]+)"', val); val = m2.group(1).replace("\\/", "/") if m2 else ""
if val: url = self._fix(val); break
if not url: return {"parse": 1, "url": pid, "header": self.headers}
return {"parse": 0, "url": url, "header": {"User-Agent": self.headers["User-Agent"], "Referer": self.host + "/"}}
tvbox/py/xlys.py
+233
View File
@@ -0,0 +1,233 @@
#!/usr/bin/python
# -*- coding: utf-8 -*-
import re, json, requests
from collections import Counter
from urllib.parse import quote
try:
from lxml import etree
except Exception:
etree = None
from base.spider import Spider
# 注意：站点存在 /aiad (AI脱衣/AI换脸类工具) 入口，本源不抓取、不引用、不提供任何访问路径
GENRES = [["动作", "dongzuo"], ["爱情", "aiqing"], ["喜剧", "xiju"], ["科幻", "kehuan"], ["恐怖", "kongbu"],
["战争", "zhanzheng"], ["武侠", "wuxia"], ["魔幻", "mohuan"], ["剧情", "juqing"], ["动画", "donghua"],
["惊悚", "jingsong"], ["3D", "3D"], ["灾难", "zainan"], ["悬疑", "xuanyi"], ["警匪", "jingfei"],
["文艺", "wenyi"], ["青春", "qingchun"], ["冒险", "maoxian"], ["犯罪", "fanzui"], ["纪录", "jilu"],
["古装", "guzhuang"], ["奇幻", "qihuan"], ["国语", "guoyu"], ["综艺", "zongyi"], ["历史", "lishi"],
["运动", "yundong"], ["原创压制", "yuanchuang"], ["美剧", "meiju"], ["韩剧", "hanju"],
["国产电视剧", "guoju"], ["日剧", "riju"], ["英剧", "yingju"], ["德剧", "deju"], ["俄剧", "eju"],
["巴剧", "baju"], ["加剧", "jiaju"], ["西剧", "spanish"], ["意大利剧", "yidaliju"], ["泰剧", "taiju"],
["港台剧", "gangtaiju"], ["法剧", "faju"], ["澳剧", "aoju"], ["短剧", "duanju"]]
ITEM_RE = re.compile(r'^/([A-Za-z0-9]+)/(\d+)\.htm$')
class Spider(Spider):
def getName(self): return "雪落影视"
def init(self, extend=""):
self.host = "https://v.xl01.eu.cc"
try: ext = json.loads(extend) if str(extend).strip().startswith("{") else {}
except Exception: ext = {}
if ext.get("host"): self.host = ext["host"].rstrip("/")
self.headers = {"User-Agent": ext.get("ua", "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/131.0.0.0 Safari/537.36"), "Referer": self.host + "/", "Accept-Language": "zh-CN,zh;q=0.9"}
if ext.get("cookie"): self.headers["Cookie"] = ext["cookie"]
self.categories = [{"type_id": "all0", "type_name": "电影"}, {"type_id": "all1", "type_name": "剧集"}] + \
[{"type_id": g[1], "type_name": g[0]} for g in GENRES]
def _fix(self, u):
if not u: return ""
if u.startswith("//"): return "https:" + u
if u.startswith("/"): return self.host + u
return u
def _get(self, path):
url = path if path.startswith("http") else self.host + path
try:
r = requests.get(url, headers=self.headers, timeout=15); r.encoding = "utf-8"
if r.status_code >= 400: print("[WARN] status=%s url=%s" % (r.status_code, url)); return None
body = r.text
if body and body[0] == "\ufeff": body = body[1:] # 去除UTF-8 BOM，否则lxml.etree.HTML会报XMLSyntaxError(USC4 little endian)
return body
except requests.exceptions.Timeout: print("[ERROR] 请求超时: %s" % url)
except requests.exceptions.ConnectionError: print("[ERROR] 连接错误: %s" % url)
except Exception as e: print("[ERROR] 请求失败: %s, %s" % (url, str(e)))
return None
def _tree(self, html, tag="页面"):
if not html: return None
if etree is None: print("[WARN] lxml 不可用"); return None
try:
tree = etree.HTML(html)
except Exception as e:
print("[WARN] %s etree解析异常: %s: %s，长度=%d 片段=%r" % (tag, type(e).__name__, e, len(html), html[:80]))
return None
if tree is None: print("[WARN] %s etree解析为空，长度=%d 片段=%r" % (tag, len(html), html[:80]))
return tree
def _regex_list(self, html):
out, seen = [], set()
for m in re.finditer(r'href="([^"]+)"[^>]*title="([^"]*)"', html):
mm = ITEM_RE.match(m.group(1))
if not mm or mm.group(0) in seen: continue
seen.add(mm.group(0))
out.append({"vod_id": "%s/%s" % (mm.group(1), mm.group(2)), "vod_name": m.group(2), "vod_pic": ""})
return out
def _parse_list(self, html):
if not html: return []
tree = self._tree(html, "列表") if etree else None
if tree is None: return self._regex_list(html)
groups = {}
for a in tree.xpath('//a[@href]'):
href = a.get("href", "")
m = ITEM_RE.match(href.split("?")[0].split("#")[0])
if not m: continue
groups.setdefault((m.group(1), m.group(2)), []).append(a)
results = []
for (genre, vid), anchors in groups.items():
name, pic, note = "", "", ""
for a in anchors:
if not name:
h4 = "".join(a.xpath('.//h4//text()')).strip()
if h4: name = h4
if not pic:
for at in ("data-original", "data-src", "src"):
v = a.xpath('.//img/@%s' % at)
if v and v[0].strip(): pic = v[0]; break
if not note: note = "".join(a.xpath('.//span//text() | .//em//text()')).strip()
if not name:
for a in anchors:
cand = (a.xpath('./following-sibling::h4[1]//text()')
or a.xpath('./preceding-sibling::h4[1]//text()')
or a.xpath('../h4//text()')
or a.xpath('../following-sibling::h4[1]//text()')
or a.xpath('../preceding-sibling::h4[1]//text()'))
if cand: name = "".join(cand).strip(); break
if not name:
for a in anchors:
if a.get("title"): name = a.get("title").strip(); break
if not name: continue
results.append({"vod_id": "%s/%s" % (genre, vid), "vod_name": name,
"vod_pic": self._fix(pic), "vod_remarks": note})
return results
def _first(self, lst): return lst[0]["vod_id"] if lst else ""
def _paged(self, base, pg):
if pg == "1": return self._get(base)
sep = "&" if "?" in base else "?"
cands = ["%s" + sep + "page=%s"] + (["%s/page/%s"] if "?" not in base else [])
first = self._first(self._parse_list(self._get(base)))
for f in cands:
html = self._get(f % (base, pg))
got = self._parse_list(html)
if got and self._first(got) != first:
print("[INFO] 分页格式确定: %s" % (f % ("{base}", "{pg}")))
return html
print("[WARN] 未能确定分页格式，仅返回首屏")
return self._get(base)
def homeContent(self, filter):
fl = {"all0": [{"key": "area", "name": "地区", "value": [{"n": "全部", "v": ""}, {"n": "中国大陆", "v": "中国大陆"}]}],
"all1": [{"key": "area", "name": "地区", "value": [{"n": "全部", "v": ""}, {"n": "中国大陆", "v": "中国大陆"}]}]}
return {"class": self.categories, "list": self._parse_list(self._get("/")), "filters": fl}
def homeVideoContent(self): return {"list": self._parse_list(self._get("/"))}
def categoryContent(self, tid, pg, filter, extend):
pg = str(pg or "1")
if tid.startswith("all"):
base = "/s/all?type=%s" % tid[3:]
area = (extend or {}).get("area", "")
if area: base += "&area=%s" % quote(area)
else:
base = "/s/%s" % tid
lst = self._parse_list(self._paged(base, pg))
return {"page": int(pg), "pagecount": int(pg) + 1 if lst else int(pg), "limit": 30, "total": 999999, "list": lst}
def searchContent(self, key, quick, pg="1"):
pg = str(pg or "1")
for p in ["/s/all?wd=%s&page=%s" % (quote(key), pg), "/search?wd=%s&page=%s" % (quote(key), pg)]:
lst = self._parse_list(self._get(p))
if lst: return {"list": lst, "page": int(pg)}
return {"list": [], "page": int(pg)}
def _field(self, text, key):
m = re.search(r'%s\s*[:：]\s*([^\n]{1,200})' % key, text)
return m.group(1).strip(" \u3000|/") if m else ""
def detailContent(self, ids):
vid = str(ids[0])
html = self._get("/%s.htm" % vid)
tree = self._tree(html, "详情页")
if tree is None: return {"list": []}
text = "\n".join(x.strip() for x in tree.xpath('//text()') if x.strip())
title_line = "".join(tree.xpath('//h1//text()')).strip()
m = re.match(r'^(.*?)\s*[\(（](\d{4})[\)）]\s*$', title_line)
name, year = (m.group(1), m.group(2)) if m else (title_line, self._field(text, "年份"))
rate_m = re.search(r'(\d\.\d)\s*豆瓣评分', text)
pic = ""
for v in tree.xpath('//img/@data-original | //img/@src'):
if v and v.strip(): pic = v; break
# 该站每个页面的导航菜单会把全部约40个类型链接完整重复渲染，频次相同；
# 正文里该条目自己标注的"类型"标签会比导航基线多出现1次，据此和导航区分，无需依赖具体的CSS/容器结构
freq = {}
for a in tree.xpath('//a[starts-with(@href,"/s/")]'):
h = a.get("href"); freq[h] = freq.get(h, 0) + 1
baseline = Counter(freq.values()).most_common(1)[0][0] if freq else 0
item_genres = [h.split("/s/")[-1] for h, c in freq.items() if c > baseline]
slug_name = {g[1]: g[0] for g in GENRES}
type_name = " ".join(slug_name.get(s, s) for s in item_genres) or self._field(text, "类型")
vod = {"vod_id": vid, "vod_name": name, "vod_pic": self._fix(pic),
"vod_year": year, "vod_area": self._field(text, "制片国家"),
"type_name": type_name,
"vod_director": " ".join(tree.xpath('//a[contains(@href,"/director/")]/text()')),
"vod_actor": " ".join(tree.xpath('//a[contains(@href,"/performer/")]/text()')),
"vod_remarks": (rate_m.group(1) + "分") if rate_m else (self._field(text, "集数") and ("共%s集" % self._field(text, "集数"))),
"vod_content": ""}
pre = text.split("在线播放", 1)[0] if "在线播放" in text else text
lines = [l.strip() for l in pre.split("\n") if l.strip()]
last_label = -1
for i, l in enumerate(lines):
if re.match(r'^[\u4e00-\u9fffA-Za-z]{2,8}[:：]', l): last_label = i
content_lines = lines[last_label + 1:] if last_label >= 0 else lines
vod["vod_content"] = " ".join(content_lines)[:500]
eps, seen = [], set()
for m2 in re.finditer(r'<a[^>]+href="([^"]*?/play/%s-\d+\.htm[^"]*)"[^>]*>([^<]*)</a>' % re.escape(vid.split("/")[-1]), html):
href, nm = m2.group(1), m2.group(2).strip()
if not nm or href in seen: continue
seen.add(href)
eps.append(nm.replace("$", "").replace("#", "") + "$" + self._fix(href))
vod["vod_play_from"] = "雪落影视"
vod["vod_play_url"] = "#".join(eps) if eps else ("播放$%s/play/%s-0.htm" % (self.host, vid.split("/")[-1]))
return {"list": [vod]}
def playerContent(self, flag, id, vipFlags):
pid = id if id.startswith("http") else self._fix(id)
html = self._get(pid) or ""
url = ""
# 专门识别 {"code":0,"data":{"url3":...}} 这种结构（若play页内联嵌入了该JSON）
# 只取 url3；若 url3 本身是逗号分隔的多个候选地址（如 "url_a,url_b,url_c"），取第一个
jm = re.search(r'\{"code"\s*:\s*0[^{}]*"data"\s*:\s*(\{[^{}]*\})\}', html.replace("\\/", "/"))
if jm:
try:
data = json.loads(jm.group(1))
url3 = data.get("url3", "")
url = url3.split(",")[0].strip() if url3 else ""
except Exception:
url = ""
if not url:
for p in [r'var\s+player_\w*\s*=\s*(\{.*?\})\s*[<;]', r'"url"\s*:\s*"([^"]+)"', r'var\s+now\s*=\s*["\']([^"\']+)["\']', r'url:\s*["\']([^"\']+\.(?:m3u8|mp4)[^"\']*)["\']', r'(https?://[^\s"\'\\<>]+\.(?:m3u8|mp4)[^\s"\'\\<>]*)']:
m = re.search(p, html.replace("\\/", "/"), re.S)
if not m: continue
val = m.group(1)
if val.startswith("{"):
try:
val = json.loads(val).get("url", "")
except Exception:
m2 = re.search(r'"url"\s*:\s*"([^"]+)"', val)
val = m2.group(1).replace("\\/", "/") if m2 else ""
if val: url = self._fix(val); break
if not url: return {"parse": 1, "url": pid, "header": self.headers}
return {"parse": 0, "url": url, "header": {"User-Agent": self.headers["User-Agent"], "Referer": self.host + "/"}}