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-26 09:36:28 +00:00
parent 0f21ecc718
commit fcaec891b1
25 changed files with 6472 additions and 4795 deletions
Download Patch File Download Diff File Expand all files Collapse all files
FGBLH/py/BadNews.py
+874
View File
@@ -0,0 +1,874 @@
# -*- coding: utf-8 -*-
# //@name:BadNews直播放
# //@id:badnews_direct
# //@version:7
import hashlib
import html as html_lib
import json
import re
import time
from urllib.parse import quote, unquote, urljoin, urlsplit
import requests
from lxml import html
from base.spider import Spider as BaseSpider
try:
from com.github.catvod import Proxy as CatVodProxy
except Exception:
CatVodProxy = None
class Spider(BaseSpider):
name = "BadNews直播放"
host = "https://bad.news"
backend_parse = False
category_mode = False
categoryMode = False
PLAY_PREFIX = "badnews-play:"
ERROR_PREFIX = "badnews-error:"
DEFAULT_PIC = "https://bad.news/favicon.ico"
CATEGORY_SPECS = (
("hot", "热门视频", "entry", "/sort-hot"),
("new", "最新视频", "entry", "/sort-new"),
("short", "短视频", "entry", "/tag/porn"),
("long", "长视频", "entry", "/tag/long-porn"),
("dm", "H动漫", "dm", "/dm"),
("dm_3d", "3D动画", "dm", "/dm/type/q-3D"),
("dm_doujin", "同人作品", "dm", "/dm/type/q-同人"),
("dm_cosplay", "Cosplay", "dm", "/dm/type/q-Cosplay"),
("better", "精选视频", "entry", "/sort-better"),
("score", "高分视频", "entry", "/sort-score"),
)
BLOCKED_HOSTS = frozenset(
{
"script-center.bad.news",
"portalfluently.com",
"vivodemisrentas.net",
"secretlygoatsarrangement.com",
"ri1.xlfn.cc",
"static.cloudflareinsights.com",
"www.google-analytics.com",
"www.googletagmanager.com",
"www.statcounter.com",
}
)
MEDIA_HOSTS = frozenset({"video.twimg.com", "static.bad.news"})
CHALLENGE_MARKERS = (
"just a moment",
"/cdn-cgi/challenge-platform",
"_cf_chl_opt",
"cf-turnstile",
"turnstile",
)
CONTENT_MARKERS = (
'class="entry',
"class='entry",
"<article",
"<video",
"data-source=",
"/dm/play/id-",
)
VIDEO_URL_RE = re.compile(r"\.(?:m3u8|mp4)(?:$|[?#])", re.I)
PAGE_RE = re.compile(r"/page-(\d+)(?:$|[/?#])", re.I)
TOPIC_ID_RE = re.compile(r"/t/(\d+)(?:$|[/?#])", re.I)
DM_ID_RE = re.compile(r"/dm/play/id-(\d+)(?:$|[/?#])", re.I)
def __init__(self):
try:
super().__init__()
except Exception:
pass
self.timeout = 15
self.verify_tls = True
self.trust_env = True
self.proxy = ""
self.cache_ttl = 30
self.prefer_progressive_mp4 = True
self.lock_hls_highest = True
self.user_agent = (
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
"AppleWebKit/537.36 (KHTML, like Gecko) "
"Chrome/131.0.0.0 Safari/537.36"
)
self._session = None
self._cache = {}
self._proxy_manifests = {}
self._reset_session()
def getName(self):
return self.name
def init(self, extend=""):
config = self._parse_config(extend)
configured_host = str(config.get("host") or self.host).strip().rstrip("/")
if configured_host.startswith(("http://", "https://")):
self.host = configured_host
self.timeout = self._bounded_int(config.get("timeout"), self.timeout, 5, 45)
self.cache_ttl = self._bounded_int(config.get("cache_ttl"), self.cache_ttl, 0, 300)
self.verify_tls = self._bool_value(config.get("verify_tls"), self.verify_tls)
self.trust_env = self._bool_value(config.get("trust_env"), self.trust_env)
self.prefer_progressive_mp4 = self._bool_value(
config.get("prefer_progressive_mp4", config.get("prefer_mp4")),
self.prefer_progressive_mp4,
)
self.lock_hls_highest = self._bool_value(
config.get("lock_hls_highest"), self.lock_hls_highest
)
self.proxy = str(config.get("proxy") or "").strip()
configured_ua = str(config.get("user_agent") or "").strip()
if configured_ua:
self.user_agent = configured_ua
self._cache.clear()
self._proxy_manifests.clear()
self._reset_session()
def destroy(self):
if self._session is not None:
try:
self._session.close()
except Exception:
pass
self._session = None
self._cache.clear()
self._proxy_manifests.clear()
def isVideoFormat(self, url):
return bool(self.VIDEO_URL_RE.search(str(url or "")))
def manualVideoCheck(self):
return False
def localProxy(self, param):
data = param if isinstance(param, dict) else self._parse_config(param)
token = str(data.get("token") or "").strip()
cached = self._proxy_manifests.get(token)
if not cached or time.time() - cached[0] > 1800:
return [404, "text/plain; charset=utf-8", b"manifest not found"]
return [
200,
"application/vnd.apple.mpegurl",
cached[1],
{"Cache-Control": "no-store", "Access-Control-Allow-Origin": "*"},
]
def homeContent(self, filter):
return {
"class": [
{"type_id": type_id, "type_name": type_name}
for type_id, type_name, _, _ in self.CATEGORY_SPECS
],
"filters": {},
}
def homeVideoContent(self):
result = self.categoryContent("new", "1", False, {})
return {"list": result.get("list", [])}
def categoryContent(self, tid, pg, filter, extend):
self._parse_config(extend)
page = self._page_number(pg)
spec = self._category_spec(tid)
if spec is None:
return self._empty_page(page, "未知分类")
_, _, parser_kind, base_path = spec
paths = [self._paged_path(base_path, page)]
if str(tid) == "hot":
paths.append("/" if page == 1 else "/page-%d" % page)
last_error = None
for path in paths:
try:
source, page_url = self._request_text(path)
if parser_kind == "dm":
return self._parse_dm_page(source, page, page_url)
return self._parse_entry_page(source, page, page_url)
except Exception as exc:
last_error = exc
print("[badnews-probe] category path=%s error=%s" % (path, exc))
return self._empty_page(page, "分类读取失败: %s" % last_error)
def searchContent(self, key, quick, pg="1"):
keyword = self._clean_text(key)
page = self._page_number(pg)
if not keyword:
return self._empty_page(page)
encoded = quote(keyword, safe="")
main_path = "/search/q-%s/type-porn" % encoded
dm_path = "/dm/search/q-%s" % encoded
if page > 1:
main_path += "/page-%d" % page
dm_path += "/page-%d" % page
items = []
pagecount = page
errors = []
for parser_kind, path in (("entry", main_path), ("dm", dm_path)):
try:
source, page_url = self._request_text(path)
parsed = (
self._parse_dm_page(source, page, page_url)
if parser_kind == "dm"
else self._parse_entry_page(source, page, page_url)
)
items.extend(parsed.get("list", []))
pagecount = max(pagecount, self._page_number(parsed.get("pagecount")))
except Exception as exc:
errors.append(str(exc))
deduped = []
seen = set()
for item in items:
vod_id = str(item.get("vod_id") or "")
if not vod_id or vod_id in seen:
continue
seen.add(vod_id)
deduped.append(item)
return {
"list": deduped,
"page": page,
"pagecount": pagecount,
"limit": len(deduped),
"total": pagecount * max(len(deduped), 1),
"msg": "; ".join(errors) if errors and not deduped else "",
}
def detailContent(self, ids):
raw_id = ids[0] if isinstance(ids, (list, tuple)) and ids else ids
value = str(raw_id or "").strip()
if value.startswith("atvp_detail:"):
value = value[len("atvp_detail:") :].strip()
if value.startswith(self.PLAY_PREFIX):
value = value[len(self.PLAY_PREFIX) :].strip()
kind, item_id = self._split_vod_id(value)
if not kind or not item_id:
return {"list": []}
path = "/t/%s" % item_id if kind == "t" else "/dm/play/id-%s" % item_id
try:
source, page_url = self._request_text(path, fresh=(kind == "dm"))
vod, _ = self._parse_detail_page(source, kind, item_id, page_url)
return {"list": [vod]}
except Exception as exc:
return {"list": [self._detail_error(value, str(exc))]}
def playerContent(self, flag, id, vipFlags):
value = str(id or "").strip()
if value.startswith(self.ERROR_PREFIX):
return self._player_error(unquote(value[len(self.ERROR_PREFIX) :]))
if value.startswith(("http://", "https://")):
if not self._is_allowed_media_url(value):
return self._player_error("播放地址域名不在媒体白名单")
return self._player_for_media(value, self._media_type(value))
if not value.startswith(self.PLAY_PREFIX):
return self._player_error("无法识别播放 ID")
kind, item_id = self._split_vod_id(value[len(self.PLAY_PREFIX) :])
if not kind or not item_id:
return self._player_error("播放 ID 不完整")
path = "/t/%s" % item_id if kind == "t" else "/dm/play/id-%s" % item_id
try:
source, page_url = self._request_text(path, fresh=(kind == "dm"))
_, media = self._parse_detail_page(source, kind, item_id, page_url)
return self._player_for_media(
media["url"], media["type"], media.get("source", "primary")
)
except Exception as exc:
return self._player_error("播放地址刷新失败: %s" % exc)
def _reset_session(self):
if self._session is not None:
try:
self._session.close()
except Exception:
pass
session = requests.Session()
session.trust_env = self.trust_env
session.headers.update(
{
"User-Agent": self.user_agent,
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
"Accept-Language": "zh-CN,zh;q=0.9,en;q=0.5",
"Cache-Control": "no-cache",
}
)
if self.proxy:
session.proxies.update({"http": self.proxy, "https": self.proxy})
self._session = session
def _request_text(self, path, fresh=False):
url = self._absolute_url(path)
if not self._is_allowed_html_url(url):
raise RuntimeError("已阻止非站点 HTML 请求")
now = time.time()
if not fresh and self.cache_ttl > 0:
cached = self._cache.get(url)
if cached and now - cached[0] <= self.cache_ttl:
return cached[1], cached[2]
last_error = None
for attempt in range(2):
try:
response = self._session.get(
url,
timeout=(min(self.timeout, 10), self.timeout),
allow_redirects=True,
verify=self.verify_tls,
)
final_url = str(response.url or url)
if not self._is_allowed_html_url(final_url):
raise RuntimeError("已阻止外域跳转: %s" % urlsplit(final_url).hostname)
text = self._response_text(response)
if self._looks_like_challenge(response.status_code, text):
raise RuntimeError("blocked_by_waf: 页面返回挑战或验证码")
if response.status_code == 429 and attempt == 0:
retry_after = self._bounded_int(
response.headers.get("Retry-After"), 1, 1, 3
)
time.sleep(retry_after)
continue
response.raise_for_status()
if not fresh and self.cache_ttl > 0:
self._cache[url] = (time.time(), text, final_url)
return text, final_url
except requests.RequestException as exc:
last_error = exc
if attempt == 0:
time.sleep(0.4)
continue
break
except RuntimeError:
raise
raise RuntimeError("网络请求失败: %s" % last_error)
def _parse_entry_page(self, source, page, page_url):
tree = self._tree(source)
items = []
entries = tree.xpath(
'//div[contains(concat(" ", normalize-space(@class), " "), " entry ")]'
)
for entry in entries:
videos = entry.xpath('.//video[@data-source or @data-id]')
if not videos:
continue
video = videos[0]
item_id = self._digits(video.get("data-id"))
if not item_id:
item_id = self._first_matching_id(entry.xpath('.//a/@href'), self.TOPIC_ID_RE)
if not item_id:
continue
title = self._first_text(
entry.xpath(
'.//h3[contains(concat(" ", normalize-space(@class), " "), " title ")]'
'/a[contains(concat(" ", normalize-space(@class), " "), " title ")][1]'
)
)
if not title:
title = "视频 %s" % item_id
pic = self._absolute_media_url(
video.get("data-poster") or video.get("poster") or "", page_url
)
duration = self._clean_text(
" ".join(entry.xpath('.//*[contains(@class,"ct-time")]//text()'))
)
tag = self._first_text(entry.xpath('.//h4[contains(@class,"label")]'))
media_type = str(video.get("data-type") or "").upper()
remarks = duration or tag or media_type
items.append(
{
"vod_id": "t:%s" % item_id,
"vod_name": title,
"vod_pic": pic or self.DEFAULT_PIC,
"vod_remarks": remarks,
}
)
return self._page_result(items, tree, page, 25)
def _parse_dm_page(self, source, page, page_url):
tree = self._tree(source)
items = []
articles = tree.xpath('//article[.//a[contains(@href,"/dm/play/id-")]]')
for article in articles:
title_links = article.xpath(
'.//a[contains(concat(" ", normalize-space(@class), " "), " title ")][1]'
)
links = title_links or article.xpath('.//a[contains(@href,"/dm/play/id-")][1]')
if not links:
continue
link = links[0]
href = str(link.get("href") or "")
item_id = self._first_matching_id([href], self.DM_ID_RE)
if not item_id:
continue
title = self._clean_text(link.get("title") or link.text_content())
images = article.xpath('.//img[1]')
pic = ""
if images:
image = images[0]
pic = self._absolute_media_url(
image.get("data-echo")
or image.get("data-src")
or image.get("src")
or "",
page_url,
)
items.append(
{
"vod_id": "dm:%s" % item_id,
"vod_name": title or "动漫 %s" % item_id,
"vod_pic": pic or self.DEFAULT_PIC,
"vod_remarks": "MP4",
}
)
return self._page_result(items, tree, page, 30)
def _parse_detail_page(self, source, kind, item_id, page_url):
tree = self._tree(source)
if kind == "t":
videos = tree.xpath('//video[@data-id="%s"]' % item_id)
if not videos:
videos = tree.xpath('//video[@data-source][1]')
else:
videos = tree.xpath('//video[@data-source][1]')
if not videos:
raise RuntimeError("详情页没有 video[data-source]")
video = videos[0]
media_url = self._absolute_media_url(
video.get("data-source") or video.get("src") or "", page_url
)
if not self._is_allowed_media_url(media_url):
raise RuntimeError("详情媒体域名不在白名单")
detected_type = self._media_type(media_url)
declared_type = str(video.get("data-type") or "").lower()
media_type = (
detected_type
if detected_type in ("mp4", "m3u8")
else declared_type
)
media_source = "primary-%s" % media_type
if self.prefer_progressive_mp4 and media_type == "m3u8":
for meta_property in ("og:video:secure_url", "og:video"):
progressive_url = self._absolute_media_url(
self._meta_content(tree, "property", meta_property), page_url
)
if (
self._media_type(progressive_url) == "mp4"
and self._is_allowed_media_url(progressive_url)
):
media_url = progressive_url
media_type = "mp4"
media_source = "target-og-progressive-mp4"
break
title = self._meta_content(tree, "property", "og:title")
if not title:
title = self._meta_content(tree, "name", "headline")
if not title:
headings = tree.xpath('//h1[1] | //h2[1]')
title = self._first_text(headings)
if not title:
title = ("动漫 " if kind == "dm" else "视频 ") + item_id
pic = self._meta_content(tree, "property", "og:image")
if not pic:
pic = video.get("data-poster") or video.get("poster") or ""
pic = self._absolute_media_url(pic, page_url) or self.DEFAULT_PIC
content = self._meta_content(tree, "name", "Description")
if not content:
content = self._meta_content(tree, "property", "og:description")
vod_id = "%s:%s" % (kind, item_id)
play_target = self.PLAY_PREFIX + vod_id
vod = {
"vod_id": vod_id,
"vod_name": title,
"vod_pic": pic,
"vod_remarks": media_type.upper(),
"vod_content": content,
"vod_play_from": "BadNews动漫" if kind == "dm" else "BadNews直连",
"vod_play_url": "播放$%s" % play_target,
}
return vod, {"url": media_url, "type": media_type, "source": media_source}
def _page_result(self, items, tree, page, default_limit):
pagecount = page
for href in tree.xpath('//a/@href'):
match = self.PAGE_RE.search(str(href or ""))
if match:
pagecount = max(pagecount, self._page_number(match.group(1)))
limit = len(items) or default_limit
return {
"list": items,
"page": page,
"pagecount": pagecount,
"limit": limit,
"total": pagecount * limit,
}
def _player_result(self, media_url, media_type):
result = {
"parse": 0,
"jx": 0,
"playUrl": "",
"url": media_url,
"header": {"User-Agent": self.user_agent},
"type": media_type,
}
if media_type == "m3u8":
result["format"] = "application/x-mpegURL"
return result
def _player_for_media(self, media_url, media_type, media_source="primary"):
self._probe_log(
"media_selected type=%s source=%s host=%s"
% (media_type, media_source, urlsplit(media_url).hostname or "")
)
if media_type == "m3u8" and self.lock_hls_highest:
try:
locked_url = self._prepare_locked_hls(media_url)
if locked_url:
self._probe_log("hls_highest_locked source=%s" % media_source)
return self._player_result(locked_url, "m3u8")
except Exception as exc:
self._probe_log("hls_lock_failed url=%s error=%s" % (media_url, exc))
self._probe_log("hls_original_fallback source=%s" % media_source)
return self._player_result(media_url, media_type)
def _prepare_locked_hls(self, master_url):
source = self._request_media_text(master_url)
manifest = self._highest_hls_manifest(source, master_url)
if not manifest:
return ""
token = hashlib.sha256(
(master_url + "\n" + manifest).encode("utf-8")
).hexdigest()[:24]
self._proxy_manifests[token] = (time.time(), manifest.encode("utf-8"))
if len(self._proxy_manifests) > 16:
oldest = min(self._proxy_manifests, key=lambda key: self._proxy_manifests[key][0])
self._proxy_manifests.pop(oldest, None)
site_key = quote(str(getattr(self, "siteKey", "") or "badnews"), safe="")
return "%s?siteKey=%s&token=%s" % (
self._proxy_base_url(),
site_key,
token,
)
def _request_media_text(self, url):
if not self._is_allowed_media_url(url) or self._media_type(url) != "m3u8":
raise RuntimeError("媒体列表地址不在白名单")
last_error = None
for attempt in range(3):
try:
response = self._session.get(
url,
headers={
"User-Agent": self.user_agent,
"Accept": "application/vnd.apple.mpegurl,application/x-mpegURL,*/*",
},
timeout=(min(self.timeout, 10), self.timeout),
verify=self.verify_tls,
allow_redirects=True,
)
if not 200 <= response.status_code < 300:
raise RuntimeError("HLS HTTP %s" % response.status_code)
source = response.content.decode("utf-8", errors="replace")
if "#EXTM3U" not in source:
raise RuntimeError("HLS 响应缺少 EXTM3U")
return source
except Exception as exc:
last_error = exc
if attempt < 2:
time.sleep(0.15 * (attempt + 1))
raise RuntimeError("HLS 主列表读取失败: %s" % last_error)
def _highest_hls_manifest(self, source, master_url):
lines = [line.strip() for line in str(source or "").splitlines() if line.strip()]
streams = []
media_lines = {}
for index, line in enumerate(lines):
if line.startswith("#EXT-X-MEDIA:"):
attrs = self._hls_attrs(line.split(":", 1)[1])
if attrs.get("TYPE") == "AUDIO" and attrs.get("GROUP-ID"):
media_lines[attrs["GROUP-ID"]] = (line, attrs)
elif line.startswith("#EXT-X-STREAM-INF:"):
attrs = self._hls_attrs(line.split(":", 1)[1])
uri = ""
for candidate in lines[index + 1 :]:
if candidate.startswith("#"):
continue
uri = candidate
break
if uri:
resolution = attrs.get("RESOLUTION", "0x0").lower().split("x")
try:
pixels = int(resolution[0]) * int(resolution[1])
except Exception:
pixels = 0
try:
bandwidth = int(attrs.get("AVERAGE-BANDWIDTH") or attrs.get("BANDWIDTH") or 0)
except Exception:
bandwidth = 0
streams.append((pixels, bandwidth, line, attrs, uri))
if not streams:
return ""
_, _, stream_line, stream_attrs, stream_uri = max(
streams, key=lambda item: (item[0], item[1])
)
audio_group = stream_attrs.get("AUDIO", "")
output = ["#EXTM3U", "#EXT-X-VERSION:6", "#EXT-X-INDEPENDENT-SEGMENTS"]
if audio_group in media_lines:
audio_line, audio_attrs = media_lines[audio_group]
audio_uri = audio_attrs.get("URI", "")
if audio_uri:
absolute_audio = urljoin(master_url, audio_uri)
audio_line = re.sub(
r'URI=(?:"[^"]*"|[^,]*)', 'URI="%s"' % absolute_audio, audio_line
)
output.append(audio_line)
output.append(stream_line)
output.append(urljoin(master_url, stream_uri))
return "\n".join(output) + "\n"
@staticmethod
def _hls_attrs(text):
attrs = {}
for match in re.finditer(r'([A-Z0-9-]+)=("[^"]*"|[^,]*)', str(text or "")):
value = match.group(2).strip()
if len(value) >= 2 and value[0] == value[-1] == '"':
value = value[1:-1]
attrs[match.group(1)] = value
return attrs
@staticmethod
def _proxy_base_url():
if CatVodProxy is not None:
return str(CatVodProxy.getUrl(True))
return "http://127.0.0.1:9978/proxy"
def _player_error(self, message):
text = self._clean_text(message) or "播放失败"
return {
"parse": 0,
"jx": 0,
"playUrl": "",
"url": "",
"header": {},
"msg": text,
"content": text,
"error": text,
}
@staticmethod
def _probe_log(message):
print("[badnews-probe] %s" % message)
def _detail_error(self, vod_id, message):
text = self._clean_text(message) or "详情读取失败"
return {
"vod_id": vod_id or "error",
"vod_name": "详情读取失败",
"vod_pic": self.DEFAULT_PIC,
"vod_content": text,
"vod_play_from": "错误",
"vod_play_url": "查看错误$%s%s" % (self.ERROR_PREFIX, quote(text, safe="")),
}
def _category_spec(self, tid):
value = str(tid or "hot").strip()
for spec in self.CATEGORY_SPECS:
if spec[0] == value:
return spec
return None
@staticmethod
def _paged_path(base_path, page):
if page <= 1:
return base_path
return base_path.rstrip("/") + "/page-%d" % page
def _absolute_url(self, path):
return urljoin(self.host.rstrip("/") + "/", str(path or "").lstrip("/"))
@staticmethod
def _absolute_media_url(value, page_url):
text = str(value or "").strip()
if not text:
return ""
return urljoin(page_url, text)
def _is_allowed_html_url(self, url):
parsed = urlsplit(str(url or ""))
host = (parsed.hostname or "").lower()
configured_host = (urlsplit(self.host).hostname or "").lower()
return (
parsed.scheme in ("http", "https")
and bool(host)
and host == configured_host
and host not in self.BLOCKED_HOSTS
)
def _is_allowed_media_url(self, url):
parsed = urlsplit(str(url or ""))
host = (parsed.hostname or "").lower()
return (
parsed.scheme in ("http", "https")
and host in self.MEDIA_HOSTS
and host not in self.BLOCKED_HOSTS
and self.isVideoFormat(url)
)
def _looks_like_challenge(self, status_code, source):
sample = str(source or "")[:200000].lower()
status = int(status_code or 0)
if 200 <= status < 300 and any(
marker in sample for marker in self.CONTENT_MARKERS
):
return False
if any(marker in sample for marker in self.CHALLENGE_MARKERS):
return True
return status in (403, 503) and "cloudflare" in sample
@staticmethod
def _response_text(response):
content = bytes(response.content or b"")
declared = str(response.encoding or "").strip()
normalized = declared.lower().replace("_", "-")
if not normalized or normalized in {
"iso-8859-1",
"utf-32",
"utf-32le",
"utf-32be",
"usc4 little endian",
"usc4 big endian",
}:
chosen = "utf-8"
else:
chosen = declared
try:
text = content.decode(chosen, errors="replace")
except (LookupError, UnicodeError):
chosen = "utf-8"
text = content.decode(chosen, errors="replace")
if chosen.lower() != normalized:
print(
"[badnews-probe] encoding declared=%s chosen=%s bytes=%d content_type=%s"
% (
declared or "none",
chosen,
len(content),
response.headers.get("Content-Type", ""),
)
)
return text
@staticmethod
def _tree(source):
if isinstance(source, bytes):
payload = source
else:
payload = str(source or "<html></html>").encode("utf-8", errors="replace")
parser = html.HTMLParser(encoding="utf-8", recover=True)
return html.fromstring(payload, parser=parser)
def _meta_content(self, tree, attr_name, attr_value):
nodes = tree.xpath(
'//meta[translate(@%s,"ABCDEFGHIJKLMNOPQRSTUVWXYZ","abcdefghijklmnopqrstuvwxyz")="%s"]/@content'
% (attr_name, attr_value.lower())
)
return self._clean_text(nodes[0]) if nodes else ""
def _first_text(self, nodes):
for node in nodes or []:
try:
value = node.text_content()
except Exception:
value = str(node or "")
value = self._clean_text(value)
if value:
return value
return ""
def _split_vod_id(self, value):
text = str(value or "").strip()
if ":" not in text:
return ("t", self._digits(text)) if self._digits(text) else ("", "")
kind, item_id = text.split(":", 1)
kind = kind.strip().lower()
item_id = self._digits(item_id)
if kind not in ("t", "dm") or not item_id:
return "", ""
return kind, item_id
@staticmethod
def _first_matching_id(values, pattern):
for value in values or []:
match = pattern.search(str(value or ""))
if match:
return match.group(1)
return ""
@staticmethod
def _digits(value):
match = re.search(r"\d+", str(value or ""))
return match.group(0) if match else ""
def _media_type(self, url):
text = str(url or "").lower()
if ".m3u8" in text:
return "m3u8"
if ".mp4" in text:
return "mp4"
return ""
@staticmethod
def _parse_config(extend):
if isinstance(extend, dict):
return dict(extend)
text = str(extend or "").strip()
if not text:
return {}
if text.startswith(("http://", "https://")):
return {"host": text}
try:
value = json.loads(text)
return value if isinstance(value, dict) else {}
except Exception:
return {}
@staticmethod
def _bool_value(value, default=False):
if isinstance(value, bool):
return value
if value is None:
return bool(default)
return str(value).strip().lower() in ("1", "true", "yes", "on")
@staticmethod
def _bounded_int(value, default, minimum=1, maximum=999999):
try:
number = int(value)
except Exception:
number = int(default)
return max(minimum, min(maximum, number))
def _page_number(self, value):
return self._bounded_int(value, 1, 1, 999999)
@staticmethod
def _clean_text(value):
text = html_lib.unescape(str(value or ""))
return re.sub(r"\s+", " ", text).strip()
@staticmethod
def _empty_page(page, message=""):
return {
"list": [],
"page": page,
"pagecount": page,
"limit": 0,
"total": 0,
"msg": message,
}
FGBLH/py/樱花传媒.py
+270
View File
@@ -0,0 +1,270 @@
# -*- coding: utf-8 -*-
import json
import re
import urllib.parse
import requests
try:
from base.spider import Spider as BaseSpider
except ImportError:
class BaseSpider:
pass
class Spider(BaseSpider):
BASE_URL = "https://x3av.com"
HEADERS = {
"User-Agent": "Mozilla/5.0 (Linux; Android 10; Mobile) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Mobile Safari/537.36",
"Referer": "https://x3av.com/",
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8"
}
def __init__(self):
self.siteUrl = self.BASE_URL
self.extend = {}
def init(self, extend=""):
if extend:
try:
self.extend = json.loads(extend) if isinstance(extend, str) else extend
self.siteUrl = self.extend.get("siteUrl", self.BASE_URL).rstrip("/")
except Exception:
self.siteUrl = self.BASE_URL
def getName(self):
return "樱花传媒"
def isVideoFormat(self, url):
return bool(re.search(r"\.(m3u8|mp4|flv|avi|mkv)(\?|$)", url or "", re.I))
def manualVideoCheck(self):
return True
def homeContent(self, filter):
classes = [
{"type_id": "1", "type_name": "有码"},
{"type_id": "2", "type_name": "无码"},
{"type_id": "3", "type_name": "素人"},
{"type_id": "4", "type_name": "中文字幕"}
]
filters = {}
for i in [x["type_id"] for x in classes]:
filters[i] = [
{"key": "by", "name": "排序", "value": [
{"n": "最新", "v": "time"},
{"n": "热门", "v": "hits"},
{"n": "评分", "v": "score"}
]}
]
return {"class": classes, "filters": filters}
def homeVideoContent(self):
html = self._get(self.siteUrl)
return {"list": self._parse_list(html)}
def categoryContent(self, tid, pg, filter, extend):
pg = str(pg or "1")
by = (extend or {}).get("by", "")
if by and by != "time":
url = self.siteUrl + "/vshow/by/{}/id/{}/page/{}.html".format(by, tid, pg)
else:
url = self.siteUrl + "/category/{}.html".format(tid) if pg == "1" else self.siteUrl + "/category/{}/page/{}.html".format(tid, pg)
html = self._get(url)
videos = self._parse_list(html)
return {"page": int(pg), "pagecount": int(pg) + 1 if videos else int(pg), "limit": 24, "total": 999999, "list": videos}
def detailContent(self, ids):
vid = ids[0] if isinstance(ids, list) else ids
url = self._full_url(vid)
html = self._get(url)
name = self._clean(self._match(html, r"<h1[^>]*>(.*?)</h1>") or self._match(html, r"<title[^>]*>(.*?)</title>") or "")
pic = self._match(html, r'background\s*:\s*url\((.*?)\)') or self._match(html, r'<img[^>]+data-original=["\']([^"\']+)') or self._match(html, r'<img[^>]+data-src=["\']([^"\']+)') or self._match(html, r'<img[^>]+src=["\']([^"\']+)')
desc = self._clean(self._match(html, r'<div[^>]+class=["\'][^"\']*(?:desc|content|video-info)[^"\']*["\'][^>]*>(.*?)</div>') or "")
actor = self._clean(self._match(html, r"主演[:：&nbsp;\s]*([^<]+)") or "")
remarks = self._clean(self._match(html, r"番号[:：&nbsp;\s]*([^<]+)") or "")
play_items = []
for m in re.finditer(r'<a[^>]+id=["\']playerserver["\'][^>]*>', html, re.I):
tag = m.group(0)
vodid = self._attr(tag, "data-vodid")
sid = self._attr(tag, "data-sid") or "1"
nid = self._attr(tag, "data-nid") or "1"
title = self._clean(self._match(html[m.end():m.end()+200], r"([^<]+)</a>") or "播放{}".format(len(play_items) + 1))
if vodid:
play_items.append("{}${}|{}|{}".format(title, vodid, sid, nid))
if not play_items:
mid = self._match(url, r"/videos/(\d+)")
if mid:
play_items.append("播放${}|1|1".format(mid))
vod = {
"vod_id": vid,
"vod_name": name,
"vod_pic": self._real_pic(pic),
"type_name": "",
"vod_year": "",
"vod_area": "",
"vod_remarks": remarks,
"vod_actor": actor,
"vod_director": "",
"vod_content": desc,
"vod_play_from": "樱花传媒",
"vod_play_url": "#".join(play_items)
}
return {"list": [vod]}
def searchContent(self, key, quick, pg="1"):
wd = urllib.parse.quote(key)
url = self.siteUrl + "/search.html?wd={}".format(wd) if str(pg) == "1" else self.siteUrl + "/search.html?wd={}&page={}".format(wd, pg)
html = self._get(url)
return {"list": self._parse_list(html)}
def playerContent(self, flag, id, vipFlags):
pp = str(id).split("|")
if len(pp) < 3:
return {"parse": 1, "playUrl": "", "url": id, "header": self.HEADERS}
data = {"ids": pp[0], "flag": "player", "sid": pp[1], "nid": pp[2]}
headers = dict(self.HEADERS)
headers["X-Requested-With"] = "XMLHttpRequest"
headers["Content-Type"] = "application/x-www-form-urlencoded; charset=UTF-8"
try:
res = requests.post(self.siteUrl + "/api.php/post/urlget/", data=data, headers=headers, timeout=15, verify=False)
text = res.text
except requests.RequestException:
text = ""
iframe = ""
try:
obj = json.loads(text)
iframe = self._match(obj.get("player", ""), r'<iframe[^>]+src=["\']([^"\']+)')
except Exception:
iframe = self._match(text, r'<iframe[^>]+src=["\']([^"\']+)')
iframe = self._full_url(iframe)
real = self._extract_player_url(iframe)
if real:
return {"parse": 0, "playUrl": "", "url": real, "header": {"User-Agent": self.HEADERS["User-Agent"], "Referer": iframe}}
return {"parse": 1, "playUrl": "", "url": iframe, "header": {"User-Agent": self.HEADERS["User-Agent"], "Referer": self.siteUrl + "/"}}
def localProxy(self, param):
return [404, "text/plain", ""]
def _parse_list(self, html):
html = re.sub(r"<!--[\s\S]*?-->", "", html or "")
arr = []
blocks = re.findall(r'<div[^>]+class=["\'][^"\']*video-elem[^"\']*["\'][\s\S]*?(?=<div[^>]+class=["\'][^"\']*video-elem|<ul[^>]+class=["\'][^"\']*pagination|</body>|$)', html, re.I)
if not blocks:
blocks = re.findall(r'<a[^>]+href=["\'][^"\']*/videos/\d+\.html[^"\']*["\'][\s\S]*?</a>', html, re.I)
for block in blocks:
href = self._match(block, r'href=["\']([^"\']*/videos/\d+\.html[^"\']*)')
name = self._clean(self._match(block, r'<a[^>]+class=["\'][^"\']*title[^"\']*["\'][^>]*>(.*?)</a>') or self._match(block, r'title=["\']([^"\']+)') or self._match(block, r'alt=["\']([^"\']+)'))
img_tag = self._match(block, r'(<img[\s\S]*?>)')
pic = self._attr(img_tag, "data-original") or self._attr(img_tag, "data-src") or self._attr(img_tag, "src")
remark = self._clean(self._match(block, r'<span[^>]+class=["\'][^"\']*(?:duration|remarks|time)[^"\']*["\'][^>]*>(.*?)</span>') or "")
pic = self._real_pic(pic)
if href and name:
arr.append({"vod_id": self._full_url(href), "vod_name": name, "vod_pic": pic, "vod_remarks": remark})
return arr
def _extract_player_url(self, iframe):
if not iframe:
return ""
html = self._get(iframe, {"Referer": self.siteUrl + "/"})
code = self._unpack(html) or html
p = {}
m = re.search(r"var\s+p\s*=\s*(\{.*?\})\s*;", code, re.S)
if m:
for k, v in re.findall(r'["\']?(hls\d+|mp4|file)["\']?\s*:\s*["\']([^"\']+)["\']', m.group(1), re.I):
p[k.lower()] = v.replace("\\/", "/")
url = p.get("hls2") or p.get("hls3") or p.get("hls4") or p.get("file") or p.get("mp4")
if not url:
sm = re.search(r'sources\s*:\s*\[\s*\{\s*file\s*:\s*(p\.(hls\d+|file|mp4)|["\']([^"\']+)["\'])', code, re.I)
if sm:
url = p.get((sm.group(2) or "").lower()) or sm.group(3) or ""
if not url:
urls = re.findall(r'https?://[^"\']+?\.(?:m3u8|mp4)(?:\?[^"\']*)?|/[A-Za-z0-9_./-]+/(?:master|index)\.(?:m3u8|mp4)(?:\?[^"\']*)?', code, re.I)
for u in urls:
if "jpg" not in u.lower() and "png" not in u.lower():
url = u
break
return self._join(iframe, url)
def _unpack(self, html):
m = re.search(r"eval\(function\(p,a,c,k,e,d\).*?\}\('(.+?)',(\d+),(\d+),'(.+?)'\.split\('\|'\)\)\)", html or "", re.S)
if not m:
return ""
p, a, c, k = m.group(1), int(m.group(2)), int(m.group(3)), m.group(4).split("|")
for i in range(c - 1, -1, -1):
if i < len(k) and k[i]:
p = re.sub(r"\b" + re.escape(self._base_n(i, a)) + r"\b", k[i], p)
return p
def _get(self, url, headers=None):
if not url:
return ""
h = dict(self.HEADERS)
if headers:
h.update(headers)
try:
r = requests.get(url, headers=h, timeout=15, verify=False)
if r.encoding == "ISO-8859-1":
r.encoding = r.apparent_encoding
return r.text
except requests.RequestException:
return ""
def _full_url(self, url):
url = (url or "").replace("&amp;", "&").strip()
if not url:
return ""
if url.startswith("//"):
return "https:" + url
if url.startswith("http"):
return url
if url.startswith("/"):
return self.siteUrl + url
return self.siteUrl + "/" + url
def _join(self, base, url):
url = (url or "").replace("\\/", "/").replace("&amp;", "&").strip()
if not url:
return ""
if url.startswith("//"):
return "https:" + url
if url.startswith("http"):
return url
return urllib.parse.urljoin(base, url)
def _real_pic(self, url):
url = self._full_url(url or "")
url = url.replace("&amp;", "&").strip()
if not url or "noimage" in url.lower() or url.endswith("/"):
return ""
if "getimages.php" in url and "src=" in url:
m = re.search(r"src=([^&]+)", url)
if m:
src = urllib.parse.unquote(m.group(1)).replace("&amp;", "&")
if src.startswith("http") and "noimage" not in src.lower():
return src
m = re.search(r"(https?://[^&'\"]+\.(?:jpg|jpeg|png|webp))", url, re.I)
if m:
return urllib.parse.unquote(m.group(1))
return url
def _match(self, text, pattern):
m = re.search(pattern, text or "", re.S | re.I)
return m.group(1).strip() if m else ""
def _attr(self, tag, key):
return self._match(tag, key + r'=["\']([^"\']+)')
def _clean(self, text):
text = re.sub(r"<[^>]+>", " ", text or "")
text = urllib.parse.unquote(text)
text = text.replace("&nbsp;", " ").replace("&amp;", "&").replace("&quot;", '"').replace("&#39;", "'")
return re.sub(r"\s+", " ", text).strip()
def _base_n(self, num, base):
chars = "0123456789abcdefghijklmnopqrstuvwxyz"
if num == 0:
return "0"
s = ""
while num:
s = chars[num % base] + s
num //= base
return s
FGBLH/py/蝶卡影视.py
+572
View File
@@ -0,0 +1,572 @@
# coding: utf-8
# 站点: 蝶卡影视网 (https://www.diekawang.com)
import json
import base64
import re
from urllib.parse import quote, urljoin, unquote
from base.spider import Spider as BaseSpider
class Spider(BaseSpider):
def __init__(self):
# __init__ 只做本地初始化，禁止网络请求，保证壳子首页秒出 class。
self.extend = ""
self.host = "https://www.diekawang.com"
self.classes = [
{"type_id": "1", "type_name": "电影"},
{"type_id": "2", "type_name": "电视剧"},
{"type_id": "3", "type_name": "综艺"},
{"type_id": "4", "type_name": "动漫"},
{"type_id": "457", "type_name": "短剧"},
{"type_id": "462", "type_name": "体育"},
]
self.filters = {
"1": [
{"key": "cate", "name": "类型", "value": [
{"n": "全部", "v": "0"},
{"n": "动作片", "v": "9"},
{"n": "喜剧片", "v": "10"},
{"n": "爱情片", "v": "11"},
{"n": "恐怖片", "v": "12"},
{"n": "剧情片", "v": "13"},
{"n": "科幻片", "v": "14"},
{"n": "惊悚片", "v": "15"},
{"n": "奇幻片", "v": "16"},
{"n": "动画片", "v": "17"},
{"n": "悬疑片", "v": "18"},
{"n": "冒险片", "v": "19"},
{"n": "纪录片", "v": "20"},
{"n": "战争片", "v": "21"},
{"n": "倫理片", "v": "460"},
]},
],
"2": [
{"key": "cate", "name": "类型", "value": [
{"n": "全部", "v": "0"},
{"n": "国产剧", "v": "22"},
{"n": "香港剧", "v": "23"},
{"n": "台湾剧", "v": "24"},
{"n": "欧美剧", "v": "25"},
{"n": "日本剧", "v": "26"},
{"n": "韩国剧", "v": "27"},
{"n": "泰国剧", "v": "28"},
{"n": "海外剧", "v": "29"},
]},
],
"3": [
{"key": "cate", "name": "类型", "value": [
{"n": "全部", "v": "0"},
{"n": "大陆综艺", "v": "30"},
{"n": "港台综艺", "v": "31"},
{"n": "日韩综艺", "v": "32"},
{"n": "欧美综艺", "v": "33"},
{"n": "海外综艺", "v": "34"},
]},
],
"4": [
{"key": "cate", "name": "类型", "value": [
{"n": "全部", "v": "0"},
{"n": "国产动漫", "v": "35"},
{"n": "日韩动漫", "v": "36"},
{"n": "欧美动漫", "v": "37"},
{"n": "海外动漫", "v": "38"},
{"n": "港台动漫", "v": "459"},
]},
],
"457": [
{"key": "cate", "name": "类型", "value": [
{"n": "全部", "v": "0"},
{"n": "漫剧", "v": "540"},
{"n": "玄幻", "v": "541"},
{"n": "剧情", "v": "542"},
{"n": "女性成长", "v": "543"},
{"n": "权谋", "v": "544"},
{"n": "豪门", "v": "545"},
{"n": "齐幻", "v": "546"},
{"n": "宫斗", "v": "547"},
{"n": "脑洞", "v": "548"},
{"n": "科幻", "v": "549"},
{"n": "冒险", "v": "550"},
{"n": "仙侠", "v": "551"},
{"n": "喜剧", "v": "552"},
{"n": "动作", "v": "553"},
{"n": "悬疑", "v": "554"},
{"n": "战神", "v": "555"},
{"n": "刑侦", "v": "556"},
{"n": "求生", "v": "557"},
{"n": "商战", "v": "558"},
{"n": "恐怖", "v": "559"},
{"n": "武侠", "v": "560"},
{"n": "爱情", "v": "561"},
{"n": "AI漫剧", "v": "562"},
]},
],
"462": [
{"key": "cate", "name": "类型", "value": [
{"n": "全部", "v": "0"},
]},
],
}
self.headers = {
"User-Agent": "Mozilla/5.0 (Linux; Android 14; 22127RK46C) AppleWebKit/537.36",
"Referer": self.host + "/",
}
def getName(self):
return "蝶卡影视"
def getDependence(self):
return []
def init(self, extend=""):
self.extend = extend or ""
# ==================== 内部工具模块 ====================
def _cleanText(self, text):
"""清洗 HTML 标签和空白字符"""
if not text:
return ""
text = re.sub(r'<[^>]+>', '', text)
text = re.sub(r'&nbsp;', ' ', text)
text = re.sub(r'\s+', ' ', text)
return text.strip()
def _absUrl(self, url):
"""补全相对 URL"""
if not url:
return ""
if url.startswith("//"):
return "https:" + url
if url.startswith("http"):
return url
return urljoin(self.host, url)
def _decodeFile(self, file_str):
"""解码播放文件 URL: 去掉3字符前缀 -> base64解码 -> URL反编码"""
if not file_str or len(file_str) <= 3:
return ""
try:
raw = file_str[3:]
decoded = base64.b64decode(raw).decode('utf-8', 'ignore')
return unquote(decoded)
except Exception:
return ""
def _makePlayHeader(self, url):
"""空防盗链头优先: 只保留 UA，避免 EXO 把错误 Referer/Origin 透传给第三方 CDN 分片"""
return {"User-Agent": self.headers.get("User-Agent", "Mozilla/5.0")}
def _parseListCards(self, html):
"""
解析列表页卡片 (多级兜底)
主选择器: div.moon-list-item > a.item.goLinklist
语义锚点: href="/vod/player/0/{id}"
兜底: 全页扫描 /vod/player/0/ 链接块
"""
result = []
seen_ids = set()
# 主选择器: 匹配完整的卡片块
# Pattern 1: 标准 moon-list-item 结构
cards = re.findall(
r'<div[^>]*class="[^"]*moon-list-item[^"]*"[^>]*>\s*'
r'<a[^>]*href="/vod/player/0/(\d+)"[^>]*>(.*?)</a>',
html, re.S
)
for vid, block in cards:
if vid in seen_ids:
continue
# 提取标题: p.name 或 .item-title
name = re.search(r'class="[^"]*name[^"]*item-title[^"]*"[^>]*>(.*?)</p>', block, re.S)
if not name:
name = re.search(r'class="[^"]*item-title[^"]*"[^>]*>(.*?)</p>', block, re.S)
name = self._cleanText(name.group(1)) if name else ""
if not name:
continue
# 提取图片: data-original > data-src > src
pic = re.search(r'data-original="([^"]+)"', block)
if not pic:
pic = re.search(r'data-src="([^"]+)"', block)
if not pic:
pic = re.search(r'src="([^"]+)"', block)
pic = self._absUrl(pic.group(1)) if pic else ""
# 提取评分/备注: label.rate
remark = re.search(r'class="[^"]*rate[^"]*"[^>]*>(.*?)</label>', block, re.S)
remark = self._cleanText(remark.group(1)) if remark else ""
# 打包轻量字段到 vod_id
vod_id = vid + '|$|' + name + '|$|' + pic + '|$|' + remark
result.append({
"vod_id": vod_id,
"vod_name": name,
"vod_pic": pic,
"vod_remarks": remark,
})
seen_ids.add(vid)
# 兜底: 如果主选择器没匹配到，全页扫描 player 链接
if not result:
blocks = re.findall(
r'<a[^>]*href="/vod/player/0/(\d+)"[^>]*class="[^"]*(?:item|goLinklist)[^"]*"[^>]*>(.*?)</a>',
html, re.S
)
for vid, block in blocks:
if vid in seen_ids:
continue
name = re.search(r'<p[^>]*>(.*?)</p>', block, re.S)
name = self._cleanText(name.group(1)) if name else ""
if not name:
continue
pic = re.search(r'data-original="([^"]+)"', block)
if not pic:
pic = re.search(r'src="([^"]+)"', block)
pic = self._absUrl(pic.group(1)) if pic else ""
remark = re.search(r'class="[^"]*rate[^"]*"[^>]*>(.*?)</label>', block, re.S)
remark = self._cleanText(remark.group(1)) if remark else ""
vod_id = vid + '|$|' + name + '|$|' + pic + '|$|' + remark
result.append({
"vod_id": vod_id,
"vod_name": name,
"vod_pic": pic,
"vod_remarks": remark,
})
seen_ids.add(vid)
return result
def _parseSearchCards(self, html):
"""
解析搜索结果卡片
主选择器: a.moon-item.seachlist
语义锚点: href="/vod/player/0/{id}"
"""
result = []
seen_ids = set()
cards = re.findall(
r'<a[^>]*href="/vod/player/0/(\d+)"[^>]*class="[^"]*seachlist[^"]*"[^>]*>(.*?)</a>',
html, re.S
)
for vid, block in cards:
if vid in seen_ids:
continue
# 标题: h2
name = re.search(r'<h2[^>]*>(.*?)</h2>', block, re.S)
name = self._cleanText(name.group(1)) if name else ""
if not name:
continue
# 图片
pic = re.search(r'data-src="([^"]+)"', block)
if not pic:
pic = re.search(r'src="([^"]+)"', block)
pic = self._absUrl(pic.group(1)) if pic else ""
# 信息: label-list 中的 span (年份/类型/地区)
info_spans = re.findall(r'<span[^>]*>(.*?)</span>', block, re.S)
info_parts = [self._cleanText(s) for s in info_spans if self._cleanText(s)]
remark = " ".join(info_parts[:3]) if info_parts else ""
vod_id = vid + '|$|' + name + '|$|' + pic + '|$|' + remark
result.append({
"vod_id": vod_id,
"vod_name": name,
"vod_pic": pic,
"vod_remarks": remark,
})
seen_ids.add(vid)
return result
def _getPageCount(self, html, default=99):
"""从"尾页"链接中提取总页数"""
# 尾页链接格式: href="/vod/list/{last_page}/{type}/{sub_type}"
last_page = re.search(r'href="/vod/list/(\d+)/\d+/\d+"[^>]*>[^<]*尾页', html)
if last_page:
try:
return int(last_page.group(1))
except ValueError:
pass
return default
def _isNoResultPage(self, html):
"""检测无结果页 (排除 Vue 模板中的暂无数据占位符)"""
no_result_patterns = [
r'没有找到您想要的结果',
r'没有找到.*结果',
r'搜索无结果',
r'暂无影片',
r'没有搜到',
r'无搜索结果',
]
for pattern in no_result_patterns:
if re.search(pattern, html):
return True
return False
# ==================== 核心接口方法 ====================
def homeContent(self, filter):
"""首页入口零网络，只返回本地 class/filters"""
return {"class": self.classes, "filters": self.filters if filter else {}}
def getHomeContent(self, filter):
return self.homeContent(filter)
def homeVideoContent(self):
"""首页推荐数据"""
try:
url = f"{self.host}/vod/list/1/1/0"
res = self.fetch(url, headers=self.headers)
html = res.text
return {"list": self._parseListCards(html)}
except Exception:
return {"list": []}
def categoryContent(self, tid, pg, filter, extend):
"""
分类列表: /vod/list/{page}/{parent_type}/{sub_type}
动态消费 pg 和 extend.cate
"""
page = str(pg) if pg else "1"
extend = extend or {}
sub_type = extend.get("cate", "0") or "0"
url = f"{self.host}/vod/list/{page}/{tid}/{sub_type}"
try:
res = self.fetch(url, headers=self.headers)
html = res.text
items = self._parseListCards(html)
# 无结果检测: 仅在卡片为空时检查无结果提示
if not items and self._isNoResultPage(html):
return {
"list": [],
"page": int(page),
"pagecount": 1,
"limit": 10,
"total": 0,
}
pagecount = self._getPageCount(html)
return {
"list": items,
"page": int(page),
"pagecount": pagecount,
"limit": 10,
"total": pagecount * 10,
}
except Exception:
return {
"list": [],
"page": int(page),
"pagecount": 1,
"limit": 10,
"total": 0,
}
def detailContent(self, ids):
"""
详情页: /vod/player/0/{vod_id}
提取 temLineList JSON 构建播放树
提取 vod 元信息 (name, pic, year, area, actor, director, score)
"""
raw = str(ids[0])
ps = raw.split('|$|')
vod_id = ps[0]
old_name = ps[1] if len(ps) > 1 else ''
old_pic = ps[2] if len(ps) > 2 else ''
old_remark = ps[3] if len(ps) > 3 else ''
url = f"{self.host}/vod/player/0/{vod_id}"
try:
res = self.fetch(url, headers=self.headers)
html = res.text
except Exception:
# 网络失败时返回列表阶段缓存的字段
return {"list": [{
"vod_id": raw,
"vod_name": old_name or "视频",
"vod_pic": old_pic,
"vod_remarks": old_remark,
"vod_play_from": "播放",
"vod_play_url": "播放$" + vod_id,
}]}
# 提取 temLineList JSON
vod_name = old_name
vod_pic = old_pic
vod_year = ""
vod_area = ""
vod_actor = ""
vod_director = ""
vod_score = ""
vod_content = ""
vod_remarks = old_remark
# 提取名称: H1 标签 > item变量 > 列表缓存
h1_match = re.search(r'<h1[^>]*>(.*?)</h1>', html, re.S)
if h1_match:
vod_name = self._cleanText(h1_match.group(1)) or vod_name
# 提取图片: item变量 imgUrl > data-original
img_match = re.search(r'imgUrl:\s*["\']([^"\']+)', html)
if img_match:
vod_pic = self._absUrl(img_match.group(1).replace('\\/', '/'))
elif not vod_pic:
img_match2 = re.search(r'data-original="([^"]+)"', html)
if img_match2:
vod_pic = self._absUrl(img_match2.group(1))
# 提取评分: label-list 中的 din-condensed
score_match = re.search(r'class="[^"]*din-condensed[^"]*"[^>]*>(.*?)</span>', html, re.S)
if score_match:
vod_score = self._cleanText(score_match.group(1))
# 提取年份和地区: label-list 中的 span
label_section = re.search(r'class="label-list"[^>]*>(.*?)</div>', html, re.S)
if label_section:
spans = re.findall(r'<span[^>]*>(.*?)</span>', label_section.group(1), re.S)
span_texts = [self._cleanText(s) for s in spans if self._cleanText(s)]
for txt in span_texts:
if re.match(r'^\d{4}$', txt):
vod_year = txt
elif txt != vod_score:
if not vod_area:
vod_area = txt
# 提取导演: worker-name 标签后的内容
director_section = re.search(
r'class="worker-name"[^>]*>\s*导演\s*</a>\s*<div[^>]*>(.*?)</div>', html, re.S
)
if director_section:
vod_director = self._cleanText(director_section.group(1))
# 提取演员: worker-name 标签后的内容
actor_section = re.search(
r'class="worker-name"[^>]*>\s*演员\s*</a>\s*<div[^>]*>(.*?)</div>', html, re.S
)
if actor_section:
vod_actor = self._cleanText(actor_section.group(1))
# 从 meta keywords 提取演员 (兜底)
if not vod_actor:
meta_kw = re.search(r'<meta\s+name="keywords"\s+content="([^"]+)"', html)
if meta_kw:
kw_parts = meta_kw.group(1).split(',')
# keywords 格式: 站名,剧名,类型,子类型,,演员列表
if len(kw_parts) >= 6:
vod_actor = kw_parts[5]
# 构建播放树
play_from_list = []
play_url_list = []
tem_line_match = re.search(r'temLineList\s*=\s*(\[.*?\])\s*;', html, re.S)
if tem_line_match:
try:
line_data = json.loads(tem_line_match.group(1))
# 按 tag 分组 (通常只有一组)
lines = {}
line_order = []
for ep in line_data:
tag = ep.get("tag", "播放") or "播放"
if tag not in lines:
lines[tag] = []
line_order.append(tag)
ep_name = ep.get("name", "") or ep.get("subTitle", "") or "播放"
ep_file = ep.get("file", "")
lines[tag].append(f"{ep_name}${ep_file}")
for tag in line_order:
play_from_list.append(tag)
play_url_list.append("#".join(lines[tag]))
except (json.JSONDecodeError, Exception):
pass
# 兜底: 如果没有提取到播放树，返回嗅探
if not play_from_list:
play_from_list.append("播放")
play_url_list.append(f"播放${vod_id}")
vod = {
"vod_id": raw,
"vod_name": vod_name or "视频",
"vod_pic": vod_pic,
"vod_year": vod_year,
"vod_area": vod_area,
"vod_actor": vod_actor,
"vod_director": vod_director,
"vod_score": vod_score,
"vod_remarks": vod_remarks or vod_score,
"vod_content": vod_content or vod_remarks,
"vod_play_from": "$$$".join(play_from_list),
"vod_play_url": "$$$".join(play_url_list),
}
return {"list": [vod]}
def searchContent(self, key, quick, pg="1"):
"""
搜索: /public/auto/search1.html?keyword={keyword}
搜索无分页，全部结果在第一页返回
"""
if not key:
return {"list": [], "page": 1}
try:
url = f"{self.host}/public/auto/search1.html?keyword={quote(key)}"
res = self.fetch(url, headers=self.headers)
html = res.text
items = self._parseSearchCards(html)
return {"list": items, "page": 1}
except Exception:
return {"list": [], "page": 1}
def playerContent(self, flag, id, vipFlags):
"""
播放解析: 解码 file 值获取 m3u8 直链
file 格式: 3字符前缀 + base64(URL编码的m3u8地址)
解码后返回 parse:0 直链
"""
# 如果 id 本身是 m3u8/mp4 直链
if id.endswith((".m3u8", ".mp4")) or id.startswith("http"):
return {
"parse": 0,
"url": id,
"header": self._makePlayHeader(id),
}
# 纯数字 ID 兜底: 嗅探
if id.isdigit():
return {
"parse": 1,
"url": f"{self.host}/vod/player/0/{id}",
"header": self.headers,
}
# 解码 file 值
decoded_url = self._decodeFile(id)
if decoded_url and decoded_url.startswith("http"):
return {
"parse": 0,
"url": decoded_url,
"header": self._makePlayHeader(decoded_url),
}
# 解码失败，降级嗅探
return {
"parse": 1,
"url": id,
"header": self.headers,
}
def localProxy(self, param):
pass
def isVideoFormat(self, url):
return bool(re.match(r'.*\.(m3u8|mp4)(\?.*)?$', url, re.I))
def manualVideoCheck(self):
return False
def destroy(self):
pass
FGBLH/py/麻豆社.py
+137
View File
@@ -0,0 +1,137 @@
# -*- coding: utf-8 -*-
import re
import urllib.parse
import requests
try:
from base.spider import Spider as BaseSpider
except ImportError:
class BaseSpider:
pass
class Spider(BaseSpider):
BASE_URL = "https://madou.club"
DASH_URL = "https://dash.madou.club"
HEADERS = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
"Referer": BASE_URL + "/",
}
def __init__(self):
super().__init__()
self.name = "麻豆社"
self.session = requests.Session()
self.session.headers.update(self.HEADERS)
self._class_cache = None
def init(self, extend="{}"):
return None
def getName(self):
return self.name
def homeContent(self, filter):
html = self._get(self.BASE_URL + "/")
return {"class": self._classes(html), "filters": {}, "list": self._parse_list(html), "parse": 0, "jx": 0}
def homeVideoContent(self):
return {"list": self._parse_list(self._get(self.BASE_URL + "/"))}
def categoryContent(self, tid, pg, filter, extend):
page = self._to_int(pg, 1)
base = tid if str(tid).startswith("http") else self.BASE_URL + "/category/" + str(tid).strip("/")
url = base.rstrip("/") if page <= 1 else base.rstrip("/") + "/page/" + str(page)
data = self._parse_list(self._get(url))
return {"page": page, "pagecount": page if len(data) < 10 else page + 1, "limit": 20, "total": 99999, "list": data, "parse": 0, "jx": 0}
def detailContent(self, ids):
result = {"list": [], "parse": 0, "jx": 0}
if not ids:
return result
url = ids[0]
html = self._get(url)
name = self._clean(self._match(html, r'<h1[^>]*class=["\']article-title["\'][^>]*>(.*?)</h1>') or self._match(html, r'<title>(.*?)</title>').split("-")[0])
pic = self._match(html, r'shareimage\s*:\s*["\']([^"\']+)') or self._match(html, r'<img[^>]+data-src=["\']([^"\']+)') or self._match(html, r'<img[^>]+src=["\']([^"\']+)')
cate = self._clean(self._match(html, r'分类：\s*<a[^>]*>(.*?)</a>'))
remarks = self._clean(self._match(html, r'观看\((.*?)\)'))
tag_block = self._match(html, r'<div[^>]+class=["\']article-tags["\'][^>]*>(.*?)</div>')
tags = ",".join([self._clean(x) for x in re.findall(r'<a[^>]*>(.*?)</a>', tag_block, re.S)])
iframe = self._match(html, r'<iframe[^>]+src=["\']?([^"\'\s>]+)')
play_id = urllib.parse.urljoin(self.BASE_URL, iframe or url)
result["list"].append({"vod_id": url, "vod_name": name, "vod_pic": urllib.parse.urljoin(self.BASE_URL, pic), "type_name": cate, "vod_year": "", "vod_area": "", "vod_remarks": remarks, "vod_actor": tags, "vod_director": "", "vod_content": name, "vod_play_from": "DPlayer", "vod_play_url": name + "$" + play_id})
return result
def searchContent(self, key, quick, pg="1"):
page = self._to_int(pg, 1)
q = urllib.parse.quote(str(key))
url = self.BASE_URL + "/?s=" + q if page <= 1 else self.BASE_URL + "/page/" + str(page) + "?s=" + q
data = self._parse_list(self._get(url))
return {"page": page, "pagecount": page if len(data) < 10 else page + 1, "limit": 20, "total": 99999, "list": data, "parse": 0, "jx": 0}
def playerContent(self, flag, id, vipFlags):
result = {"parse": 0, "playUrl": "", "url": id or "", "jx": 0, "header": {"User-Agent": self.HEADERS["User-Agent"], "Referer": self.BASE_URL + "/"}}
if not id:
return result
play_page = id
if "dash.madou.club/share/" not in play_page:
html = self._get(play_page)
play_page = urllib.parse.urljoin(self.BASE_URL, self._match(html, r'<iframe[^>]+src=["\']?([^"\'\s>]+)') or play_page)
html = self._get(play_page, {"Referer": self.BASE_URL + "/"})
token = self._match(html, r'var\s+token\s*=\s*["\']([^"\']*)')
m3u8 = self._match(html, r'var\s+m3u8\s*=\s*["\']([^"\']+\.m3u8)["\']')
if m3u8:
url = urllib.parse.urljoin(self.DASH_URL, m3u8)
result["url"] = url + (("&" if "?" in url else "?") + "token=" + token if token else "")
result["header"] = {"User-Agent": self.HEADERS["User-Agent"], "Referer": play_page, "Origin": self.BASE_URL}
return result
def _classes(self, html=None):
if self._class_cache:
return self._class_cache
html = html or self._get(self.BASE_URL + "/")
classes, seen = [], set()
for href, name in re.findall(r'<a[^>]+href=["\'](https://madou\.club/category/[^"\']+)["\'][^>]*>(.*?)</a>', html, re.S):
name = self._clean(name)
key = href.rstrip("/")
if key not in seen and name:
seen.add(key)
classes.append({"type_id": href, "type_name": name})
self._class_cache = classes
return classes
def _parse_list(self, html):
data = []
blocks = re.findall(r'<article\b.*?</article>', html, re.S) or re.findall(r'<li>.*?</li>', html, re.S)
for item in blocks:
href = self._match(item, r'<a[^>]+href=["\']([^"\']+\.html)["\']')
name = self._clean(self._match(item, r'<h2[^>]*>.*?<a[^>]*>(.*?)</a>') or self._match(item, r'<a[^>]*>(?:<span.*?</span>)?\s*(.*?)</a>'))
pic = self._match(item, r'<img[^>]+data-src=["\']([^"\']+)') or self._match(item, r'<img[^>]+src=["\']([^"\']+)')
remarks = self._clean(self._match(item, r'<time[^>]*>(.*?)</time>') or self._match(item, r'观看\((.*?)\)'))
if href and name:
data.append({"vod_id": urllib.parse.urljoin(self.BASE_URL, href), "vod_name": name, "vod_pic": urllib.parse.urljoin(self.BASE_URL, pic), "vod_remarks": remarks})
return data
def _get(self, url, headers=None):
h = dict(self.HEADERS)
if headers:
h.update(headers)
try:
return self.session.get(url, headers=h, timeout=15, verify=False).text
except Exception:
return ""
def _match(self, text, pattern):
m = re.search(pattern, text or "", re.S | re.I)
return m.group(1).strip() if m else ""
def _clean(self, text):
text = re.sub(r'<.*?>', '', text or '')
text = text.replace('&nbsp;', ' ').replace('&amp;', '&').replace('&#038;', '&').replace('"', '"')
return re.sub(r'\s+', ' ', text).strip()
def _to_int(self, value, default=0):
try:
return int(value)
except Exception:
return default