XC-Pro/xvip
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-15 10:30:42 +02:00
parent 03ddf98142
commit 11e71aa836
5 changed files with 2070 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
py/华数.py
+546
View File
@@ -0,0 +1,546 @@
# coding=utf-8
# !/usr/bin/python
"""
作者 精彩一瞬间 内容均从互联网收集而来 仅供交流学习使用 严禁用于商业用途 请于24小时内删除
====================Diudiumiao====================
"""
from Crypto.Util.Padding import unpad
from Crypto.Util.Padding import pad
from urllib.parse import urlparse
from urllib.parse import unquote
from Crypto.Cipher import ARC4
from urllib.parse import quote
from base.spider import Spider
from Crypto.Cipher import AES
from datetime import datetime
from bs4 import BeautifulSoup
from base64 import b64decode
import concurrent.futures
import urllib.request
import urllib.parse
import datetime
import binascii
import requests
import hashlib
import base64
import json
import time
import hmac
import sys
import re
import os
sys.path.append('..')
xurl = "https://www.wasu.cn"
headerx = {
'User-Agent': 'Mozilla/5.0 (Windows NT 6.1; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/50.0.2661.87 Safari/537.36'
}
headerz = {
'accept': '*/*',
'accept-language': 'zh-CN,zh;q=0.9,en;q=0.8,en-GB;q=0.7,en-US;q=0.6',
'cache-control': 'no-cache',
'origin': xurl,
'pragma': 'no-cache',
'priority': 'u=1, i',
'referer': xurl,
'sec-ch-ua': '"Microsoft Edge";v="143", "Chromium";v="143", "Not A(Brand";v="24"',
'sec-ch-ua-mobile': '?0',
'sec-ch-ua-platform': '"Windows"',
'sec-fetch-dest': 'empty',
'sec-fetch-mode': 'cors',
'sec-fetch-site': 'cross-site',
'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/143.0.0.0 Safari/537.36 Edg/143.0.0.0',
}
class Spider(Spider):
def getName(self):
return "精彩"
def init(self, extend):
pass
def isVideoFormat(self, url):
pass
def manualVideoCheck(self):
pass
def homeVideoContent(self):
pass
def homeContent(self, filter):
result = {}
result = {"class": [{"type_id": "961", "type_name": "电影"},
{"type_id": "962", "type_name": "剧集"},
{"type_id": "963", "type_name": "少儿"},
{"type_id": "965", "type_name": "栏目"},
{"type_id": "966", "type_name": "新闻"}],
"list": [],
"filters": {"961": [{"key": "地区",
"name": "地区",
"value": [{"n": "全部", "v": ""},
{"n": "内地", "v": "内地"},
{"n": "港台", "v": "港台"},
{"n": "欧美", "v": "欧美"},
{"n": "日韩", "v": "日韩"},
{"n": "泰国", "v": "泰国"},
{"n": "其他", "v": "其他"}]},
{"key": "类型",
"name": "类型",
"value": [{"n": "全部", "v": ""},
{"n": "动作", "v": "动作"},
{"n": "科幻", "v": "科幻"},
{"n": "惊悚", "v": "惊悚"},
{"n": "冒险", "v": "冒险"},
{"n": "剧情", "v": "剧情"},
{"n": "励志", "v": "励志"},
{"n": "爱情", "v": "爱情"},
{"n": "喜剧", "v": "喜剧"},
{"n": "家庭", "v": "家庭"},
{"n": "历史", "v": "历史"},
{"n": "魔幻", "v": "魔幻"},
{"n": "恐怖", "v": "恐怖"},
{"n": "战争", "v": "战争"},
{"n": "武侠", "v": "武侠"}]},
{"key": "年代",
"name": "年代",
"value": [{"n": "全部", "v": ""},
{"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"}]}],
"962": [{"key": "地区",
"name": "地区",
"value": [{"n": "全部", "v": ""},
{"n": "内地", "v": "内地"},
{"n": "港台", "v": "港台"},
{"n": "日韩", "v": "日韩"},
{"n": "欧美", "v": "欧美"},
{"n": "泰国", "v": "泰国"},
{"n": "其他", "v": "其他"}]},
{"key": "类型",
"name": "类型",
"value": [{"n": "全部", "v": ""},
{"n": "都市", "v": "都市"},
{"n": "爱情", "v": "爱情"},
{"n": "战争", "v": "战争"},
{"n": "家庭", "v": "家庭"},
{"n": "悬疑", "v": "悬疑"},
{"n": "古装", "v": "古装"},
{"n": "短剧", "v": "短剧"},
{"n": "谍战", "v": "谍战"},
{"n": "喜剧", "v": "喜剧"},
{"n": "农村", "v": "农村"},
{"n": "刑侦", "v": "刑侦"},
{"n": "武侠", "v": "武侠"},
{"n": "历史", "v": "历史"}]},
{"key": "年代",
"name": "年代",
"value": [{"n": "全部", "v": ""},
{"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"}]}],
"963": [{"key": "地区",
"name": "地区",
"value": [{"n": "全部", "v": ""},
{"n": "内地", "v": "内地"},
{"n": "日韩", "v": "日韩"},
{"n": "欧美", "v": "欧美"},
{"n": "港台", "v": "港台"},
{"n": "其他", "v": "其他"}]},
{"key": "类型",
"name": "类型",
"value": [{"n": "全部", "v": ""},
{"n": "动作", "v": "动作"},
{"n": "冒险", "v": "冒险"},
{"n": "益智", "v": "益智"},
{"n": "亲子", "v": "亲子"},
{"n": "热血", "v": "热血"},
{"n": "剧情", "v": "剧情"},
{"n": "魔幻", "v": "魔幻"},
{"n": "励志", "v": "励志"},
{"n": "机战", "v": "机战"},
{"n": "搞笑", "v": "搞笑"},
{"n": "科幻", "v": "科幻"},
{"n": "治愈", "v": "治愈"},
{"n": "儿歌", "v": "儿歌"},
{"n": "教育", "v": "教育"},
{"n": "校园", "v": "校园"},
{"n": "童话", "v": "童话"},
{"n": "推理", "v": "推理"},
{"n": "怀旧", "v": "怀旧"},
{"n": "宠物", "v": "宠物"},
{"n": "舞蹈", "v": "舞蹈"}]},
{"key": "年代",
"name": "年代",
"value": [{"n": "全部", "v": ""},
{"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"}]}],
"965": [{"key": "地区",
"name": "地区",
"value": [{"n": "全部", "v": ""},
{"n": "内地", "v": "内地"},
{"n": "欧美", "v": "欧美"}]},
{"key": "类型",
"name": "类型",
"value": [{"n": "全部", "v": ""},
{"n": "文化", "v": "文化"},
{"n": "纪实", "v": "纪实"},
{"n": "访谈", "v": "访谈"},
{"n": "历史", "v": "历史"},
{"n": "美食", "v": "美食"},
{"n": "旅游", "v": "旅游"},
{"n": "时尚", "v": "时尚"},
{"n": "情感", "v": "情感"},
{"n": "生活", "v": "生活"},
{"n": "真人秀", "v": "真人秀"}]},
{"key": "年代",
"name": "年代",
"value": [{"n": "全部", "v": ""},
{"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"}]}],
"966": [{"key": "类型",
"name": "类型",
"value": [{"n": "全部", "v": ""},
{"n": "国内视野", "v": "国内视野"},
{"n": "国际纵览", "v": "国际纵览"},
{"n": "军事话题", "v": "军事话题"},
{"n": "社会百态", "v": "社会百态"},
{"n": "央视频", "v": "央视频"}]}]}}
return result
def generate_x_sign(self, secret_b64):
secret_key = self.decode_secret_key(secret_b64)
data = self.get_data_string()
signature = self.compute_signature(secret_key, data)
x_sign = self.encode_signature(signature)
return x_sign
def decode_secret_key(self, secret_b64):
return base64.b64decode(secret_b64)
def get_data_string(self):
return "{}"
def compute_signature(self, secret_key, data):
return hmac.new(secret_key, data.encode('utf-8'), hashlib.sha256).digest()
def encode_signature(self, signature):
return base64.b64encode(signature).decode('utf-8')
def get_current_app_key(self, xurl, headerx):
resp = self.fetch_index_page(xurl, headerx)
js_path = self.extract_js_path(resp.text)
js_url = f"{xurl}{js_path}"
js_content = self.fetch_js_content(js_url, headerx)
target_key_b64 = self.extract_target_key(js_content)
return target_key_b64
def fetch_index_page(self, xurl, headerx):
resp = requests.get(xurl, headers=headerx)
resp.raise_for_status()
return resp
def extract_js_path(self, html_content):
js_path_pattern = r'src="(/[\d\.]+/assets/js/index-[\w\.-]+\.js)"'
match = re.search(js_path_pattern, html_content)
return match.group(1)
def fetch_js_content(self, js_url, headerx):
js_resp = requests.get(js_url, headers=headerx)
js_resp.raise_for_status()
return js_resp.text
def extract_target_key(self, js_content):
key_pattern = r'const \w+="([^"]+)",\w+="([^"]+)",\w+="([^"]+)";'
key_match = re.search(key_pattern, js_content)
return key_match.group(2)
def get_headers(self, sign1, xurl):
return {
'accept': 'application/json, text/plain, */*',
'accept-language': 'zh-CN,zh;q=0.9,en;q=0.8,en-GB;q=0.7,en-US;q=0.6',
'cache-control': 'no-cache',
'launchchannel': 'web_channel',
'origin': xurl,
'pragma': 'no-cache',
'priority': 'u=1, i',
'referer': xurl,
'sec-ch-ua': '"Microsoft Edge";v="143", "Chromium";v="143", "Not A(Brand";v="24"',
'sec-ch-ua-mobile': '?0',
'sec-ch-ua-platform': '"Windows"',
'sec-fetch-dest': 'empty',
'sec-fetch-mode': 'cors',
'sec-fetch-site': 'cross-site',
'siteid': '1000101',
'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/143.0.0.0 Safari/537.36 Edg/143.0.0.0',
'x-sign': sign1,
}
def categoryContent(self, cid, pg, filter, ext):
secret_b64 = self.get_current_app_key(xurl, headerx)
sign1 = self.generate_x_sign(secret_b64)
page = self.get_page_number(pg)
NdType, DqType, LxType = self.extract_filter_types(ext)
headers = self.get_headers(sign1, xurl)
params = self.build_category_params(cid, page, NdType, DqType, LxType)
data = self.fetch_category_data(params, headers)
videos = self.parse_category_videos(data)
return self.build_category_result(videos, pg)
def get_page_number(self, pg):
return int(pg) if pg else 1
def extract_filter_types(self, ext):
NdType = ext.get('年代', '全部')
DqType = ext.get('地区', '全部')
LxType = ext.get('类型', '全部')
return NdType, DqType, LxType
def build_category_params(self, cid, page, NdType, DqType, LxType):
return {'functionName': 'getNewsSearchedByCondition','nodeId': cid,'nodeTag': LxType,'yearTag': NdType,'countryTag': DqType,'orderType': '0','pageSize': '40','page': page,'keyword': '','siteId': '1000101',}
def fetch_category_data(self, params, headers):
detail = requests.get('https://ups.5g.wasu.tv/rmp-user-suggest/1000101/hzhs/searchServlet', params=params,headers=headers)
detail.encoding = "utf-8"
return detail.json()
def parse_category_videos(self, data):
videos = []
for vod in data['data']:
videos.append(self.parse_category_video(vod))
return videos
def parse_category_video(self, vod):
return {"vod_id": f"{vod['nodeId']}@{vod['newsId']}","vod_name": vod['title'],"vod_pic": vod['hPic'],"vod_year": vod.get('pubTime', '暂无日期'),"vod_remarks": vod.get('episodeDesc', '暂无备注')}
def build_category_result(self, videos, pg):
result = {'list': videos}
result['page'] = pg
result['pagecount'] = 9999
result['limit'] = 90
result['total'] = 999999
return result
def detailContent(self, ids):
did = ids[0]
secret_b64 = self.get_current_app_key(xurl, headerx)
sign1 = self.generate_x_sign(secret_b64)
fenge = did.split("@")
headers = self.get_headers(sign1, xurl)
params = self.build_detail_params(fenge)
data = self.fetch_detail_data(params, headers)
content = self.build_content(data)
director = self.extract_detail_field(data, 'director')
actor = self.extract_detail_field(data, 'actor')
remarks = self.extract_detail_field(data, 'episodeDesc')
year = self.extract_detail_field(data, 'pubTime')
area = self.extract_detail_field(data, 'countryTag')
bofang = self.build_play_url(data)
videos = [self.build_video_data(did, director, actor, remarks, year, area, content, bofang)]
return self.build_result(videos)
def build_detail_params(self, fenge):
return {'siteId': '1000101', 'functionName': 'getCurrentNews', 'nodeId': fenge[0], 'newsId': fenge[1],'platform': 'web',}
def fetch_detail_data(self, params, headers):
detail = requests.get('https://mcspapp.5g.wasu.tv/bvradio_app/hzhs/newsServlet', params=params, headers=headers)
detail.encoding = "utf-8"
return detail.json()
def build_content(self, data):
return '介绍剧情📢' + data.get('data', {}).get('newsAbstract', '')
def extract_detail_field(self, data, field_name):
return data.get('data', {}).get(field_name, '')
def build_play_url(self, data):
bofang = ''
for vod in data['data']['vodList']:
name = vod['title']
if len(vod['fileList']) > 1:
id = vod['fileList'][1]['playUrl']
else:
id = vod['fileList'][0]['playUrl']
bofang += name + '$' + id + '#'
return bofang[:-1]
def build_video_data(self, did, director, actor, remarks, year, area, content, bofang):
return {"vod_id": did, "vod_director": director, "vod_actor": actor, "vod_remarks": remarks, "vod_year": year, "vod_area": area, "vod_content": content, "vod_play_from": "华数专线", "vod_play_url": bofang}
def build_result(self, videos):
result = {}
result['list'] = videos
return result
def get_x_sign_for_post(self, secret_b64, playUrl):
secret_key = self.decode_secret_key(secret_b64)
payload_dict = self.build_payload_dict(playUrl)
data_string = self.serialize_payload(payload_dict)
signature = self.compute_hmac_signature(secret_key, data_string)
x_sign = self.encode_signature(signature)
return x_sign
def decode_secret_key(self, secret_b64):
return base64.b64decode(secret_b64)
def build_payload_dict(self, playUrl):
return {"playUrl": playUrl, "platform": "web"}
def serialize_payload(self, payload_dict):
return json.dumps(payload_dict, separators=(',', ':'), ensure_ascii=False)
def compute_hmac_signature(self, secret_key, data_string):
return hmac.new(secret_key, data_string.encode('utf-8'), hashlib.sha256).digest()
def encode_signature(self, signature):
return base64.b64encode(signature).decode('utf-8')
def playerContent(self, flag, id, vipFlags):
secret_b64 = self.get_current_app_key(xurl, headerx)
id = self.normalize_play_url(id)
sign2 = self.get_x_sign_for_post(secret_b64, id)
headers = self.get_headers(sign2, xurl)
json_data = self.build_request_json(id)
result = self.post_play_url_request(headers, json_data)
play_url = result['data']['playUrl']
return self.build_player_result(play_url)
def normalize_play_url(self, id):
return id.replace('.mp4', '/playlist.m3u8')
def build_request_json(self, playUrl):
return {'playUrl': playUrl, 'platform': 'web', }
def post_play_url_request(self, headers, json_data):
response = requests.post('https://mcspapp.5g.wasu.tv/thirdApiFile/file/getPlayUrl', headers=headers, json=json_data)
return response.json()
def build_player_result(self, play_url):
result = {}
result["parse"] = 0
result["playUrl"] = ''
result["url"] = play_url
result["header"] = headerz
return result
def searchContentPage(self, key, quick, pg):
page = self.get_page_number(pg)
secret_b64 = self.get_current_app_key(xurl, headerx)
sign1 = self.generate_x_sign(secret_b64)
headers = self.get_headers(sign1, xurl)
params = self.build_search_params(key, page)
data = self.fetch_search_data(params, headers)
videos = self.parse_search_videos(data)
return self.build_search_result(videos, pg)
def get_page_number(self, pg):
return int(pg) if pg else 1
def build_search_params(self, key, page):
return {'functionName': 'getNewsSearched', 'searchNewsType': '3,4,5', 'keyword': key, 'pageSize': 10, 'page': page, 'siteId': 1000101}
def fetch_search_data(self, params, headers):
detail = requests.get("https://ups.5g.wasu.tv/rmp-user-suggest/1000101/hzhs/searchServlet", params=params, headers=headers)
detail.encoding = "utf-8"
return detail.json()
def parse_search_videos(self, data):
videos = []
for vod in data['data']:
videos.append(self.parse_search_video(vod))
return videos
def parse_search_video(self, vod):
return {"vod_id": f"{vod['nodeId']}@{vod['newsId']}", "vod_name": vod['title'], "vod_pic": vod['hPic'], "vod_year": vod.get('pubTime', '暂无日期'), "vod_remarks": vod.get('episodeDesc', '暂无备注')}
def build_search_result(self, videos, pg):
result = {'list': videos}
result['page'] = pg
result['pagecount'] = 9999
result['limit'] = 90
result['total'] = 999999
return result
def searchContent(self, key, quick, pg="1"):
return self.searchContentPage(key, quick, '1')
def localProxy(self, params):
if params['type'] == "m3u8":
return self.proxyM3u8(params)
elif params['type'] == "media":
return self.proxyMedia(params)
elif params['type'] == "ts":
return self.proxyTs(params)
return None
py/可可.py
+426
View File
@@ -0,0 +1,426 @@
#coding=utf-8
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
TVBox / 影视仓 Python源脚本
站点: 可可影视 (103.51.147.112:51120)
"""
import sys
import re
import json
import requests
import urllib3
from urllib.parse import quote
from pyquery import PyQuery as pq
sys.path.append('..')
from base.spider import Spider
# 禁用SSL警告
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
class Spider(Spider):
def __init__(self):
super().__init__()
self.site = 'https://103.51.147.112:51120'
self.session = requests.Session()
self.session.headers.update({
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36',
'Referer': 'https://103.51.147.112:51120/'
})
self.cateManual = {
'电影': '1',
'连续剧': '2',
'动漫': '3',
'综艺纪录': '4',
'短剧': '6'
}
# 隐晦的站点标识
self._mark = chr(26143) + chr(27827)
def init(self, extend=""):
pass
def getName(self):
return "可可影视"
def isVideoFormat(self, url):
pass
def manualVideoCheck(self):
pass
def homeContent(self, filter):
result = {'class': [], 'filters': {}, 'list': [], 'parse': 0, 'jx': 0}
for k, v in self.cateManual.items():
result['class'].append({
'type_id': str(v),
'type_name': k
})
return result
def homeVideoContent(self):
videos = []
try:
url = f'{self.site}/channel/1.html'
r = self.session.get(url, timeout=15, verify=False)
r.encoding = 'utf-8'
doc = pq(r.text)
items = doc('.module-item')
seen = set()
for item in items.items():
a = item.find('.v-item')
href = a.attr('href') or ''
vid = self.getVid(href)
if not vid or vid in seen:
continue
seen.add(vid)
# 标题
titles = item.find('.v-item-title')
title = ''
for j in range(len(titles)):
t = titles.eq(j).text().strip()
if t and t != '可可影视-kekys.com':
title = t
break
# 图片
pic = ''
imgs = item.find('img')
for j in range(len(imgs)):
img = imgs.eq(j)
src = img.attr('data-original') or ''
if src and 'placeholder' not in src and 'logo_placeholder' not in src:
pic = src
break
if pic and pic.startswith('/'):
pic = 'https://vres.zyxpedu.com' + pic
# 备注
note = ''
bottom = item.find('.v-item-bottom span')
if bottom.length:
note = bottom.text().strip()
if title:
videos.append({
'vod_id': vid,
'vod_name': title,
'vod_pic': pic,
'vod_remarks': note
})
except Exception as e:
print(f'homeVideoContent error: {e}')
return {'list': videos[:24], 'parse': 0, 'jx': 0}
def categoryContent(self, tid, pg, filter, extend):
result = {'list': [], 'parse': 0, 'jx': 0}
page = int(pg) if pg else 1
try:
url = f'{self.site}/channel/{tid}.html?page={page}'
r = self.session.get(url, timeout=15, verify=False)
r.encoding = 'utf-8'
doc = pq(r.text)
items = doc('.module-item')
for item in items.items():
a = item.find('.v-item')
href = a.attr('href') or ''
vid = self.getVid(href)
if not vid:
continue
# 标题
titles = item.find('.v-item-title')
title = ''
for j in range(len(titles)):
t = titles.eq(j).text().strip()
if t and t != '可可影视-kekys.com':
title = t
break
# 图片
pic = ''
imgs = item.find('img')
for j in range(len(imgs)):
img = imgs.eq(j)
src = img.attr('data-original') or ''
if src and 'placeholder' not in src and 'logo_placeholder' not in src:
pic = src
break
if pic and pic.startswith('/'):
pic = 'https://vres.zyxpedu.com' + pic
# 备注
note = ''
bottom = item.find('.v-item-bottom span')
if bottom.length:
note = bottom.text().strip()
if title:
result['list'].append({
'vod_id': vid,
'vod_name': title,
'vod_pic': pic,
'vod_remarks': note
})
except Exception as e:
print(f'categoryContent error: {e}')
result['page'] = page
# 修复：满页24条说明还有下一页，否则已到尾页
result['pagecount'] = page + 1 if len(result['list']) >= 24 else page
result['limit'] = len(result['list'])
result['total'] = len(result['list'])
return result
def detailContent(self, ids):
result = {'list': [], 'parse': 0, 'jx': 0}
vid = ids[0] if ids else ''
if not vid:
return result
try:
url = f'{self.site}/detail/{vid}.html'
r = self.session.get(url, timeout=15, verify=False)
r.encoding = 'utf-8'
html = r.text
# 标题：从title标签提取，最可靠
title = ''
title_match = re.search(r'<title>(.+?)</title>', html)
if title_match:
title = title_match.group(1).split('-')[0].strip()
# 去掉特殊字符水印
title = re.sub(r'[𝕜𝕜𝕪𝕤𝟘𝟙𝕔𝕠𝕞.\s]+', ' ', title).strip()
title = re.sub(r'\s+', ' ', title).strip()
# 图片：从og:image提取
pic = ''
og_img = re.search(r'<meta\s+property="og:image"\s+content="([^"]+)"', html)
if og_img:
pic = og_img.group(1)
if pic.startswith('/'):
pic = 'https://vres.zyxpedu.com' + pic
# 简介：从meta description提取
desc = ''
desc_match = re.search(r'<meta\s+name="description"\s+content="([^"]+)"', html)
if desc_match:
desc = desc_match.group(1).strip()
# 播放线路和集数
play_from = []
play_url = []
episodes_by_sid = {}
sids_in_order = []
seen_sids = set()
# 纯正则提取所有播放链接
all_play = re.findall(r'<a[^>]+href="(/play/\d+-(\d+)-(\d+)\.html)"[^>]+class="episode-item"[^>]*>(.*?)</a>', html, re.DOTALL)
# 如果没匹配到，试试class在href前面的情况
if not all_play:
all_play = re.findall(r'<a[^>]+class="episode-item"[^>]+href="(/play/\d+-(\d+)-(\d+)\.html)"[^>]*>(.*?)</a>', html, re.DOTALL)
for href, sid, nid, link_html in all_play:
text = re.sub(r'<[^>]+>', '', link_html).strip()
if not text:
continue
if sid not in episodes_by_sid:
episodes_by_sid[sid] = []
episodes_by_sid[sid].append(f'{text}${href}')
if sid not in seen_sids:
seen_sids.add(sid)
sids_in_order.append(sid)
# 线路名称
source_labels = []
all_labels = re.findall(r'class="source-item-label"[^>]*>([^<]+)</', html)
for label in all_labels:
label = label.strip()
if label:
source_labels.append(label)
for i, sid in enumerate(sids_in_order):
if sid in episodes_by_sid and episodes_by_sid[sid]:
if i < len(source_labels):
line_name = source_labels[i]
else:
line_name = f'线路{sid}'
# 跳过4K线路（只有APP端能用）
if line_name == '4K':
continue
play_from.append(line_name)
play_url.append('#'.join(episodes_by_sid[sid]))
vod = {
'vod_id': vid,
'vod_name': title,
'vod_pic': pic,
'type_name': '',
'vod_year': '',
'vod_area': '',
'vod_remarks': '',
'vod_actor': '',
'vod_director': self._mark,
'vod_content': desc,
'vod_play_from': '$$$'.join(play_from) if play_from else '',
'vod_play_url': '$$$'.join(play_url) if play_url else ''
}
result['list'].append(vod)
except Exception as e:
print(f'detailContent error: {e}')
return result
def playerContent(self, flag, id, vipFlags):
result = {}
try:
play_url = id
if id and not id.startswith('http'):
play_url = self.site + id
r = self.session.get(play_url, timeout=15, verify=False)
r.encoding = 'utf-8'
video_url = ''
patterns = [
r"src:\s*['\"]([^'\"]+\.(m3u8|mp4)[^'\"]*)['\"]",
r'"url"\s*:\s*"([^"]+\.(m3u8|mp4)[^"]*)"',
r"url\s*:\s*'([^']+\.(m3u8|mp4)[^']*)'",
]
for pat in patterns:
m = re.search(pat, r.text, re.DOTALL)
if m:
video_url = m.group(1)
break
# 修复：使用非捕获组 (?:m3u8|mp4)，避免 re.findall 只返回捕获组内容
if not video_url:
all_urls = re.findall(r'https?://[^\s"\'<>]+\.(?:m3u8|mp4)[^\s"\'<>]*', r.text)
if all_urls:
for u in all_urls:
if 'index.m3u8' in u or 'video.m3u8' in u or '.mp4' in u:
video_url = u
break
if not video_url:
video_url = all_urls[0]
# 修复：处理相对路径视频URL
if video_url and not video_url.startswith('http'):
if video_url.startswith('/'):
video_url = self.site + video_url
else:
video_url = self.site + '/' + video_url
if video_url:
result['parse'] = 0
result['url'] = video_url
result['jx'] = 0
result['header'] = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36',
'Referer': self.site + '/'
}
else:
result['parse'] = 1
result['url'] = play_url
result['jx'] = 0
result['header'] = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36',
'Referer': self.site + '/'
}
except Exception as e:
print(f'playerContent error: {e}')
result['parse'] = 1
result['url'] = id if id.startswith('http') else self.site + id
result['jx'] = 0
result['header'] = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36',
'Referer': self.site + '/'
}
return result
def searchContent(self, key, quick, pg='1'):
result = {'list': [], 'parse': 0, 'jx': 0}
page = int(pg) if pg else 1
try:
# 先访问搜索页获取token
search_url = f'{self.site}/search?k={quote(key)}'
r = self.session.get(search_url, timeout=15, verify=False)
r.encoding = 'utf-8'
t_match = re.search(r'name="t" value="([^"]+)"', r.text)
t = t_match.group(1) if t_match else ''
if t:
url = f'{self.site}/search?k={quote(key)}&t={quote(t)}'
if page > 1:
url += f'&page={page}'
r = self.session.get(url, timeout=15, verify=False)
r.encoding = 'utf-8'
doc = pq(r.text)
items = doc('.search-result-item')
for item in items.items():
# 修复：兼容 href 在元素本身或子级 <a> 标签上的情况
href = item.attr('href') or ''
if not href:
a_tag = item.find('a')
if a_tag.length:
href = a_tag.attr('href') or ''
vid = self.getVid(href)
if not vid:
continue
# 标题
title = ''
title_elem = item.find('.title')
if title_elem.length:
title = title_elem.text().strip()
if not title:
img = item.find('img')
if img.length:
title = img.attr('alt') or img.attr('title') or ''
title = title.strip()
# 图片
pic = ''
imgs = item.find('img')
for j in range(len(imgs)):
img = imgs.eq(j)
src = img.attr('data-original') or img.attr('src') or ''
if src and 'placeholder' not in src and 'logo_placeholder' not in src:
pic = src
break
if pic and pic.startswith('/'):
pic = 'https://vres.zyxpedu.com' + pic
if title:
result['list'].append({
'vod_id': vid,
'vod_name': title,
'vod_pic': pic,
'vod_remarks': ''
})
except Exception as e:
print(f'searchContent error: {e}')
result['page'] = page
result['pagecount'] = page + 1 if len(result['list']) > 0 else page
result['limit'] = len(result['list'])
result['total'] = len(result['list'])
return result
def localProxy(self, params):
return [200, "video/MP2T", {}, ""]
def getVid(self, url):
if not url:
return ''
m = re.search(r'/detail/(\d+)\.html', url)
if m:
return m.group(1)
m = re.search(r'/play/(\d+)-', url)
if m:
return m.group(1)
return ''
py/咖啡直播.py
+375
View File
@@ -0,0 +1,375 @@
# coding=utf-8
#!/usr/bin/env python3
# @name 咖啡直播
# @author 转自 OmniBox JS
# @description 体育赛事录像回放 + 直播（足球/篮球/NBA）
# @version 2.0.0
import json
import requests
class Spider:
def getName(self):
return "咖啡直播"
def getDependence(self):
return []
def init(self, extend=""):
self.host = "https://kafeizhibo.cc"
self.headers = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/134.0.0.0 Safari/537.36",
"Accept": "application/json, text/plain, */*",
"Referer": "https://kafeizhibo.com/live/all"
}
def log(self, msg):
print("[咖啡直播] " + str(msg))
# =========================================================
# 工具
# =========================================================
def _normalize_url(self, path):
if not path:
return ""
if path.startswith("http"):
return path
if path.startswith("//"):
return "https:" + path
return self.host + ("" if path.startswith("/") else "/") + path
def _get(self, path, params=None, referer=None):
h = dict(self.headers)
if referer:
h["Referer"] = referer
resp = requests.get(self.host + path, headers=h, params=params, timeout=10)
return resp.json()
# =========================================================
# 直播部分
# =========================================================
def _fetch_live_all(self):
"""GET /api/v1/archor — 返回所有正在直播的频道"""
return self._get("/api/v1/archor", referer=self.host + "/live/all")
def _parse_live_list(self, items, category_filter=None):
"""
category: 1=足球, 2=篮球, None=全部
每个 archor 代表一个独立直播频道（同一场球可能有多个频道）
合并同 match_id 的频道到一个 vod，多线路在 detail 里处理
"""
# 按 room_id 去重（同一 room_id 只取第一个，避免重复）
seen_rooms = set()
result = []
for item in items:
if category_filter and item.get("category") != category_filter:
continue
room_id = str(item.get("room_id", ""))
if room_id in seen_rooms:
continue
seen_rooms.add(room_id)
home = item.get("home_team", "")
away = item.get("away_team", "")
league = item.get("league_name", "")
h_score = item.get("home_score", 0)
a_score = item.get("away_score", 0)
title = "{} vs {} ({})".format(home, away, league)
pic = self._normalize_url(item.get("screenshot", ""))
if not pic or "default" in pic:
mi = item.get("match_info") or {}
pic = mi.get("home_team_logo", "")
result.append({
"vod_id": "live_{}".format(room_id),
"vod_name": "🔴 " + title,
"vod_pic": pic,
"vod_remarks": "{} - {} | {}".format(h_score, a_score, item.get("name", "")),
})
return result
def _detail_live(self, room_id):
"""GET /api/v1/room/{room_id} — 获取直播间多线路"""
try:
data = self._get(
"/api/v1/room/{}".format(room_id),
referer=self.host + "/room/{}".format(room_id)
)
if data.get("code") != 200 or not data.get("data"):
return {"list": []}
d = data["data"]
room_info = d.get("room_info", {})
signals = d.get("signals", [])
home = room_info.get("home_team", "")
away = room_info.get("away_team", "")
league = room_info.get("league", "")
h_score = room_info.get("home_score", 0)
a_score = room_info.get("away_score", 0)
title = "{} vs {} ({})".format(home, away, league)
teams = d.get("teams", {})
pic = (teams.get("home") or {}).get("logo", "")
# 每条 signal 是一个线路（官方直播/原声直播）
episodes = []
for sig in signals:
url = sig.get("stream_url", "")
if url:
name = sig.get("name", "线路")
episodes.append("{}${}".format(name, url))
# 如果 signals 为空，fallback 到 archor
if not episodes:
archor = d.get("archor", {})
url = archor.get("stream_url", "")
if url:
episodes.append("{}${}".format(archor.get("name", "直播"), url))
vod = {
"vod_id": "live_{}".format(room_id),
"vod_name": "🔴 " + title,
"vod_pic": pic,
"vod_content": "{} {} vs {}，比分 {} - {}".format(
league, home, away, h_score, a_score
),
"vod_play_from": "直播线路",
"vod_play_url": "#".join(episodes),
}
return {"list": [vod]}
except Exception as e:
self.log("直播详情失败: " + str(e))
return {"list": []}
# =========================================================
# 录像部分
# =========================================================
def _fetch_recordings(self, page=1, size=30, league=None, type_id=None):
params = {"page": page, "size": size}
if league:
params["league"] = league
elif type_id and type_id not in ("all", "nba", "live_all", "live_1", "live_2"):
params["type"] = type_id
h = dict(self.headers)
h["Referer"] = self.host + "/pc/replay"
resp = requests.get(self.host + "/api/v1/recordings", headers=h, params=params, timeout=10)
return resp.json()
def _parse_video_list(self, items):
result = []
for item in items:
title = "{} vs {} ({})".format(
item["home_team"], item["away_team"], item["league_name"]
)
score = "{} - {}".format(item["home_score"], item["away_score"])
pic = item.get("cover_image", "")
if pic and not pic.startswith("http"):
pic = self._normalize_url(pic)
if not pic or "default_cover" in pic:
pic = item.get("home_team_logo", "")
remarks = "{} | {} | {}个录像".format(
score, item["start_time"], item.get("recording_count", 0)
)
result.append({
"vod_id": str(item["match_id"]),
"vod_name": title,
"vod_pic": pic,
"vod_remarks": remarks,
})
return result
def _detail_recording(self, vid):
try:
h = dict(self.headers)
h["Referer"] = self.host + "/pc/replay"
resp = requests.get(
"{}/api/v1/match/{}/recordings".format(self.host, vid),
headers=h,
timeout=10
)
data = resp.json()
if data.get("code") != 200 or not data.get("data"):
return {"list": []}
match = data["data"]["match"]
replays = data["data"].get("replays", [])
highlights = data["data"].get("highlights", [])
title = "{} vs {} ({})".format(
match["home_team"], match["away_team"], match["league_name"]
)
pic = match.get("home_team_logo") or match.get("away_team_logo") or ""
episodes = []
for idx, rec in enumerate(replays):
if rec.get("video_url"):
name = rec.get("title") or "录像{}".format(idx + 1)
episodes.append("{}${}".format(name, rec["video_url"]))
for idx, rec in enumerate(highlights):
if rec.get("video_url"):
name = rec.get("title") or "集锦{}".format(idx + 1)
episodes.append("{}${}".format(name, rec["video_url"]))
vod = {
"vod_id": str(vid),
"vod_name": title,
"vod_pic": pic,
"vod_content": "{} {} {} vs {}，比分 {} - {}，比赛时间：{}".format(
match["league_name"], match.get("match_round", ""),
match["home_team"], match["away_team"],
match["home_score"], match["away_score"],
match["start_time"]
),
"vod_play_from": "录像源",
"vod_play_url": "#".join(episodes),
}
return {"list": [vod]}
except Exception as e:
self.log("录像详情失败: " + str(e))
return {"list": []}
# =========================================================
# FongMi 接口
# =========================================================
def homeContent(self, filter):
categories = [
# 直播分类
{"type_id": "live_all", "type_name": "🔴 直播全部"},
{"type_id": "live_1", "type_name": "🔴 直播足球"},
{"type_id": "live_2", "type_name": "🔴 直播篮球"},
# 录像分类
{"type_id": "all", "type_name": "录像全部"},
{"type_id": "1", "type_name": "录像足球"},
{"type_id": "2", "type_name": "录像篮球"},
{"type_id": "nba", "type_name": "录像NBA"},
]
# 首页展示直播列表
try:
data = self._fetch_live_all()
vod_list = self._parse_live_list(data.get("data", [])) if data.get("code") == 200 else []
except Exception as e:
self.log("首页失败: " + str(e))
vod_list = []
return {"class": categories, "list": vod_list}
def homeVideoContent(self):
return {"list": []}
def categoryContent(self, tid, pg, filter, extend):
pg = int(pg) if pg else 1
# ---- 直播分类 ----
if tid in ("live_all", "live_1", "live_2"):
try:
data = self._fetch_live_all()
if data.get("code") == 200:
cat = None if tid == "live_all" else int(tid.split("_")[1])
vod_list = self._parse_live_list(data.get("data", []), category_filter=cat)
else:
vod_list = []
except Exception as e:
self.log("直播分类失败: " + str(e))
vod_list = []
return {"list": vod_list, "page": 1, "pagecount": 1, "limit": 100, "total": len(vod_list)}
# ---- 录像分类 ----
try:
if tid == "nba":
data = self._fetch_recordings(pg, 20, league="NBA")
size = 20
elif tid == "all":
data = self._fetch_recordings(pg, 30)
size = 30
else:
data = self._fetch_recordings(pg, 30, type_id=tid)
size = 30
vod_list = []
pagecount = 1
if data.get("code") == 200 and data.get("data"):
vod_list = self._parse_video_list(data["data"])
pagecount = pg + 1 if len(data["data"]) == size else pg
except Exception as e:
self.log("录像分类失败: " + str(e))
vod_list = []
pagecount = 1
return {"list": vod_list, "page": pg, "pagecount": pagecount, "limit": 30, "total": len(vod_list)}
def detailContent(self, ids):
vid = ids[0] if isinstance(ids, list) and ids else str(ids)
if vid.startswith("live_"):
room_id = vid[5:] # 去掉 "live_" 前缀
return self._detail_live(room_id)
else:
return self._detail_recording(vid)
def searchContent(self, key, quick, pg=1):
if not key:
return {"list": []}
keyword = key.lower()
result = []
# 搜索直播
try:
data = self._fetch_live_all()
if data.get("code") == 200:
for item in data.get("data", []):
if (keyword in item.get("home_team", "").lower()
or keyword in item.get("away_team", "").lower()
or keyword in item.get("league_name", "").lower()
or keyword in item.get("title", "").lower()):
room_id = str(item.get("room_id", ""))
home = item.get("home_team", "")
away = item.get("away_team", "")
league = item.get("league_name", "")
result.append({
"vod_id": "live_{}".format(room_id),
"vod_name": "🔴 {} vs {} ({})".format(home, away, league),
"vod_pic": "",
"vod_remarks": "直播中",
})
except Exception as e:
self.log("搜索直播失败: " + str(e))
# 搜索录像
try:
data = self._fetch_recordings(1, 100)
if data.get("code") == 200:
for item in data["data"]:
if (keyword in item["home_team"].lower()
or keyword in item["away_team"].lower()
or keyword in item["league_name"].lower()):
title = "{} vs {} ({})".format(
item["home_team"], item["away_team"], item["league_name"]
)
result.append({
"vod_id": str(item["match_id"]),
"vod_name": title,
"vod_pic": "",
"vod_remarks": "{} - {}".format(item["home_score"], item["away_score"]),
})
except Exception as e:
self.log("搜索录像失败: " + str(e))
return {"list": result, "page": 1, "pagecount": 1}
def playerContent(self, flag, id, vipFlags):
return {
"parse": 0,
"playUrl": "",
"url": id,
"header": json.dumps({
"User-Agent": self.headers["User-Agent"],
"Referer": self.host,
"Origin": self.host,
})
}
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
+225
View File
@@ -0,0 +1,225 @@
# -*- coding: utf-8 -*-
# by @嗷呜
import json
import sys
import threading
import uuid
import requests
sys.path.append('..')
from base.spider import Spider
import time
from Crypto.Hash import MD5, SHA1
class Spider(Spider):
'''
配置示例：
{
"key": "xxxx",
"name": "xxxx",
"type": 3,
"api": ".所在路径/金牌.py",
"searchable": 1,
"quickSearch": 1,
"filterable": 1,
"changeable": 1,
"ext": {
"site": "https://www.jiabaide.cn,域名2,域名3"
}
},
'''
def init(self, extend=""):
if extend:
hosts=json.loads(extend)['site']
self.host = self.host_late(hosts)
pass
def getName(self):
pass
def isVideoFormat(self, url):
pass
def manualVideoCheck(self):
pass
def destroy(self):
pass
def homeContent(self, filter):
cdata = self.fetch(f"{self.host}/api/mw-movie/anonymous/get/filer/type", headers=self.getheaders()).json()
fdata = self.fetch(f"{self.host}/api/mw-movie/anonymous/v1/get/filer/list", headers=self.getheaders()).json()
result = {}
classes = []
filters={}
for k in cdata['data']:
classes.append({
'type_name': k['typeName'],
'type_id': str(k['typeId']),
})
sort_values = [{"n": "最近更新", "v": "2"},{"n": "人气高低", "v": "3"}, {"n": "评分高低", "v": "4"}]
for tid, d in fdata['data'].items():
current_sort_values = sort_values.copy()
if tid == '1':
del current_sort_values[0]
filters[tid] = [
{"key": "type", "name": "类型",
"value": [{"n": i["itemText"], "v": i["itemValue"]} for i in d["typeList"]]},
*([] if not d["plotList"] else [{"key": "v_class", "name": "剧情",
"value": [{"n": i["itemText"], "v": i["itemText"]}
for i in d["plotList"]]}]),
{"key": "area", "name": "地区",
"value": [{"n": i["itemText"], "v": i["itemText"]} for i in d["districtList"]]},
{"key": "year", "name": "年份",
"value": [{"n": i["itemText"], "v": i["itemText"]} for i in d["yearList"]]},
{"key": "lang", "name": "语言",
"value": [{"n": i["itemText"], "v": i["itemText"]} for i in d["languageList"]]},
{"key": "sort", "name": "排序", "value": current_sort_values}
]
result['class'] = classes
result['filters'] = filters
return result
def homeVideoContent(self):
data1 = self.fetch(f"{self.host}/api/mw-movie/anonymous/v1/home/all/list", headers=self.getheaders()).json()
data2=self.fetch(f"{self.host}/api/mw-movie/anonymous/home/hotSearch",headers=self.getheaders()).json()
data=[]
for i in data1['data'].values():
data.extend(i['list'])
data.extend(data2['data'])
vods=self.getvod(data)
return {'list':vods}
def categoryContent(self, tid, pg, filter, extend):
params = {
"area": extend.get('area', ''),
"filterStatus": "1",
"lang": extend.get('lang', ''),
"pageNum": pg,
"pageSize": "30",
"sort": extend.get('sort', '1'),
"sortBy": "1",
"type": extend.get('type', ''),
"type1": tid,
"v_class": extend.get('v_class', ''),
"year": extend.get('year', '')
}
data = self.fetch(f"{self.host}/api/mw-movie/anonymous/video/list?{self.js(params)}", headers=self.getheaders(params)).json()
result = {}
result['list'] = self.getvod(data['data']['list'])
result['page'] = pg
result['pagecount'] = 9999
result['limit'] = 90
result['total'] = 999999
return result
def detailContent(self, ids):
data=self.fetch(f"{self.host}/api/mw-movie/anonymous/video/detail?id={ids[0]}",headers=self.getheaders({'id':ids[0]})).json()
vod=self.getvod([data['data']])[0]
vod['vod_play_from']='金牌'
vod['vod_play_url'] = '#'.join(
f"{i['name'] if len(vod['episodelist']) > 1 else vod['vod_name']}${ids[0]}@@{i['nid']}" for i in
vod['episodelist'])
vod.pop('episodelist', None)
return {'list':[vod]}
def searchContent(self, key, quick, pg="1"):
params = {
"keyword": key,
"pageNum": pg,
"pageSize": "8",
"sourceCode": "1"
}
data=self.fetch(f"{self.host}/api/mw-movie/anonymous/video/searchByWord?{self.js(params)}",headers=self.getheaders(params)).json()
vods=self.getvod(data['data']['result']['list'])
return {'list':vods,'page':pg}
def playerContent(self, flag, id, vipFlags):
self.header = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64; ) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/126.0.6478.61 Chrome/126.0.6478.61 Not/A)Brand/8 Safari/537.36',
'sec-ch-ua-platform': '"Windows"',
'DNT': '1',
'sec-ch-ua': '"Not/A)Brand";v="8", "Chromium";v="126", "Google Chrome";v="126"',
'sec-ch-ua-mobile': '?0',
'Origin': self.host,
'Referer': f'{self.host}/'
}
ids=id.split('@@')
pdata = self.fetch(f"{self.host}/api/mw-movie/anonymous/v2/video/episode/url?clientType=1&id={ids[0]}&nid={ids[1]}",headers=self.getheaders({'clientType':'1','id': ids[0], 'nid': ids[1]})).json()
vlist=[]
for i in pdata['data']['list']:vlist.extend([i['resolutionName'],i['url']])
return {'parse':0,'url':vlist,'header':self.header}
def localProxy(self, param):
pass
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:
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 md5(self, sign_key):
md5_hash = MD5.new()
md5_hash.update(sign_key.encode('utf-8'))
md5_result = md5_hash.hexdigest()
return md5_result
def js(self, param):
return '&'.join(f"{k}={v}" for k, v in param.items())
def getheaders(self, param=None):
if param is None:param = {}
t=str(int(time.time()*1000))
param['key']='cb808529bae6b6be45ecfab29a4889bc'
param['t']=t
sha1_hash = SHA1.new()
sha1_hash.update(self.md5(self.js(param)).encode('utf-8'))
sign = sha1_hash.hexdigest()
deviceid = str(uuid.uuid4())
headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64; ) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/126.0.6478.61 Chrome/126.0.6478.61 Not/A)Brand/8 Safari/537.36',
'Accept': 'application/json, text/plain, */*',
'sign': sign,
't': t,
'deviceid':deviceid
}
return headers
def convert_field_name(self, field):
field = field.lower()
if field.startswith('vod') and len(field) > 3:
field = field.replace('vod', 'vod_')
if field.startswith('type') and len(field) > 4:
field = field.replace('type', 'type_')
return field
def getvod(self, array):
return [{self.convert_field_name(k): v for k, v in item.items()} for item in array]