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-23 07:37:22 +00:00
parent 339fbf5fdf
commit 34d265c1c5
27 changed files with 3901 additions and 1955 deletions
Download Patch File Download Diff File Expand all files Collapse all files
jaychouqq/tvbox/file.json
+1 -1
View File
File diff suppressed because one or more lines are too long
jaychouqq/tvbox/file1.json
+1 -1
View File
File diff suppressed because one or more lines are too long
jaychouqq/tvbox/file2.json
+1 -1
View File
File diff suppressed because one or more lines are too long
jaychouqq/tvbox/jaychouqq
+1 -1
View File
File diff suppressed because one or more lines are too long
jaychouqq/yingshi/py9/hanime1.py
+565
View File
@@ -0,0 +1,565 @@
# -*- coding: utf-8 -*-
# //@name:Hanime1
# //@id:hanime1
# //@version:2
# //wab201 学习研究用
import base64
import json
import re
import time
from urllib.parse import parse_qs, urljoin, urlsplit
import requests
from lxml import html
from base.spider import Spider
class Spider(Spider):
HOST = "https://hanime1.me"
PLAY_PREFIX = "hanime1://play/"
CATEGORIES = (
("all", "全部", ""),
("hentai", "裏番", "裏番"),
("short", "泡麵番", "泡麵番"),
("motion", "Motion Anime", "Motion Anime"),
("cg3d", "3DCG", "3DCG"),
("d25", "2.5D", "2.5D"),
("d2", "2D動畫", "2D動畫"),
("ai", "AI生成", "AI生成"),
("mmd", "MMD", "MMD"),
("cosplay", "Cosplay", "Cosplay"),
)
SORTS = (
"最新上市",
"最新上傳",
"本日排行",
"本週排行",
"本月排行",
"觀看次數",
"讚好比例",
"時長最長",
"他們在看",
)
DURATIONS = (
"",
"1 分鐘 +",
"5 分鐘 +",
"10 分鐘 +",
"20 分鐘 +",
"30 分鐘 +",
"60 分鐘 +",
"0 - 10 分鐘",
"0 - 20 分鐘",
)
VIDEO_RE = re.compile(r"\.(?:mp4|m3u8|mkv|webm)(?:$|[?#])", re.I)
def __init__(self):
self.name = "Hanime1"
self.host = self.HOST
self.timeout = 20
self.retries = 2
self.preferred_quality = 0
self.trust_env = True
self.cookie = ""
self.user_agent = (
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
"AppleWebKit/537.36 (KHTML, like Gecko) "
"Chrome/138.0.0.0 Safari/537.36"
)
self.backend_parse = False
self.category_mode = False
self.categoryMode = False
self.session = None
def getName(self):
return self.name
def init(self, extend=""):
config = self._config(extend)
self.host = self._host(config.get("host") or self.HOST)
self.timeout = self._bounded_int(config.get("timeout"), 20, 8, 45)
self.retries = self._bounded_int(config.get("retries"), 2, 0, 4)
self.preferred_quality = self._bounded_int(
config.get("preferred_quality"), 0, 0, 4320
)
self.trust_env = self._bool(config.get("trust_env"), True)
self.cookie = str(
config.get("cookie") or config.get("cf_cookie") or ""
).strip()
self.user_agent = str(config.get("user_agent") or self.user_agent).strip()
self._reset_session()
def destroy(self):
if self.session is not None:
try:
self.session.close()
except Exception:
pass
self.session = None
def isVideoFormat(self, url):
return bool(self.VIDEO_RE.search(str(url or "")))
def manualVideoCheck(self):
return False
def homeContent(self, filter):
classes = [
{"type_id": type_id, "type_name": type_name}
for type_id, type_name, _ in self.CATEGORIES
]
filters = {}
sort_values = [{"n": value, "v": value} for value in self.SORTS]
duration_values = [
{"n": value or "全部", "v": value} for value in self.DURATIONS
]
for type_id, _, _ in self.CATEGORIES:
filters[type_id] = [
{"key": "sort", "name": "排序", "value": sort_values},
{"key": "duration", "name": "時長", "value": duration_values},
]
return {"class": classes, "filters": filters}
def homeVideoContent(self):
result = self.categoryContent("all", "1", False, {"sort": "最新上傳"})
return {"list": result.get("list", [])}
def categoryContent(self, tid, pg, filter, extend):
page = self._page(pg)
genre = self._category_genre(tid)
options = self._config(extend)
params = {
"sort": str(options.get("sort") or "最新上傳"),
"page": page,
}
duration = str(options.get("duration") or "").strip()
if genre:
params["genre"] = genre
if duration:
params["duration"] = duration
try:
source, _ = self._request_text("/search", params=params)
return self._parse_listing(source, page)
except Exception as exc:
return self._empty_page(page, "分類讀取失敗: %s" % exc)
def searchContent(self, key, quick, pg="1"):
page = self._page(pg)
keyword = self._clean(key)
if not keyword:
return self._empty_page(page)
try:
source, _ = self._request_text(
"/search", params={"query": keyword, "page": page}
)
return self._parse_listing(source, page)
except Exception as exc:
return self._empty_page(page, "搜尋失敗: %s" % exc)
def detailContent(self, ids):
raw_id = ids[0] if isinstance(ids, (list, tuple)) and ids else ids
video_id = self._video_id(raw_id)
if not video_id:
return {"list": []}
try:
source, page_url = self._request_text(
"/watch", params={"v": video_id}
)
return {"list": [self._parse_detail(source, video_id, page_url)]}
except Exception as exc:
message = "詳情讀取失敗: %s" % exc
return {
"list": [
{
"vod_id": video_id,
"vod_name": "Hanime1 %s" % video_id,
"vod_remarks": message,
"vod_content": message,
"vod_play_from": "Hanime1直鏈",
"vod_play_url": "重試$%s%s/0" % (self.PLAY_PREFIX, video_id),
}
]
}
def playerContent(self, flag, id, vipFlags):
video_id, requested_quality = self._play_id(id)
if not video_id:
return self._player_error("無法識別播放 ID")
try:
source, page_url = self._request_text(
"/watch", params={"v": video_id}
)
sources = self._parse_sources(self._document(source))
selected = self._select_source(sources, requested_quality)
if not selected:
return self._player_error("播放頁沒有可用的直鏈")
return {
"parse": 0,
"jx": 0,
"playUrl": "",
"url": selected[1],
"header": {
"User-Agent": self.user_agent,
"Referer": page_url,
"Origin": self.host,
},
}
except Exception as exc:
return self._player_error("播放解析失敗: %s" % exc)
def _reset_session(self):
self.destroy()
self.session = requests.Session()
self.session.trust_env = self.trust_env
self.session.headers.update(
{
"User-Agent": self.user_agent,
"Accept": (
"text/html,application/xhtml+xml,application/xml;q=0.9,"
"image/avif,image/webp,*/*;q=0.8"
),
"Accept-Language": "zh-TW,zh;q=0.9,en;q=0.7",
"Cache-Control": "no-cache",
}
)
if self.cookie:
self.session.headers["Cookie"] = self.cookie.removeprefix("Cookie:").strip()
def _request_text(self, path, params=None):
if self.session is None:
self._reset_session()
url = urljoin(self.host + "/", str(path or "").lstrip("/"))
last_error = None
for attempt in range(self.retries + 1):
try:
response = self.session.get(
url,
params=params,
headers={"Referer": self.host + "/"},
timeout=self.timeout,
allow_redirects=True,
)
source = self._decode_response(response)
if self._is_challenge(source) or response.status_code in (403, 429, 503):
raise RuntimeError(
"Cloudflare 驗證未通過；請在同一出口的瀏覽器完成驗證，"
"再於 Extend 填入 Cookie 和相同 User-Agent"
)
response.raise_for_status()
return source, response.url
except (requests.RequestException, RuntimeError) as exc:
last_error = exc
if attempt < self.retries:
time.sleep(0.35 * (attempt + 1))
raise RuntimeError(str(last_error or "請求失敗"))
def _parse_listing(self, source, page):
doc = self._document(source)
cards = []
seen = set()
nodes = doc.xpath(
"//a[contains(@href,'/watch') and "
".//div[contains(concat(' ',normalize-space(@class),' '),' search-videos ')]]"
)
if not nodes:
nodes = doc.xpath("//a[contains(@class,'video-link') and contains(@href,'/watch')]")
for node in nodes:
video_id = self._video_id(node.get("href"))
if not video_id or video_id in seen:
continue
seen.add(video_id)
title = self._clean(
node.xpath("string(.//div[contains(@class,'home-rows-videos-title')][1])")
or node.xpath("string(.//div[contains(@class,'title')][1])")
or node.get("title")
)
image = node.xpath("string(.//img[1]/@src)")
duration = self._clean(
node.xpath("string(.//div[contains(@class,'duration')][1])")
)
cards.append(
{
"vod_id": video_id,
"vod_name": title or "Hanime1 %s" % video_id,
"vod_pic": urljoin(self.host + "/", image),
"vod_remarks": duration,
}
)
page_numbers = []
for text_value in doc.xpath("//ul[contains(@class,'pagination')]//li//text()"):
value = self._clean(text_value)
if value.isdigit():
page_numbers.append(int(value))
pagecount = max([page] + page_numbers)
limit = len(cards)
return {
"list": cards,
"page": page,
"pagecount": pagecount,
"limit": limit,
"total": pagecount * limit if limit else 0,
}
def _parse_detail(self, source, video_id, page_url):
doc = self._document(source)
title = self._clean(
doc.xpath("string(//meta[@property='og:title']/@content)")
or doc.xpath("string(//h3[@id='shareBtn-title'])")
or doc.xpath("string(//title)")
)
title = re.sub(r"\s*-\s*Hanime1\.me\s*$", "", title, flags=re.I)
cover = doc.xpath("string(//meta[@property='og:image']/@content)")
if not cover:
cover = doc.xpath("string(//video[@id='player']/@poster)")
content = self._clean(doc.xpath("string(//meta[@name='description']/@content)"))
actor = self._clean(doc.xpath("string(//a[@id='video-artist-name'])"))
genre = self._clean(
doc.xpath("string((//a[contains(@href,'genre=')])[last()])")
)
tag_values = []
for node in doc.xpath("//div[contains(@class,'single-video-tag')]/a"):
value = re.sub(r"\s*\(\d+\)\s*$", "", self._clean(node.text_content()))
if value and value not in tag_values:
tag_values.append(value)
date_match = re.search(
r"\b(20\d{2})-\d{2}-\d{2}\b", self._clean(doc.text_content())
)
sources = self._parse_sources(doc)
play_items = []
for quality, _ in sources:
label = "%sP" % quality if quality else "自動"
play_items.append(
"%s$%s%s/%s" % (label, self.PLAY_PREFIX, video_id, quality or 0)
)
if not play_items:
play_items.append("自動$%s%s/0" % (self.PLAY_PREFIX, video_id))
return {
"vod_id": video_id,
"vod_name": title or "Hanime1 %s" % video_id,
"vod_pic": urljoin(page_url, cover),
"type_name": genre,
"vod_year": date_match.group(1) if date_match else "",
"vod_actor": actor,
"vod_content": content,
"vod_tag": ",".join(tag_values[:24]),
"vod_play_from": "Hanime1直鏈",
"vod_play_url": "#".join(play_items),
}
def _parse_sources(self, doc):
values = {}
for node in doc.xpath("//video//source[@src]"):
url = urljoin(self.host + "/", node.get("src"))
quality = self._quality(node.get("size"), url)
if url.startswith(("http://", "https://")):
values[quality] = url
if not values:
for node in doc.xpath("//link[@rel='preload' and @as='video']/@href"):
url = urljoin(self.host + "/", node)
values[self._quality("", url)] = url
return sorted(values.items(), key=lambda item: item[0], reverse=True)
def _document(self, source):
if isinstance(source, (bytes, bytearray)):
text = self._decode_bytes(bytes(source))
else:
text = str(source or "")
if "\x00" in text[:256]:
try:
text = self._decode_bytes(text.encode("latin-1"))
except (UnicodeEncodeError, UnicodeDecodeError):
text = text.replace("\x00", "")
parser = html.HTMLParser(encoding="utf-8", recover=True)
return html.fromstring(text.encode("utf-8"), parser=parser)
def _decode_response(self, response):
return self._decode_bytes(response.content, response.encoding)
@staticmethod
def _decode_bytes(raw, declared_encoding=""):
if not raw:
return ""
signatures = (
(b"\xff\xfe\x00\x00", "utf-32-le"),
(b"\x00\x00\xfe\xff", "utf-32-be"),
(b"\xff\xfe", "utf-16-le"),
(b"\xfe\xff", "utf-16-be"),
)
for signature, encoding in signatures:
if raw.startswith(signature):
return raw.decode(encoding).lstrip("\ufeff")
sample = raw[:512]
if len(sample) >= 16:
groups = len(sample) // 4
le32_zeros = sum(
sample[index] == 0
for index in range(1, groups * 4)
if index % 4 in (1, 2, 3)
)
be32_zeros = sum(
sample[index] == 0
for index in range(0, groups * 4)
if index % 4 in (0, 1, 2)
)
if le32_zeros >= groups * 2:
return raw.decode("utf-32-le")
if be32_zeros >= groups * 2:
return raw.decode("utf-32-be")
if raw[:4] == b"<\x00\x00\x00":
return raw.decode("utf-32-le")
if raw[:4] == b"\x00\x00\x00<":
return raw.decode("utf-32-be")
if raw[:2] == b"<\x00":
return raw.decode("utf-16-le")
if raw[:2] == b"\x00<":
return raw.decode("utf-16-be")
candidates = [str(declared_encoding or "").strip(), "utf-8-sig", "utf-8"]
for encoding in candidates:
if not encoding:
continue
try:
return raw.decode(encoding)
except (LookupError, UnicodeDecodeError):
continue
return raw.decode("utf-8", errors="replace")
def _select_source(self, sources, requested_quality):
if not sources:
return None
target = requested_quality or self.preferred_quality
if target:
for item in sources:
if item[0] == target:
return item
return min(sources, key=lambda item: abs(item[0] - target))
return sources[0]
def _play_id(self, value):
text = str(value or "").strip()
match = re.match(r"^hanime1://play/(\d+)/(\d+)$", text)
if match:
return match.group(1), int(match.group(2))
video_id = self._video_id(text)
return (video_id, 0) if video_id else ("", 0)
def _video_id(self, value):
text = str(value or "").strip()
if text.startswith("atvp_detail:"):
text = text[len("atvp_detail:") :].strip()
if text.isdigit():
return text
match = re.search(r"hanime1://(?:video|play)/(\d+)", text)
if match:
return match.group(1)
try:
values = parse_qs(urlsplit(text).query).get("v") or []
if values and str(values[0]).isdigit():
return str(values[0])
except Exception:
pass
match = re.search(r"(?:[?&]v=|/watch/)(\d+)", text)
return match.group(1) if match else ""
def _category_genre(self, tid):
value = str(tid or "all")
for type_id, _, genre in self.CATEGORIES:
if value == type_id or value == genre:
return genre
return ""
def _quality(self, value, url):
match = re.search(r"(\d{3,4})", str(value or ""))
if not match:
match = re.search(r"[-_](\d{3,4})p(?:\.|[/?])", str(url or ""), re.I)
return int(match.group(1)) if match else 0
def _is_challenge(self, source):
text = str(source or "")[:80000].lower()
signals = (
"<title>just a moment...</title>",
"id=\"challenge-form\"",
"cf-browser-verification",
"cf-chl-captcha",
"attention required! | cloudflare",
)
return any(signal in text for signal in signals)
def _player_error(self, message):
text = self._clean(message) or "播放失敗"
return {
"parse": 0,
"jx": 0,
"playUrl": "",
"url": "",
"header": {},
"msg": text,
"error": text,
"content": text,
}
def _empty_page(self, page, message=""):
result = {
"list": [],
"page": page,
"pagecount": page,
"limit": 0,
"total": 0,
}
if message:
result["msg"] = message
return result
def _config(self, extend):
if isinstance(extend, dict):
return extend
text = str(extend or "").strip()
if not text:
return {}
candidates = [text]
try:
candidates.append(
base64.urlsafe_b64decode(text + "=" * (-len(text) % 4)).decode("utf-8")
)
except Exception:
pass
for candidate in candidates:
try:
value = json.loads(candidate)
if isinstance(value, dict):
return value
except Exception:
continue
return {"cookie": text} if "=" in text else {}
@staticmethod
def _host(value):
text = str(value or "").strip().rstrip("/")
return text if text.startswith(("http://", "https://")) else Spider.HOST
@staticmethod
def _clean(value):
return " ".join(str(value or "").replace("\xa0", " ").split())
@staticmethod
def _page(value):
try:
return max(1, int(value))
except Exception:
return 1
@staticmethod
def _bounded_int(value, default, low, high):
try:
return min(high, max(low, int(value)))
except Exception:
return default
@staticmethod
def _bool(value, default):
if isinstance(value, bool):
return value
if value is None:
return default
return str(value).strip().lower() in ("1", "true", "yes", "on")
jaychouqq/yingshi/py9/spider.py
+243
View File
@@ -0,0 +1,243 @@
import requests
from bs4 import BeautifulSoup
import re
import subprocess
import platform
import os
import sys
# ==================== 播放器配置 ====================
# 这里配置你的 PotPlayer 所在目录
POTPLAYER_DIR = r"E:\potplayer"
def get_potplayer_path():
"""自动在指定目录下寻找 PotPlayer 的可执行文件"""
if platform.system() != "Windows":
return "open -a IINA" # Mac 默认使用 IINA
paths_to_try = [
os.path.join(POTPLAYER_DIR, "PotPlayer64.exe"),
os.path.join(POTPLAYER_DIR, "PotPlayerMini64.exe"),
os.path.join(POTPLAYER_DIR, "PotPlayer.exe"),
os.path.join(POTPLAYER_DIR, "PotPlayerMini.exe"),
]
for p in paths_to_try:
if os.path.exists(p):
return p
return None
class HanxiaoquanPlayer:
def __init__(self):
self.base_url = 'https://www.jennyhow.com'
self.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',
'Referer': self.base_url
}
self.target_categories = [
"今日推荐", "最新韩剧", "韩国电影", "韩国综艺", "韩国动漫",
"最新韩剧•月榜", "韩国电影•月榜", "韩国综艺•月榜", "韩国动漫•月榜"
]
self.menu_data = {} # 存放首页菜单数据
self.player_exe = get_potplayer_path()
def clear_screen(self):
os.system('cls' if os.name == 'nt' else 'clear')
def fetch_html(self, url):
"""通用网络请求"""
try:
response = requests.get(url, headers=self.headers, timeout=10)
response.raise_for_status()
response.encoding = 'utf-8'
return response.text
except Exception as e:
print(f"[-] 网络请求失败: {e}")
return None
def load_index(self):
"""1. 爬取首页，加载所有分类和片单（轻量级，秒出）"""
print("[*] 正在连接韩小圈服务器，获取最新片单...")
html = self.fetch_html(self.base_url)
if not html:
sys.exit(1)
soup = BeautifulSoup(html, 'lxml')
titles = soup.find_all(class_='module-title')
for title_tag in titles:
category_name = title_tag.get_text(strip=True)
if category_name not in self.target_categories: continue
videos = []
if "月榜" not in category_name:
# 左侧图文卡片
heading_div = title_tag.find_parent('div', class_='module-heading')
list_div = heading_div.find_next_sibling('div', class_='module-list')
if list_div:
for item in list_div.find_all('div', class_='module-item'):
a_tag = item.find('a', class_='module-item-title')
if a_tag:
name = a_tag.get_text(strip=True)
href = a_tag.get('href', '')
videos.append({"名称": name, "链接": self.base_url + href if href else ""})
else:
# 右侧月榜纯文字
heading_div = title_tag.find_parent('div', class_='module-heading')
list_div = heading_div.find_next_sibling('div', class_='module-side-list')
if list_div:
for item in list_div.find_all('a', class_='text-list-item'):
name = item.get('title', '未知')
href = item.get('href', '')
videos.append({"名称": name, "链接": self.base_url + href if href else ""})
if videos:
self.menu_data[category_name] = videos
def load_detail(self, detail_url):
"""2. 用户点进某部剧后，实时抓取详情页解析集数"""
print("\n[*] 正在获取播放线路...")
html = self.fetch_html(detail_url)
if not html: return {}
soup = BeautifulSoup(html, 'lxml')
lines_data = {}
# 寻找选项卡
tabs_ul = soup.find('ul', class_='nav-tabs')
if tabs_ul:
for li in tabs_ul.find_all('li'):
a_tag = li.find('a')
if not a_tag: continue
line_name = a_tag.get_text(strip=True)
target_id = a_tag.get('href', '').replace('#', '')
playlist_div = soup.find('div', id=target_id)
if playlist_div:
episodes = []
for ep in playlist_div.find_all('a'):
ep_name = ep.get('title') or ep.get_text(strip=True)
ep_href = ep.get('href', '')
episodes.append({"剧集": ep_name, "链接": self.base_url + ep_href if ep_href else ""})
lines_data[line_name] = episodes
return lines_data
def get_m3u8(self, play_url):
"""3. 用户选择集数后，实时去播放页提取 m3u8 直链"""
print("\n[*] 正在破解高速视频流直链...")
html = self.fetch_html(play_url)
if not html: return None
# 使用正则提取 var now="..."; 里的内容，兼容单双引号
match = re.search(r'var now=[\'"](.*?)[\'"];', html)
if match:
return match.group(1)
return None
def play_video(self, m3u8_url):
"""4. 唤起 PotPlayer"""
print(f"[+] 获取直链成功！\n -> {m3u8_url}")
print("\n🚀 正在呼叫本地播放器...")
if platform.system() == "Windows":
if not self.player_exe:
print(f"[-] 未在 {POTPLAYER_DIR} 找到 PotPlayer！请检查路径配置！")
print("你可以手动复制上面的链接，用任何播放器打开。")
return
try:
# 隐藏控制台黑框调用 PotPlayer
subprocess.Popen([self.player_exe, m3u8_url], creationflags=subprocess.CREATE_NO_WINDOW)
except Exception as e:
print(f"[-] 播放器启动失败: {e}")
else:
os.system(f'{self.player_exe} "{m3u8_url}"')
def start(self):
"""主交互循环"""
self.clear_screen()
self.load_index()
while True:
self.clear_screen()
print("=" * 45)
print(" 🍿 韩小圈 VIP免广专属点播器 🍿")
print("=" * 45)
categories = list(self.menu_data.keys())
for i, cat in enumerate(categories):
print(f" [{i:02d}] {cat}")
print("-" * 45)
cat_idx = input("👉 请选择分类编号 (输入 q 退出): ")
if cat_idx.lower() == 'q': break
if not cat_idx.isdigit() or int(cat_idx) >= len(categories):
input("[-] 输入有误，按回车重试...")
continue
selected_cat = categories[int(cat_idx)]
videos = self.menu_data[selected_cat]
# --- 选剧 ---
self.clear_screen()
print(f"=== 当前板块: {selected_cat} ===")
for i, v in enumerate(videos):
print(f" [{i:02d}] {v['名称']}")
vid_idx = input("\n👉 请选择影片编号 (输入 b 返回): ")
if vid_idx.lower() == 'b': continue
if not vid_idx.isdigit() or int(vid_idx) >= len(videos):
input("[-] 输入有误，按回车重试...")
continue
selected_video = videos[int(vid_idx)]
# --- 加载并选线路 ---
lines = self.load_detail(selected_video['链接'])
if not lines:
input("[-] 抱歉，该影片暂无可播放的资源，按回车返回...")
continue
line_names = list(lines.keys())
print("\n" + "-" * 30)
for i, line in enumerate(line_names):
print(f" [{i}] {line} (共 {len(lines[line])} 集)")
line_idx = input("\n👉 请选择线路编号 (默认0, b返回): ")
if line_idx.lower() == 'b': continue
line_idx = 0 if line_idx == "" else int(line_idx)
if line_idx >= len(line_names): continue
selected_line = line_names[line_idx]
episodes = lines[selected_line]
# --- 选集并播放 ---
print("\n" + "-" * 30)
for i, ep in enumerate(episodes):
# 每排打印多个集数，看着更整齐
print(f"[{i:02d}] {ep['剧集'][:8]:<8}", end="\t")
if (i + 1) % 4 == 0: print()
print() # 换行
ep_idx = input("\n👉 请输入要播放的集数编号 (默认0, b返回): ")
if ep_idx.lower() == 'b': continue
ep_idx = 0 if ep_idx == "" else int(ep_idx)
if ep_idx >= len(episodes): continue
selected_ep = episodes[ep_idx]
# 获取 m3u8 并播放
m3u8_url = self.get_m3u8(selected_ep['链接'])
if m3u8_url:
self.play_video(m3u8_url)
else:
print("[-] 提取视频流失败，可能是网站修改了规则。")
input("\n[ 观影愉快！看完按回车键返回主菜单... ]")
if __name__ == '__main__':
app = HanxiaoquanPlayer()
app.start()
jaychouqq/yingshi/py9/太乙影视.py
+156
View File
@@ -0,0 +1,156 @@
# coding=utf-8
import sys
sys.path.append('..')
from base.spider import Spider
import json
import urllib.parse
import re
from lxml import etree
class Spider(Spider):
def init(self, extend=""):
self.homeUrl = "https://ww98.taiee.xyz"
self.headers = {
"User-Agent": "Mozilla/5.0 (Android; Mobile) AppleWebKit/537.36 Chrome/120 Mobile Safari/537.36",
"Referer": self.homeUrl
}
# 首页推荐数据
def homeContent(self, filter):
result = {
"class": [
{"type_id": "20", "type_name": "电影"},
{"type_id": "21", "type_name": "剧集"},
{"type_id": "22", "type_name": "综艺"},
{"type_id": "23", "type_name": "动漫"}
],
"list": [],
"page": 1
}
# 首页推荐区域（相关影视板块）
html = self.fetch(self.homeUrl, headers=self.headers).text
tree = etree.HTML(html)
vod_list = tree.xpath('//div[contains(@class,"public-list-box")]')
for item in vod_list:
vod = {}
# 详情链接
link = item.xpath('.//a[@class="public-list-exp"]/@href')
if not link:
continue
vod["vod_id"] = link[0].split("/id/")[-1].replace(".html", "")
vod["vod_detail_url"] = self.homeUrl + link[0]
# 片名
name = item.xpath('.//a[@class="time-title"]/text()')
vod["vod_name"] = name[0].strip() if name else "未知影片"
# 封面图 data-src懒加载
pic = item.xpath('.//img/@data-src')
vod["vod_pic"] = pic[0] if pic else ""
result["list"].append(vod)
return result
# 分类列表
def categoryContent(self, tid, pg, filter, ext):
result = {"list": [], "page": pg, "pagecount": 10, "limit": 24}
url = f"{self.homeUrl}/index.php/vod/type/id/{tid}/page/{pg}.html"
html = self.fetch(url, headers=self.headers).text
tree = etree.HTML(html)
vod_list = tree.xpath('//div[contains(@class,"public-list-box")]')
for item in vod_list:
vod = {}
link = item.xpath('.//a[@class="public-list-exp"]/@href')
if not link:
continue
vod["vod_id"] = link[0].split("/id/")[-1].replace(".html", "")
vod["vod_detail_url"] = self.homeUrl + link[0]
name = item.xpath('.//a[@class="time-title"]/text()')
vod["vod_name"] = name[0].strip() if name else ""
pic = item.xpath('.//img/@data-src')
vod["vod_pic"] = pic[0] if pic else ""
result["list"].append(vod)
return result
# 搜索影片
def searchContent(self, key, quick, pg):
result = {"list": [], "page": pg, "pagecount": 5}
search_url = f"{self.homeUrl}/index.php/vod/search.html?wd={urllib.parse.quote(key)}"
html = self.fetch(search_url, headers=self.headers).text
tree = etree.HTML(html)
vod_list = tree.xpath('//div[contains(@class,"public-list-box")]')
for item in vod_list:
vod = {}
link = item.xpath('.//a[@class="public-list-exp"]/@href')
if not link:
continue
vod["vod_id"] = link[0].split("/id/")[-1].replace(".html", "")
vod["vod_detail_url"] = self.homeUrl + link[0]
name = item.xpath('.//a[@class="time-title"]/text()')
vod["vod_name"] = name[0].strip() if name else ""
pic = item.xpath('.//img/@data-src')
vod["vod_pic"] = pic[0] if pic else ""
result["list"].append(vod)
return result
# 详情页：简介、演员、年份、分集列表
def detailContent(self, ids):
vod_id = ids[0]
detail_url = f"{self.homeUrl}/index.php/vod/detail/id/{vod_id}.html"
html = self.fetch(detail_url, headers=self.headers).text
tree = etree.HTML(html)
vod = {}
vod["vod_id"] = vod_id
# 片名
title = tree.xpath('//h2[@class="player-title-link"]/text()')
vod["vod_name"] = title[0].strip() if title else ""
# 封面
pic = tree.xpath('//div[@class="card-top cf"]//img/@data-src')
vod["vod_pic"] = pic[0] if pic else ""
# 年份、地区、类型
year = tree.xpath('//a[contains(@href,"year")]/text()')
vod["vod_year"] = year[0] if year else ""
area = tree.xpath('//a[contains(@href,"area")]/text()')
vod["vod_area"] = area[0] if area else ""
type_list = tree.xpath('//a[contains(@href,"class")]/text()')
vod["vod_type"] = ",".join(type_list) if type_list else ""
# 演员
actor_list = tree.xpath('//div[@class="card-top cf"]//a[contains(@href,"actor")]/text()')
vod["vod_actor"] = ",".join(actor_list) if actor_list else "未知演员"
# 简介
desc = tree.xpath('//div[@class="card-text"]/text()')
vod["vod_content"] = desc[0].strip() if desc else "暂无简介"
# 分集播放列表
play_list = []
episode_items = tree.xpath('//ul[@class="anthology-list-play"]/li')
for ep in episode_items:
ep_name = ep.xpath('.//span/text()')
ep_link = ep.xpath('.//a/@href')
if ep_name and ep_link:
play_list.append({
"name": ep_name[0].strip(),
"url": self.homeUrl + ep_link[0]
})
vod["vod_play_from"] = ["TX线路"]
vod["vod_play_url"] = ["$".join([f"{item['name']}${item['url']}" for item in play_list])]
return {"list": [vod]}
# 播放页解析真实m3u8（核心，提取iframe内腾讯视频地址）
def playerContent(self, link, vodId, playFrom, pg):
html = self.fetch(link, headers=self.headers).text
# 提取player_aaaa里的url（腾讯原链接）
player_data = re.search(r'var player_aaaa=(\{.*?\})', html, re.S)
real_m3u8 = ""
if player_data:
json_str = player_data.group(1)
data = json.loads(json_str)
real_m3u8 = data.get("url", "")
return {
"parse": 0,
"url": real_m3u8,
"header": self.headers,
"playErr": ""
}
def searchable(self):
return True
def isVideo(self):
return True
jaychouqq/yingshi/py9/搜TV啦.py
+481
View File
@@ -0,0 +1,481 @@
# coding=utf-8
import re
import json
import requests
from urllib.parse import quote, urlparse, parse_qs
from lxml import etree
from base.spider import Spider
class Spider(Spider):
# 这些资源站返回的是第三方平台页面，需要解析器，普通播放器无法直接播放
PARSER_SOURCES = {'QY', 'QQ', 'YK', 'MG', 'BZ', 'RR'}
# 搜索结果中的来源文案黑名单（对应 PARSER_SOURCES 的资源站）
PARSER_REMARKS = {'无广告资源', '爱奇艺资源', '腾讯资源', '优酷资源', '芒果资源', '百赞资源', '人人资源'}
# 该站是搜索引擎，没有真正的分类页。用子类关键词搜索，可得到带直链的结果。
CATEGORY_KEYWORDS = {
"1": ["动作片", "喜剧片", "爱情片", "科幻片", "恐怖片", "剧情片"], # 电影
"2": ["韩剧", "美剧", "日剧", "泰剧", "港剧"], # 电视剧
"3": ["真人秀", "脱口秀", "演唱会"], # 综艺
"4": ["日本动画", "动画电影", "宫崎骏", "奥特曼", "火影忍者"], # 动漫
"26": ["霸道总裁", "赘婿", "穿越短剧", "甜宠短剧"] # 短剧
}
def __init__(self):
self.name = "sotvla"
self.host = "https://www.sotvla.cc"
self.header = {
'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,image/webp,*/*;q=0.8',
'Accept-Language': 'zh-CN,zh;q=0.9',
'Referer': self.host
}
self.session = requests.Session()
self.session.headers.update(self.header)
def getName(self):
return self.name
def init(self, extend=''):
pass
def _get(self, url, params=None):
r = self.session.get(url, params=params, timeout=15)
r.encoding = 'utf-8'
return r.text
def _post(self, url, data=None):
r = self.session.post(url, data=data, timeout=15)
r.encoding = 'utf-8'
return r.text
def _fix_url(self, url):
if not url:
return ''
if url.startswith('//'):
return 'https:' + url
if url.startswith('/'):
return self.host + url
return url
def _parse_text(self, elem):
if elem is None:
return ''
return ''.join(elem.itertext()).strip()
def _parse_pic(self, elem):
pic = ''
if elem is None:
return pic
if elem.tag == 'img':
pic = elem.get('src') or ''
else:
imgs = elem.xpath('.//img')
if imgs:
pic = imgs[0].get('src') or ''
if pic and pic.startswith('data:image'):
pic = ''
return self._fix_url(pic)
def _extract_vod_id(self, href):
"""detail.php?api_id=1&vod_id=88345 -> 1-88345"""
m = re.search(r'[?&]api_id=(\d+)&vod_id=(\d+)', href)
if m:
return f"{m.group(1)}-{m.group(2)}"
return None
def _split_vod_id(self, vod_id):
"""1-88345 -> (1, 88345)"""
if '-' in vod_id:
parts = vod_id.split('-', 1)
if parts[0].isdigit() and parts[1].isdigit():
return parts[0], parts[1]
return None, None
def _is_parser_remark(self, remark):
"""判断搜索结果来源文案是否来自需要解析器的资源站"""
if not remark:
return False
for bad in self.PARSER_REMARKS:
if bad in remark:
return True
return False
def _is_parser_source(self, source_name):
"""判断详情页线路代码是否来自需要解析器的资源站"""
code = source_name.strip().upper()
# 去掉常见后缀，如 "QY高清" -> "QY"
code = re.sub(r'[^A-Z0-9]', '', code)
return code in self.PARSER_SOURCES
def _parse_search_item(self, article):
"""解析搜索结果条目"""
a = article.xpath('.//a[@class="sr-title"]')
if not a:
a = article.xpath('.//a[contains(@href, "detail.php")]')
if not a:
return None
a = a[0]
href = a.get('href', '')
vod_id = self._extract_vod_id(href)
if not vod_id:
return None
vod_name = self._parse_text(a)
# 封面
poster = article.xpath('.//a[@class="sr-poster"]//img')
vod_pic = self._parse_pic(poster[0]) if poster else ''
# 更新日期 / 播放源
remark = ''
src_line = article.xpath('.//div[contains(@class, "sr-source-line")]//span[@class="sr-value"]/text()')
if src_line:
remark = src_line[0].strip()
if not remark:
date_line = article.xpath('.//div[contains(@class, "sr-meta-grid")]//span[@class="sr-value"]/text()')
if date_line:
remark = date_line[-1].strip()
return {
"vod_id": vod_id,
"vod_name": vod_name,
"vod_pic": vod_pic,
"vod_remarks": remark
}
def homeContent(self, filter):
result = {"class": []}
classes = [
{"type_name": "电影", "type_id": "1"},
{"type_name": "电视剧", "type_id": "2"},
{"type_name": "综艺", "type_id": "3"},
{"type_name": "动漫", "type_id": "4"},
{"type_name": "短剧", "type_id": "26"}
]
result["class"] = classes
# 搜索引擎站点，筛选条件仅做展示/学习用途
filters = {}
year_vals = [
{"n": "全部", "v": ""},
{"n": "2026", "v": "2026"},
{"n": "2025", "v": "2025"},
{"n": "2024", "v": "2024"},
{"n": "2023", "v": "2023"},
{"n": "2022", "v": "2022"}
]
for c in classes:
filters[c['type_id']] = [
{"key": "year", "name": "年份", "value": year_vals}
]
result["filters"] = filters
return result
def homeVideoContent(self):
"""首页热播：调用热榜 API，再反查第一条搜索结果获取 vod_id，并过滤掉只有解析源的资源"""
videos = []
try:
url = f"{self.host}/api/hot_movie.php"
html = self._get(url)
data = json.loads(html)
items = data.get('items', [])
for item in items:
try:
title = item.get('title', '').strip()
pic = self._fix_url(item.get('pic', ''))
if not title:
continue
# 通过搜索反查 vod_id
search_url = f"{self.host}/search.php"
params = {"q": title}
search_html = self._get(search_url, params=params)
root = etree.HTML(search_html)
arts = root.xpath('//article[@class="search-result-item"]')
if not arts:
continue
for art in arts:
video = self._parse_search_item(art)
if not video:
continue
# 过滤掉来源文案明显是解析源的结果
if self._is_parser_remark(video.get('vod_remarks', '')):
continue
# 热榜 API 的封面质量更高，优先使用
if pic:
video['vod_pic'] = pic
videos.append(video)
break
except Exception:
pass
except Exception:
pass
return {"list": videos}
def categoryContent(self, tid, pg, filter, extend):
"""分类：用子类关键词搜索，并过滤掉只有解析源（无法直接播放）的资源"""
videos = []
try:
keywords = self.CATEGORY_KEYWORDS.get(str(tid), [])
if not keywords:
return {'list': [], 'page': int(pg), 'pagecount': 0, 'limit': 0, 'total': 0}
# 按页码轮询子类关键词，让不同页显示不同内容
idx = (int(pg) - 1) % len(keywords)
keyword = keywords[idx]
url = f"{self.host}/search.php"
params = {"q": keyword, "page": str(pg)}
html = self._get(url, params=params)
root = etree.HTML(html)
arts = root.xpath('//article[@class="search-result-item"]')
for art in arts:
try:
video = self._parse_search_item(art)
if not video:
continue
# 搜索结果文案是解析源，直接跳过
if self._is_parser_remark(video.get('vod_remarks', '')):
continue
# 进一步校验详情页是否至少有一条可直接播放的线路
detail = self.detailContent([video['vod_id']])
if not detail.get('list'):
continue
d = detail['list'][0]
if not d.get('vod_play_url'):
continue
videos.append(video)
except Exception:
pass
total = 0
total_elem = root.xpath('//span[@id="search-result-total"]/text()')
if total_elem:
total = int(re.sub(r'\D', '', total_elem[0]) or '0')
limit = max(len(videos), 16)
pagecount = (total + limit - 1) // limit if total > 0 else 1
return {
'list': videos,
'page': int(pg),
'pagecount': pagecount,
'limit': limit,
'total': total
}
except Exception:
return {'list': [], 'page': 1, 'pagecount': 0, 'limit': 0, 'total': 0}
def detailContent(self, ids):
try:
vod_id = ids[0]
api_id, vid = self._split_vod_id(vod_id)
if not api_id or not vid:
return {'list': []}
detail_url = f"{self.host}/detail.php?api_id={api_id}&vod_id={vid}"
html = self._get(detail_url)
root = etree.HTML(html)
# 标题
vod_name = ''
h1 = root.xpath('//div[contains(@class, "detail-info")]//h1/text()')
if h1:
vod_name = h1[0].strip()
if not vod_name:
title = root.xpath('//title/text()')
if title:
vod_name = title[0].split('·')[0].strip()
# 封面
vod_pic = ''
poster = root.xpath('//div[contains(@class, "detail-poster")]//img')
if poster:
vod_pic = self._parse_pic(poster[0])
# 年代 / 地区 / 演员 / 导演 / 简介
vod_year = ''
vod_area = ''
vod_actor = ''
vod_director = ''
vod_content = ''
info = root.xpath('//div[contains(@class, "detail-info")]')
if info:
info_text = self._parse_text(info[0])
# 年代
m = re.search(r'年代\s*([0-9]{4})', info_text)
if m:
vod_year = m.group(1)
# 地区
m = re.search(r'地区\s*([^\s\n]+)', info_text)
if m:
vod_area = m.group(1).strip()
# meta-list 中逐行提取
for div in root.xpath('//div[contains(@class, "detail-info")]//div[@class="meta-list"]/div'):
txt = self._parse_text(div)
if txt.startswith('演员'):
vod_actor = txt.replace('演员', '').strip().strip('：').strip(':').strip()
elif txt.startswith('导演'):
vod_director = txt.replace('导演', '').strip().strip('：').strip(':').strip()
blurb = root.xpath('//div[contains(@class, "detail-info")]//p[contains(@class, "blurb")]')
if blurb:
vod_content = self._parse_text(blurb[0])
# 播放源与选集：过滤掉只能走解析器的线路
vod_play_from = []
vod_play_url = []
section = root.xpath('//section[contains(@class, "play-list-section")]')
if section:
sec = section[0]
# 线路按钮
tabs = sec.xpath('.//button[contains(@class, "play-source-tab")]')
# 选集面板
panels = sec.xpath('.//div[contains(@class, "play-ep-panel")]')
for idx, tab in enumerate(tabs):
source_name = self._parse_text(tab)
source_name = re.sub(r'\s+', ' ', source_name).strip()
if not source_name:
source_name = f"线路{idx + 1}"
# 跳过爱奇艺/腾讯/优酷/芒果/百赞/人人等解析源
if self._is_parser_source(source_name):
continue
panel = None
for p in panels:
if p.get('data-src-panel') == str(idx):
panel = p
break
if panel is None and idx < len(panels):
panel = panels[idx]
if panel is None:
continue
links = panel.xpath('.//a[contains(@href, "play.php")]')
play_list = []
for a in links:
ep_name = self._parse_text(a)
href = a.get('href', '')
if not ep_name or not href:
continue
play_url = self._fix_url(href)
play_list.append(f"{ep_name}${play_url}")
if play_list:
vod_play_from.append(source_name)
vod_play_url.append("#".join(play_list))
if vod_play_from:
vod_play_from_str = "$$$".join(vod_play_from)
vod_play_url_str = "$$$".join(vod_play_url)
else:
vod_play_from_str = ""
vod_play_url_str = ""
detail = {
"vod_id": vod_id,
"vod_name": vod_name,
"vod_pic": vod_pic,
"vod_year": vod_year,
"vod_area": vod_area,
"vod_actor": vod_actor,
"vod_director": vod_director,
"vod_content": vod_content,
"vod_play_from": vod_play_from_str,
"vod_play_url": vod_play_url_str
}
return {'list': [detail]}
except Exception:
return {'list': []}
def playerContent(self, flag, id, vipFlags):
"""播放：该站为聚合搜索，播放页 iframe 嵌入第三方解析。
若 iframe 参数里携带 m3u8/mp4 直链则直接播放，否则返回解析地址。"""
try:
html = self._get(id)
real_url = None
# 1. 优先提取 JS 里的 iframe 地址
m = re.search(r'var\s+embedAutoOn\s*=\s*"([^"]+)"', html)
if m:
real_url = m.group(1).encode('utf-8').decode('unicode_escape')
real_url = real_url.replace('\\/', '/')
else:
# 2. 兼容 iframe 标签
iframe_match = re.search(r'<iframe[^>]+src\s*=\s*"([^"]+)"', html, re.I)
if iframe_match:
real_url = iframe_match.group(1)
if real_url:
real_url = self._fix_url(real_url)
# 尝试从 iframe 参数中提取直链
parsed = urlparse(real_url)
qs = parse_qs(parsed.query)
direct_url = ''
for k in ['url', 'v', 'src']:
if k in qs:
for v in qs[k]:
if v and v.strip():
direct_url = v.strip()
# 如果参数里就是直链视频地址，直接播放
if direct_url and self.isVideoFormat(direct_url):
return {"parse": 0, "playUrl": "", "url": direct_url, "header": json.dumps(self.header)}
return {"parse": 1, "playUrl": "", "url": real_url, "header": json.dumps(self.header)}
# 兜底：返回播放页地址让播放器自行解析
return {"parse": 1, "playUrl": "", "url": id, "header": json.dumps(self.header)}
except Exception:
return {"parse": 0, "playUrl": "", "url": ""}
def searchContent(self, key, quick, pg='1'):
videos = []
try:
url = f"{self.host}/search.php"
params = {"q": key, "page": str(pg)}
html = self._get(url, params=params)
root = etree.HTML(html)
arts = root.xpath('//article[@class="search-result-item"]')
for art in arts:
try:
video = self._parse_search_item(art)
if not video:
continue
# 根据搜索结果来源文案过滤明显无法直接播放的解析源
if self._is_parser_remark(video.get('vod_remarks', '')):
continue
videos.append(video)
except Exception:
pass
total = 0
total_elem = root.xpath('//span[@id="search-result-total"]/text()')
if total_elem:
total = int(re.sub(r'\D', '', total_elem[0]) or '0')
limit = max(len(videos), 16)
pagecount = (total + limit - 1) // limit if total > 0 else 1
return {
'list': videos,
'page': int(pg),
'pagecount': pagecount,
'limit': limit,
'total': total
}
except Exception:
return {'list': [], 'page': 1, 'pagecount': 0, 'limit': 0, 'total': 0}
def isVideoFormat(self, url):
return any(url.lower().endswith(fmt) for fmt in ['.m3u8', '.mp4', '.flv', '.ts'])
def manualVideoCheck(self):
pass
def localProxy(self, params):
return None
def destroy(self):
pass
jaychouqq/yingshi/py9/聚影网盘 (1).py
+230
View File
@@ -0,0 +1,230 @@
import re
import json
import base64
import requests
from urllib.parse import quote, unquote
from base.spider import Spider as BaseSpider
class Spider(BaseSpider):
def getName(self):
return "聚影网盘"
def init(self, extend=""):
self.ext=self._json(extend); self.host=str(self.ext.get("host") or "https://www.jying.top").rstrip("/"); self.username=str(self.ext.get("username") or ""); self.password=str(self.ext.get("password") or ""); self.token=str(self.ext.get("token") or ""); self.cookie=str(self.ext.get("cookie") or ""); self.timeout=int(self.ext.get("timeout") or 15); self.limit=int(self.ext.get("limit") or 24); self.cache={}; self.session=requests.Session(); self.session.verify=False; self.session.headers.update({"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","Accept":"application/json, text/plain, */*","Content-Type":"application/json","X-Requested-With":"XMLHttpRequest","Referer":self.host+"/login"})
if self.cookie: self.session.headers.update({"Cookie":self.cookie})
if self.token: self.session.headers.update({"X-App-User-Token":self.token})
self._login(); return {}
def homeContent(self, filter):
d=self._overview(); cls=[{"type_id":"all","type_name":"影视库"},{"type_id":"movie","type_name":"电影"},{"type_id":"tv","type_name":"电视剧"},{"type_id":"anime","type_name":"动漫"},{"type_id":"documentary","type_name":"纪录片"},{"type_id":"other","type_name":"其他"}]
return {"class":cls,"filters":self._filters(d) if filter else {}}
def homeVideoContent(self):
d=self._api("/api/app/home-initial-data/",{"page_size":12}); arr=[]
for k in ["default_movies","hero_candidates"]: arr+=d.get(k) or []
for v in (d.get("featured_sections") or {}).values(): arr+=v or []
return {"list":[self._vod(x) for x in self._dedupe(arr)[:24]]}
def categoryContent(self, tid, pg, filter, extend):
page=self._page(pg); tid=str(tid or "all"); ext=self._json(extend); typ,data=self._pid(tid)
if typ=="group": return self._group_page(data,page)
if typ=="movie": return self._movie_res_page(data,page)
p={"page":page,"page_size":self.limit,"count":1,"ordering":ext.get("ordering") or ext.get("sort") or "-created_at"}
if tid not in ["all",""]: p["type"]=tid
for k in ["category","region","tag","year","resource_type"]:
if ext.get(k): p[k]=ext.get(k)
d=self._api("/api/app/movies/",p); arr=d.get("results") or []
return {"list":[self._vod(x) for x in arr],"page":page,"pagecount":int(d.get("total_pages") or page+(1 if arr else 0)),"limit":self.limit,"total":int(d.get("total_count") or d.get("count") or len(arr))}
def detailContent(self, ids):
vid=ids[0] if isinstance(ids,list) and ids else ids; typ,data=self._pid(str(vid or ""))
if typ=="res": return self._detail_res(data)
mid=data.get("id") if typ in ["movie","group"] else vid
d=self._detail(mid); m=d.get("data") or d.get("movie") or d
res=self._resources(mid); groups={}
for x in res: groups.setdefault(self._rtype(x),[]).append(x)
froms=[]; urls=[]
for k in self._order(groups):
arr=groups.get(k) or []
if arr: froms.append(self._rname(k,arr[0])); urls.append("#".join([self._clean(x.get("title") or x.get("resource_description") or self._rname(k,x))+"$"+self._eid("res",x) for x in arr]))
if not froms:
froms=["提示"]; urls=["暂无资源$__EMPTY__"]
vod={"vod_id":str(mid),"vod_name":m.get("title") or "资源","vod_pic":m.get("cover") or m.get("cover_url") or "","vod_remarks":self._remarks(m),"type_name":m.get("movie_type_display") or m.get("category_name") or "","vod_year":str(m.get("release_year") or m.get("year") or ""),"vod_area":m.get("region") or "","vod_actor":m.get("actors") or "","vod_director":m.get("director") or "","vod_content":m.get("description") or "","vod_play_from":"$$$".join(froms),"vod_play_url":"$$$".join(urls)}
return {"list":[vod]}
def searchContent(self, key, quick=False, pg="1"):
return self.searchContentPage(key,quick,pg)
def searchContentPage(self, key, quick=False, pg="1"):
page=self._page(pg); kw=str(key or "").strip()
if not kw: return {"list":[],"page":page,"pagecount":1,"limit":self.limit,"total":0}
d=self._api("/api/app/movies/",{"q":kw,"page":page,"page_size":self.limit,"count":1,"exact":1}); arr=d.get("results") or []
return {"list":[self._vod(x) for x in arr],"page":page,"pagecount":int(d.get("total_pages") or 1),"limit":self.limit,"total":int(d.get("total_count") or d.get("count") or len(arr))}
def playerContent(self, flag, id, vipFlags):
if str(id or "") in ["__EMPTY__",""]: return {"parse":0,"playUrl":"","url":"","header":self._h()}
typ,data=self._pid(id)
if typ=="res":
url=self._access(data); url=unquote(str(url or "")).replace("&amp;","&").strip()
if url.lower().startswith("magnet:?") or url.lower().startswith("ed2k://"): return {"parse":0,"playUrl":"","url":url,"header":self._h()}
if re.search(r"\.(m3u8|mp4|flv|mkv|ts)(\?|$)",url,re.I): return {"parse":0,"playUrl":"","url":url,"header":self._h()}
return {"parse":0,"playUrl":"","url":"push://"+url if url.startswith("http") else url,"header":self._h()}
return {"parse":0,"playUrl":"","url":id,"header":self._h()}
def localProxy(self, params):
return None
def isVideoFormat(self, url):
return bool(re.search(r"\.(m3u8|mp4|flv|mkv|ts)(\?|$)",str(url or ""),re.I))
def manualVideoCheck(self):
return False
def destroy(self):
return ""
def _login(self):
if self.token: return True
if not self.username or not self.password: return False
try:
self.session.get(self.host+"/api/csrf/",timeout=self.timeout); csrf=self.session.cookies.get("csrftoken") or ""; h=self._h(); h.update({"X-CSRFToken":csrf,"Origin":self.host,"Referer":self.host+"/login"}); r=self.session.post(self.host+"/api/app/login/",headers=h,json={"username":self.username,"password":self.password},timeout=self.timeout).json(); self.token=r.get("token") or ""; self.session.headers.update({"X-App-User-Token":self.token}) if self.token else None; return bool(self.token)
except Exception: return False
def _api(self,path,params=None,post=None):
key=("p" if post is not None else "g")+path+json.dumps(params or {},ensure_ascii=False,sort_keys=True)+json.dumps(post or {},ensure_ascii=False,sort_keys=True)
if key in self.cache: return self.cache[key]
try:
url=self.host+path; h=self._h(); h.update({"Referer":self.host+"/categories"}); r=self.session.post(url,headers=h,json=post or {},timeout=self.timeout) if post is not None else self.session.get(url,headers=h,params=params or {},timeout=self.timeout); d=r.json() if r.text else {}
except Exception: d={}
self.cache[key]=d if isinstance(d,dict) else {}; return self.cache[key]
def _overview(self):
return self._api("/api/app/categories/")
def _detail(self,mid):
return self._api("/api/app/movie/%s/detail/"%quote(str(mid)))
def _resources(self,mid):
key="res|"+str(mid)
if key in self.cache: return self.cache[key]
arr=[]; page=1
while page<=10:
d=self._api("/api/app/movie/%s/resources/"%quote(str(mid)),{"page":page,"page_size":120}); a=d.get("resources") or []
if not a: break
arr+=a
if not d.get("has_more"): break
page+=1
self.cache[key]=self._dedupe(arr); return self.cache[key]
def _access(self,x):
if x.get("target"): return x.get("target")
u=x.get("share_link") or x.get("raw_share_link") or x.get("share_link_with_code") or ""
if u: return u
rid=x.get("id"); ticket=x.get("access_ticket") or ""
if not rid or not ticket: return ""
d=self._api("/api/app/resource/%s/access/"%quote(str(rid)),post={"access_ticket":str(ticket)})
target=d.get("target") or d.get("share_link") or ""; code=d.get("access_code") or d.get("extraction_code") or ""
return (target+(" 提取码:"+code if code and code not in target else "")).strip()
def _group_page(self,data,page):
arr=[x for x in self._resources(data.get("id")) if self._rtype(x)==data.get("group")]; total=len(arr); pc=max(1,(total+self.limit-1)//self.limit); page=max(1,min(page,pc)); s=(page-1)*self.limit
return {"list":[{"vod_id":self._eid("res",x),"vod_name":self._clean(x.get("title") or x.get("resource_description") or self._rname(self._rtype(x),x)),"vod_pic":data.get("pic") or "","vod_remarks":self._rname(self._rtype(x),x),"vod_tag":"file","vod_content":x.get("description") or x.get("resource_description") or ""} for x in arr[s:s+self.limit]],"page":page,"pagecount":pc,"limit":self.limit,"total":total}
def _movie_res_page(self,data,page):
groups={}
for x in self._resources(data.get("id")): groups.setdefault(self._rtype(x),[]).append(x)
lst=[{"vod_id":self._eid("group",{"id":data.get("id"),"group":k,"pic":data.get("pic") or ""}),"vod_name":self._rname(k,(groups.get(k) or [{}])[0]),"vod_pic":data.get("pic") or "","vod_remarks":"%s条资源"%len(groups.get(k) or []),"vod_tag":"folder"} for k in self._order(groups)]
return {"list":lst,"page":1,"pagecount":1,"limit":len(lst),"total":len(lst)}
def _detail_res(self,x):
name=self._clean(x.get("title") or x.get("resource_description") or self._rname(self._rtype(x),x)); pic=x.get("pic") or ""; return {"list":[{"vod_id":str(x.get("id") or "res"),"vod_name":name,"vod_pic":pic,"vod_remarks":self._rname(self._rtype(x),x),"vod_content":x.get("description") or x.get("resource_description") or "","vod_play_from":self._rname(self._rtype(x),x),"vod_play_url":name+"$"+self._eid("res",x)}]}
def _filters(self,d):
years=[{"n":"全部","v":""}]+[{"n":str(y),"v":str(y)} for y in range(2028,1989,-1)]
types=[{"n":"全部","v":""},{"n":"电影","v":"movie"},{"n":"电视剧","v":"tv"},{"n":"动漫","v":"anime"},{"n":"纪录片","v":"documentary"},{"n":"其他","v":"other"}]
regions=[{"n":"全部","v":""}]+self._vals(d.get("region_stats") or [],"name","key")
tags=[{"n":"全部","v":""}]+self._vals(d.get("tags") or [],"name","name")
cats=[{"n":"全部","v":""}]+self._vals(d.get("categories") or [],"name","slug")
disks=[{"n":"全部","v":""}]+self._vals(d.get("resource_type_stats") or [],"name","key")
if len(disks)==1: disks+=[{"n":"百度网盘","v":"baidu"},{"n":"115网盘","v":"115"},{"n":"123云盘","v":"123"},{"n":"迅雷云盘","v":"xunlei"},{"n":"夸克网盘","v":"quark"},{"n":"阿里云盘","v":"aliyun"},{"n":"磁力","v":"magnet"}]
sort=[{"n":"最新","v":"-created_at"},{"n":"最多资源","v":"-resource_count"},{"n":"年份新","v":"-release_year"},{"n":"热度","v":"-views"}]
f=[{"key":"type","name":"类型","value":types},{"key":"region","name":"国家","value":regions[:80]},{"key":"year","name":"年份","value":years},{"key":"tag","name":"标签","value":tags[:120]},{"key":"category","name":"分类","value":cats[:120]},{"key":"resource_type","name":"网盘","value":disks[:80]},{"key":"ordering","name":"排序","value":sort}]
return {k:f for k in ["all","movie","tv","anime","documentary","other"]}
def _vals(self,arr,nk,vk):
out=[]; seen=set()
for x in arr:
n=str(x.get(nk) or x.get("display_name") or x.get("label") or x.get("resource_type_display") or x.get(vk) or "").strip(); v=str(x.get(vk) or x.get("slug") or x.get("key") or x.get("resource_type") or n).strip()
if n and v and v not in seen: seen.add(v); out.append({"n":n,"v":v})
return out
def _vod(self,x):
return {"vod_id":self._eid("movie",{"id":x.get("id"),"pic":x.get("cover") or x.get("cover_url") or ""}),"vod_name":x.get("title") or "资源","vod_pic":x.get("cover") or x.get("cover_url") or "","vod_remarks":self._remarks(x),"vod_content":x.get("description") or "","vod_tag":"folder"}
def _remarks(self,x):
a=[]
for k in ["movie_type_display","release_year","year","category_name","region"]:
if x.get(k) and str(x.get(k)) not in a: a.append(str(x.get(k)))
if x.get("resource_count") is not None: a.append("%s源"%x.get("resource_count"))
return " / ".join(a[:4])
def _rtype(self,x):
t=str(x.get("resource_type") or "other").lower()
if t=="magnetlink": return "magnet"
if "123" in t: return "123"
if "baidu" in t: return "baidu"
if "xunlei" in t: return "xunlei"
if "quark" in t: return "quark"
if "aliyun" in t or "alipan" in t: return "aliyun"
if "115" in t: return "115"
if "uc"==t: return "uc"
if "ed2k" in t: return "ed2k"
return t or "other"
def _rname(self,k,x=None):
return (x or {}).get("resource_type_display") or {"123":"123云盘","baidu":"百度网盘","xunlei":"迅雷云盘","quark":"夸克网盘","aliyun":"阿里云盘","115":"115网盘","uc":"UC网盘","magnet":"磁力","ed2k":"电驴","other":"其他"}.get(k,k)
def _order(self,groups):
keys=["baidu","115","123","xunlei","quark","aliyun","uc","magnet","ed2k","other"]
return [x for x in keys if groups.get(x)]+[x for x in groups.keys() if x not in keys]
def _h(self):
h={"User-Agent":self.session.headers.get("User-Agent","Mozilla/5.0"),"Accept":"application/json, text/plain, */*","Content-Type":"application/json","X-Requested-With":"XMLHttpRequest"}
if self.token: h["X-App-User-Token"]=self.token
csrf=self.session.cookies.get("csrftoken")
if csrf: h["X-CSRFToken"]=csrf
return h
def _eid(self,typ,data):
return "jy|%s|%s"%(typ,base64.urlsafe_b64encode(json.dumps(data or {},ensure_ascii=False,separators=(",",":")).encode()).decode().rstrip("="))
def _pid(self,s):
a=str(s or "").split("|",2)
if len(a)>=3 and a[0]=="jy":
try: return a[1],json.loads(base64.urlsafe_b64decode((a[2]+"="*(-len(a[2])%4)).encode()).decode())
except Exception: return "",{}
return "",{}
def _json(self,s):
if isinstance(s,dict): return s
for x in [str(s or "")]:
try: return json.loads(x) if x.strip().startswith("{") else {}
except Exception: pass
try: y=unquote(x); return json.loads(y) if y.strip().startswith("{") else {}
except Exception: pass
try: y=base64.b64decode(x+"="*(-len(x)%4)).decode(); return json.loads(y) if y.strip().startswith("{") else {}
except Exception: pass
return {}
def _page(self,pg):
return int(pg) if str(pg).isdigit() and int(pg)>0 else 1
def _clean(self,s):
return re.sub(r"\s+"," ",str(s or "资源")).replace("$","＄").replace("#","＃").strip()[:120]
def _dedupe(self,arr):
out=[]; seen=set()
for x in arr or []:
k=str(x.get("id") or x.get("share_link") or x.get("title") or x)
if k and k not in seen: seen.add(k); out.append(x)
return out
jaychouqq/yingshi/py9/🙈heiliao🔞[密].py
+289
View File
@@ -0,0 +1,289 @@
#!/usr/bin/python
# -*- coding: utf-8 -*-
import sys, json, re, requests
from urllib.parse import quote, unquote
sys.path.append('..')
from base.spider import Spider as BaseSpider
class Spider(BaseSpider):
def getName(self):
return "黑料网"
def isVideoFormat(self, url):
if not url:
return False
if url.startswith(('novel://', 'text://', 'pics://', 'book_', 'comic_')):
return False
return '.mp4' in url or '.m3u8' in url or '.ts' in url
def manualVideoCheck(self):
return False
def destroy(self):
pass
headers = {
'User-Agent': 'Mozilla/5.0 (iPhone; CPU iPhone OS 16_6 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/16.6 Mobile/15E148 Safari/604.1',
'Referer': 'https://heiliao.com/',
}
host = 'https://heiliao.com'
cat_map = {
'hlcg': '最新黑料', 'jrrs': '今日热瓜', 'jqrm': '热门黑料', 'lsdg': '经典黑料',
'xycg': '校园黑料', 'whhl': '网红黑料', 'fczq': '反差专区', 'ycsq': '原创社区',
'mxcw': '明星丑闻', 'mrds': '每日大赛', 'qqqw': '全球奇闻', 'ttsq': '推特社区',
'ysdj': '影视短剧', 'whhj': '网黄合集', 'shxw': '社会新闻', 'thzq': '探花专区',
'cpcd': '厕拍抄底', 'yqby': '有求必应', 'syzy': '深夜综艺', 'djbl': '独家爆料',
'jqxs': '黑料小说', 'gchl': '官场爆料', 'hlkt': '黑料课堂', 'hlbg': '黑料爆改',
'ttzz': '桃图杂志', 'mrrb': '日榜黑料', 'zbjx': '周榜精选', 'ybrg': '月榜热瓜',
}
ad_ids = {'39668', '8148', '8147', '8150', '8146', '38757', '41525', '109358', '109356'}
ad_keywords = ('黑料网最新入口', '黑料网海外主站', '黑料APP', '获取最新地址', '发送任意内容至')
def init(self, extend=""):
self.session = requests.Session()
def _get(self, url):
try:
r = self.session.get(url, headers=self.headers, timeout=20)
r.encoding = r.apparent_encoding or 'utf-8'
return r.text
except:
return ''
def _clean_pic(self, pic):
if not pic:
return ''
if pic.startswith('http'):
abs_url = pic
elif pic.startswith('/'):
abs_url = f'{self.host}{pic}'
else:
abs_url = f'{self.host}/{pic}'
return f"{self.getProxyUrl()}&url={quote(abs_url, safe='')}"
def _parse_list(self, html):
items = []
if not html:
return items
pat = r'<div[^>]*class="video-item"[^>]*>(.*?)(?=<div[^>]*class="video-item"[^>]*>|<div[^>]*class="[^"]*page[^"]*"|<div[^>]*class="[^"]*pagination[^"]*"|$)'
for block in re.finditer(pat, html, re.S):
b = block.group(1)
pid_m = re.search(r'archives/(\d+)/', b)
if not pid_m:
continue
pid = pid_m.group(1)
if pid in self.ad_ids:
continue
pic_m = re.search(r'z-image-loader-url=["\']([^"\']+)["\']', b)
pic = pic_m.group(1).strip() if pic_m else ''
if not pic:
continue
alt_m = re.search(r'alt=["\']([^"\']+)["\']', b)
title = alt_m.group(1).strip() if alt_m else ''
if not title:
continue
items.append({"vod_id": pid, "vod_name": title, "vod_pic": self._clean_pic(pic)})
return items
def _parse_pagecount(self, html):
if not html:
return 1
nums = re.findall(r'/page/(\d+)/', html)
return max(int(n) for n in nums) if nums else 1
def homeContent(self, filter):
try:
cats = [{'type_id': k, 'type_name': v} for k, v in self.cat_map.items()]
return {"class": cats, "list": [], "filters": {}}
except:
return {"class": [], "filters": {}, "list": [], "page": 1, "pagecount": 1}
def homeVideoContent(self):
try:
html = self._get(self.host)
return {"list": self._parse_list(html)[:20]}
except:
return {"list": []}
def categoryContent(self, tid, pg, filter, extend):
try:
page = int(pg) if pg else 1
url = f"{self.host}/{tid}/" if page == 1 else f"{self.host}/{tid}/page/{page}/"
html = self._get(url)
return {"page": page, "pagecount": self._parse_pagecount(html), "list": self._parse_list(html)}
except:
return {"page": int(pg) if pg else 1, "pagecount": 1, "list": []}
def detailContent(self, ids):
try:
vid = ids[0]
html = self._get(f"{self.host}/archives/{vid}")
if not html:
return {"list": []}
title_m = re.search(r'<h1[^>]*class="[^"]*detail-title[^"]*"[^>]*>(.*?)</h1>', html, re.S)
title = re.sub(r'<[^>]+>', '', title_m.group(1)).strip() if title_m else f"黑料{vid}"
cs = re.search(r'<div[^>]*class="[^"]*editormd-preview[^"]*"[^>]*>(.*?)(?=<div[^>]*class="[^"]*article-tags|<div[^>]*class="[^"]*article-meta|</article|$)', html, re.S)
content = cs.group(1) if cs else ''
img_urls = []
for m in re.finditer(r'z-image-loader-url=["\']([^"\']+)["\']', content):
u = m.group(1).strip()
if not u or not u.startswith('http'):
continue
if 'pic.uforxk.cn' not in u and 'upload_01' not in u:
continue
am = re.search(r'alt=["\']([^"\']*)["\']', content[m.end():m.end()+200])
alt = am.group(1).strip() if am else ''
if alt and len(alt) >= 20 and re.fullmatch(r'[0-9a-fA-F]+', alt):
continue
img_urls.append(self._clean_pic(u))
video_eps_main = []
video_eps_backup = []
cover_pic = ''
dp_idx = 0
for dp in re.finditer(r"<div[^>]*class=\"[^\"]*dplayer[^\"]*\"[^>]*config='([^']*)'", html):
cfg_str = dp.group(1).replace('&quot;', '"').replace('&amp;', '&')
try:
cfg = json.loads(cfg_str)
v = cfg.get('video', {})
if dp_idx == 0 and v.get('pic'):
cover_pic = self._clean_pic(v.get('pic'))
urls = v.get('urls', [])
ep_name = f"第{dp_idx+1}集"
ep_raw_title = v.get('title', '') or ''
m_ep = re.search(r'(\d+)$', ep_raw_title)
if m_ep:
ep_name = f"第{int(m_ep.group(1))}集"
if urls:
video_eps_main.append(f"{ep_name}${urls[0].get('url', '')}")
if len(urls) > 1:
video_eps_backup.append(f"{ep_name}${urls[1].get('url', '')}")
elif v.get('url'):
video_eps_main.append(f"{ep_name}${v.get('url')}")
dp_idx += 1
except:
continue
raw_text = re.sub(r'<[^>]+>', '\n', content)
raw_text = re.sub(r'\n{3,}', '\n\n', raw_text).strip()
text_parts = []
for line in raw_text.split('\n'):
line = line.strip()
if not line:
continue
if any(kw in line for kw in self.ad_keywords):
continue
if line.startswith('黑料网') and '最新入口' in line:
continue
if '海外主站' in line or '中转' in line:
continue
text_parts.append(line)
full_text = '\n'.join(text_parts)
pic = cover_pic or (img_urls[0] if img_urls else '')
from_names = []
ep_parts = []
if video_eps_main:
from_names.append('视频')
ep_parts.append('#'.join(video_eps_main))
if video_eps_backup:
from_names.append('备用')
ep_parts.append('#'.join(video_eps_backup))
if img_urls:
from_names.append('图文')
ep_parts.append(f'图片$pics://{"&&".join(img_urls)}')
if from_names:
vod = {
"vod_id": vid, "vod_name": title, "vod_remarks": "",
"vod_pic": pic, "vod_content": full_text[:500],
"vod_play_from": "$$$".join(from_names),
"vod_play_url": "$$$".join(ep_parts),
}
else:
short = full_text[:300]
vod = {
"vod_id": vid, "vod_name": title, "vod_remarks": "",
"vod_pic": pic, "vod_content": full_text[:500],
"vod_play_from": "文字",
"vod_play_url": f"阅读${short}",
}
return {"list": [vod]}
except:
return {"list": []}
def searchContent(self, key, quick, pg="1"):
try:
page = int(pg) if pg else 1
kw = quote(key)
url = f"{self.host}/?s={kw}" if page == 1 else f"{self.host}/page/{page}/?s={kw}"
html = self._get(url)
return {"list": self._parse_list(html), "page": page}
except:
return {"list": [], "page": int(pg) if pg else 1}
def playerContent(self, flag, id, vipFlags):
try:
if flag == '视频' or flag == '备用':
return {"parse": 0, "url": id, "header": self.headers, "position": "0"}
if flag == '图文' or id.startswith('pics://'):
return {"parse": 0, "playUrl": "", "url": id.replace('图片', ''), "header": self.headers, "position": "0"}
if flag == '文字' or '阅读' in id:
content = id.replace('阅读', '')
if '$$$' in content:
content = content.split('$$$')[-1]
nj = json.dumps({"title": content[:50], "content": content}, ensure_ascii=False)
return {"parse": 0, "url": f"novel://{nj}", "header": "", "vod_player": "书", "position": "0"}
if id.startswith('pics://'):
return {"parse": 0, "playUrl": "", "url": id, "header": self.headers, "position": "0"}
if id.startswith('http'):
return {"parse": 0, "url": id, "header": self.headers, "position": "0"}
return {"parse": 0, "url": id, "header": self.headers, "position": "0"}
except:
return {"parse": 0, "url": "", "position": "0"}
def _img_decrypt(self, data):
try:
from Crypto.Cipher import AES
from Crypto.Util.Padding import unpad
key = ''.join(chr(int(c)) for c in '102_53_100_57_54_53_100_102_55_53_51_51_54_50_55_48'.split('_')).encode('utf-8')
iv = ''.join(chr(int(c)) for c in '57_55_98_54_48_51_57_52_97_98_99_50_102_98_101_49'.split('_')).encode('utf-8')
cipher = AES.new(key, AES.MODE_CBC, iv)
dec = cipher.decrypt(data)
dec = unpad(dec, AES.block_size)
return dec
except:
return data
def localProxy(self, param):
try:
url = param.get('url', '')
if not url:
return [404, 'text/plain', b'not found']
url = unquote(url)
r = self.session.get(url, headers={'User-Agent': self.headers['User-Agent'], 'Referer': self.host + '/'}, timeout=15, verify=False)
data = r.content
if not data:
return [404, 'text/plain', b'not found']
dec = self._img_decrypt(data)
if dec[:2] == b'\xff\xd8':
data, ct = dec, 'image/jpeg'
elif dec[:4] == b'\x89PNG':
data, ct = dec, 'image/png'
elif dec[:4] == b'RIFF' and dec[8:12] == b'WEBP':
data, ct = dec, 'image/webp'
elif data[:2] == b'\xff\xd8':
ct = 'image/jpeg'
elif data[:4] == b'\x89PNG':
ct = 'image/png'
elif data[:4] == b'RIFF' and data[8:12] == b'WEBP':
ct = 'image/webp'
else:
ct = r.headers.get('Content-Type', 'image/jpeg')
return [200, ct, data, {'Content-Length': str(len(data))}]
except:
return [404, 'text/plain', b'not found']