XC-Pro/sg
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

上传文件至「py」

Browse Source
This commit is contained in:
XC-Pro
2026-08-14 11:25:02 +02:00
parent 9e90ad0c97
commit 1cd5c93451
5 changed files with 1913 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
py/共青春.py
+259
View File
@@ -0,0 +1,259 @@
# -*- coding: utf-8 -*-
import sys
import re
import json
from urllib.parse import urljoin, quote
sys.path.append('..')
try:
from base.spider import Spider
except ImportError:
class Spider:
def fetch(self, url, headers=None, **kw):
import requests as rq
kw.pop('timeout', None)
r = rq.get(url, headers=headers, timeout=15, **kw)
r.encoding = 'utf-8'
return r
HOST = "https://gqc.ink"
UA = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
CATEGORIES = {
"dianying": "电影", "lianxuju": "连续剧", "zongyi": "综艺",
"dongman": "动漫", "duanju": "短剧",
}
class Spider(Spider):
def init(self, extend=""):
global HOST
try:
r = self.fetch(HOST, headers={"User-Agent": UA}, timeout=15000)
if hasattr(r, 'url') and r.url and r.url != HOST.rstrip("/"):
HOST = r.url.rstrip("/")
except:
pass
def homeContent(self, filter=False):
r = {"class": [], "list": []}
for k, v in CATEGORIES.items():
r["class"].append({"type_id": k, "type_name": v})
return r
def homeVideoContent(self):
return {"list": []}
def categoryContent(self, tid, pg=1, filter=False, extend=""):
pn = 1
try: pn = max(int(str(pg)), 1)
except: pass
cat = str(tid)
if cat not in CATEGORIES:
cat = "dianying"
slug = cat
try:
if pn > 1:
url = f"{HOST}/vodshow/{slug}--------{pn}---.html"
else:
url = f"{HOST}/vodshow/{slug}--------1---.html"
r = self.fetch(url, headers={"User-Agent": UA}, timeout=30000)
html = r.text if hasattr(r, 'text') else str(r)
items = self._items(html)
return {"page": pn, "pagecount": self._pagecount(html), "limit": 50, "total": len(items), "list": items}
except:
return {"page": pn, "pagecount": 1, "limit": 50, "total": 0, "list": []}
def detailContent(self, ids):
if isinstance(ids, list):
vid = ids[0] if ids else ""
else:
vid = str(ids) if ids else ""
m = re.search(r'(\d+)', str(vid))
vid = m.group(1) if m else ""
if not vid:
return {"list": []}
try:
r = self.fetch(f"{HOST}/neirong/{vid}.html", headers={"User-Agent": UA}, timeout=30000)
h = r.text if hasattr(r, 'text') else str(r)
except:
return {"list": []}
d = {"vod_id": vid, "vod_name": "", "vod_pic": "", "vod_year": "",
"vod_area": "", "vod_class": "", "vod_director": "", "vod_actor": "",
"vod_content": "", "vod_remarks": "", "vod_play_from": "", "vod_play_url": ""}
t1 = re.search(r'<h1[^>]*>(.*?)</h1>', h, re.S)
if t1:
clean = re.sub(r'<[^>]+>', ' ', t1.group(1)).strip()
d["vod_name"] = clean.split('\n')[0].strip()
if not d["vod_name"]:
t2 = re.search(r'<title>(.*?)</title>', h)
if t2: d["vod_name"] = t2.group(1).split("-")[0].strip()
p = re.search(r'data-original="(https?://[^"]+\.(?:jpg|jpeg|png|webp))"', h, re.I)
if not p:
p = re.search(r'data-background="(https?://[^"]+\.(?:jpg|jpeg|png|webp))"', h, re.I)
if not p:
p = re.search(r'<img[^>]*src="(https?://[^"]+\.(?:jpg|jpeg|png|webp))"', h, re.I)
if p: d["vod_pic"] = p.group(1)
desc_m = re.search(r'class="fed-part-esan[^"]*"[\s\S]*?<span[^>]*>简介[：:]?</span>\s*&nbsp;.*?(?:<a[^>]*>[\s\S]*?</a>)\s*([^<]+)', h)
if not desc_m:
desc_m = re.search(r'class="fed-part-esan[^"]*"[\s\S]*?<span[^>]*>简介[：:]?</span>\s*&nbsp;\s*([^<]+)', h)
if desc_m:
desc = re.sub(r'\s+', ' ', desc_m.group(1)).strip()[:500]
d["vod_content"] = desc
for mi in re.finditer(r'class="fed-deta-info[^"]*"[\s\S]*?>(.*?)</li>', h, re.S):
t = re.sub(r'<[^>]+>', '', mi.group(1)).strip()
if "导演" in t: d["vod_director"] = t.split("：")[-1].split(":")[-1].strip()
elif "主演" in t: d["vod_actor"] = t.split("：")[-1].split(":")[-1].strip()
elif "年份" in t or "上映" in t:
ym = re.search(r'(\d{4})', t)
if ym: d["vod_year"] = ym.group(1)
elif "备注" in t or "更新" in t: d["vod_remarks"] = t
try:
pf, pu = [], []
tops = re.search(r'fed-drop-tops(.*?)fed-drop-btms', h, re.S)
route_buttons = []
if tops:
for bm in re.finditer(r'href="/bofang/\d+-(\d+)-1\.html"[\s\S]*?>(.*?)<span[^>]*>(\d+)<', tops.group(1), re.S):
route = bm.group(1)
name = re.sub(r'<[^>]+>', '', bm.group(2)).strip()
count = int(bm.group(3))
if name and count > 0:
route_buttons.append((route, name, count))
vis = re.search(r'fed-play-item fed-drop-item fed-visible(.*?)</ul>\s*</div>', h, re.S)
default_eps = []
default_route = None
if vis:
for em in re.finditer(r'href="/bofang/\d+-(\d+)-(\d+)\.html"[\s\S]*?>([^<]+)<', vis.group(1)):
r2, nid, name = em.group(1), em.group(2), em.group(3).strip()
if not default_route:
default_route = r2
default_eps.append((int(nid), name))
if not route_buttons:
items = re.findall(r'href="(/bofang/\d+-\d+-\d+\.html)"[\s\S]*?>([^<]+)<', h)
routes = {}
for href, name in items:
name = name.strip()
if not name or "报错" in name or "剧情" in name or "简介" in name or "立即播放" in name:
continue
rm = re.search(r'/bofang/\d+-(\d+)-(\d+)\.html', href)
if rm:
rt, nid = rm.group(1), rm.group(2)
if rt not in routes:
routes[rt] = []
routes[rt].append(f"{name}${urljoin(HOST, href)}")
for rt in sorted(routes):
pf.append(f"源{len(pf)+1}")
pu.append("#".join(routes[rt]))
else:
for route, name, count in route_buttons:
eps_list = []
if default_route == route and default_eps:
for nid, ename in default_eps:
eps_list.append(f"{ename}${urljoin(HOST, f'/bofang/{vid}-{route}-{nid}.html')}")
else:
if default_eps:
for nid, ename in default_eps:
if nid <= count:
eps_list.append(f"{ename}${urljoin(HOST, f'/bofang/{vid}-{route}-{nid}.html')}")
else:
for nid in range(1, count + 1):
eps_list.append(f"第{nid}集${urljoin(HOST, f'/bofang/{vid}-{route}-{nid}.html')}")
if eps_list:
pf.append(name)
pu.append("#".join(eps_list))
if pf:
d["vod_play_from"] = "$$$".join(pf)
d["vod_play_url"] = "$$$".join(pu)
except:
pass
return {"list": [d]}
def searchContent(self, key, quick=False, pg="1"):
return {"list": []}
def playerContent(self, flag, id, vipFlags=None):
a, b = str(flag), str(id) if id else ""
if a.startswith("http") or "/bofang/" in a:
url = a
elif b.startswith("http") or "/bofang/" in b:
url = b
elif a.startswith("/"):
url = urljoin(HOST, a)
elif b.startswith("/"):
url = urljoin(HOST, b)
else:
url = a
try:
r = self.fetch(url, headers={"User-Agent": UA}, timeout=30000)
h = r.text if hasattr(r, 'text') else str(r)
except:
return {"url": ""}
dm = re.search(r'"dmId"\s*:\s*"([^"]+)"', h)
if dm:
u = dm.group(1)
if u.startswith("http"):
if ".m3u8" in u:
return {"url": u}
return {"url": u}
m3u8 = re.search(r'(https?://[^\s"\'<>]+\.m3u8)', h)
if m3u8:
return {"url": m3u8.group(1)}
return {"url": ""}
def localProxy(self, param):
pass
def _pagecount(self, html):
pc = 1
# 优先用 data-total 属性 (fed模板分页跳转按钮)
dt = re.findall(r'data-total="(\d+)"', html)
for t in dt:
try: pc = max(pc, int(t))
except: pass
# /vodshow/ 格式分页: href="/vodshow/dianying--------2---.html"
pages_vod = re.findall(r'href="/vodshow/[^"]*-(\d+)-*--*\.html"', html)
for p in pages_vod:
try:
n = int(p)
if n > 1 and n < 10000:
pc = max(pc, n)
except: pass
# /fenlei/ 格式分页 (兼容旧格式)
pages_fen = re.findall(r'href="/fenlei/[^"]*-?(\d+)\.html"', html)
for p in pages_fen:
try:
n = int(p)
if n > 1 and n < 10000:
pc = max(pc, n)
except: pass
return pc
def _items(self, html):
items, seen = [], set()
# 只取 data-original 的列表项，不取 data-background 的轮播项
# 轮播项标题和封面不在同一个结构里，混入会导致错位
cover_map = {}
for m in re.finditer(r'data-original="(https?://[^"]+)"', html, re.I):
# 找到 data-original 前面最近的 href (取最后一个，不是第一个)
before = html[max(0, m.start()-500):m.start()]
hms = re.findall(r'href="(/neirong/(\d+)\.html)"', before)
if hms:
hm = hms[-1]
vid = hm[1]
if vid not in cover_map:
cover_map[vid] = (hm[0], m.group(1))
# 标题: fed-list-title
titles = {}
for tm in re.finditer(r'class="[^"]*fed-list-title[^"]*"[\s\S]*?href="/neirong/(\d+)\.html"[\s\S]*?>\s*(.*?)\s*</a>', html, re.S):
name = re.sub(r'<[^>]+>', '', tm.group(2)).strip()
if name:
titles[tm.group(1)] = name[:50]
# 合并: 只用列表项的封面
for vid, (href, pic) in cover_map.items():
if vid in seen:
continue
seen.add(vid)
items.append({
"vod_id": vid, "vod_name": titles.get(vid, ""),
"vod_pic": pic, "vod_remarks": "", "vod_url": urljoin(HOST, href),
})
return items
py/枫叶影院3.py
+375
View File
@@ -0,0 +1,375 @@
"""
@header({
searchable: 1,
filterable: 1,
quickSearch: 1,
title: '茶杯狐-TJTCDL',
lang: 'hipy',
})
"""
# -*- coding: utf-8 -*-
# !/usr/bin/python
import requests
import base64
import random
import re
import json
import sys
import urllib.parse
import ssl
import urllib3
import hashlib
from bs4 import BeautifulSoup
from requests.adapters import HTTPAdapter
from urllib3.util.ssl_ import create_urllib3_context
urllib3.disable_warnings()
sys.path.append('..')
from base.spider import Spider
class TLSAdapter(HTTPAdapter):
def init_poolmanager(self, *args, **kwargs):
ciphers = (
'ECDHE-ECDSA-AES128-GCM-SHA256:ECDHE-RSA-AES128-GCM-SHA256:'
'ECDHE-ECDSA-AES256-GCM-SHA384:ECDHE-RSA-AES256-GCM-SHA384:'
'ECDHE-ECDSA-CHACHA20-POLY1305:ECDHE-RSA-CHACHA20-POLY1305:'
'DHE-RSA-AES128-GCM-SHA256:DHE-RSA-AES256-GCM-SHA384'
)
context = create_urllib3_context(ciphers=ciphers)
context.check_hostname = False
context.verify_mode = ssl.CERT_NONE
kwargs['ssl_context'] = context
return super(TLSAdapter, self).init_poolmanager(*args, **kwargs)
class Spider(Spider):
def __init__(self):
super(Spider, self).__init__()
self.session = requests.Session()
self.session.verify = False
self.session.mount('https://', TLSAdapter())
self.host = "https://www.tjtcdl.com"
self.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': f'{self.host}/',
'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8',
'Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.8',
}
def getName(self):
return "茶杯狐-TJTCDL"
def init(self, extend):
pass
def homeContent(self, filter):
classes = [
{"type_id": "1", "type_name": "电影"},
{"type_id": "2", "type_name": "电视剧"},
{"type_id": "4", "type_name": "动漫"},
{"type_id": "3", "type_name": "综艺"},
{"type_id": "5", "type_name": "热门短剧"},
]
filter_dict = {}
years = [{"n": "全部", "v": ""}] + [{"n": str(y), "v": str(y)} for y in range(2026, 2003, -1)]
orders = [
{"n": "按最新", "v": "time"},
{"n": "按最热", "v": "hits"},
{"n": "按评分", "v": "score"}
]
movie_classes = ["动作", "喜剧", "爱情", "科幻", "恐怖", "剧情", "战争", "惊悚", "悬疑", "犯罪", "奇幻", "冒险",
"动画", "武侠"]
movie_areas = ["大陆", "香港", "台湾", "美国", "韩国", "日本", "泰国", "新加坡", "马来西亚", "印度", "英国",
"法国", "加拿大", "西班牙", "俄罗斯", "其它"]
tv_classes = ["古装", "战争", "青春偶像", "喜剧", "家庭", "犯罪", "动作", "奇幻", "剧情", "历史", "经典",
"乡村", "情景", "商战", "网剧", "其他"]
tv_areas = ["内地", "韩国", "香港", "台湾", "日本", "美国", "泰国", "英国", "新加坡", "其他"]
comic_classes = ["科幻", "热血", "推理", "搞笑", "冒险", "萝莉", "校园", "动作", "机战", "运动", "战争", "少年",
"少女"]
show_classes = ["脱口秀", "真人秀", "搞笑", "访谈", "生活", "晚会", "美食", "游戏", "亲子", "旅游", "音乐",
"舞蹈"]
def create_filter(classes_list, areas_list):
return [
{"key": "class", "name": "类型",
"value": [{"n": "全部", "v": ""}] + [{"n": c, "v": c} for c in classes_list]},
{"key": "area", "name": "地区",
"value": [{"n": "全部", "v": ""}] + [{"n": a, "v": a} for a in areas_list]},
{"key": "year", "name": "年份", "value": years},
{"key": "by", "name": "排序", "value": orders}
]
filter_dict["1"] = create_filter(movie_classes, movie_areas)
filter_dict["2"] = create_filter(tv_classes, tv_areas)
filter_dict["4"] = create_filter(comic_classes, tv_areas)
filter_dict["3"] = create_filter(show_classes, tv_areas)
filter_dict["5"] = create_filter(["女频", "男频", "复仇", "甜宠", "穿越", "逆袭", "战神", "脑洞"],
["内地", "其他"])
return {"class": classes, "filters": filter_dict}
def homeVideoContent(self):
return {'list': []}
def categoryContent(self, cid, pg, filter, ext):
page = int(pg)
ext = ext or {}
area = urllib.parse.quote(ext.get('area', '')) if ext.get('area') else ''
by = ext.get('by', '')
class_name = urllib.parse.quote(ext.get('class', '')) if ext.get('class') else ''
lang = urllib.parse.quote(ext.get('lang', '')) if ext.get('lang') else ''
letter = ext.get('letter', '')
year = ext.get('year', '')
url = f"{self.host}/cupfox-list/{cid}-{area}-{by}-{class_name}-{lang}-{letter}---{page}---{year}.html"
res = self.session.get(url, headers=self.headers)
soup = BeautifulSoup(res.text, 'html.parser')
videos = []
for item in soup.select('.public-list-box'):
link = item.select_one('.public-list-exp')
if not link: continue
vid_match = re.search(r'/chabeihu/(\d+)\.html', link.get('href', ''))
if not vid_match: continue
pic_img = link.select_one('img')
note_tag = item.select_one('.public-list-prb')
videos.append({
"vod_id": vid_match.group(1),
"vod_name": link.get('title', '').strip(),
"vod_pic": pic_img.get('data-src') or pic_img.get('src') or '' if pic_img else '',
"vod_remarks": note_tag.text.strip() if note_tag else ''
})
return {'list': videos, 'page': page, 'pagecount': 9999, 'limit': 90, 'total': 999999}
def detailContent(self, ids):
did = ids[0]
url = f"{self.host}/chabeihu/{did}.html"
res = self.session.get(url, headers=self.headers, timeout=10)
soup = BeautifulSoup(res.text, 'html.parser')
name, state, actor, director, year, content = "", "", "", "", "", ""
for li in soup.select('.info-parameter li'):
em = li.select_one('em')
if not em: continue
em_text = em.text.strip()
val = li.text.replace(em_text, '').replace('\xa0', ' ').strip()
if '片名' in em_text:
name = val
elif '状态' in em_text:
state = val
elif '主演' in em_text:
actor = val
elif '导演' in em_text:
director = val
elif '年份' in em_text:
year = val
elif '简介' in em_text:
content = val
if not name:
title_tag = soup.select_one('.this-desc-title')
name = title_tag.text.strip() if title_tag else ''
play_from, play_url = [], []
sources = [s.text.replace(s.select_one('.badge').text, '').strip() if s.select_one('.badge') else s.text.strip()
for s in soup.select('.anthology-tab .swiper-slide')]
for idx, box in enumerate(soup.select('.anthology-list-box')):
eps = [
f"{a.text.strip()}${self.host + a.get('href') if not a.get('href').startswith('http') else a.get('href')}"
for a in box.select('li a') if a.get('href')]
if eps:
eps.reverse()
play_from.append(sources[idx] if idx < len(sources) else f"线路{idx + 1}")
play_url.append('#'.join(eps))
return {'list': [{
"vod_id": did,
"vod_name": name,
"vod_actor": actor,
"vod_director": director,
"vod_content": content,
"vod_remarks": state,
"vod_year": year,
"vod_play_from": '$$$'.join(play_from),
"vod_play_url": '$$$'.join(play_url)
}]}
def playerContent(self, flag, id, vipFlags):
try:
res = self.session.get(id, headers=self.headers, timeout=5)
match = re.search(r'var player_aaaa=(.*?)</script>', res.text)
if not match:
return {'parse': 0, 'url': ''}
player_data = json.loads(match.group(1))
durl = player_data.get('url', '')
encrypt = player_data.get('encrypt', 0)
from_flag = player_data.get('from', '')
if encrypt == 1:
durl = urllib.parse.unquote(durl)
elif encrypt == 2:
durl = urllib.parse.unquote(durl)
durl = base64.b64decode(durl).decode('utf-8')
durl = urllib.parse.unquote(durl)
if durl.startswith('http') and ('.m3u8' in durl or '.mp4' in durl):
return {'parse': 0, 'url': durl}
config_url = f"{self.host}/static/js/playerconfig.js"
config_res = self.session.get(config_url, headers=self.headers, verify=False, timeout=5)
parse_api = ""
if from_flag:
m = re.search(f'"{from_flag}":\\{{[^}}]*"parse":"([^"]+)"', config_res.text)
if m: parse_api = m.group(1).replace('\\/', '/')
if not parse_api:
m = re.search(r'"parse":"(http[^"]+)"', config_res.text)
if m: parse_api = m.group(1).replace('\\/', '/')
if not parse_api:
parse_api = "https://fgsrg.hzqingshan.com/player/?url="
iframe_url = parse_api + durl
print(f"[*] 2. 锁定隐藏解析服务器(iframe): {iframe_url}")
iframe_headers = self.headers.copy()
iframe_headers['Referer'] = id
iframe_res = self.session.get(iframe_url, headers=iframe_headers, verify=False, timeout=5)
iframe_soup = BeautifulSoup(iframe_res.text, 'html.parser')
player_data_tag = iframe_soup.select_one('#player-data')
if not player_data_tag:
return {'parse': 1, 'url': iframe_url}
token = player_data_tag.get('data-te', '')
bt = player_data_tag.get('data-bt', '/player/')
api_base = urllib.parse.urlparse(parse_api)
api_host = f"{api_base.scheme}://{api_base.netloc}"
api_url = f"{api_host}{bt}mplayer.php"
post_data = {'url': durl, 'token': token}
api_headers = self.headers.copy()
api_headers['Content-Type'] = 'application/x-www-form-urlencoded; charset=UTF-8'
api_headers['X-Requested-With'] = 'XMLHttpRequest'
api_headers['Referer'] = iframe_url
api_headers['Origin'] = api_host
api_res = self.session.post(api_url, data=post_data, headers=api_headers, verify=False, timeout=10)
if api_res.status_code == 200:
api_json = api_res.json()
print(f"[*] 5. 接口解密成功: {api_json}")
real_url = api_json.get('url') or api_json.get('data', {}).get('url', '')
urlmode = str(api_json.get('urlmode') or api_json.get('data', {}).get('urlmode', ''))
if urlmode == '1':
real_url = self.js_decrypt1(real_url)
elif urlmode == '2':
real_url = self.js_decrypt2(real_url)
elif urlmode == '3':
real_url = self.js_decrypt3(real_url)
for _ in range(3):
if real_url.startswith('WyJ') and '/' in real_url:
real_url = self.js_decrypt3(real_url)
else:
break
if real_url:
print(f"[*] 6. 🎉斩获最终真实 M3U8: {real_url}")
return {'parse': 0 if ('.m3u8' in real_url or '.mp4' in real_url) else 1, 'url': real_url}
return {'parse': 1, 'url': iframe_url}
except Exception as e:
return {'parse': 1, 'url': id}
def searchContent(self, key, quick, pg="1"):
search_url = f'{self.host}/cupfox-search/-------------.html'
res = self.session.get(search_url, params={'wd': key}, headers=self.headers)
soup = BeautifulSoup(res.text, 'html.parser')
videos = []
for item in soup.select('.public-list-box'):
link = item.select_one('.public-list-exp')
if not link: continue
vid_match = re.search(r'/chabeihu/(\d+)\.html', link.get('href', ''))
if not vid_match: continue
pic_img = link.select_one('img')
note_tag = item.select_one('.public-list-prb')
videos.append({
"vod_id": vid_match.group(1),
"vod_name": link.get('title', '').strip(),
"vod_pic": pic_img.get('data-src') or pic_img.get('src') or '' if pic_img else '',
"vod_remarks": note_tag.text.strip() if note_tag else ''
})
return {'list': videos, 'page': int(pg), 'pagecount': 1, 'limit': len(videos), 'total': len(videos)}
def js_decrypt1(self, data):
try:
key = hashlib.md5(b'test').hexdigest()
dec1 = base64.b64decode(data)
code = bytearray([dec1[i] ^ ord(key[i % len(key)]) for i in range(len(dec1))])
return base64.b64decode(code).decode('utf-8')
except:
return data
def js_decrypt2(self, data):
staticchars = "PXhw7UT1B0a9kQDKZsjIASmOezxYG4CHo5Jyfg2b8FLpEvRr3WtVnlqMidu6cN"
try:
dec = base64.b64decode(data).decode('utf-8', errors='ignore')
return "".join(
[staticchars[(staticchars.find(dec[i]) + 59) % 62] if staticchars.find(dec[i]) != -1 else dec[i] for i
in range(1, len(dec), 3)])
except:
return data
def js_decrypt3(self, data):
def fix_b64(s):
return s + '=' * (4 - len(s) % 4) if len(s) % 4 else s
try:
parts = data.split('/')
if len(parts) >= 3:
arr1 = json.loads(base64.b64decode(fix_b64(parts[0])).decode('utf-8'))
arr2 = json.loads(base64.b64decode(fix_b64(parts[1])).decode('utf-8'))
cipher = base64.b64decode(fix_b64('/'.join(parts[2:]))).decode('utf-8', errors='ignore')
return "".join([arr1[arr2.index(c)] if c in arr2 else c for c in cipher])
except:
pass
return data
py/瓜娃子.py
+510
View File
@@ -0,0 +1,510 @@
# coding = utf-8
#!/usr/bin/python
import re
import sys
import json
import time
import base64
import requests
import hashlib
import random
import string
import urllib.parse
from Crypto.Cipher import AES
from Crypto.Util.Padding import pad, unpad
from Crypto.PublicKey import RSA
from Crypto.Cipher import PKCS1_v1_5
from base.spider import Spider
sys.path.append('..')
class Spider(Spider):
def __init__(self):
self.name = "瓜子"
self.hosts = [
'https://apinew.uozvr.com',
'https://api.w32z7vtd.com',
'https://api.6a7nnf7.com',
'https://api.umygrx3.com',
'https://api.rmedphk.com'
]
self.host_index = 0
self.host = self.hosts[self.host_index]
# AES 固定密钥（与Java版一致）
self.AES_KEY = 'OITxa5OqAYjhswxx'
self.AES_IV = 'rCMNwZASNBKZ8mXV'
# RSA 公钥/私钥
self.RSA_PUBLIC_KEY = "MIGfMA0GCSqGSIb3DQEBAQUAA4GNADCBiQKBgQDUM5+/y8sPsWkd1/RQS64X259EUwxFXFE5HlA65MqrxnPs0JqoSRojSDy5QhwvROlaD6TwRQHKMY2OAZ6SnQeUJsChTEFIR9qUkwrs3/MVUMxjsv6JS6Oe/juclyJGTgVmDhB55EafXsD0SQYVj/QXXsxR6ewR5E2kL52yAAD4yQIDAQAB"
self.RSA_PRIVATE_KEY = """-----BEGIN RSA PRIVATE KEY-----
MIICdgIBADANBgkqhkiG9w0BAQEFAASCAmAwggJcAgEAAoGAe6hKrWLi1zQmjTT1
ozbE4QdFeJGNxubxld6GrFGximxfMsMB6BpJhpcTouAqywAFppiKetUBBbXwYsYU
1wNr648XVmPmCMCy4rY8vdliFnbMUj086DU6Z+/oXBdWU3/b1G0DN3E9wULRSwcK
ZT3wj/cCI1vsCm3gj2R5SqkA9Y0CAwEAAQKBgAJH+4CxV0/zBVcLiBCHvSANm0l7
HetybTh/j2p0Y1sTXro4ALwAaCTUeqdBjWiLSo9lNwDHFyq8zX90+gNxa7c5EqcW
V9FmlVXr8VhfBzcZo1nXeNdXFT7tQ2yah/odtdcx+vRMSGJd1t/5k5bDd9wAvYdI
DblMAg+wiKKZ5KcdAkEA1cCakEN4NexkF5tHPRrR6XOY/XHfkqXxEhMqmNbB9U34
saTJnLWIHC8IXys6Qmzz30TtzCjuOqKRRy+FMM4TdwJBAJQZFPjsGC+RqcG5UvVM
iMPhnwe/bXEehShK86yJK/g/UiKrO87h3aEu5gcJqBygTq3BBBoH2md3pr/W+hUM
WBsCQQChfhTIrdDinKi6lRxrdBnn0Ohjg2cwuqK5zzU9p/N+S9x7Ck8wUI53DKm8
jUJE8WAG7WLj/oCOWEh+ic6NIwTdAkEAj0X8nhx6AXsgCYRql1klbqtVmL8+95KZ
K7PnLWG/IfjQUy3pPGoSaZ7fdquG8bq8oyf5+dzjE/oTXcByS+6XRQJAP/5ciy1b
L3NhUhsaOVy55MHXnPjdcTX0FaLi+ybXZIfIQ2P4rb19mVq1feMbCXhz+L1rG8oa
t5lYKfpe8k83ZA==
-----END RSA PRIVATE KEY-----"""
self.DEVICE_OLD_KEY = "aLFBMWpxBrIDAD1Si/KVvm41"
# 设备信息（随机生成）
self.deviceId = str(864150060000000 + random.randint(0, 9999))
self.deviceKey = ''.join(random.choices('0123456789ABCDEF', k=40)) # 20字节hex大写
self.token = ""
self.token_id = ""
self.registered = False
self.header = {
'User-Agent': 'Lavf/57.83.100',
'code': 'GZ0369',
'deviceId': self.deviceId,
'lang': 'zh_cn',
'Cache-Control': 'no-cache',
'Content-Type': 'application/x-www-form-urlencoded',
'Version': '2604028',
'PackageName': 'com.ae06aebdbb.y286327f5a.ofe849883320260517',
'Ver': '3.0.3.2',
'api-ver': '3.0.3.2',
'Referer': self.host
}
self.cache = {}
self.cache_timeout = 300
# 初始化token
self.init_token()
def getName(self):
return self.name
def init(self, extend=''):
pass
# ---------- 设备注册与认证 ----------
def init_token(self):
"""初始化token：注册设备 -> 刷新"""
print("===== 初始化设备认证 =====")
try:
if not self.registered:
self.sign_up()
# 刷新获取最终token
self.refresh_token()
except Exception as e:
print(f"初始化token失败: {e}")
# 兜底使用原有硬编码（几乎没用）
self.token = '024212ef0975c5306a1434e113a46463.bc77313e11a248558a6ca244ca980944ec3421fa480c50e0229ad91f1cb15aea582603202cd71796885c9e5163e500f1b72f737059aff1ddb8beea47c5a331d6760540345b7f88b2302a0e6e09589f9dcf3ff9175d8c905f990203f5fc04748008ea7a366571cbf5b09509a873dcfba3cf1d5590385f5f7ef6e01d1850974aa220eb5178c89e61c24411af9b9a19435e.06fde789ece48d9b33c5dc857e04e9b5838f08264d928b87237d3476c4484b46'
def sign_up(self):
"""注册设备"""
print("注册新设备...")
params = {
"new_key": self.deviceKey,
"old_key": self.DEVICE_OLD_KEY,
"phone_type": 1,
"code": ""
}
result = self._auth_request('/App/Authentication/Device/signUp', params)
self._apply_auth(result)
self.registered = True
def sign_in(self):
"""登录设备"""
print("设备登录...")
params = {
"new_key": self.deviceKey,
"old_key": self.DEVICE_OLD_KEY
}
result = self._auth_request('/App/Authentication/Device/signIn', params)
self._apply_auth(result)
def _apply_auth(self, result):
"""从认证响应中提取token"""
new_token = result.get('token', '')
if not new_token:
raise Exception("认证失败，无token返回: {}".format(result))
self.token = new_token
new_token_id = result.get('app_user_id', '')
if new_token_id:
self.token_id = new_token_id
print(f"获取token成功, token前缀: {self.token[:30]}...")
def refresh_token(self):
"""刷新token"""
print("刷新token...")
result = self._auth_request('/App/Authentication/Authenticator/refresh', {})
self._apply_auth(result)
def _auth_request(self, path, params):
"""认证类请求（不需要ensure_token）"""
return self._send_encrypted_request(params, path, is_auth=True)
# ---------- 业务请求核心（修复加密与签名） ----------
def ensure_token(self):
"""确保token有效，如未就绪则重新获取"""
if not self.token or not self.token_id:
if self.registered:
self.sign_in()
else:
self.sign_up()
self.refresh_token()
def _send_encrypted_request(self, data, path, is_auth=False):
"""
发送加密请求，返回解密后的字典
:param data: 业务参数字典
:param path: 请求路径
:param is_auth: 是否为认证类请求（signUp/signIn/refresh），此时不使用ensure_token
"""
try:
if not is_auth:
self.ensure_token()
# 1. 将参数转为JSON并AES加密
json_params = json.dumps(data)
encrypted = self.aes_encrypt(json_params, self.AES_KEY, self.AES_IV)
request_key = encrypted.upper() # Java中是bytesToHex(encrypted).toUpperCase()
# 2. 生成keys (RSA加密 iv/key JSON)
key_json = json.dumps({"iv": self.AES_IV, "key": self.AES_KEY})
keys = self.rsa_encrypt(key_json, self.RSA_PUBLIC_KEY)
# 3. 生成签名
t = str(int(time.time()))
sign_str = f"token_id=,token={self.token},phone_type=1,request_key={request_key},app_id=1,time={t},keys={keys}*&zvdvdvddbfikkkumtmdwqppp?|4Y!s!2br"
signature = self.get_md5(sign_str) # 已改为大写
# 4. 构建请求体
body = {
'token': self.token,
'token_id': '',
'phone_type': '1',
'time': t,
'phone_model': 'xiaomi-25031', # 与Java版保持一致
'keys': keys,
'request_key': request_key,
'signature': signature,
'app_id': '1',
'ad_version': '1'
}
# 5. 发送请求
url = f"{self.host}{path}"
response = self.post(url, headers=self.header, data=body, timeout=10)
if response.status_code != 200:
raise Exception(f"HTTP {response.status_code}")
resp_json = response.json()
# 检查业务code（若不为200可能token过期）
if 'code' in resp_json and resp_json['code'] != 200:
print(f"业务错误码: {resp_json['code']}, 信息: {resp_json}")
# 如果不是认证请求，尝试重新获取token后重试一次（这里简单处理，外层get_data已有重试）
raise Exception("业务错误")
data_section = resp_json.get('data')
if not data_section:
raise Exception("响应缺少data字段")
encrypted_response = data_section.get('response_key', '')
encrypted_keys = data_section.get('keys', '')
# 6. 解密响应
decrypted_keys_json = self.rsa_decrypt(encrypted_keys, self.RSA_PRIVATE_KEY)
key_info = json.loads(decrypted_keys_json)
resp_key = key_info['key']
resp_iv = key_info['iv']
decrypted_data = self.aes_decrypt(encrypted_response, resp_key, resp_iv)
return json.loads(decrypted_data)
except Exception as e:
print(f"请求失败 [{path}]: {e}")
return None
def get_data(self, data, path, use_cache=True):
"""带重试和域名轮询的数据获取（保持原框架）"""
try:
cache_key = f"{path}_{hash(str(data))}" if use_cache else None
if use_cache and cache_key in self.cache:
cached_data, timestamp = self.cache[cache_key]
if time.time() - timestamp < self.cache_timeout:
return cached_data
for attempt in range(3):
tried = 0
while tried < len(self.hosts):
self.host = self.hosts[self.host_index]
self.header['Referer'] = self.host
result = self._send_encrypted_request(data, path)
if result is not None:
print(f"请求成功: {path}, 域名: {self.host}")
if use_cache and cache_key:
self.cache[cache_key] = (result, time.time())
return result
# 切换到下一个域名
self.host_index = (self.host_index + 1) % len(self.hosts)
tried += 1
# 所有域名失败，尝试重新认证并重试
if attempt < 2:
print("所有域名失败，尝试重新认证...")
try:
self.ensure_token()
except:
pass
self.host_index = 0
else:
break
return None
except Exception as e:
print(f"get_data异常: {e}")
return None
# ---------- 加解密工具 ----------
def aes_encrypt(self, text, key, iv):
try:
key_bytes = key.encode('utf-8')
iv_bytes = iv.encode('utf-8')
cipher = AES.new(key_bytes, AES.MODE_CBC, iv_bytes)
encrypted = cipher.encrypt(pad(text.encode('utf-8'), AES.block_size))
return encrypted.hex().upper()
except Exception as e:
print(f"AES加密失败: {e}")
return ""
def aes_decrypt(self, text, key, iv):
try:
key_bytes = key.encode('utf-8')
iv_bytes = iv.encode('utf-8')
cipher = AES.new(key_bytes, AES.MODE_CBC, iv_bytes)
encrypted_bytes = bytes.fromhex(text)
decrypted = unpad(cipher.decrypt(encrypted_bytes), AES.block_size)
return decrypted.decode('utf-8')
except Exception as e:
print(f"AES解密失败: {e}")
return ""
def rsa_encrypt(self, text, public_key_str):
"""RSA公钥加密（PKCS1v1.5）"""
try:
key = RSA.import_key("-----BEGIN PUBLIC KEY-----\n" + public_key_str + "\n-----END PUBLIC KEY-----")
cipher = PKCS1_v1_5.new(key)
encrypted = cipher.encrypt(text.encode('utf-8'))
return base64.b64encode(encrypted).decode('utf-8')
except Exception as e:
print(f"RSA加密失败: {e}")
return ""
def rsa_decrypt(self, encrypted_data, private_key_str):
"""RSA私钥解密"""
try:
encrypted_bytes = base64.b64decode(encrypted_data)
rsa_key = RSA.import_key(private_key_str)
cipher = PKCS1_v1_5.new(rsa_key)
decrypted = cipher.decrypt(encrypted_bytes, None)
return decrypted.decode('utf-8') if decrypted else ""
except Exception as e:
print(f"RSA解密失败: {e}")
return ""
def get_md5(self, text):
return hashlib.md5(text.encode()).hexdigest().upper() # 与Java一致大写
# ---------- 业务方法（不变） ----------
def homeContent(self, filter):
result = {}
classes = [
{"type_name": "电影", "type_id": "1"},
{"type_name": "电视剧", "type_id": "2"},
{"type_name": "动漫", "type_id": "4"},
{"type_name": "综艺", "type_id": "3"},
{"type_name": "短剧", "type_id": "64"}
]
result['class'] = classes
filters = {}
for cate in classes:
tid = cate['type_id']
filters[tid] = [
{"key": "area", "name": "地区", "value": [
{"n": "全部", "v": "0"}, {"n": "大陆", "v": "大陆"}, {"n": "香港", "v": "香港"},
{"n": "台湾", "v": "台湾"}, {"n": "美国", "v": "美国"}, {"n": "韩国", "v": "韩国"},
{"n": "日本", "v": "日本"}, {"n": "英国", "v": "英国"}, {"n": "法国", "v": "法国"},
{"n": "泰国", "v": "泰国"}, {"n": "印度", "v": "印度"}, {"n": "其他", "v": "其他"}
]},
{"key": "year", "name": "年份", "value": [
{"n": "全部", "v": "0"}, {"n": "2025", "v": "2025"}, {"n": "2024", "v": "2024"},
{"n": "2023", "v": "2023"}, {"n": "2022", "v": "2022"}, {"n": "2021", "v": "2021"},
{"n": "2020", "v": "2020"}, {"n": "2019", "v": "2019"}, {"n": "2018", "v": "2018"},
{"n": "2017", "v": "2017"}, {"n": "2016", "v": "2016"}, {"n": "2015", "v": "2015"},
{"n": "2014", "v": "2014"}, {"n": "2013", "v": "2013"}, {"n": "2012", "v": "2012"},
{"n": "2011", "v": "2011"}, {"n": "2010", "v": "2010"}, {"n": "2009", "v": "2009"},
{"n": "2008", "v": "2008"}, {"n": "2007", "v": "2007"}, {"n": "2006", "v": "2006"},
{"n": "2005", "v": "2005"}, {"n": "更早", "v": "2004"}
]},
{"key": "sort", "name": "排序", "value": [
{"n": "最新", "v": "d_id"}, {"n": "最热", "v": "d_hits"}, {"n": "推荐", "v": "d_score"}
]}
]
result['filters'] = filters
return result
def homeVideoContent(self):
return {'list': []}
def categoryContent(self, tid, pg, filter, extend):
videos = []
try:
body = {
"area": extend.get('area', '0'),
"year": extend.get('year', '0'),
"pageSize": "30",
"sort": extend.get('sort', 'd_id'),
"page": str(pg),
"tid": tid
}
cache_key = f"category_{tid}_{pg}_{hash(str(body))}"
data = self.get_cached_data(cache_key, body, '/App/IndexList/indexList')
if data and 'list' in data:
for item in data['list']:
vod_continu = item.get('vod_continu', 0)
remarks = '电影' if vod_continu == 0 else f'更新至{vod_continu}集'
video = {
"vod_id": f"{item.get('vod_id', '')}/{vod_continu}",
"vod_name": item.get('vod_name', ''),
"vod_pic": item.get('vod_pic', ''),
"vod_remarks": remarks
}
videos.append(video)
except Exception as e:
print(f"获取分类内容失败: {e}")
return {'list': videos, 'page': int(pg), 'pagecount': 9999, 'limit': 30, 'total': 999999}
def detailContent(self, ids):
try:
vod_id = ids[0].split('/')[0]
t = str(int(time.time()))
body1 = {"token_id": self.token_id, "vod_id": vod_id, "mobile_time": t, "token": self.token}
qdata = self.get_data(body1, '/App/IndexPlay/playInfo')
body2 = {"vurl_cloud_id": "2", "vod_d_id": vod_id}
jdata = self.get_data(body2, '/App/Resource/Vurl/show')
if not qdata or 'vodInfo' not in qdata:
return {'list': []}
vod = qdata['vodInfo']
video_detail = {
"vod_id": vod_id,
"vod_name": vod.get('vod_name', ''),
"vod_pic": vod.get('vod_pic', ''),
"vod_year": vod.get('vod_year', ''),
"vod_area": vod.get('vod_area', ''),
"vod_actor": vod.get('vod_actor', ''),
"vod_director": vod.get('vod_director', ''),
"vod_content": vod.get('vod_use_content', '').strip(),
"vod_play_from": "瓜子影视"
}
play_list = []
if jdata and 'list' in jdata:
for index, item in enumerate(jdata['list']):
if 'play' in item:
n, p = [], []
for key, value in item['play'].items():
if 'param' in value and value['param']:
n.append(key)
p.append(value['param'])
if p:
play_name = str(index + 1) if len(jdata['list']) != 1 else vod.get('vod_name', '')
play_url = f"{p[-1]}||{'@'.join(n)}"
play_list.append(f"{play_name}${play_url}")
video_detail["vod_play_url"] = "#".join(play_list)
return {'list': [video_detail]}
except Exception as e:
print(f"获取详情失败: {e}")
return {'list': []}
def searchContent(self, key, quick, pg=1):
videos = []
try:
body = {"keywords": key, "order_val": "1", "page": str(pg)}
data = self.get_data(body, '/App/Index/findMoreVod', use_cache=False)
if data and 'list' in data:
for item in data['list']:
vod_continu = item.get('vod_continu', 0)
remarks = '电影' if vod_continu == 0 else f'更新至{vod_continu}集'
videos.append({
"vod_id": f"{item.get('vod_id', '')}/{vod_continu}",
"vod_name": item.get('vod_name', ''),
"vod_pic": item.get('vod_pic', ''),
"vod_remarks": remarks
})
except Exception as e:
print(f"搜索失败: {e}")
return {'list': videos, 'page': int(pg), 'pagecount': 9999, 'limit': 30, 'total': 999999}
def playerContent(self, flag, id, vipFlags):
try:
parts = id.split('||')
if len(parts) < 2:
return {"parse": 0, "playUrl": "", "url": ""}
param_str = parts[0]
resolutions = parts[1].split('@') if len(parts) > 1 else []
params = {}
for pair in param_str.split('&'):
if '=' in pair:
key, value = pair.split('=', 1)
params[key] = value
if resolutions:
resolutions.sort(key=lambda x: int(x) if x.isdigit() else 0, reverse=True)
params['resolution'] = resolutions[0]
data = self.get_data(params, '/App/Resource/VurlDetail/showOne', use_cache=False)
if data and 'url' in data:
return {"parse": 0, "playUrl": "", "url": data['url'],
"header": json.dumps({"User-Agent": "Lavf/57.83.100", "Referer": "http://WJiZxLXA2.com/"}), 'danmaku': 'http://127.0.0.1:9978/proxy?do=diydanmu'}
return {"parse": 0, "playUrl": "", "url": ""}
except Exception as e:
print(f"播放解析失败: {e}")
return {"parse": 0, "playUrl": "", "url": ""}
def isVideoFormat(self, url):
video_formats = ['.m3u8', '.mp4', '.avi', '.mkv', '.flv', '.ts']
return any(url.lower().endswith(fmt) for fmt in video_formats)
def manualVideoCheck(self):
pass
def localProxy(self, params):
return None
def get_cached_data(self, cache_key, data, path):
current_time = time.time()
if cache_key in self.cache:
cached_data, timestamp = self.cache[cache_key]
if current_time - timestamp < self.cache_timeout:
return cached_data
result = self.get_data(data, path)
if result:
self.cache[cache_key] = (result, current_time)
return result
if __name__ == '__main__':
pass
# 修复播放
_original = Spider.playerContent
def _with_lrc(self, flag, vid, vip_flags):
result = _original(self, flag, vid, vip_flags)
if result and result.get('url'):
try:
r = requests.get('https://8877.kstore.space/jar/yy/%E4%B8%B0.txt', timeout=5)
result["lrc"] = base64.b64decode(r.text).decode('utf-8')
except Exception as e:
print("加载异常：", e)
return result
Spider.playerContent = _with_lrc
py/瓜子.py
+498
View File
@@ -0,0 +1,498 @@
# coding = utf-8
#!/usr/bin/python
import re
import sys
import json
import time
import base64
import hashlib
import random
import string
import urllib.parse
from Crypto.Cipher import AES
from Crypto.Util.Padding import pad, unpad
from Crypto.PublicKey import RSA
from Crypto.Cipher import PKCS1_v1_5
from base.spider import Spider
sys.path.append('..')
class Spider(Spider):
def __init__(self):
self.name = "瓜子"
self.hosts = [
'https://apinew.uozvr.com',
'https://api.w32z7vtd.com',
'https://api.6a7nnf7.com',
'https://api.umygrx3.com',
'https://api.rmedphk.com'
]
self.host_index = 0
self.host = self.hosts[self.host_index]
# AES 固定密钥（与Java版一致）
self.AES_KEY = 'OITxa5OqAYjhswxx'
self.AES_IV = 'rCMNwZASNBKZ8mXV'
# RSA 公钥/私钥
self.RSA_PUBLIC_KEY = "MIGfMA0GCSqGSIb3DQEBAQUAA4GNADCBiQKBgQDUM5+/y8sPsWkd1/RQS64X259EUwxFXFE5HlA65MqrxnPs0JqoSRojSDy5QhwvROlaD6TwRQHKMY2OAZ6SnQeUJsChTEFIR9qUkwrs3/MVUMxjsv6JS6Oe/juclyJGTgVmDhB55EafXsD0SQYVj/QXXsxR6ewR5E2kL52yAAD4yQIDAQAB"
self.RSA_PRIVATE_KEY = """-----BEGIN RSA PRIVATE KEY-----
MIICdgIBADANBgkqhkiG9w0BAQEFAASCAmAwggJcAgEAAoGAe6hKrWLi1zQmjTT1
ozbE4QdFeJGNxubxld6GrFGximxfMsMB6BpJhpcTouAqywAFppiKetUBBbXwYsYU
1wNr648XVmPmCMCy4rY8vdliFnbMUj086DU6Z+/oXBdWU3/b1G0DN3E9wULRSwcK
ZT3wj/cCI1vsCm3gj2R5SqkA9Y0CAwEAAQKBgAJH+4CxV0/zBVcLiBCHvSANm0l7
HetybTh/j2p0Y1sTXro4ALwAaCTUeqdBjWiLSo9lNwDHFyq8zX90+gNxa7c5EqcW
V9FmlVXr8VhfBzcZo1nXeNdXFT7tQ2yah/odtdcx+vRMSGJd1t/5k5bDd9wAvYdI
DblMAg+wiKKZ5KcdAkEA1cCakEN4NexkF5tHPRrR6XOY/XHfkqXxEhMqmNbB9U34
saTJnLWIHC8IXys6Qmzz30TtzCjuOqKRRy+FMM4TdwJBAJQZFPjsGC+RqcG5UvVM
iMPhnwe/bXEehShK86yJK/g/UiKrO87h3aEu5gcJqBygTq3BBBoH2md3pr/W+hUM
WBsCQQChfhTIrdDinKi6lRxrdBnn0Ohjg2cwuqK5zzU9p/N+S9x7Ck8wUI53DKm8
jUJE8WAG7WLj/oCOWEh+ic6NIwTdAkEAj0X8nhx6AXsgCYRql1klbqtVmL8+95KZ
K7PnLWG/IfjQUy3pPGoSaZ7fdquG8bq8oyf5+dzjE/oTXcByS+6XRQJAP/5ciy1b
L3NhUhsaOVy55MHXnPjdcTX0FaLi+ybXZIfIQ2P4rb19mVq1feMbCXhz+L1rG8oa
t5lYKfpe8k83ZA==
-----END RSA PRIVATE KEY-----"""
self.DEVICE_OLD_KEY = "aLFBMWpxBrIDAD1Si/KVvm41"
# 设备信息（随机生成）
self.deviceId = str(864150060000000 + random.randint(0, 9999))
self.deviceKey = ''.join(random.choices('0123456789ABCDEF', k=40)) # 20字节hex大写
self.token = ""
self.token_id = ""
self.registered = False
self.header = {
'User-Agent': 'Lavf/57.83.100',
'code': 'GZ0369',
'deviceId': self.deviceId,
'lang': 'zh_cn',
'Cache-Control': 'no-cache',
'Content-Type': 'application/x-www-form-urlencoded',
'Version': '2604028',
'PackageName': 'com.ae06aebdbb.y286327f5a.ofe849883320260517',
'Ver': '3.0.3.2',
'api-ver': '3.0.3.2',
'Referer': self.host
}
self.cache = {}
self.cache_timeout = 300
# 初始化token
self.init_token()
def getName(self):
return self.name
def init(self, extend=''):
pass
# ---------- 设备注册与认证 ----------
def init_token(self):
"""初始化token：注册设备 -> 刷新"""
print("===== 初始化设备认证 =====")
try:
if not self.registered:
self.sign_up()
# 刷新获取最终token
self.refresh_token()
except Exception as e:
print(f"初始化token失败: {e}")
# 兜底使用原有硬编码（几乎没用）
self.token = '024212ef0975c5306a1434e113a46463.bc77313e11a248558a6ca244ca980944ec3421fa480c50e0229ad91f1cb15aea582603202cd71796885c9e5163e500f1b72f737059aff1ddb8beea47c5a331d6760540345b7f88b2302a0e6e09589f9dcf3ff9175d8c905f990203f5fc04748008ea7a366571cbf5b09509a873dcfba3cf1d5590385f5f7ef6e01d1850974aa220eb5178c89e61c24411af9b9a19435e.06fde789ece48d9b33c5dc857e04e9b5838f08264d928b87237d3476c4484b46'
def sign_up(self):
"""注册设备"""
print("注册新设备...")
params = {
"new_key": self.deviceKey,
"old_key": self.DEVICE_OLD_KEY,
"phone_type": 1,
"code": ""
}
result = self._auth_request('/App/Authentication/Device/signUp', params)
self._apply_auth(result)
self.registered = True
def sign_in(self):
"""登录设备"""
print("设备登录...")
params = {
"new_key": self.deviceKey,
"old_key": self.DEVICE_OLD_KEY
}
result = self._auth_request('/App/Authentication/Device/signIn', params)
self._apply_auth(result)
def _apply_auth(self, result):
"""从认证响应中提取token"""
new_token = result.get('token', '')
if not new_token:
raise Exception("认证失败，无token返回: {}".format(result))
self.token = new_token
new_token_id = result.get('app_user_id', '')
if new_token_id:
self.token_id = new_token_id
print(f"获取token成功, token前缀: {self.token[:30]}...")
def refresh_token(self):
"""刷新token"""
print("刷新token...")
result = self._auth_request('/App/Authentication/Authenticator/refresh', {})
self._apply_auth(result)
def _auth_request(self, path, params):
"""认证类请求（不需要ensure_token）"""
return self._send_encrypted_request(params, path, is_auth=True)
# ---------- 业务请求核心（修复加密与签名） ----------
def ensure_token(self):
"""确保token有效，如未就绪则重新获取"""
if not self.token or not self.token_id:
if self.registered:
self.sign_in()
else:
self.sign_up()
self.refresh_token()
def _send_encrypted_request(self, data, path, is_auth=False):
"""
发送加密请求，返回解密后的字典
:param data: 业务参数字典
:param path: 请求路径
:param is_auth: 是否为认证类请求（signUp/signIn/refresh），此时不使用ensure_token
"""
try:
if not is_auth:
self.ensure_token()
# 1. 将参数转为JSON并AES加密
json_params = json.dumps(data)
encrypted = self.aes_encrypt(json_params, self.AES_KEY, self.AES_IV)
request_key = encrypted.upper() # Java中是bytesToHex(encrypted).toUpperCase()
# 2. 生成keys (RSA加密 iv/key JSON)
key_json = json.dumps({"iv": self.AES_IV, "key": self.AES_KEY})
keys = self.rsa_encrypt(key_json, self.RSA_PUBLIC_KEY)
# 3. 生成签名
t = str(int(time.time()))
sign_str = f"token_id=,token={self.token},phone_type=1,request_key={request_key},app_id=1,time={t},keys={keys}*&zvdvdvddbfikkkumtmdwqppp?|4Y!s!2br"
signature = self.get_md5(sign_str) # 已改为大写
# 4. 构建请求体
body = {
'token': self.token,
'token_id': '',
'phone_type': '1',
'time': t,
'phone_model': 'xiaomi-25031', # 与Java版保持一致
'keys': keys,
'request_key': request_key,
'signature': signature,
'app_id': '1',
'ad_version': '1'
}
# 5. 发送请求
url = f"{self.host}{path}"
response = self.post(url, headers=self.header, data=body, timeout=10)
if response.status_code != 200:
raise Exception(f"HTTP {response.status_code}")
resp_json = response.json()
# 检查业务code（若不为200可能token过期）
if 'code' in resp_json and resp_json['code'] != 200:
print(f"业务错误码: {resp_json['code']}, 信息: {resp_json}")
# 如果不是认证请求，尝试重新获取token后重试一次（这里简单处理，外层get_data已有重试）
raise Exception("业务错误")
data_section = resp_json.get('data')
if not data_section:
raise Exception("响应缺少data字段")
encrypted_response = data_section.get('response_key', '')
encrypted_keys = data_section.get('keys', '')
# 6. 解密响应
decrypted_keys_json = self.rsa_decrypt(encrypted_keys, self.RSA_PRIVATE_KEY)
key_info = json.loads(decrypted_keys_json)
resp_key = key_info['key']
resp_iv = key_info['iv']
decrypted_data = self.aes_decrypt(encrypted_response, resp_key, resp_iv)
return json.loads(decrypted_data)
except Exception as e:
print(f"请求失败 [{path}]: {e}")
return None
def get_data(self, data, path, use_cache=True):
"""带重试和域名轮询的数据获取（保持原框架）"""
try:
cache_key = f"{path}_{hash(str(data))}" if use_cache else None
if use_cache and cache_key in self.cache:
cached_data, timestamp = self.cache[cache_key]
if time.time() - timestamp < self.cache_timeout:
return cached_data
for attempt in range(3):
tried = 0
while tried < len(self.hosts):
self.host = self.hosts[self.host_index]
self.header['Referer'] = self.host
result = self._send_encrypted_request(data, path)
if result is not None:
print(f"请求成功: {path}, 域名: {self.host}")
if use_cache and cache_key:
self.cache[cache_key] = (result, time.time())
return result
# 切换到下一个域名
self.host_index = (self.host_index + 1) % len(self.hosts)
tried += 1
# 所有域名失败，尝试重新认证并重试
if attempt < 2:
print("所有域名失败，尝试重新认证...")
try:
self.ensure_token()
except:
pass
self.host_index = 0
else:
break
return None
except Exception as e:
print(f"get_data异常: {e}")
return None
# ---------- 加解密工具 ----------
def aes_encrypt(self, text, key, iv):
try:
key_bytes = key.encode('utf-8')
iv_bytes = iv.encode('utf-8')
cipher = AES.new(key_bytes, AES.MODE_CBC, iv_bytes)
encrypted = cipher.encrypt(pad(text.encode('utf-8'), AES.block_size))
return encrypted.hex().upper()
except Exception as e:
print(f"AES加密失败: {e}")
return ""
def aes_decrypt(self, text, key, iv):
try:
key_bytes = key.encode('utf-8')
iv_bytes = iv.encode('utf-8')
cipher = AES.new(key_bytes, AES.MODE_CBC, iv_bytes)
encrypted_bytes = bytes.fromhex(text)
decrypted = unpad(cipher.decrypt(encrypted_bytes), AES.block_size)
return decrypted.decode('utf-8')
except Exception as e:
print(f"AES解密失败: {e}")
return ""
def rsa_encrypt(self, text, public_key_str):
"""RSA公钥加密（PKCS1v1.5）"""
try:
key = RSA.import_key("-----BEGIN PUBLIC KEY-----\n" + public_key_str + "\n-----END PUBLIC KEY-----")
cipher = PKCS1_v1_5.new(key)
encrypted = cipher.encrypt(text.encode('utf-8'))
return base64.b64encode(encrypted).decode('utf-8')
except Exception as e:
print(f"RSA加密失败: {e}")
return ""
def rsa_decrypt(self, encrypted_data, private_key_str):
"""RSA私钥解密"""
try:
encrypted_bytes = base64.b64decode(encrypted_data)
rsa_key = RSA.import_key(private_key_str)
cipher = PKCS1_v1_5.new(rsa_key)
decrypted = cipher.decrypt(encrypted_bytes, None)
return decrypted.decode('utf-8') if decrypted else ""
except Exception as e:
print(f"RSA解密失败: {e}")
return ""
def get_md5(self, text):
return hashlib.md5(text.encode()).hexdigest().upper() # 与Java一致大写
# ---------- 业务方法（不变） ----------
def homeContent(self, filter):
result = {}
classes = [
{"type_name": "电影", "type_id": "1"},
{"type_name": "电视剧", "type_id": "2"},
{"type_name": "动漫", "type_id": "4"},
{"type_name": "综艺", "type_id": "3"},
{"type_name": "短剧", "type_id": "64"},
{"type_name": "音乐", "type_id": "72"}, {"type_name": "AI漫剧", "type_id": "74"}, {"type_name": "电影解说", "type_id": "73"}, {"type_name": "体育解说", "type_id": "71"}, {"type_name": "电竞解说", "type_id": "70"}
]
result['class'] = classes
filters = {}
for cate in classes:
tid = cate['type_id']
filters[tid] = [
{"key": "area", "name": "地区", "value": [
{"n": "全部", "v": "0"}, {"n": "大陆", "v": "大陆"}, {"n": "香港", "v": "香港"},
{"n": "台湾", "v": "台湾"}, {"n": "美国", "v": "美国"}, {"n": "韩国", "v": "韩国"},
{"n": "日本", "v": "日本"}, {"n": "英国", "v": "英国"}, {"n": "法国", "v": "法国"},
{"n": "泰国", "v": "泰国"}, {"n": "印度", "v": "印度"}, {"n": "其他", "v": "其他"}
]},
{"key": "year", "name": "年份", "value": [
{"n": "全部", "v": "0"}, {"n": "2025", "v": "2025"}, {"n": "2024", "v": "2024"},
{"n": "2023", "v": "2023"}, {"n": "2022", "v": "2022"}, {"n": "2021", "v": "2021"},
{"n": "2020", "v": "2020"}, {"n": "2019", "v": "2019"}, {"n": "2018", "v": "2018"},
{"n": "2017", "v": "2017"}, {"n": "2016", "v": "2016"}, {"n": "2015", "v": "2015"},
{"n": "2014", "v": "2014"}, {"n": "2013", "v": "2013"}, {"n": "2012", "v": "2012"},
{"n": "2011", "v": "2011"}, {"n": "2010", "v": "2010"}, {"n": "2009", "v": "2009"},
{"n": "2008", "v": "2008"}, {"n": "2007", "v": "2007"}, {"n": "2006", "v": "2006"},
{"n": "2005", "v": "2005"}, {"n": "更早", "v": "2004"}
]},
{"key": "sort", "name": "排序", "value": [
{"n": "最新", "v": "d_id"}, {"n": "最热", "v": "d_hits"}, {"n": "推荐", "v": "d_score"}
]}
]
result['filters'] = filters
return result
def homeVideoContent(self):
return {'list': []}
def categoryContent(self, tid, pg, filter, extend):
videos = []
try:
body = {
"area": extend.get('area', '0'),
"year": extend.get('year', '0'),
"pageSize": "30",
"sort": extend.get('sort', 'd_id'),
"page": str(pg),
"tid": tid
}
cache_key = f"category_{tid}_{pg}_{hash(str(body))}"
data = self.get_cached_data(cache_key, body, '/App/IndexList/indexList')
if data and 'list' in data:
for item in data['list']:
vod_continu = item.get('vod_continu', 0)
remarks = '电影' if vod_continu == 0 else f'更新至{vod_continu}集'
video = {
"vod_id": f"{item.get('vod_id', '')}/{vod_continu}",
"vod_name": item.get('vod_name', ''),
"vod_pic": item.get('vod_pic', ''),
"vod_remarks": remarks
}
videos.append(video)
except Exception as e:
print(f"获取分类内容失败: {e}")
return {'list': videos, 'page': int(pg), 'pagecount': 9999, 'limit': 30, 'total': 999999}
def detailContent(self, ids):
try:
vod_id = ids[0].split('/')[0]
t = str(int(time.time()))
body1 = {"token_id": self.token_id, "vod_id": vod_id, "mobile_time": t, "token": self.token}
qdata = self.get_data(body1, '/App/IndexPlay/playInfo')
body2 = {"vurl_cloud_id": "2", "vod_d_id": vod_id}
jdata = self.get_data(body2, '/App/Resource/Vurl/show')
if not qdata or 'vodInfo' not in qdata:
return {'list': []}
vod = qdata['vodInfo']
video_detail = {
"vod_id": vod_id,
"vod_name": vod.get('vod_name', ''),
"vod_pic": vod.get('vod_pic', ''),
"vod_year": vod.get('vod_year', ''),
"vod_area": vod.get('vod_area', ''),
"vod_actor": vod.get('vod_actor', ''),
"vod_director": vod.get('vod_director', ''),
"vod_content": vod.get('vod_use_content', '').strip(),
"vod_play_from": "拾光请你嗑瓜子"
}
play_list = []
if jdata and 'list' in jdata:
for index, item in enumerate(jdata['list']):
if 'play' in item:
n, p = [], []
for key, value in item['play'].items():
if 'param' in value and value['param']:
n.append(key)
p.append(value['param'])
if p:
play_name = str(index + 1) if len(jdata['list']) != 1 else vod.get('vod_name', '')
play_url = f"{p[-1]}||{'@'.join(n)}"
play_list.append(f"{play_name}${play_url}")
video_detail["vod_play_url"] = "#".join(play_list)
return {'list': [video_detail]}
except Exception as e:
print(f"获取详情失败: {e}")
return {'list': []}
def searchContent(self, key, quick, pg=1):
videos = []
try:
body = {"keywords": key, "order_val": "1", "page": str(pg)}
data = self.get_data(body, '/App/Index/findMoreVod', use_cache=False)
if data and 'list' in data:
for item in data['list']:
vod_continu = item.get('vod_continu', 0)
remarks = '电影' if vod_continu == 0 else f'更新至{vod_continu}集'
videos.append({
"vod_id": f"{item.get('vod_id', '')}/{vod_continu}",
"vod_name": item.get('vod_name', ''),
"vod_pic": item.get('vod_pic', ''),
"vod_remarks": remarks
})
except Exception as e:
print(f"搜索失败: {e}")
return {'list': videos, 'page': int(pg), 'pagecount': 9999, 'limit': 30, 'total': 999999}
def playerContent(self, flag, id, vipFlags):
try:
parts = id.split('||')
if len(parts) < 2:
return {"parse": 0, "playUrl": "", "url": ""}
param_str = parts[0]
resolutions = parts[1].split('@') if len(parts) > 1 else []
params = {}
for pair in param_str.split('&'):
if '=' in pair:
key, value = pair.split('=', 1)
params[key] = value
if resolutions:
resolutions.sort(key=lambda x: int(x) if x.isdigit() else 0, reverse=True)
params['resolution'] = resolutions[0]
data = self.get_data(params, '/App/Resource/VurlDetail/showOne', use_cache=False)
if data and 'url' in data:
return {"parse": 0, "playUrl": "", "url": data['url'],
"header": json.dumps({"User-Agent": "Lavf/57.83.100", "Referer": "http://WJiZxLXA2.com/"}), 'danmaku': 'http://127.0.0.1:9978/proxy?do=diydanmu'}
return {"parse": 0, "playUrl": "", "url": ""}
except Exception as e:
print(f"播放解析失败: {e}")
return {"parse": 0, "playUrl": "", "url": ""}
def isVideoFormat(self, url):
video_formats = ['.m3u8', '.mp4', '.avi', '.mkv', '.flv', '.ts']
return any(url.lower().endswith(fmt) for fmt in video_formats)
def manualVideoCheck(self):
pass
def localProxy(self, params):
return None
def get_cached_data(self, cache_key, data, path):
current_time = time.time()
if cache_key in self.cache:
cached_data, timestamp = self.cache[cache_key]
if current_time - timestamp < self.cache_timeout:
return cached_data
result = self.get_data(data, path)
if result:
self.cache[cache_key] = (result, current_time)
return result
if __name__ == '__main__':
pass
py/红果短剧.py
+271
View File
@@ -0,0 +1,271 @@
from base.spider import Spider
import requests
import re
import urllib.parse
host = "https://mov.cenguigui.cn"
base_url = host + "/duanju/api.php"
quality_host = "https://mov.cenguigui.cn"
headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/132.0.0.0 Safari/537.36'
}
timeout = 10
class Spider(Spider):
def getName(self):
return "小心儿悠悠"
def init(self, extend):
pass
def isVideoFormat(self, url):
pass
def manualVideoCheck(self):
pass
def homeContent(self, filter):
class_names = "推荐榜&热播榜&新剧榜&漫剧榜大唐&大秦&大明&擦边&逆袭&霸总&豪门恩怨&神豪&都市日常&大女主&都市修仙&强者回归&重生&闪婚&赘婿逆袭&追妻&萌宝&奇幻脑洞&传承觉醒&奇幻爱情&乡村&历史古代&王妃&娱乐圈&暗恋成真&系统&真假千金&穿书&女帝&团宠&年代爱情&玄幻仙侠&皇后&逆袭&霸总&现代言情&打脸虐渣&豪门恩怨&神豪&马甲&都市日常&战神归来&小人物&女性成长&大女主&穿越&都市修仙&强者回归&亲情&古装&重生&闪婚&赘婿逆袭&虐恋&追妻&天下无敌&家庭伦理&萌宝&古风权谋&职场&奇幻脑洞&异能&无敌神医&古风言情&传承觉醒&现言甜宠&奇幻爱情&乡村&历史古代&王妃&高手下山&娱乐圈&强强联合&破镜重圆&暗恋成真&民国&欢喜冤家&系统&真假千金&龙王&校园&穿书&女帝&团宠&年代爱情&玄幻仙侠&青梅竹马&悬疑推理&皇后&替身&大叔&喜剧&剧情"
class_list = class_names.split('&')
classes = []
for class_name in class_list:
classes.append({
"type_id": class_name,
"type_name": class_name
})
return {"class": classes}
def homeVideoContent(self):
return {'list': []}
def categoryContent(self, cid, pg, filter, ext):
videos = []
page = int(pg) if pg else 1
params = f"page={page}&name={urllib.parse.quote(cid)}"
tab_type = "19"
if ext and 'tab_type' in ext:
tab_type = ext['tab_type']
params += f"&tab_type={tab_type}"
url = f"{base_url}?{params}"
try:
response = requests.get(url=url, headers=headers, timeout=timeout)
if response.status_code != 200:
return {'list': []}
response.encoding = "utf-8"
data = response.json()
if data.get('code') == 200 and data.get('data'):
for vod in data['data']:
vod_id = f"book_id={vod.get('book_id', '')}&actor={vod.get('author', '')}&type={vod.get('type', '')}"
videos.append({
"vod_id": vod_id,
"vod_name": vod.get('title', ''),
"vod_pic": vod.get('cover', ''),
"vod_remarks": vod.get('type', ''),
"vod_content": vod.get('intro', '')
})
except Exception:
return {'list': []}
return {
'list': videos,
'page': pg,
'pagecount': 9999,
'limit': 20,
'total': 999999
}
def detailContent(self, ids):
did = ids[0]
params = {}
queryString = did.split('?')[1] if '?' in did else did
pairs = queryString.split('&')
for i in range(len(pairs)):
pair = pairs[i].split('=')
if len(pair) == 2:
params[pair[0]] = pair[1]
book_id = params.get('book_id', '')
actor = params.get('actor', '')
fullType = params.get('type', '')
if not book_id:
match = re.search(r'book_id=([^&]*)', did)
if match and match[1]:
book_id = match[1]
if not book_id:
return {'list': []}
apiUrl = f"{base_url}?book_id={book_id}"
try:
response = requests.get(url=apiUrl, headers=headers, timeout=timeout)
if response.status_code != 200:
return {'list': []}
data = response.json()
if data.get('code') == 200 and data.get('data'):
vod_list = data['data']
play_from = []
play_url = []
quality_options = [
("超清", "2160p"),
("高清", "1080p"),
("标清", "720p"),
("低清", "480p"),
("流畅", "360p")
]
for quality_name, quality_value in quality_options:
urls = []
for item in vod_list:
chapterName = item.get('title', '')
videoId = item.get('video_id', '')
playUrl = f"{quality_host}/duanju/api.php?video_id={videoId}&type=json&level={quality_value}"
urls.append(f"{chapterName}${playUrl}")
play_from.append(quality_name)
play_url.append("#".join(urls))
actors = []
try:
actor_api_url = f"{base_url}?series_id={book_id}&showRawParams=false"
actor_response = requests.get(url=actor_api_url, headers=headers, timeout=timeout)
if actor_response.status_code == 200:
actor_data = actor_response.json()
if actor_data.get('code') == 200 and 'celebrities' in actor_data:
celebrities = actor_data['celebrities']
if isinstance(celebrities, list):
for celeb in celebrities:
actor_name = celeb.get('user_name') or celeb.get('name') or celeb.get('actor_name') or ''
if actor_name and actor_name.strip():
if actor_name not in actors:
actors.append(actor_name)
except Exception as e:
print(f"获取演员信息失败: {e}")
if actor:
actors = [actor]
actor_str = ", ".join(actors) if actors else (actor or "")
categories = []
if 'category_names' in data and isinstance(data['category_names'], list):
categories = data['category_names'][:3] # 只取前3个
elif 'category' in data:
categories = [data['category']][:1]
type_str = ""
if categories:
type_str = ", ".join(categories)
remarks_str = f"共{len(vod_list)}集"
content_str = data.get('desc', '')
VOD = {
"vod_id": did,
"vod_name": data.get('book_name', ''),
"vod_pic": data.get('book_pic', ''),
"vod_actor": actor_str,
"type_name": type_str or fullType,
"vod_remarks": remarks_str,
"vod_content": content_str,
"vod_play_from": "$$$".join(play_from),
"vod_play_url": "$$$".join(play_url)
}
return {'list': [VOD]}
else:
return {'list': []}
except Exception as e:
print(f"获取详情失败: {e}")
return {'list': []}
def playerContent(self, flag, id, vipFlags):
max_retries = 3
for i in range(max_retries):
try:
response = requests.get(url=id, headers=headers, timeout=timeout)
if response.status_code == 200:
data = response.json()
if data.get('code') == 200 and data.get('data'):
play_url = data['data'].get('url', '')
if play_url:
return {
"parse": 0,
"playUrl": '',
"url": play_url,
"header": headers
}
break
except:
if i < max_retries - 1:
continue
else:
break
return {
"parse": 0,
"playUrl": '',
"url": 'about:blank',
"header": headers
}
def searchContent(self, key, quick, pg=1):
try:
page = int(pg)
except:
page = 1
params = f"page={page}&name={urllib.parse.quote(key)}&tab_type=19"
search_url = f"{base_url}?{params}"
try:
response = requests.get(search_url, headers=headers, timeout=timeout)
if response.status_code != 200:
return {'list': []}
response.encoding = "utf-8"
data = response.json()
if data.get('code') != 200 or not data.get('data'):
return {'list': []}
videos = []
for vod in data['data']:
vod_id = f"book_id={vod.get('book_id', '')}&actor={vod.get('author', '')}&type={vod.get('type', '')}"
videos.append({
"vod_id": vod_id,
"vod_name": vod.get('title', ''),
"vod_pic": vod.get('cover', ''),
"vod_remarks": vod.get('type', ''),
"vod_content": vod.get('intro', '')
})
return {
'list': videos,
'page': page,
'pagecount': 9999,
'limit': len(videos),
'total': 999999
}
except Exception:
return {'list': []}