feat: add libvio spider

This commit is contained in:
Harold
2026-04-18 13:13:49 +08:00
parent b5cfd5dac1
commit b319bfc785
4 changed files with 1252 additions and 0 deletions
+292
View File
@@ -0,0 +1,292 @@
# coding=utf-8
import json
import re
import sys
from urllib.parse import quote
from base.spider import Spider as BaseSpider
sys.path.append("..")
class Spider(BaseSpider):
def __init__(self):
self.name = "LibVIO"
self.host = "https://libvio.site"
self.headers = {
"User-Agent": (
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
"AppleWebKit/537.36 (KHTML, like Gecko) "
"Chrome/124.0.0.0 Safari/537.36"
)
}
self.categories = [
{"type_name": "最近更新", "type_id": "index"},
{"type_name": "电影", "type_id": "movie"},
{"type_name": "电视剧", "type_id": "series"},
{"type_name": "动漫", "type_id": "anime"},
{"type_name": "日韩剧", "type_id": "jpandkr"},
{"type_name": "欧美剧", "type_id": "euandus"},
]
self.category_paths = {
"index": "/",
"movie": "/type/1-{pg}.html",
"series": "/type/2-{pg}.html",
"anime": "/type/4-{pg}.html",
"jpandkr": "/type/15-{pg}.html",
"euandus": "/type/16-{pg}.html",
}
def init(self, extend=""):
return None
def getName(self):
return self.name
def homeContent(self, filter):
return {"class": self.categories}
def _build_url(self, href):
raw = str(href or "").strip()
if not raw:
return ""
if raw.startswith(("http://", "https://")):
return raw
if raw.startswith("//"):
return "https:" + raw
return self.host + "/" + raw.lstrip("/")
def _extract_vod_id(self, href):
raw = str(href or "").strip()
matched = re.search(r"/detail/(\d+)\.html", raw)
if matched:
return matched.group(1)
if re.fullmatch(r"\d+", raw):
return raw
return ""
def _extract_play_id(self, href):
raw = str(href or "").strip()
matched = re.search(r"/play/([^./]+-\d+-\d+)\.html", raw)
if matched:
return matched.group(1)
if re.fullmatch(r"[^./]+-\d+-\d+", raw):
return raw
return ""
def _build_detail_request_url(self, vod_id):
return f"{self.host}/detail/{self._extract_vod_id(vod_id)}.html"
def _build_play_request_url(self, play_id):
return f"{self.host}/play/{self._extract_play_id(play_id)}.html"
def _parse_list_cards(self, html):
root = self.html(html)
results = []
if root is None:
return results
seen = set()
for card in root.xpath("//*[contains(@class,'stui-vodlist__box')]"):
href = ((card.xpath(".//a[@href][1]/@href") or [""])[0]).strip()
vod_id = self._extract_vod_id(href)
title = ((card.xpath(".//a[@title][1]/@title") or [""])[0]).strip()
pic = (
(card.xpath(".//a[@data-original][1]/@data-original") or [""])[0].strip()
or (card.xpath(".//img[@data-original][1]/@data-original") or [""])[0].strip()
or (card.xpath(".//img[@src][1]/@src") or [""])[0].strip()
)
remarks = "".join(card.xpath(".//*[contains(@class,'pic-text')][1]//text()")).strip()
if not vod_id or vod_id in seen or not title:
continue
seen.add(vod_id)
results.append(
{
"vod_id": vod_id,
"vod_name": title,
"vod_pic": self._build_url(pic),
"vod_remarks": remarks,
}
)
return results
def _request_html(self, path_or_url, expect_xpath=None, referer=None):
target = path_or_url if str(path_or_url).startswith("http") else self._build_url(path_or_url)
headers = dict(self.headers)
headers["Referer"] = referer or (self.host + "/")
response = self.fetch(target, headers=headers, timeout=10)
if response.status_code != 200:
return ""
return response.text or ""
def _page_result(self, items, pg):
page = int(pg)
pagecount = page + 1 if items else page
return {
"list": items,
"page": page,
"pagecount": pagecount,
"limit": len(items),
"total": pagecount * max(len(items), 1),
}
def _clean_text(self, text):
return re.sub(r"\s+", " ", str(text or "").replace("\xa0", " ")).strip()
def _extract_detail_field(self, root, label, joiner=""):
if root is None:
return ""
nodes = root.xpath(f'.//span[contains(normalize-space(.), "{label}")]')
if not nodes:
return ""
values = []
label_node = nodes[0]
if self._clean_text(label_node.tail):
values.append(self._clean_text(label_node.tail))
for sibling in label_node.itersiblings():
if sibling.tag == "span" and "text-muted" in " ".join(sibling.xpath("./@class")):
break
text = self._clean_text("".join(sibling.xpath(".//text()")))
if text:
values.append(text)
tail = self._clean_text(sibling.tail)
if tail:
values.append(tail)
cleaned = []
for value in values:
if value and value not in cleaned:
cleaned.append(value)
return joiner.join(cleaned) if joiner else "".join(cleaned)
def _parse_detail_page(self, html, vod_id):
root = self.html(html)
detail_root = (root.xpath("//*[contains(@class,'stui-content__detail')][1]") or [root])[0]
title = ((detail_root.xpath(".//*[contains(@class,'title')][1]//text()") or [""])[0]).strip()
pic = (
(root.xpath("//*[contains(@class,'stui-content__thumb')]//img/@data-original") or [""])[0].strip()
or (root.xpath("//*[contains(@class,'stui-content__thumb')]//img/@src") or [""])[0].strip()
)
episodes = []
seen = set()
for playlist in root.xpath("//*[contains(@class,'stui-content__playlist')]"):
heading = self._clean_text("".join(playlist.xpath("./preceding-sibling::*[1]//text()")))
if any(keyword in heading for keyword in ("夸克", "UC", "网盘")):
continue
for anchor in playlist.xpath(".//a[@href]"):
play_id = self._extract_play_id((anchor.xpath("./@href") or [""])[0])
name = self._clean_text("".join(anchor.xpath(".//text()")))
if not play_id or not name or play_id in seen:
continue
seen.add(play_id)
episodes.append(f"{name}${play_id}")
vod = {
"vod_id": vod_id,
"path": self._build_detail_request_url(vod_id),
"vod_name": title,
"vod_pic": self._build_url(pic),
"vod_tag": "",
"vod_time": "",
"vod_remarks": "",
"vod_play_from": "LibVIO",
"vod_play_url": "#".join(episodes),
"type_name": self._extract_detail_field(detail_root, "类型"),
"vod_content": self._extract_detail_field(detail_root, "简介"),
"vod_year": self._extract_detail_field(detail_root, "年份"),
"vod_area": self._extract_detail_field(detail_root, "地区"),
"vod_lang": "",
"vod_director": self._extract_detail_field(detail_root, "导演", joiner=","),
"vod_actor": self._extract_detail_field(detail_root, "主演", joiner=","),
}
return {"list": [vod]}
def homeVideoContent(self):
html = self._request_html("/", expect_xpath="//*[contains(@class,'stui-vodlist__box')]")
return {"list": self._parse_list_cards(html)}
def categoryContent(self, tid, pg, filter, extend):
path = self.category_paths.get(tid, self.category_paths["movie"]).format(pg=pg)
html = self._request_html(path, expect_xpath="//*[contains(@class,'stui-vodlist__box')]")
return self._page_result(self._parse_list_cards(html), pg)
def searchContent(self, key, quick, pg="1"):
path = "/search/-------------.html?wd={0}".format(quote(key))
html = self._request_html(path, expect_xpath="//*[contains(@class,'stui-vodlist__box')]")
return self._page_result(self._parse_list_cards(html), pg)
def detailContent(self, ids):
vod_id = ids[0]
html = self._request_html(
self._build_detail_request_url(vod_id),
expect_xpath="//*[contains(@class,'stui-content__playlist')]",
)
return self._parse_detail_page(html, vod_id)
def _parse_player_config(self, html):
matched = re.search(r"player_[a-z0-9_]+\s*=\s*(\{[\s\S]*?\})\s*;?", html, re.I)
if not matched:
return None
try:
return json.loads(matched.group(1))
except Exception:
return None
def _extract_play_api_base(self, body):
matched = re.search(r'src\s*=\s*["\']([^"\']+)["\']', body, re.I)
if not matched:
return ""
return self._build_url(matched.group(1))
def _extract_playable_url(self, body):
patterns = [
r'["\']?urls?["\']?\s*:\s*["\']([^"\']+)["\']',
r'(?:var|let|const)\s+urls?\s*=\s*["\']([^"\']+)["\']',
r'["\']url["\']\s*:\s*["\']([^"\']+\.(?:m3u8|mp4)[^"\']*)["\']',
r'url\s*=\s*["\']([^"\']+\.(?:m3u8|mp4)[^"\']*)["\']',
]
text = str(body or "")
for pattern in patterns:
matched = re.search(pattern, text, re.I)
if matched:
return self._build_url(matched.group(1).replace("\\/", "/").replace("&", "&").strip())
return ""
def _request_player_js(self, source):
return self._request_html(f"/static/player/{source}.js", referer=self.host + "/")
def playerContent(self, flag, id, vipFlags):
play_page_url = self._build_play_request_url(id)
detail_html = self._request_html(play_page_url, referer=self.host + "/")
config = self._parse_player_config(detail_html)
if not config or config.get("from") in ("kuake", "uc"):
return {"parse": 0, "playUrl": "", "url": ""}
if config.get("from") == "ty_new1":
api_body = self._request_html(f"/vid/ty4.php?url={config.get('url', '')}", referer=self.host + "/")
final_url = self._extract_playable_url(api_body)
else:
player_js = self._request_player_js(config.get("from", ""))
api_base = self._extract_play_api_base(player_js)
api_url = "{base}{url}&next={next}&id={id}&nid={nid}".format(
base=api_base,
url=config.get("url", ""),
next=config.get("link_next", ""),
id=config.get("id", ""),
nid=config.get("nid", ""),
)
api_body = self._request_html(api_url, referer=self.host + "/")
final_url = self._extract_playable_url(api_body)
if not final_url:
return {"parse": 0, "playUrl": "", "url": ""}
return {
"parse": 0,
"playUrl": "",
"url": final_url,
"header": {
"User-Agent": self.headers["User-Agent"],
"Referer": self.host + "/",
},
}