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-02 17:50:05 +00:00
parent 484b1d8d12
commit 61a3dffcb0
13 changed files with 4745 additions and 3322 deletions
Download Patch File Download Diff File Expand all files Collapse all files
FGBLH/py/MiFun动漫.py
+248
View File
@@ -0,0 +1,248 @@
import re
import sys
import threading
import requests
from Crypto.Hash import MD5
sys.path.append("..")
from Crypto.Cipher import AES
from Crypto.Util.Padding import pad, unpad
from urllib.parse import quote, urlparse
from base64 import b64encode, b64decode
import json
import time
from base.spider import Spider
class Spider(Spider):
def init(self, extend=""):
self.host = self.gethost()
self.did=self.getdid()
pass
def isVideoFormat(self, url):
pass
def manualVideoCheck(self):
pass
def action(self, action):
pass
def destroy(self):
pass
def homeContent(self, filter):
data = self.getdata("/api.php/getappapi.index/initV119")
dy = {"class": "类型", "area": "地区", "lang": "语言", "year": "年份", "letter": "字母", "by": "排序",
"sort": "排序"}
filters = {}
classes = []
json_data = data["type_list"]
homedata = data["banner_list"][8:]
for item in json_data:
if item["type_name"] == "全部":
continue
has_non_empty_field = False
jsontype_extend = json.loads(item["type_extend"])
homedata.extend(item["recommend_list"])
jsontype_extend["sort"] = "最新,最热,最赞"
classes.append({"type_name": item["type_name"], "type_id": item["type_id"]})
for key in dy:
if key in jsontype_extend and jsontype_extend[key].strip() != "":
has_non_empty_field = True
break
if has_non_empty_field:
filters[str(item["type_id"])] = []
for dkey in jsontype_extend:
if dkey in dy and jsontype_extend[dkey].strip() != "":
values = jsontype_extend[dkey].split(",")
value_array = [{"n": value.strip(), "v": value.strip()} for value in values if
value.strip() != ""]
filters[str(item["type_id"])].append({"key": dkey, "name": dy[dkey], "value": value_array})
result = {}
result["class"] = classes
result["filters"] = filters
result["list"] = homedata[1:]
return result
def homeVideoContent(self):
pass
def categoryContent(self, tid, pg, filter, extend):
body = {"area": extend.get('area', '全部'), "year": extend.get('year', '全部'), "type_id": tid, "page": pg,
"sort": extend.get('sort', '最新'), "lang": extend.get('lang', '全部'),
"class": extend.get('class', '全部')}
result = {}
data = self.getdata("/api.php/getappapi.index/typeFilterVodList", body)
result["list"] = data["recommend_list"]
result["page"] = pg
result["pagecount"] = 9999
result["limit"] = 90
result["total"] = 999999
return result
def detailContent(self, ids):
body = f"vod_id={ids[0]}"
data = self.getdata("/api.php/getappapi.index/vodDetail", body)
vod = data["vod"]
play = []
names = []
for itt in data["vod_play_list"]:
a = []
names.append(itt["player_info"]["show"])
for it in itt['urls']:
it['user_agent'] = itt["player_info"].get("user_agent")
it["parse"] = itt["player_info"].get("parse")
a.append(f"{it['name']}${self.e64(json.dumps(it))}")
play.append("#".join(a))
vod["vod_play_from"] = "$$$".join(names)
vod["vod_play_url"] = "$$$".join(play)
result = {"list": [vod]}
return result
def searchContent(self, key, quick, pg="1"):
body = f"keywords={key}&type_id=0&page={pg}"
data = self.getdata("/api.php/getappapi.index/searchList", body)
result = {"list": data["search_list"], "page": pg}
return result
def playerContent(self, flag, id, vipFlags):
ids = json.loads(self.d64(id))
h = {"User-Agent": (ids['user_agent'] or "okhttp/3.14.9")}
try:
if re.search(r'url=', ids['parse_api_url']):
data = self.fetch(ids['parse_api_url'], headers=h, timeout=10).json()
url = data.get('url') or data['data'].get('url')
else:
body = f"parse_api={ids.get('parse') or ids['parse_api_url'].replace(ids['url'], '')}&url={quote(self.aes(ids['url'], True))}&token={ids.get('token')}"
b = self.getdata("/api.php/getappapi.index/vodParse", body)['json']
url = json.loads(b)['url']
if 'error' in url: raise ValueError(f"解析失败: {url}")
p = 0
except Exception as e:
print('错误信息：', e)
url, p = ids['url'], 1
if re.search(r'\.jpg|\.png|\.jpeg', url):
url = self.Mproxy(url)
result = {}
result["parse"] = p
result["url"] = url
result["header"] = h
return result
def localProxy(self, param):
return self.Mlocal(param)
def gethost(self):
headers = {
'User-Agent': 'okhttp/3.14.9'
}
response = self.fetch('https://miget-1313189639.cos.ap-guangzhou.myqcloud.com/mifun.txt',headers=headers).text
return self.host_late(response.split('\n'))
def host_late(self, url_list):
if isinstance(url_list, str):
urls = [u.strip() for u in url_list.split(',')]
else:
urls = url_list
if len(urls) <= 1:
return urls[0] if urls else ''
results = {}
threads = []
def test_host(url):
try:
url = url.strip()
start_time = time.time()
response = requests.head(url, timeout=1.0, allow_redirects=False)
delay = (time.time() - start_time) * 1000
results[url] = delay
except Exception as e:
results[url] = float('inf')
for url in urls:
t = threading.Thread(target=test_host, args=(url,))
threads.append(t)
t.start()
for t in threads:
t.join()
return min(results.items(), key=lambda x: x[1])[0]
def getdid(self):
did=self.getCache('did')
if not did:
t = str(int(time.time()))
did = self.md5(t)
self.setCache('did', did)
return did
def aes(self, text, b=None):
key = b"GETMIFUNGEIMIFUN"
cipher = AES.new(key, AES.MODE_CBC, key)
if b:
ct_bytes = cipher.encrypt(pad(text.encode("utf-8"), AES.block_size))
ct = b64encode(ct_bytes).decode("utf-8")
return ct
else:
pt = unpad(cipher.decrypt(b64decode(text)), AES.block_size)
return pt.decode("utf-8")
def header(self):
t = str(int(time.time()))
header = {"Referer": self.host,
"User-Agent": "okhttp/3.14.9", "app-version-code": "516", "app-ui-mode": "light",
"app-api-verify-time": t, "app-user-device-id": self.did,
"app-api-verify-sign": self.aes(t, True),
"Content-Type": "application/x-www-form-urlencoded; charset=UTF-8"}
return header
def getdata(self, path, data=None):
vdata = self.post(f"{self.host}{path}", headers=self.header(), data=data, timeout=10).json()['data']
data1 = self.aes(vdata)
return json.loads(data1)
def Mproxy(self, url):
return f"{self.getProxyUrl()}&url={self.e64(url)}&type=m3u8"
def Mlocal(self, param, header=None):
url = self.d64(param["url"])
ydata = self.fetch(url, headers=header, allow_redirects=False)
data = ydata.content.decode('utf-8')
if ydata.headers.get('Location'):
url = ydata.headers['Location']
data = self.fetch(url, headers=header).content.decode('utf-8')
parsed_url = urlparse(url)
durl = parsed_url.scheme + "://" + parsed_url.netloc
lines = data.strip().split('\n')
for index, string in enumerate(lines):
if '#EXT' not in string and 'http' not in string:
last_slash_index = string.rfind('/')
lpath = string[:last_slash_index + 1]
lines[index] = durl + ('' if lpath.startswith('/') else '/') + lpath
data = '\n'.join(lines)
return [200, "application/vnd.apple.mpegur", data]
def e64(self, text):
try:
text_bytes = text.encode('utf-8')
encoded_bytes = b64encode(text_bytes)
return encoded_bytes.decode('utf-8')
except Exception as e:
print(f"Base64编码错误: {str(e)}")
return ""
def d64(self, encoded_text):
try:
encoded_bytes = encoded_text.encode('utf-8')
decoded_bytes = b64decode(encoded_bytes)
return decoded_bytes.decode('utf-8')
except Exception as e:
print(f"Base64解码错误: {str(e)}")
return ""
def md5(self, text):
h = MD5.new()
h.update(text.encode('utf-8'))
return h.hexdigest()
FGBLH/py/hanime1.py
-300
View File
@@ -1,300 +0,0 @@
# coding=utf-8
# hanime1.py —— TVBox / FongMi Python 爬虫
import re
import sys
import urllib.parse
sys.path.append('..')
from base.spider import Spider
class Spider(Spider):
host = 'https://hanime1.me'
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',
'Referer': 'https://hanime1.me/',
'Accept-Language': 'zh-TW,zh;q=0.9,zh-CN;q=0.8',
}
cates = [
('最新里番', 'genre_sort:裏番|最新上市'),
('最新上市', 'sort:最新上市'),
('最新上传', 'sort:最新上傳'),
('里番', '裏番'),
('泡面番', '泡麵番'),
('Motion Anime', 'Motion Anime'),
('3DCG', '3DCG'),
('2.5D', '2.5D'),
('2D动画', '2D動畫'),
('AI生成', 'AI生成'),
('MMD', 'MMD'),
('Cosplay', 'Cosplay'),
('新番预告', '新番預告'),
]
sorts = [
{'n': '最新上市', 'v': '最新上市'},
{'n': '最新上传', 'v': '最新上傳'},
{'n': '本日排行', 'v': '本日排行'},
{'n': '本周排行', 'v': '本週排行'},
{'n': '本月排行', 'v': '本月排行'},
]
# ---------- 基础 ----------
def getName(self):
return 'hanime1'
def init(self, extend=''):
pass
def isVideoFormat(self, url):
pass
def manualVideoCheck(self):
pass
def destroy(self):
pass
def localProxy(self, param):
return None
def _get(self, url):
try:
return self.fetch(url, headers=self.headers, timeout=15).text
except Exception:
import requests
return requests.get(url, headers=self.headers, timeout=15).text
def _meta(self, html, prop):
m = re.search(r'<meta[^>]+property="%s"[^>]+content="([^"]*)"' % prop, html)
if not m:
m = re.search(r'<meta[^>]+content="([^"]*)"[^>]+property="%s"' % prop, html)
return m.group(1).strip() if m else ''
# ---------- 列表解析 ----------
def parse_list(self, html):
videos = []
seen = set()
card_pattern = re.compile(
r'<a[^>]+href=["\'](?:https?://hanime1\.me)?/watch\?v=(\d+)["\'][^>]*>(.*?)</a>',
re.S
)
for m in card_pattern.finditer(html):
vid = m.group(1)
if vid in seen:
continue
inner = m.group(2)
title = ''
t = re.search(
r'class="[^"]*(?:title|name)[^"]*"[^>]*>\s*([^<]+?)\s*<',
inner, re.S
)
if t:
title = t.group(1).strip()
if not title:
t = re.search(r'<img[^>]+alt="([^"]+)"', inner)
if t:
title = t.group(1).strip()
if not title:
t = re.search(r'title="([^"]+)"', m.group(0))
if t:
title = t.group(1).strip()
if not title:
continue
pic = ''
for img in re.findall(r'<img[^>]+(?:data-src|src)="(http[^"]+)"', inner):
if '.gif' in img or 'icon' in img or 'avatar' in img:
continue
pic = img
break
seen.add(vid)
videos.append({
'vod_id': vid,
'vod_name': title,
'vod_pic': pic,
'vod_remarks': ''
})
if not videos:
videos = self._parse_list_fallback(html)
return videos
def _parse_list_fallback(self, html):
videos = []
seen = set()
for m in re.finditer(
r'<a[^>]+href=["\'][^"\']*watch\?v=(\d+)["\']([^>]*)>(.*?)</a>',
html, re.S
):
vid = m.group(1)
if vid in seen:
continue
attrs = m.group(2)
inner = m.group(3)
t = re.search(r'title="([^"]+)"', attrs)
title = t.group(1).strip() if t else ''
if not title:
title = re.sub(r'<[^>]+>', '', inner).strip()
if not title:
continue
seen.add(vid)
videos.append({'vod_id': vid, 'vod_name': title, 'vod_pic': '', 'vod_remarks': ''})
return videos
# ---------- 首页 ----------
def homeContent(self, filter):
classes = [{'type_name': n, 'type_id': v} for n, v in self.cates]
filters = {}
for _, v in self.cates:
if not v.startswith('sort:') and not v.startswith('genre_sort:'):
filters[v] = [{'key': 'sort', 'name': '排序', 'value': self.sorts}]
return {'class': classes, 'filters': filters}
def homeVideoContent(self):
html = self._get(self.host)
return {'list': self.parse_list(html)[:30]}
# ---------- 分类 ----------
def categoryContent(self, tid, pg, filter, extend):
if tid.startswith('genre_sort:'):
parts = tid[len('genre_sort:'):].split('|', 1)
genre = parts[0]
sort = parts[1] if len(parts) > 1 else ''
url = '{}/search?genre={}&page={}'.format(
self.host, urllib.parse.quote(genre), pg)
if sort:
url += '&sort=' + urllib.parse.quote(sort)
elif tid.startswith('sort:'):
sort_val = tid[len('sort:'):]
url = '{}/search?sort={}&page={}'.format(
self.host, urllib.parse.quote(sort_val), pg)
else:
url = '{}/search?genre={}&page={}'.format(
self.host, urllib.parse.quote(tid), pg)
if extend and extend.get('sort'):
url += '&sort=' + urllib.parse.quote(extend['sort'])
html = self._get(url)
videos = self.parse_list(html)
return {
'list': videos,
'page': int(pg),
'pagecount': int(pg) + (1 if len(videos) >= 20 else 0),
'limit': 30,
'total': 999999
}
# ---------- 详情（核心修复：把当前请求的这一集放到播放列表首位） ----------
def detailContent(self, ids):
vid = ids[0]
html = self._get('{}/watch?v={}'.format(self.host, vid))
title = self._meta(html, 'og:title')
pic = self._meta(html, 'og:image')
desc = self._meta(html, 'og:description')
episodes = self._extract_playlist(html, vid, title)
play_url = '#'.join(
'{}${}'.format(ep['name'], ep['vid'])
for ep in episodes
)
vod = {
'vod_id' : vid,
'vod_name' : title,
'vod_pic' : pic,
'vod_content' : desc,
'vod_play_from': 'Hanime1',
'vod_play_url' : play_url,
}
return {'list': [vod]}
def _extract_playlist(self, html, current_vid, current_title):
"""
提取播放列表（选集），并把当前请求的这一集放到列表最前面，
这样无论用户从外层列表点的是第几集，进详情页后默认播放的
都是用户实际点击的那一集，而不是固定播放第一集。
"""
episodes = []
seen = set()
ep_pattern = re.compile(
r'<a class="overlay" href="https?://hanime1\.me/watch\?v=(\d+)"></a>\s*'
r'<div class="card-mobile-panel inner">.*?'
r'<div class="card-mobile-title"[^>]*>([^<]+)</div>',
re.S
)
for m in ep_pattern.finditer(html):
ep_vid = m.group(1)
ep_name = m.group(2).strip()
if ep_vid in seen:
continue
seen.add(ep_vid)
episodes.append({'name': ep_name, 'vid': ep_vid})
# 没有播放列表说明是单集视频，直接返回当前这一集
if not episodes:
return [{'name': current_title or current_vid, 'vid': current_vid}]
# 确保当前请求的这一集一定在列表里（理论上必然存在，兜底一下）
if not any(e['vid'] == current_vid for e in episodes):
episodes.append({'name': current_title or current_vid, 'vid': current_vid})
# 先按集数数字升序排好，方便用户在选集列表里查看顺序
def _ep_num(ep):
m = re.search(r'(\d+)\s*$', ep['name'])
return int(m.group(1)) if m else 0
try:
episodes.sort(key=_ep_num)
except Exception:
pass
# 核心修复：把当前请求的这一集挪到最前面，作为默认播放项
current_index = next(
(i for i, e in enumerate(episodes) if e['vid'] == current_vid), None
)
if current_index is not None and current_index != 0:
current_ep = episodes.pop(current_index)
episodes.insert(0, current_ep)
return episodes
# ---------- 搜索 ----------
def searchContent(self, key, quick, pg='1'):
url = '{}/search?query={}&page={}'.format(
self.host, urllib.parse.quote(key), pg)
html = self._get(url)
return {'list': self.parse_list(html), 'page': int(pg)}
# ---------- 播放 ----------
def playerContent(self, flag, id, vipFlags):
html = self._get('{}/watch?v={}'.format(self.host, id))
play = ''
sources = re.findall(r'<source[^>]+src="([^"]+)"[^>]*size="(\d+)"', html)
if sources:
sources.sort(key=lambda x: int(x[1]), reverse=True)
play = sources[0][0]
if not play:
m = re.search(r'"contentUrl"\s*:\s*"([^"]+)"', html)
if m:
play = m.group(1).replace('\\/', '/')
if not play:
m = re.search(r'(https?://[^\s\'"]+\.(?:m3u8|mp4)[^\s\'"]*)', html)
if m:
play = m.group(1)
return {
'parse': 0 if play else 1,
'url' : play or '{}/watch?v={}'.format(self.host, id),
'header': self.headers
}
FGBLH/py/ppx.py
+169
View File
@@ -0,0 +1,169 @@
# -*- coding: utf-8 -*-
# 本资源来源于互联网公开渠道，仅可用于个人学习及爬虫技术交流。
# 严禁将其用于任何商业用途，下载后请于 24 小时内删除，搜索结果均来自源站，本人不承担任何责任。
import re,sys,uuid
from base.spider import Spider
sys.path.append('..')
class Spider(Spider):
host,config,local_uuid,parsing_config = '','','',[]
# 头部添加token认证
headers = {
'User-Agent': "Dart/2.19 (dart:io)",
'Accept-Encoding': "gzip",
'appto-local-uuid': local_uuid,
'token': "eyJhbGciOiJIUzI1NiJ9.eyJkYXRhIjp7InVzZXJfY2hlY2siOiI4ZTEyNDE1Y2UyOGQzMGM4MWE3MDBiNWYxMDgzZTU2OCIsInVzZXJfaWQiOjM0NTYsInVzZXJfbmFtZSI6IjEwMTAxMiJ9LCJleHAiOjE4MDQ3MzkyODAuNjA4MTA4MywiaWF0IjoxNzczMjAzMjgxLCJpc3MiOiJBcHBUbyIsImp0aSI6ImZmZDMyYjk4N2VkMTg1ZjNiNGQ5Zjc5NzU2YWRjNGQ5IiwibmJmIjoxNzczMjAzMjgxLCJzdWIiOiJBcHBUbyJ9.tDhURwWVzsPy0-yXvo_d3bgsmoq9Ri5n0Y4fQsvxKy0"
}
def init(self, extend=''):
try:
host = extend.strip()
if not host.startswith('http'):
return {}
if not re.match(r'^https?://[a-zA-Z0-9-]+(\.[a-zA-Z0-9-]+)*(:\d+)?/?$', host):
host_=self.fetch(host).json()
self.host = host_['domain']
else:
self.host = host
self.local_uuid = str(uuid.uuid4())
# 动态更新headers中的uuid（避免初始化时uuid为空）
self.headers['appto-local-uuid'] = self.local_uuid
response = self.fetch(f'{self.host}/apptov5/v1/config/get?p=android&__platform=android', headers=self.headers).json()
config = response['data']
self.config = config
parsing_conf = config['get_parsing']['lists']
parsing_config = {}
for i in parsing_conf:
if len(i['config']) != 0:
label = []
for j in i['config']:
if j['type'] == 'json':
label.append(j['label'])
parsing_config.update({i['key']:label})
self.parsing_config = parsing_config
return None
except Exception as e:
print(f'初始化异常：{e}')
return {}
def detailContent(self, ids):
response = self.fetch(f"{self.host}/apptov5/v1/vod/getVod?id={ids[0]}",headers=self.headers).json()
data3 = response['data']
videos = []
vod_play_url = ''
vod_play_from = ''
for i in data3['vod_play_list']:
play_url = ''
for j in i['urls']:
play_url += f"{j['name']}${i['player_info']['from']}@{j['url']}#"
vod_play_from += i['player_info']['show'] + '$$$'
vod_play_url += play_url.rstrip('#') + '$$$'
vod_play_url = vod_play_url.rstrip('$$$')
vod_play_from = vod_play_from.rstrip('$$$')
videos.append({
'vod_id': data3.get('vod_id'),
'vod_name': data3.get('vod_name'),
'vod_content': data3.get('vod_content'),
'vod_remarks': data3.get('vod_remarks'),
'vod_director': data3.get('vod_director'),
'vod_actor': data3.get('vod_actor'),
'vod_year': data3.get('vod_year'),
'vod_area': data3.get('vod_area'),
'vod_play_from': vod_play_from,
'vod_play_url': vod_play_url
})
return {'list': videos}
def searchContent(self, key, quick, pg='1'):
url = f"{self.host}/apptov5/v1/search/lists?wd={key}&page={pg}&type=&__platform=android"
response = self.fetch(url, headers=self.headers).json()
data = response['data']['data']
for i in data:
if i.get('vod_pic').startswith('mac://'):
i['vod_pic'] = i['vod_pic'].replace('mac://', 'http://', 1)
return {'list': data, 'page': pg, 'total': response['data']['total']}
def playerContent(self, flag, id, vipflags):
default_ua = 'Mozilla/5.0 (iPhone; CPU iPhone OS 13_2_3 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/13.0.3 Mobile/15E148 Safari/604.1'
parsing_config = self.parsing_config
parts = id.split('@')
if len(parts) != 2:
return {'parse': 0, 'url': id, 'header': {'User-Agent': default_ua}}
playfrom, rawurl = parts
label_list = parsing_config.get(playfrom)
if not label_list:
return {'parse': 0, 'url': rawurl, 'header': {'User-Agent': default_ua}}
result = {'parse': 1, 'url': rawurl, 'header': {'User-Agent': default_ua}}
for label in label_list:
payload = {
'play_url': rawurl,
'label': label,
'key': playfrom
}
try:
response = self.post(
f"{self.host}/apptov5/v1/parsing/proxy?__platform=android",
data=payload,
headers=self.headers
).json()
except Exception as e:
print(f"请求异常: {e}")
continue
if not isinstance(response, dict):
continue
if response.get('code') == 422:
continue
data = response.get('data')
if not isinstance(data, dict):
continue
url = data.get('url')
if not url:
continue
ua = data.get('UA') or data.get('UserAgent') or default_ua
result = {
'parse': 0,
'url': url,
'header': {'User-Agent': ua}
}
break
return result
def homeContent(self, filter):
config = self.config
if not config:
return {}
home_cate = config['get_home_cate']
classes = []
for i in home_cate:
if isinstance(i.get('extend', []),dict):
classes.append({'type_id': i['cate'], 'type_name': i['title']})
return {'class': classes}
def homeVideoContent(self):
response = self.fetch(f'{self.host}/apptov5/v1/home/data?id=1&mold=1&__platform=android',headers=self.headers).json()
data = response['data']
vod_list = []
for i in data['sections']:
for j in i['items']:
vod_pic = j.get('vod_pic')
if vod_pic.startswith('mac://'):
vod_pic = vod_pic.replace('mac://', 'http://', 1)
vod_list.append({
"vod_id": j.get('vod_id'),
"vod_name": j.get('vod_name'),
"vod_pic": vod_pic,
"vod_remarks": j.get('vod_remarks')
})
return {'list': vod_list}
def categoryContent(self, tid, pg, filter, extend):
response = self.fetch(f"{self.host}/apptov5/v1/vod/lists?area={extend.get('area','')}&lang={extend.get('lang','')}&year={extend.get('year','')}&order={extend.get('sort','time')}&type_id={tid}&type_name=&page={pg}&pageSize=21&__platform=android", headers=self.headers).json()
data = response['data']
data2 = data['data']
for i in data['data']:
if i.get('vod_pic','').startswith('mac://'):
i['vod_pic'] = i['vod_pic'].replace('mac://', 'http://', 1)
return {'list': data2, 'page': pg, 'total': data['total']}
def getName(self):
pass
def isVideoFormat(self, url):
pass
def manualVideoCheck(self):
pass
def destroy(self):
pass
def localProxy(self, param):
pass
FGBLH/py/七味.py
+948
View File
@@ -0,0 +1,948 @@
# -*- coding: utf-8 -*-
import re
import ssl
import json
import html
import base64
import urllib3
import threading
import time
import sys
from urllib.parse import quote, unquote, urljoin, urlparse
import requests
from requests.adapters import HTTPAdapter
from pyquery import PyQuery as pq
sys.path.append("..")
from base.spider import Spider as BaseSpider
urllib3.disable_warnings()
class SSLAdapter(HTTPAdapter):
def init_poolmanager(self, connections, maxsize, block=False, **kwargs):
ctx = ssl.create_default_context()
ctx.check_hostname = False
ctx.verify_mode = ssl.CERT_NONE
kwargs["ssl_context"] = ctx
return super().init_poolmanager(connections, maxsize, block=block, **kwargs)
def proxy_manager_for(self, proxy, **kwargs):
ctx = ssl.create_default_context()
ctx.check_hostname = False
ctx.verify_mode = ssl.CERT_NONE
kwargs["ssl_context"] = ctx
return super().proxy_manager_for(proxy, **kwargs)
class Spider(BaseSpider):
hosts = [
"https://www.qwmkv.com",
"https://www.qwnull.com",
"https://www.qwfilm.com",
"https://www.qnmp4.com",
"https://www.qn63.com"
]
host = hosts[0]
headers = {
"User-Agent": "Mozilla/5.0 (Linux; Android 12; M2012K11AC) AppleWebKit/537.36 "
"(KHTML, like Gecko) Chrome/124.0.0.0 Mobile Safari/537.36",
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
"Accept-Language": "zh-CN,zh;q=0.9",
"Accept-Encoding": "gzip, deflate",
"Connection": "keep-alive",
"Upgrade-Insecure-Requests": "1"
}
CATEGORY_IDS = {"电影": 1, "剧集": 2, "综艺": 3, "动漫": 4, "短剧": 30}
KEYWORDS = ["杜比", "dolby", "原盘", "高码", "remux", "蓝光", "hdr10+", "hdr10", "hdr", "4k", "2160p", "uhd"]
QUARK_CHECK_LIMIT = 100
CHECK_TIME_BUDGET = 12.0 # 检测最多12秒
def getName(self):
return "七味-最终稳定版(含大屏分组排序+防串位修复)"
def init(self, extend=""):
self.session = requests.Session()
adapter = SSLAdapter(max_retries=2)
self.session.mount("http://", adapter)
self.session.mount("https://", adapter)
self.session.verify = False
self.session.headers.update(dict(self.headers))
self.last_vod_pic = ""
self.vod_pic_cache = {}
self.pan_115_cookie = ""
self.ack_mp4 = "https://vd2.bdstatic.com/mda-nj5kxa8kr7wgq6ie/sc/cae_h264_nowatermark/1653272065989267185/mda-nj5kxa8kr7wgq6ie.mp4"
if extend:
try:
ext = json.loads(extend)
self.pan_115_cookie = ext.get("pan_115_cookie", "")
self.ack_mp4 = ext.get("ack_mp4", self.ack_mp4)
except Exception as e:
print(f"init extend error: {e}")
self._probe_host()
def destroy(self):
try:
self.session.close()
except:
pass
# ---------------- utils ----------------
def _probe_host(self):
for h in self.hosts:
try:
r = self.session.get(h + "/", timeout=6, headers=self.headers, verify=False)
if r.status_code == 200:
self.host = h
return
except:
pass
def _full_url(self, path):
if not path:
return ""
path = html.unescape(str(path)).strip()
if path.startswith("//"):
return "https:" + path
if path.startswith(("http://", "https://", "magnet:?")):
return path
return urljoin(self.host + "/", path)
def _full_url_by_host(self, host, path):
if not path:
return ""
path = html.unescape(str(path)).strip()
if path.startswith("//"):
return "https:" + path
if path.startswith(("http://", "https://", "magnet:?")):
return path
return urljoin(host.rstrip("/") + "/", path.lstrip("/"))
def _fetch(self, url, timeout=10):
tries = [self._full_url(url)]
if isinstance(url, str) and not url.startswith(("http://", "https://", "magnet:?")):
for h in self.hosts:
u = urljoin(h + "/", url)
if u not in tries:
tries.append(u)
for u in tries:
try:
h = dict(self.headers)
h["Referer"] = self.host + "/"
r = self.session.get(u, timeout=timeout, headers=h, verify=False)
r.encoding = r.apparent_encoding or "utf-8"
if r.status_code == 200 and len(r.text or "") > 30:
for hh in self.hosts:
if u.startswith(hh):
self.host = hh
break
return r
except:
continue
return None
def _pq(self, url, timeout=10):
r = self._fetch(url, timeout=timeout)
return pq(r.text if r else "")
def _clean_text(self, s):
return re.sub(r"\s+", " ", html.unescape(s or "")).strip()
def _clean_name(self, s, max_len=120):
s = html.unescape(s or "").replace("#", "＃").replace("$", "＄")
s = re.sub(r"\s+", " ", s).strip()
return s[:max_len]
def _img_src(self, img):
return img.attr("data-src") or img.attr("data-original") or img.attr("src") or ""
def _is_pan(self, u):
u = (u or "").lower()
return any(k in u for k in [
"pan.quark.cn/s/", "pan.baidu.com/s/", "drive.uc.cn/s/",
"pan.xunlei.com/s/", "aliyundrive.com/s/", "alipan.com/s/",
"cloud.189.cn/", "caiyun.139.com/", "123pan.com/s/",
"115.com/s/", "lanzou", "lanzoui", "lanzoux", "lanzoub"
])
def _b64e(self, obj):
txt = json.dumps(obj, ensure_ascii=False, separators=(",", ":"))
return base64.urlsafe_b64encode(txt.encode()).decode().rstrip("=")
def _b64d(self, s):
try:
s += "=" * (-len(s) % 4)
return json.loads(base64.urlsafe_b64decode(s.encode()).decode())
except:
return {}
def _get_mid(self, tid):
if str(tid).isdigit():
return int(tid)
m = re.search(r"/vt/(\d+)", str(tid))
if m:
return int(m.group(1))
return 1
def _score_name(self, name):
n = (name or "").lower()
score = 0
for i, kw in enumerate(self.KEYWORDS):
if kw.lower() in n:
score += (len(self.KEYWORDS) - i)
return score
def _extract_video_list(self, doc):
videos, seen = [], set()
selectors = ["ul.pic-list li", "ul.content-list li", ".pic-list li", ".content-list li"]
nodes = []
for sel in selectors:
n = list(doc(sel).items())
if n:
nodes = n
break
for li in nodes:
a = li("a[href]").eq(0)
href = a.attr("href")
if not href:
continue
vid = self._full_url(href)
if vid in seen:
continue
seen.add(vid)
img = li("img").eq(0)
pic = self._full_url(self._img_src(img))
title = a.attr("title") or img.attr("alt") or li("h3 b").text() or li("h3").text() or ""
remark = self._clean_text(li("span.s1").text() or li("span.s2").text() or li("p").text() or li(".tag").text())
if pic:
self.vod_pic_cache[vid] = pic
videos.append({
"vod_id": vid,
"vod_name": self._clean_name(title, 80),
"vod_pic": pic,
"vod_remarks": remark
})
return videos
def _is_bad_cover(self, u):
if not u:
return True
s = u.lower()
return ("logo.png" in s) or ("loading" in s) or ("/template/piankuwap/image/logo" in s)
def _normalize_magnet(self, href):
try:
if not href:
return ""
href = str(href).strip().replace("&amp;", "&")
if href.startswith("push://"):
href = href.replace("push://", "", 1).replace("#0agent", "")
if "%3A" in href or "%3F" in href or "%26" in href:
href = unquote(href)
href = re.sub(r"\s+", "", href)
if not href.startswith("magnet:") or "urn:btih:" not in href:
return ""
return href
except:
return ""
def _magnet_btih(self, magnet):
m = self._normalize_magnet(magnet)
if not m:
return ""
g = re.search(r"xt=urn:btih:([a-zA-Z0-9]+)", m, re.I)
return g.group(1).lower() if g else ""
def _is_verify_page(self, text):
t = (text or "").lower()
return (
("系统安全验证" in t) or
("verify_check" in t) or
("mac_verify_img" in t) or
("请输入验证码" in t)
)
def _mk_vod_id(self, h, raw_id, raw_url=""):
if raw_url:
u = self._full_url_by_host(h, raw_url)
if "/mv/" in u and u.endswith(".html"):
return u
if str(raw_id).isdigit():
return f"{h}/mv/{raw_id}.html"
m = re.search(r"/mv/(\d+)\.html", u)
if m:
return f"{h}/mv/{m.group(1)}.html"
return u
rid = str(raw_id or "").strip()
if rid.isdigit():
return f"{h}/mv/{rid}.html"
if rid.startswith(("http://", "https://", "/")):
return self._full_url_by_host(h, rid)
return f"{h}/mv/{rid}.html" if rid else ""
# ---------------- only check quark/115 ----------------
def _check_pan_valid(self, url, provider, timeout=3):
if not url:
return False
if provider not in ("quark", "115"):
return True
try:
h = dict(self.headers)
h["Referer"] = self.host + "/"
r = requests.get(url, headers=h, timeout=timeout, verify=False, allow_redirects=True)
if r.status_code >= 400:
return False
text = (r.text or "").lower()
if provider == "quark":
keys = ["分享已失效", "不存在", "已被取消", "取消", "删除", "已被删除", "来晚了", "违规", "无法访问"]
else:
keys = ["分享已失效", "不存在", "404", "已取消", "链接错误"]
return not any(k in text for k in keys)
except:
return False
# ---------------- home ----------------
def homeContent(self, filter):
classes = [
{"type_name": "大陆电影", "type_id": "https://www.qwmkv.com/ms/1-大陆-time---------.html"},
{"type_name": "大陆剧集", "type_id": "https://www.qwmkv.com/ms/2-大陆-time---------.html"},
{"type_name": "大陆综艺", "type_id": "https://www.qwmkv.com/ms/3-大陆-time---------.html"},
{"type_name": "大陆动漫", "type_id": "https://www.qwmkv.com/ms/4-大陆-time---------.html"},
{"type_name": "电影", "type_id": "/vt/1.html"},
{"type_name": "综艺", "type_id": "/vt/3.html"},
{"type_name": "剧集", "type_id": "/vt/2.html"},
{"type_name": "动漫", "type_id": "/vt/4.html"},
{"type_name": "短剧", "type_id": "/vt/30.html"},
]
return {"class": classes}
def homeVideoContent(self):
doc = self._pq(self.host + "/")
videos = self._extract_video_list(doc)
return {"list": videos, "page": 1, "pagecount": 1, "limit": len(videos), "total": len(videos)}
# ---------------- category ----------------
def _build_category_url(self, tid, pg, fdict):
if isinstance(tid, str) and tid.startswith(("http://", "https://")):
if pg <= 1:
return tid
if "---------.html" in tid:
return tid.replace("---------.html", f"------{pg}---.html")
return tid.replace(".html", f"-{pg}.html")
mid = self._get_mid(tid)
if not fdict:
return f"{self.host}/vt/{mid}.html" if pg <= 1 else f"{self.host}/vt/{mid}-{pg}.html"
area = quote(fdict.get("地区", ""), safe="")
sort = ""
sv = fdict.get("排序", "")
if sv == "按时间":
sort = "time"
elif sv == "按人气":
sort = "hits"
elif sv == "按评分":
sort = "score"
typ = quote(fdict.get("类型", ""), safe="")
lang = quote(fdict.get("语言", ""), safe="")
year = fdict.get("年代", "")
fields = [area, sort, typ, lang, "", "", "", "", year]
base = f"{self.host}/ms/{mid}-" + "-".join(fields)
return base + ".html" if pg <= 1 else base + f"-{pg}.html"
def categoryContent(self, tid, pg, filter, extend):
pg = int(pg) if str(pg).isdigit() else 1
fdict = extend if isinstance(extend, dict) else {}
url = self._build_category_url(tid, pg, fdict)
doc = self._pq(url)
if len(doc("ul.pic-list li")) == 0 and len(doc("ul.content-list li")) == 0 and pg > 1:
doc = self._pq(url.replace(".html", f".html?page={pg}"))
videos = self._extract_video_list(doc)
page_count = pg
for a in doc(".pages a").items():
t = (a.text() or "").strip()
href = a.attr("href") or ""
if t.isdigit():
page_count = max(page_count, int(t))
else:
m = re.search(r"-(\d+)\.html", href)
if m:
page_count = max(page_count, int(m.group(1)))
return {
"list": videos,
"page": pg,
"pagecount": max(page_count, pg),
"limit": 30,
"total": max(page_count, pg) * 30
}
# ---------------- detail ----------------
def detailContent(self, ids):
try:
vod_id = ids[0] if isinstance(ids, list) and ids else ids
vod_id = self._full_url(vod_id)
doc = self._pq(vod_id)
raw = doc.html() or ""
# 1. 抓取基本影视信息
title = self._clean_text(doc("h1").eq(0).text())
if not title:
tt = self._clean_text(doc("title").text())
title = tt.split("在线观看")[0] if tt else "七味资源"
cover = ""
og = self._full_url(doc('meta[property="og:image"]').attr("content") or "")
if og and not self._is_bad_cover(og):
cover = og
if not cover:
c1 = self._full_url(self._img_src(doc(".main-left .img img").eq(0)))
if c1 and not self._is_bad_cover(c1):
cover = c1
if not cover:
for im in doc("img").items():
src = self._full_url(self._img_src(im))
if src and not self._is_bad_cover(src):
cover = src
break
if not cover:
cover = self.vod_pic_cache.get(vod_id, "")
if not cover:
cover = self._full_url("/template/piankuwap/image/logo.png")
self.last_vod_pic = cover
content = self._clean_text(
doc(".movie-introduce .sqjj_a").text() or
doc(".movie-introduce .zkjj_a").text() or
doc(".content").text()
)
# =====【优化修复】2. 多线路合并遍历抓取在线资源（彻底杜绝选择器冲突覆盖） =====
online_routes = {} # 格式: {"在线线路1": ["第1集$payload", "第2集$payload"]}
player_uls = doc("div#url .bd ul.player, ul.player")
line_no = 1
for ul_node in player_uls.items():
links = list(ul_node("a[href]").items())
if not links:
continue
route_name = f"📺在线播放-线路{line_no}"
episodes = []
for a in links:
href = a.attr("href")
if not href:
continue
src_name = self._clean_name(self._clean_text(a.text()) or "播放", 50)
payload = self._b64e({"type": "py", "url": self._full_url(href), "pic": cover})
episodes.append(f"{src_name}${payload}")
if episodes:
online_routes[route_name] = episodes
line_no += 1
# ===== 3. 规整网盘与磁力资源链接 =====
pan_resources = []
magnet_raw = []
seen_pan = set()
for a in doc("a[href]").items():
u = html.unescape(a.attr("href") or "").strip()
if not u:
continue
txt = self._clean_text(a.text())
if u.lower().startswith("magnet:?"):
magnet_raw.append((u, self._clean_name(txt or "磁力资源", 60)))
continue
if self._is_pan(u):
low = u.lower()
pv = "other"
if "pan.quark" in low: pv = "quark"
elif "115.com" in low: pv = "115"
elif "pan.baidu" in low: pv = "baidu"
elif "drive.uc.cn" in low: pv = "uc"
elif "pan.xunlei" in low: pv = "xunlei"
elif "aliyundrive" in low or "alipan" in low: pv = "ali"
elif "cloud.189" in low: pv = "189"
elif "123pan" in low: pv = "pan123"
if u not in seen_pan:
seen_pan.add(u)
pan_resources.append({
"provider": pv,
"url": u,
"name": txt or "网盘资源",
"checked_valid": False
})
for m in re.finditer(r"magnet:\?[^\s\"'<>]+", raw, re.I):
magnet_raw.append((html.unescape(m.group(0)), "磁力资源"))
# 磁力资源精简去重并打分排序
magnet_unified = []
btih_seen = set()
for u, n in magnet_raw:
mu = self._normalize_magnet(u)
if not mu:
continue
btih = self._magnet_btih(mu)
key = btih if btih else mu.lower()
if key in btih_seen:
continue
btih_seen.add(key)
magnet_unified.append({
"url": mu,
"name": self._clean_name(n or "磁力资源", 60)
})
magnet_unified.sort(key=lambda x: -self._score_name(x.get("name", "")))
# ===== 4. 网盘有效性探针（保持原有超时/计数策略） =====
check_begin = time.monotonic()
quark_checked = 0
valid_pan = []
for p in pan_resources:
if time.monotonic() - check_begin >= self.CHECK_TIME_BUDGET:
valid_pan.append(p)
continue
pv = p["provider"]
if pv == "quark":
if quark_checked >= self.QUARK_CHECK_LIMIT:
valid_pan.append(p)
continue
quark_checked += 1
if self._check_pan_valid(p["url"], "quark"):
p["checked_valid"] = True
valid_pan.append(p)
elif pv == "115":
if self._check_pan_valid(p["url"], "115"):
p["checked_valid"] = True
valid_pan.append(p)
else:
valid_pan.append(p)
# =====【功能修复】5. 网盘资源渠道完全独立隔离，防止错乱混杂 =====
pan_routes = {
"quark": {"name": "🟢夸克网盘", "list": []},
"ali": {"name": "☁️阿里云盘", "list": []},
"115": {"name": "固定115网盘", "list": []},
"baidu": {"name": "📘百度网盘", "list": []},
"uc": {"name": "📱UC网盘", "list": []},
"xunlei": {"name": "⚡迅雷网盘", "list": []},
"189": {"name": "☎️天翼云盘", "list": []},
"pan123": {"name": "📦123网盘", "list": []},
"other": {"name": "📦其它网盘", "list": []}
}
for r in valid_pan:
prov = r["provider"]
if prov not in pan_routes:
prov = "other"
ep_name = self._clean_name(r.get('name', '网盘提取资源'), 60)
payload = self._b64e({"type": "pan", "url": r["url"], "pic": cover})
pan_routes[prov]["list"].append(f"{ep_name}${payload}")
# =====【核心修复】6. 按标准大屏壳子1:1规则完美有序组装，防串位 =====
play_from = []
play_url = []
# 分支 A：写入网盘分类线路（依照预设的网盘体验优先级高低呈现）
drive_order = ["quark", "ali", "115", "baidu", "uc", "xunlei", "189", "pan123", "other"]
for d_key in drive_order:
route_info = pan_routes[d_key]
if route_info["list"]:
play_from.append(route_info["name"])
play_url.append("#".join(route_info["list"]))
# 分支 B：写入磁力解析相关线路（保持互相隔离）
if magnet_unified:
lines_115 = []
lines_play = []
for i, m in enumerate(magnet_unified, start=1):
nm = self._clean_name(f"磁力源-{i:02d} {m['name']}", 60)
encoded_mag = base64.urlsafe_b64encode(m['url'].encode()).decode().rstrip("=")
p_payload = self._b64e({"type": "magnet", "url": m['url'], "pic": cover})
lines_115.append(f"{nm}${encoded_mag}")
lines_play.append(f"{nm}${p_payload}")
# 独立线路一：115离线专用线
play_from.append("📥115云下载")
play_url.append("#".join(lines_115))
# 独立一条空白ACK确认交互线
play_from.append("0")
play_url.append("已提交请到115离线任务查看$__ACK__")
# 独立线路二：自带流播或通过本地壳嗅探弹磁力
play_from.append("🧲磁力播放")
play_url.append("#".join(lines_play))
# 分支 C：写入在线直连/网页采集线路
for r_name, r_eps in online_routes.items():
play_from.append(r_name)
play_url.append("#".join(r_eps))
# ===== 当前站搜索入口 =====
try:
search_payload = self._b64e({
"type": "search",
"wd": title,
"pic": cover
})
play_from.insert(0, "🔍点击选择")
play_url.insert(0, f"当前站搜索${search_payload}")
except Exception as e:
print(f"search line add error: {e}")
# 兜底处理
if not play_from:
play_from.append("🌐原网页查看")
play_url.append(f"点击跳转原详情页${self._b64e({'type': 'web', 'url': vod_id, 'pic': cover})}")
# 7. 构建标准影视输出字典
vod = {
"vod_id": vod_id,
"vod_name": self._clean_name(title, 100),
"vod_pic": cover,
"vod_content": content,
"vod_play_from": "$$$".join(play_from),
"vod_play_url": "$$$".join(play_url)
}
return {"list": [vod]}
except Exception as e:
print(f"detailContent error: {e}")
return {"list": []}
# ---------------- player ----------------
def _parse_py_page(self, py_url):
r = self._fetch(py_url, timeout=10)
if not r:
return ""
txt = r.text or ""
m = re.search(r"player_aaaa\s*=\s*(\{.*?\})\s*<", txt, re.S)
if not m:
m = re.search(r"player_aaaa\s*=\s*(\{.*?\})\s*;", txt, re.S)
if not m:
return ""
js = m.group(1)
try:
js2 = re.sub(r"(\w+)\s*:", r'"\1":', js)
obj = json.loads(js2)
except:
try:
obj = json.loads(js)
except:
return ""
u = obj.get("url", "") or ""
enc = str(obj.get("encrypt", "0"))
if enc == "1":
u = unquote(u)
elif enc == "2":
try:
u = unquote(base64.b64decode(u).decode("utf-8", "ignore"))
except:
pass
if u.startswith("//"):
u = "https:" + u
elif u.startswith("/"):
u = self._full_url(u)
return u
def _return_ack_video(self):
ret = {
"parse": 0,
"playUrl": "",
"url": self.ack_mp4,
"header": {
"User-Agent": self.headers.get("User-Agent", ""),
"Referer": self.host + "/"
}
}
if self.last_vod_pic:
ret["pic"] = self.last_vod_pic
ret["poster"] = self.last_vod_pic
return ret
def _add_to_115(self, magnet):
if not self.pan_115_cookie:
print("115添加失败: 未配置 pan_115_cookie")
return
magnet = self._normalize_magnet(magnet)
if not magnet:
print("115添加失败: 非法磁力")
return
headers = {
"User-Agent": self.headers.get("User-Agent", ""),
"Cookie": self.pan_115_cookie,
"Origin": "https://115.com",
"Referer": "https://115.com/web/lixian/",
"Accept": "application/json, text/javascript, */*; q=0.01",
"X-Requested-With": "XMLHttpRequest"
}
try:
pan_sess = requests.Session()
pan_sess.verify = False
pan_sess.mount("http://", SSLAdapter(max_retries=2))
pan_sess.mount("https://", SSLAdapter(max_retries=2))
space_resp = pan_sess.get("https://115.com/?ct=offline&ac=space", headers=headers, timeout=10)
try:
space_json = space_resp.json()
except:
print(f"115获取签名失败(非JSON): {space_resp.text[:200]}")
return
if not space_json.get("state"):
print(f"115获取签名失败(可能Cookie过期): {space_json}")
return
sign = space_json.get("sign", "")
req_time = space_json.get("time", "")
if not sign or not req_time:
print(f"115签名数据异常: {space_json}")
return
uid_match = re.search(r'UID=(\d+)', self.pan_115_cookie)
uid = uid_match.group(1) if uid_match else ""
add_url = "https://115.com/web/lixian/?ct=lixian&ac=add_task_url"
post_data = {"url": magnet, "uid": uid, "sign": sign, "time": req_time}
headers["Content-Type"] = "application/x-www-form-urlencoded; charset=UTF-8"
add_resp = pan_sess.post(add_url, data=post_data, headers=headers, timeout=10)
try:
add_json = add_resp.json()
except:
print(f"115添加失败(非JSON): {add_resp.text[:200]}")
return
if add_json.get("state") or add_json.get("errcode") == 0:
print(f"115离线添加成功: {magnet[:100]}...")
else:
err = add_json.get("error_msg") or add_json.get("msg") or add_json.get("error") or str(add_json)
print(f"115添加失败: {err}")
except Exception as e:
print(f"115离线网络异常: {e}")
def playerContent(self, flag, id, vipFlags):
if flag == "0" or id == "__ACK__":
return self._return_ack_video()
if flag == "📥115云下载":
try:
decoded = base64.urlsafe_b64decode(id.encode() + b"==").decode()
magnet = self._normalize_magnet(decoded)
if not magnet:
return self._return_ack_video()
if not self.pan_115_cookie:
print("115未配置Cookie")
return self._return_ack_video()
threading.Thread(target=self._add_to_115, args=(magnet,), daemon=True).start()
return self._return_ack_video()
except Exception as e:
print(f"115云下载处理异常: {e}")
return self._return_ack_video()
if flag == "🧲磁力播放":
data = self._b64d(id)
if data and data.get("type") == "magnet":
mu = self._normalize_magnet(data.get("url", ""))
if mu:
pic = data.get("pic") or self.last_vod_pic
return {"parse": 0, "url": "push://" + mu, "pic": pic, "poster": pic}
if isinstance(id, str) and id.startswith("push://"):
return {"parse": 0, "url": id, "pic": self.last_vod_pic, "poster": self.last_vod_pic}
mu = self._normalize_magnet(id)
if mu:
return {"parse": 0, "url": "push://" + mu, "pic": self.last_vod_pic, "poster": self.last_vod_pic}
return {"parse": 1, "url": id, "pic": self.last_vod_pic, "poster": self.last_vod_pic}
data = self._b64d(id)
if not data:
if isinstance(id, str) and id.startswith("push://"):
return {"parse": 0, "url": id, "pic": self.last_vod_pic, "poster": self.last_vod_pic}
return {"parse": 1, "url": id, "pic": self.last_vod_pic, "poster": self.last_vod_pic}
typ = data.get("type", "")
url = data.get("url", "")
pic = data.get("pic") or self.last_vod_pic
if not url:
return {"parse": 1, "url": id, "pic": pic, "poster": pic}
if typ == "search":
wd = data.get("wd", "").strip()
if not wd:
return {
"parse": 1,
"url": self.host,
"pic": pic,
"poster": pic
}
search_url = f"{self.host}/vodsearch/{quote(wd)}----------1---.html"
return {
"parse": 0,
"url": "push://" + search_url,
"pic": pic,
"poster": pic
}
if typ == "pan":
return {"parse": 0, "url": "push://" + url, "pic": pic, "poster": pic}
if typ == "magnet":
mu = self._normalize_magnet(url)
if mu:
return {"parse": 0, "url": "push://" + mu, "pic": pic, "poster": pic}
return {"parse": 1, "url": url, "pic": pic, "poster": pic}
if typ == "py":
real = self._parse_py_page(url)
if real:
return {"parse": 0, "url": real, "pic": pic, "poster": pic}
return {"parse": 1, "url": url, "pic": pic, "poster": pic}
if typ == "web":
return {"parse": 1, "url": url, "pic": pic, "poster": pic}
return {"parse": 1, "url": url, "pic": pic, "poster": pic}
# ---------------- search ----------------
def searchContent(self, key, quick, pg="1"):
pg = int(pg) if str(pg).isdigit() else 1
wd = quote(key)
for h in self.hosts:
suggest_api = f"{h}/index.php/ajax/suggest?mid=1&limit=20&wd={wd}"
try:
r = self.session.get(suggest_api, timeout=8, headers=self.headers, verify=False)
txt = r.text or ""
if not self._is_verify_page(txt) and r.status_code == 200:
data = r.json()
lst = data.get("list") or []
videos = []
for it in lst:
vid = it.get("id") or it.get("vod_id")
name = it.get("name") or it.get("vod_name") or ""
pic = self._full_url_by_host(h, it.get("pic") or it.get("vod_pic") or "")
remarks = self._clean_text(it.get("en") or it.get("remark") or "")
jump_url = it.get("url") or it.get("link") or ""
vid_url = self._mk_vod_id(h, vid, jump_url)
if not vid_url:
continue
if pic:
self.vod_pic_cache[vid_url] = pic
videos.append({
"vod_id": vid_url,
"vod_name": self._clean_name(name, 80),
"vod_pic": pic,
"vod_remarks": remarks
})
if videos:
self.host = h
return {
"list": videos,
"page": pg,
"pagecount": pg + 1 if len(videos) >= 20 else pg,
"limit": len(videos),
"total": len(videos)
}
except:
pass
api_list = [
f"{h}/api.php/provide/vod/?ac=detail&wd={wd}&pg={pg}",
f"{h}/api.php/provide/vod?ac=detail&wd={wd}&pg={pg}",
]
for api in api_list:
try:
r = self.session.get(api, timeout=8, headers=self.headers, verify=False)
txt = r.text or ""
if self._is_verify_page(txt):
continue
if r.status_code != 200:
continue
data = r.json()
lst = data.get("list") or data.get("data") or []
videos = []
for it in lst:
vid = it.get("vod_id") or it.get("id")
name = it.get("vod_name") or it.get("name") or ""
pic = self._full_url_by_host(h, it.get("vod_pic") or it.get("pic") or "")
remarks = self._clean_text(it.get("vod_remarks") or it.get("remarks") or "")
jump_url = it.get("vod_play_url") or it.get("url") or it.get("link") or ""
vid_url = self._mk_vod_id(h, vid, jump_url)
if not vid_url:
continue
if pic:
self.vod_pic_cache[vid_url] = pic
videos.append({
"vod_id": vid_url,
"vod_name": self._clean_name(name, 80),
"vod_pic": pic,
"vod_remarks": remarks
})
if videos:
self.host = h
return {
"list": videos,
"page": int(data.get("page", pg) or pg),
"pagecount": int(data.get("pagecount", 1) or 1),
"limit": len(videos),
"total": int(data.get("total", len(videos)) or len(videos))
}
except:
pass
return {
"list": [],
"page": pg,
"pagecount": pg,
"limit": 0,
"total": 0
}
FGBLH/py/采集聚合.py
+302
View File
@@ -0,0 +1,302 @@
# -*- coding: utf-8 -*-
# 专属全网聚合 Python版
# 适配常见 Cat/TVBox Python Spider
#本地py适配 😂
import json
import requests
from concurrent.futures import ThreadPoolExecutor, as_completed
from base.spider import Spider
class Spider(Spider):
sources = {
's1': {'name': '🎬电影天堂', 'api': 'http://caiji.dyttzyapi.com/api.php/provide/vod/from/dyttm3u8/at/json'},
's2': {'name': '💧无水印', 'api': 'https://api.wsyzy.net/api.php/provide/vod'},
's3': {'name': '🧸量子', 'api': 'https://cj.lziapi.com/api.php/provide/vod'},
's4': {'name': '📺1080资源', 'api': 'https://api.1080zyku.com/inc/api_mac10.php'},
's5': {'name': '🔥155资源', 'api': 'https://155api.com/api.php/provide/vod'},
's6': {'name': '📺天涯', 'api': 'https://tyyszy.com/api.php/provide/vod'},
's7': {'name': '📺暴风', 'api': 'https://bfzyapi.com/api.php/provide/vod'},
's8': {'name': '⚡索尼闪电', 'api': 'https://xsd.sdzyapi.com/api.php/provide/vod'},
's9': {'name': '📺索尼', 'api': 'https://suoniapi.com/api.php/provide/vod'},
's10': {'name': '📺红牛', 'api': 'https://www.hongniuzy2.com/api.php/provide/vod'},
's11': {'name': '📺茅台', 'api': 'https://caiji.maotaizy.cc/api.php/provide/vod'},
's12': {'name': '🐯虎牙', 'api': 'https://www.huyaapi.com/api.php/provide/vod'},
's13': {'name': '📺豆瓣', 'api': 'https://caiji.dbzy.tv/api.php/provide/vod'},
's14': {'name': '📺豆瓣2', 'api': 'https://dbzy.tv/api.php/provide/vod'},
's15': {'name': '📺豪华', 'api': 'https://hhzyapi.com/api.php/provide/vod'},
's16': {'name': '📺CK资源', 'api': 'https://ckzy.me/api.php/provide/vod'},
's17': {'name': '📺U酷', 'api': 'https://api.ukuapi.com/api.php/provide/vod'},
's18': {'name': '📺ikun', 'api': 'https://ikunzyapi.com/api.php/provide/vod'},
's19': {'name': '📺无尽', 'api': 'https://api.wujinapi.cc/api.php/provide/vod'},
's20': {'name': '🌕光速', 'api': 'https://api.guangsuapi.com/api.php/provide/vod'},
's21': {'name': '📺卧龙', 'api': 'https://collect.wolongzyw.com/api.php/provide/vod'},
's22': {'name': '📺新浪', 'api': 'https://api.xinlangapi.com/xinlangapi.php/provide/vod'},
's23': {'name': '📺旺旺', 'api': 'https://api.wwzy.tv/api.php/provide/vod'},
's24': {'name': '📺最大', 'api': 'https://api.zuidapi.com/api.php/provide/vod'},
's25': {'name': '🌸樱花', 'api': 'https://m3u8.apiyhzy.com/api.php/provide/vod'},
's26': {'name': '🐮牛牛', 'api': 'https://api.niuniuzy.me/api.php/provide/vod'},
's27': {'name': '☁️百度云', 'api': 'https://api.apibdzy.com/api.php/provide/vod'},
's28': {'name': '🏎速播', 'api': 'https://subocaiji.com/api.php/provide/vod'},
's29': {'name': '🦅金鹰', 'api': 'https://jinyingzy.com/api.php/provide/vod'},
's30': {'name': '⚡闪电', 'api': 'https://sdzyapi.com/api.php/provide/vod'},
's31': {'name': '👑非凡', 'api': 'https://cj.ffzyapi.com/api.php/provide/vod'},
's32': {'name': '🍃飘零', 'api': 'https://p2100.net/api.php/provide/vod'},
's33': {'name': '🐾魔爪', 'api': 'https://mozhuazy.com/api.php/provide/vod'},
's34': {'name': '📺魔都', 'api': 'https://www.mdzyapi.com/api.php/provide/vod'},
}
headers = {
"User-Agent": "Mozilla/5.0"
}
def getName(self):
return "影视+专属全网聚合"
def init(self, extend=""):
pass
def fetch(self, url, timeout=8):
try:
r = requests.get(
url,
headers=self.headers,
timeout=timeout,
verify=False
)
return r.text
except Exception:
return ""
def clean_item(self, item, source_key, source_name, is_detail=False):
item = dict(item)
if not is_detail:
item["vod_id"] = f"{source_key}@@{item.get('vod_id', '')}"
remarks = item.get("vod_remarks", "")
item["vod_remarks"] = f"{source_name} | {remarks}"
if item.get("vod_play_from"):
froms = item["vod_play_from"].split("$$$")
froms = [f"{source_name}-{x}" for x in froms]
item["vod_play_from"] = "$$$".join(froms)
item.pop("vod_down_from", None)
item.pop("vod_down_url", None)
return item
def homeContent(self, filter):
classes = []
filters = {}
def load_class(key, source):
url = f"{source['api']}?ac=list"
html = self.fetch(url, 4)
try:
data = json.loads(html)
except:
data = {}
vals = [{"n": "全部(最新)", "v": ""}]
for c in data.get("class", []):
vals.append({
"n": c.get("type_name", ""),
"v": c.get("type_id", "")
})
return key, vals
with ThreadPoolExecutor(max_workers=16) as executor:
futures = []
for key, source in self.sources.items():
classes.append({
"type_id": key,
"type_name": source["name"]
})
futures.append(executor.submit(load_class, key, source))
for future in as_completed(futures):
try:
key, vals = future.result()
filters[key] = [{
"key": "cateId",
"name": "分类",
"value": vals
}]
except:
pass
return {
"class": classes,
"filters": filters,
"list": []
}
def categoryContent(self, tid, pg, filter, extend):
if tid not in self.sources:
return {"list": []}
source = self.sources[tid]
cate_id = ""
if isinstance(extend, dict):
cate_id = extend.get("cateId", "")
url = f"{source['api']}?ac=detail&pg={pg}"
if cate_id:
url += f"&t={cate_id}"
html = self.fetch(url)
try:
data = json.loads(html)
except:
data = {}
result = []
for item in data.get("list", []):
result.append(
self.clean_item(
item,
tid,
source["name"],
False
)
)
return {
"list": result,
"page": data.get("page", pg),
"pagecount": data.get("pagecount", 1),
"limit": data.get("limit", 20),
"total": data.get("total", len(result))
}
def detailContent(self, ids):
if isinstance(ids, list):
ids = ids[0]
if "@@" not in ids:
return {"list": []}
source_key, real_id = ids.split("@@", 1)
if source_key not in self.sources:
return {"list": []}
source = self.sources[source_key]
url = f"{source['api']}?ac=detail&ids={real_id}"
html = self.fetch(url)
try:
data = json.loads(html)
except:
data = {}
result = []
for item in data.get("list", []):
cleaned = self.clean_item(
item,
source_key,
source["name"],
True
)
cleaned["vod_id"] = ids
result.append(cleaned)
return {"list": result}
def search_one(self, source_key, source, keyword, pg):
url = f"{source['api']}?ac=detail&wd={keyword}&pg={pg}"
html = self.fetch(url, 6)
try:
data = json.loads(html)
except:
data = {}
result = []
for item in data.get("list", []):
result.append(
self.clean_item(
item,
source_key,
source["name"],
False
)
)
return {
"list": result,
"pagecount": data.get("pagecount", 1)
}
def searchContent(self, key, quick=False, pg=1):
result = []
max_page = 1
with ThreadPoolExecutor(max_workers=20) as executor:
futures = []
for source_key, source in self.sources.items():
futures.append(
executor.submit(
self.search_one,
source_key,
source,
key,
pg
)
)
for future in as_completed(futures):
try:
data = future.result()
result.extend(data["list"])
if data["pagecount"] > max_page:
max_page = data["pagecount"]
except:
pass
return {
"list": result,
"page": pg,
"pagecount": max_page,
"limit": 40,
"total": 9999
}
def playerContent(self, flag, id, vipFlags):
return {
"parse": 0,
"playUrl": "",
"url": id,
"header": self.headers
}
def localProxy(self, param):
return [200, "text/plain", "ok"]
if __name__ == "__main__":
Spider().run()
FGBLH/海豚666.json
+24 -19
View File
@@ -46,17 +46,17 @@
"ext": "https://ghfast.top/https://raw.githubusercontent.com/IY-CPU/IY/main/lib/茫茫.png"
},
{
"key": "MGtv",
"name": "🐬芒果TV.py",
"type": 3,
"api": "https://ghfast.top/https://raw.githubusercontent.com/FGBLH/GHK/refs/heads/main/py/芒果TV.py"
"key": "MGtv",
"name": "🐬芒果TV.py",
"type": 3,
"api": "https://ghfast.top/https://raw.githubusercontent.com/FGBLH/GHK/refs/heads/main/py/芒果TV.py"
},
{
"key": "ppx",
"name": "🐬皮皮虾.py",
"type": 3,
"api": "https://ghfast.top/https://raw.githubusercontent.com/FGBLH/GHK/refs/heads/main/py/皮皮虾.py",
"ext": "http://43.248.117.123:4680"
"key": "ppx",
"name": "🐬皮皮虾.py",
"type": 3,
"api": "https://ghfast.top/https://raw.githubusercontent.com/FGBLH/GHK/refs/heads/main/py/皮皮虾.py",
"ext": "http://43.248.117.123:4680"
},
{
"key": "fY",
@@ -112,6 +112,18 @@
"type": 3,
"api": "https://ghfast.top/https://raw.githubusercontent.com/FGBLH/GHK/refs/heads/main/py/毒舌影视.py"
},
{
"key": "qw",
"name": "🐬七味.py(关梯)",
"type": 3,
"api": "https://ghfast.top/https://raw.githubusercontent.com/FGBLH/GHK/refs/heads/main/py/七味.py"
},
{
"key": "cjjh",
"name": "🐬采集聚合.py",
"type": 3,
"api": "https://ghfast.top/https://raw.githubusercontent.com/FGBLH/GHK/refs/heads/main/py/采集聚合.py"
},
{
"key": "kf",
"name": "🐬咖啡体育直播.py",
@@ -387,10 +399,10 @@
]
},
{
"key": "XFDM",
"name": "🐬稀饭动漫.py",
"key": "MiFun",
"name": "🐬MiFun动漫.py",
"type": 3,
"api": "https://ghfast.top/https://raw.githubusercontent.com/FGBLH/GHK/refs/heads/main/py/稀饭动漫.py"
"api": "https://ghfast.top/https://raw.githubusercontent.com/FGBLH/GHK/refs/heads/main/py/MiFun动漫.py"
},
{
"key": "cj_360资源",
@@ -536,13 +548,6 @@
"type": 3,
"api": "https://ghfast.top/https://raw.githubusercontent.com/FGBLH/GHK/refs/heads/main/py/123AV.py"
},
{
"key": "hanime1",
"name": "🐬hanime1.py｜🔞",
"type": 3,
"api": "https://ghfast.top/https://raw.githubusercontent.com/FGBLH/GHK/refs/heads/main/py/hanime1.py"
},
{
"key": "Xvideos",
"name": "🐬Xvideos.py｜🔞",
FGBLH/海豚无18加555.json
+14 -2
View File
@@ -78,7 +78,7 @@
},
{
"key": "GZ",
"name": "🐬瓜子APP.py(关梯)",
"name": "🐬瓜子APP.py",
"type": 3,
"api": "https://ghfast.top/https://raw.githubusercontent.com/FGBLH/GHK/refs/heads/main/py/瓜子APP.py"
},
@@ -112,6 +112,12 @@
"type": 3,
"api": "https://ghfast.top/https://raw.githubusercontent.com/FGBLH/GHK/refs/heads/main/py/毒舌影视.py"
},
{
"key": "qw",
"name": "🐬七味.py(关梯)",
"type": 3,
"api": "https://ghfast.top/https://raw.githubusercontent.com/FGBLH/GHK/refs/heads/main/py/七味.py"
},
{
"key": "kf",
"name": "🐬咖啡体育直播.py",
@@ -364,7 +370,13 @@
"港台动漫",
"海外动漫"
]
},
},
{
"key": "MiFun",
"name": "🐬MiFun动漫.py",
"type": 3,
"api": "https://ghfast.top/https://raw.githubusercontent.com/FGBLH/GHK/refs/heads/main/py/MiFun动漫.py"
},
{
"key": "cj_360资源",
"name": "🐬360丨短剧",
FGBLH/鱼壳海豚.json
+19 -1
View File
@@ -139,10 +139,22 @@
},
{
"key": "dsys",
"name": "🐬毒舌影视.py(关梯)",
"name": "🐬毒舌影视.py(关梯)[追剧]",
"type": 3,
"api": "https://ghfast.top/https://raw.githubusercontent.com/FGBLH/GHK/refs/heads/main/py/毒舌影视.py"
},
{
"key": "qw",
"name": "🐬七味.py(关梯)[追剧]",
"type": 3,
"api": "https://ghfast.top/https://raw.githubusercontent.com/FGBLH/GHK/refs/heads/main/py/七味.py"
},
{
"key": "cjjh",
"name": "🐬采集聚合.py[追剧]",
"type": 3,
"api": "https://ghfast.top/https://raw.githubusercontent.com/FGBLH/GHK/refs/heads/main/py/采集聚合.py"
},
{
"key": "fY",
"name": "🐬枫叶影院.py(关梯)[追剧]",
@@ -469,6 +481,12 @@
"海外动漫"
]
},
{
"key": "MiFun",
"name": "🐬MiFun动漫.py",
"type": 3,
"api": "https://ghfast.top/https://raw.githubusercontent.com/FGBLH/GHK/refs/heads/main/py/MiFun动漫.py"
},
{
"key": "cj_360资源",
"name": "🐬360[短剧]",
FGBLH/鱼壳海豚无18加.json
+14 -2
View File
@@ -102,7 +102,7 @@
},
{
"key": "GZ",
"name": "🐬瓜子APP.py(关梯)",
"name": "🐬瓜子APP.py",
"type": 3,
"api": "https://ghfast.top/https://raw.githubusercontent.com/FGBLH/GHK/refs/heads/main/py/瓜子APP.py"
},
@@ -136,6 +136,12 @@
"type": 3,
"api": "https://ghfast.top/https://raw.githubusercontent.com/FGBLH/GHK/refs/heads/main/py/毒舌影视.py"
},
{
"key": "qw",
"name": "🐬七味.py(关梯)",
"type": 3,
"api": "https://ghfast.top/https://raw.githubusercontent.com/FGBLH/GHK/refs/heads/main/py/七味.py"
},
{
"key": "fY",
"name": "🐬枫叶影院.py(关梯子使用)",
@@ -440,7 +446,13 @@
"港台动漫",
"海外动漫"
]
},
},
{
"key": "MiFun",
"name": "🐬MiFun动漫.py",
"type": 3,
"api": "https://ghfast.top/https://raw.githubusercontent.com/FGBLH/GHK/refs/heads/main/py/MiFun动漫.py"
},
{
"key": "cj_360资源",
"name": "🐬360丨短剧",