yeyin/tvbox_collect
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Sync all projects

Browse Source
This commit is contained in:
github-actions[bot]
2026-07-07 18:31:13 +00:00
parent 14a2f7cc9b
commit 9f853387bd
46 changed files with 4383 additions and 4320 deletions
Download Patch File Download Diff File Expand all files Collapse all files
FGBLH/Web鱼壳海豚.json
+6
View File
@@ -50,6 +50,12 @@
"searchable": 1,
"changeable": 1,
"ext": "https://fastlink.cokey.xyz/f/1MOEc5/%E8%85%BE%E8%85%BE.js"
},
{
"key": "txvm",
"name": "🐬腾讯视频.py[追剧]",
"type": 3,
"api": "https://ghfast.top/https://raw.githubusercontent.com/FGBLH/HKL/refs/heads/main/py/腾讯视频.py"
},
{"key":"MG",
"name":"🐬芒果TV[追剧]",
FGBLH/Web鱼壳海豚py.json
+6
View File
@@ -15,6 +15,12 @@
"name": "🐬芒果TV.py海豚影视交流群 TG：@hshsjk9[追剧]",
"type": 3,
"api": "https://ghfast.top/https://raw.githubusercontent.com/FGBLH/HKL/refs/heads/main/py/芒果TV.py"
},
{
"key": "txvm",
"name": "🐬腾讯视频.py[追剧]",
"type": 3,
"api": "https://ghfast.top/https://raw.githubusercontent.com/FGBLH/HKL/refs/heads/main/py/腾讯视频.py"
},
{
"key": "YK",
FGBLH/Web鱼壳海豚无18加.json
+6
View File
@@ -50,6 +50,12 @@
"searchable": 1,
"changeable": 1,
"ext": "https://fastlink.cokey.xyz/f/1MOEc5/%E8%85%BE%E8%85%BE.js"
},
{
"key": "txvm",
"name": "🐬腾讯视频.py",
"type": 3,
"api": "https://ghfast.top/https://raw.githubusercontent.com/FGBLH/HKL/refs/heads/main/py/腾讯视频.py"
},
{"key":"MG",
"name":"🐬芒果TV",
FGBLH/ok海豚.json
+6
View File
@@ -37,6 +37,12 @@
"searchable": 1,
"changeable": 1,
"ext": "https://fastlink.cokey.xyz/f/1MOEc5/%E8%85%BE%E8%85%BE.js"
},
{
"key": "txvm",
"name": "🐬腾讯视频.py",
"type": 3,
"api": "https://ghfast.top/https://raw.githubusercontent.com/FGBLH/HKL/refs/heads/main/py/腾讯视频.py"
},
{"key":"MG",
"name":"🐬芒果TV",
FGBLH/ok海豚py.json
+6
View File
@@ -9,6 +9,12 @@
"name": "🐬芒果TV.py 海豚影视交流群 TG：@hshsjk9",
"type": 3,
"api": "https://ghfast.top/https://raw.githubusercontent.com/FGBLH/HKL/refs/heads/main/py/芒果TV.py"
},
{
"key": "txvm",
"name": "🐬腾讯视频.py",
"type": 3,
"api": "https://ghfast.top/https://raw.githubusercontent.com/FGBLH/HKL/refs/heads/main/py/腾讯视频.py"
},
{
"key": "YK",
FGBLH/ok海豚无18加.json
+6
View File
@@ -37,6 +37,12 @@
"searchable": 1,
"changeable": 1,
"ext": "https://fastlink.cokey.xyz/f/1MOEc5/%E8%85%BE%E8%85%BE.js"
},
{
"key": "txvm",
"name": "🐬腾讯视频.py",
"type": 3,
"api": "https://ghfast.top/https://raw.githubusercontent.com/FGBLH/HKL/refs/heads/main/py/腾讯视频.py"
},
{"key":"MG",
"name":"🐬芒果TV",
FGBLH/py/stripchat.py
+24 -150
View File
@@ -1,35 +1,17 @@
# coding=utf-8
# !/usr/bin/python
import sys
import base64
import hashlib
import requests
from typing import Tuple
from base.spider import Spider
from datetime import datetime, timedelta
from urllib.parse import quote, unquote
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
sys.path.append('..')
# 搜索用户名，关键词格式为“类别+空格+关键词”
# 类别在标签上已注明，比如“女主播g”，则搜索类别为“g”
# 搜索“g per”，则在“女主播”中搜索“per”, 关键词不区分大小写，但至少3位，否则空结果
from base.spider import Spider
class Spider(Spider):
def init(self, extend="{}"):
origin = 'https://zh.stripchat.com'
self.host = origin
self.host='https://zh.stripol.com/'
self.headers = {
'Origin': origin,
'Referer': f"{origin}/",
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:144.0) Gecko/20100101 Firefox/144.0'
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:142.0) Gecko/20100101 Firefox/142.0'
}
self.stripchat_key = self.decode_key_compact()
# 缓存字典
self._hash_cache = {}
self.create_session_with_retry()
def getName(self):
pass
@@ -45,7 +27,7 @@ class Spider(Spider):
def homeContent(self, filter):
result = {}
classes = [{'type_name': '女主播g', 'type_id': 'girls'}, {'type_name': '情侣c', 'type_id': 'couples'}, {'type_name': '男主播m', 'type_id': 'men'}, {'type_name': '跨性别t', 'type_id': 'trans'}]
classes = [{'type_name': '女主播', 'type_id': 'girls'}, {'type_name': '情侣', 'type_id': 'couples'}, {'type_name': '男主播', 'type_id': 'men'}, {'type_name': '跨性别', 'type_id': 'trans'}]
filters = {}
value = [{'n': '中国', 'v': 'tagLanguageChinese'}, {'n': '亚洲', 'v': 'ethnicityAsian'}, {'n': '白人', 'v': 'ethnicityWhite'}, {'n': '拉丁', 'v': 'ethnicityLatino'}, {'n': '混血', 'v': 'ethnicityMultiracial'}, {'n': '印度', 'v': 'ethnicityIndian'}, {'n': '阿拉伯', 'v': 'ethnicityMiddleEastern'}, {'n': '黑人', 'v': 'ethnicityEbony'}]
value_gay = [{'n': '情侣', 'v': 'sexGayCouples'}, {'n': '直男', 'v': 'orientationStraight'}]
@@ -64,7 +46,7 @@ class Spider(Spider):
def categoryContent(self, tid, pg, filter, extend):
limit = 60
offset = limit * (int(pg) - 1)
domain = f"{self.host}/api/front/models?improveTs=false&removeShows=false&limit={limit}&offset={offset}&primaryTag={tid}&sortBy=stripRanking&rcmGrp=A&rbCnGr=true&prxCnGr=false&nic=false"
domain = f"{self.host}api/front/models?improveTs=false&removeShows=false&limit={limit}&offset={offset}&primaryTag={tid}&sortBy=viewersRating&rcmGrp=A&rbCnGr=true&prxCnGr=false&nic=false"
if 'tag' in extend:
domain += "&filterGroupTags=%5B%5B%22" + extend['tag'] + "%22%5D%5D"
rsp = requests.get(domain, headers=self.headers).json()
@@ -72,16 +54,13 @@ class Spider(Spider):
videos = []
for vod in vodList:
id = str(vod['id'])
name = str(vod['username']).strip()
title = str(vod['username']).strip()
stamp = vod['snapshotTimestamp']
country = str(vod['country']).strip()
flag = self.country_code_to_flag(country)
remark = "🎫" if vod['status'] == "groupShow" else ""
videos.append({
"vod_id": name,
"vod_name": f"{flag}{name}",
"vod_id": title,
"vod_name": title,
"vod_pic": f"https://img.doppiocdn.net/thumbs/{stamp}/{id}",
"vod_remarks": remark
"vod_remarks": "购票表演中" if vod['groupShowType'] else ""
})
total = int(rsp['filteredCount'])
result = {}
@@ -94,68 +73,32 @@ class Spider(Spider):
def detailContent(self, array):
username = array[0]
domain = f"{self.host}/api/front/v2/models/username/{username}/cam"
domain = f"{self.host}api/front/v2/models/username/{username}/cam"
rsp = requests.get(domain, headers=self.headers).json()
info = rsp['cam']
user = rsp['user']['user']
id = str(user['id'])
country = str(user['country']).strip()
isLive = "" if user['isLive'] else " 已下播"
flag = self.country_code_to_flag(country)
remark = ''
if info['show']:
show = info['show']['details']['groupShow']
BJtime = (datetime.strptime(show["startAt"], "%Y-%m-%dT%H:%M:%SZ") + timedelta(hours=8)).strftime("%m月%d日 %H:%M")
remark = f"🎫 始于 {BJtime}"
vod = [{
vod = {
"vod_id": id,
"vod_name": str(info['topic']).strip(),
"vod_pic": str(user['avatarUrl']),
"vod_director": f"{flag}{username}{isLive}",
"vod_remarks": remark,
'vod_play_from': 'StripChat',
"vod_director": username,
"vod_area": str(user['country']),
'vod_play_from': '直播线路',
'vod_play_url': f"{id}${id}"
}]
result = {}
result['list'] = vod
}
result = {
'list': [
vod
]
}
return result
def process_key(self, key: str) -> Tuple[str, str]:
tags = {'G': 'girls', 'C': 'couples', 'M': 'men', 'T': 'trans'}
parts = key.split(maxsplit=1) # 仅分割第一个空格
if len(parts) > 1 and tags.get(parts[0].upper(), ''):
return tags[parts[0].upper()], parts[1].strip()
return 'girls', key.strip()
def searchContent(self, key, quick, pg="1"):
result = {}
if int(pg) > 1:
return result
tag, key = self.process_key(key)
domain = f"{self.host}/api/front/v4/models/search/group/username?query={key}&limit=900&primaryTag={tag}"
rsp = requests.get(domain, headers=self.headers).json()
users = rsp['models']
videos = []
for user in users:
if not user['isLive']:
continue
id = str(user['id'])
name = str(user['username']).strip()
stamp = user['snapshotTimestamp']
country = str(user['country']).strip()
flag = self.country_code_to_flag(country)
remark = "🎫" if user['status'] == "groupShow" else ""
videos.append({
"vod_id": name,
"vod_name": f"{flag}{name}",
"vod_pic": f"https://img.doppiocdn.net/thumbs/{stamp}/{id}",
"vod_remarks": remark
})
result['list'] = videos
return result
pass
def playerContent(self, flag, id, vipFlags):
domain = f"https://edge-hls.doppiocdn.net/hls/{id}/master/{id}_auto.m3u8?playlistType=lowLatency"
domain = f"https://edge-hls.growcdnssedge.com/hls/{id}/master/{id}_auto.m3u8?playlistType=lowLatency"
rsp = requests.get(domain, headers=self.headers).text
lines = rsp.strip().split('\n')
psch = ''
@@ -175,10 +118,9 @@ class Spider(Spider):
url_base = lines[i + 1]
# 组合最终的URL，并加上psch和pkey参数
full_url = f"{url_base}&psch={psch}&pkey={pkey}"
proxy_url = f"{self.getProxyUrl()}&url={quote(full_url)}"
# 将画质和URL添加到列表中
url.append(qn)
url.append(proxy_url)
url.append(full_url)
result = {}
result["url"] = url
result["parse"] = '0'
@@ -187,72 +129,4 @@ class Spider(Spider):
return result
def localProxy(self, param):
url = unquote(param['url'])
data = self.session.get(url, headers=self.headers, timeout=10)
if data.status_code != 200:
return [404, "text/plain", ""]
data = data.text
if "#EXT-X-MOUFLON:FILE" in data:
data = self.process_m3u8_content_v2(data)
return [200, "application/vnd.apple.mpegur", data]
def process_m3u8_content_v2(self, m3u8_content):
lines = m3u8_content.strip().split('\n')
for i, line in enumerate(lines):
if (line.startswith('#EXT-X-MOUFLON:FILE:') and 'media.mp4' in lines[i + 1]):
encrypted_data = line.split(':', 2)[2].strip()
try:
decrypted_data = self.decrypt(encrypted_data, self.stripchat_key)
except Exception as e:
decrypted_data = self.decrypt(encrypted_data, "Zokee2OhPh9kugh4")
lines[i + 1] = lines[i + 1].replace('media.mp4', decrypted_data)
return '\n'.join(lines)
def country_code_to_flag(self, country_code):
if len(country_code) != 2 or not country_code.isalpha():
return country_code
flag_emoji = ''.join([chr(ord(c.upper()) - ord('A') + 0x1F1E6) for c in country_code])
return flag_emoji
def decode_key_compact(self):
base64_str = "NTEgNzUgNjUgNjEgNmUgMzQgNjMgNjEgNjkgMzkgNjIgNmYgNGEgNjEgMzUgNjE="
decoded = base64.b64decode(base64_str).decode('utf-8')
key_bytes = bytes(int(hex_str, 16) for hex_str in decoded.split(" "))
return key_bytes.decode('utf-8')
def compute_hash(self, key: str) -> bytes:
"""计算并缓存SHA-256哈希"""
if key not in self._hash_cache:
sha256 = hashlib.sha256()
sha256.update(key.encode('utf-8'))
self._hash_cache[key] = sha256.digest()
return self._hash_cache[key]
def decrypt(self, encrypted_b64: str, key: str) -> str:
"""解密Base64编码的密文"""
# 修复Base64填充
padding = len(encrypted_b64) % 4
if padding:
encrypted_b64 += '=' * (4 - padding)
# 计算哈希并解密
hash_bytes = self.compute_hash(key)
encrypted_data = base64.b64decode(encrypted_b64)
# 异或解密
decrypted_bytes = bytearray()
for i, cipher_byte in enumerate(encrypted_data):
key_byte = hash_bytes[i % len(hash_bytes)]
decrypted_bytes.append(cipher_byte ^ key_byte)
return decrypted_bytes.decode('utf-8')
def create_session_with_retry(self, retries=3, backoff_factor=0.3):
self.session = requests.Session()
retry_strategy = Retry(
total=retries,
backoff_factor=backoff_factor,
status_forcelist=[429, 500, 502, 503, 504] # 需要重试的状态码
)
adapter = HTTPAdapter(max_retries=retry_strategy)
self.session.mount("http://", adapter)
self.session.mount("https://", adapter)
pass
FGBLH/py/腾讯视频.py
+365
View File
@@ -0,0 +1,365 @@
# coding=utf-8
# !/usr/bin/python
import json
import sys
import uuid
import copy
sys.path.append('..')
from base.spider import Spider
from pyquery import PyQuery as pq
from concurrent.futures import ThreadPoolExecutor, as_completed
class Spider(Spider):
def init(self, extend=""):
self.dbody = {
"page_params": {
"channel_id": "",
"filter_params": "sort=75",
"page_type": "channel_operation",
"page_id": "channel_list_second_page"
}
}
self.body = self.dbody
pass
def getName(self):
pass
def isVideoFormat(self, url):
pass
def manualVideoCheck(self):
pass
def destroy(self):
pass
host = 'https://v.qq.com'
apihost = 'https://pbaccess.video.qq.com'
headers = {
'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/109.0.5410.0 Safari/537.36',
'origin': host,
'referer': f'{host}/'
}
def homeContent(self, filter):
cdata = {
"电视剧": "100113",
"电影": "100173",
"综艺": "100109",
"纪录片": "100105",
"动漫": "100119",
"少儿": "100150",
"短剧": "110755"
}
result = {}
classes = []
filters = {}
for k in cdata:
classes.append({
'type_name': k,
'type_id': cdata[k]
})
with ThreadPoolExecutor(max_workers=len(classes)) as executor:
futures = [executor.submit(self.get_filter_data, item['type_id']) for item in classes]
for future in futures:
cid, data = future.result()
if not data.get('data', {}).get('module_list_datas'):
continue
filter_dict = {}
try:
items = data['data']['module_list_datas'][-1]['module_datas'][-1]['item_data_lists']['item_datas']
for item in items:
if not item.get('item_params', {}).get('index_item_key'):
continue
params = item['item_params']
filter_key = params['index_item_key']
if filter_key not in filter_dict:
filter_dict[filter_key] = {
'key': filter_key,
'name': params['index_name'],
'value': []
}
filter_dict[filter_key]['value'].append({
'n': params['option_name'],
'v': params['option_value']
})
except (IndexError, KeyError):
continue
filters[cid] = list(filter_dict.values())
result['class'] = classes
result['filters'] = filters
return result
def homeVideoContent(self):
vlist = []
data = self.gethtml(self.host)
its = data('script')
s = None
for it in its.items():
if 'window.__INITIAL_STATE__' in it.text():
s = it.text()
break
if s:
index = s.find('=')
if index != -1:
sd = json.loads(s[index + 1:])
if sd.get('storeModulesData', {}).get('channelsModulesMap', {}).get('choice', {}).get('cardListData'):
for its in sd['storeModulesData']['channelsModulesMap']['choice']['cardListData']:
if its and its.get('children_list', {}).get('list', {}).get('cards'):
for it in its['children_list']['list']['cards']:
if it and it.get('params'):
p = it['params']
tag = json.loads(p.get('uni_imgtag', '{}') or p.get('imgtag', '{}') or '{}')
id = it.get('id') or p.get('cid')
name = p.get('mz_title') or p.get('title')
if name and 'http' not in id:
vlist.append({
'vod_id': id,
'vod_name': name,
'vod_pic': p.get('image_url'),
'vod_year': tag.get('tag_2', {}).get('text'),
'vod_remarks': tag.get('tag_4', {}).get('text')
})
return {'list': vlist}
def categoryContent(self, tid, pg, filter, extend):
result = {}
params = {
"sort": extend.get('sort', '75'),
"attr": extend.get('attr', '-1'),
"itype": extend.get('itype', '-1'),
"ipay": extend.get('ipay', '-1'),
"iarea": extend.get('iarea', '-1'),
"iyear": extend.get('iyear', '-1'),
"theater": extend.get('theater', '-1'),
"award": extend.get('award', '-1'),
"recommend": extend.get('recommend', '-1')
}
if pg == '1':
self.body = self.dbody.copy()
self.body['page_params']['channel_id'] = tid
self.body['page_params']['filter_params'] = self.josn_to_params(params)
data = self.post(
f'{self.apihost}/trpc.universal_backend_service.page_server_rpc.PageServer/GetPageData?video_appid=1000005&vplatform=2&vversion_name=8.9.10&new_mark_label_enabled=1',
json=self.body, headers=self.headers).json()
ndata = data['data']
if ndata['has_next_page']:
result['pagecount'] = 9999
self.body['page_context'] = ndata['next_page_context']
else:
result['pagecount'] = int(pg)
vlist = []
for its in ndata['module_list_datas'][-1]['module_datas'][-1]['item_data_lists']['item_datas']:
id = its.get('item_params', {}).get('cid')
if id:
p = its['item_params']
tag = json.loads(p.get('uni_imgtag', '{}') or p.get('imgtag', '{}') or '{}')
name = p.get('mz_title') or p.get('title')
pic = p.get('new_pic_hz') or p.get('new_pic_vt')
vlist.append({
'vod_id': id,
'vod_name': name,
'vod_pic': pic,
'vod_year': tag.get('tag_2', {}).get('text'),
'vod_remarks': tag.get('tag_4', {}).get('text')
})
result['list'] = vlist
result['page'] = pg
result['limit'] = 90
result['total'] = 999999
return result
def detailContent(self, ids):
vbody = {
"page_params": {
"req_from": "web",
"cid": ids[0],
"vid": "",
"lid": "",
"page_type": "detail_operation",
"page_id": "detail_page_introduction"
},
"has_cache": 1
}
body = {
"page_params": {
"req_from": "web_vsite",
"page_id": "vsite_episode_list",
"page_type": "detail_operation",
"id_type": "1",
"page_size": "",
"cid": ids[0],
"vid": "",
"lid": "",
"page_num": "",
"page_context": "",
"detail_page_type": "1"
},
"has_cache": 1
}
with ThreadPoolExecutor(max_workers=2) as executor:
future_detail = executor.submit(self.get_vdata, vbody)
future_episodes = executor.submit(self.get_vdata, body)
vdata = future_detail.result()
data = future_episodes.result()
pdata = self.process_tabs(data, body, ids)
if not pdata:
return self.handle_exception(None, "No pdata available")
try:
star_list = vdata['data']['module_list_datas'][0]['module_datas'][0]['item_data_lists']['item_datas'][
0].get('sub_items', {}).get('star_list', {}).get('item_datas', [])
actors = [star['item_params']['name'] for star in star_list]
names = ['腾讯视频', '预告片']
plist, ylist = self.process_pdata(pdata, ids)
if not plist:
del names[0]
if not ylist:
del names[1]
vod = self.build_vod(vdata, actors, plist, ylist, names)
return {'list': [vod]}
except Exception as e:
return self.handle_exception(e, "Error processing detail")
def searchContent(self, key, quick, pg="1"):
body = {"version": "24072901", "clientType": 1, "filterValue": "", "uuid": str(uuid.uuid4()), "retry": 0,
"query": key, "pagenum": int(pg) - 1, "pagesize": 30, "queryFrom": 0, "searchDatakey": "",
"transInfo": "", "isneedQc": True, "preQid": "", "adClientInfo": "",
"extraInfo": {"isNewMarkLabel": "1", "multi_terminal_pc": "1"}}
data = self.post(f'{self.apihost}/trpc.videosearch.mobile_search.MultiTerminalSearch/MbSearch?vplatform=2',
json=body, headers=self.headers).json()
vlist = []
for k in data['data']['areaBoxList'][-1]['itemList']:
if k.get('doc', {}).get('id'):
img_tag = k.get('videoInfo', {}).get('imgTag')
if img_tag is not None and isinstance(img_tag, str):
try:
tag = json.loads(img_tag)
except json.JSONDecodeError as e:
tag = {}
else:
tag = {}
pic = k.get('videoInfo', {}).get('imgUrl')
vlist.append({
'vod_id': k['doc']['id'],
'vod_name': k['videoInfo']['title'],
'vod_pic': pic,
'vod_year': tag.get('tag_2', {}).get('text', ''),
'vod_remarks': tag.get('tag_4', {}).get('text', '')
})
return {'list': vlist, 'page': pg}
def playerContent(self, flag, id, vipFlags):
ids = id.split('@')
url = f"{self.host}/x/cover/{ids[0]}/{ids[1]}.html"
parse_url = f"https://jx.xmflv.com/?url={url}"
return {'parse': 1, 'url': parse_url, 'header': ''}
def localProxy(self, param):
pass
def gethtml(self, url):
rsp = self.fetch(url, headers=self.headers)
rsp = self.cleanText(rsp.text)
return pq(rsp)
def get_filter_data(self, cid):
hbody = self.dbody.copy()
hbody['page_params']['channel_id'] = cid
data = self.post(
f'{self.apihost}/trpc.universal_backend_service.page_server_rpc.PageServer/GetPageData?video_appid=1000005&vplatform=2&vversion_name=8.9.10&new_mark_label_enabled=1',
json=hbody, headers=self.headers).json()
return cid, data
def get_vdata(self, body):
try:
vdata = self.post(
f'{self.apihost}/trpc.universal_backend_service.page_server_rpc.PageServer/GetPageData?video_appid=3000010&vplatform=2&vversion_name=8.2.96',
json=body, headers=self.headers
).json()
# print(body)
return vdata
except Exception as e:
print(f"Error in get_vdata: {str(e)}")
return {'data': {'module_list_datas': []}}
def process_pdata(self, pdata, ids):
plist = []
ylist = []
for k in pdata:
if k.get('item_id'):
pid = f"{k['item_params']['union_title']}${ids[0]}@{k['item_id']}"
if '预告' in k['item_params']['union_title']:
ylist.append(pid)
else:
plist.append(pid)
return plist, ylist
def build_vod(self, vdata, actors, plist, ylist, names):
d = vdata['data']['module_list_datas'][0]['module_datas'][0]['item_data_lists']['item_datas'][0]['item_params']
urls = []
if plist:
urls.append('#'.join(plist))
if ylist:
urls.append('#'.join(ylist))
vod = {
'type_name': d.get('sub_genre', ''),
'vod_name': d.get('title', ''),
'vod_year': d.get('year', ''),
'vod_area': d.get('area_name', ''),
'vod_remarks': d.get('holly_online_time', '') or d.get('hotval', ''),
'vod_actor': ','.join(actors),
'vod_content': d.get('cover_description', ''),
'vod_play_from': '$$$'.join(names),
'vod_play_url': '$$$'.join(urls)
}
return vod
def handle_exception(self, e, message):
print(f"{message}: {str(e)}")
return {'list': [{'vod_play_from': '哎呀翻车啦', 'vod_play_url': '翻车啦#555'}]}
def process_tabs(self, data, body, ids):
try:
pdata = data['data']['module_list_datas'][-1]['module_datas'][-1]['item_data_lists']['item_datas']
tabs = data['data']['module_list_datas'][-1]['module_datas'][-1]['module_params'].get('tabs')
if tabs and len(json.loads(tabs)):
tabs = json.loads(tabs)
remaining_tabs = tabs[1:]
task_queue = []
for tab in remaining_tabs:
nbody = copy.deepcopy(body)
nbody['page_params']['page_context'] = tab['page_context']
task_queue.append(nbody)
with ThreadPoolExecutor(max_workers=10) as executor:
future_map = {executor.submit(self.get_vdata, task): idx for idx, task in enumerate(task_queue)}
results = [None] * len(task_queue)
for future in as_completed(future_map.keys()):
idx = future_map[future]
results[idx] = future.result()
for result in results:
if result:
page_data = result['data']['module_list_datas'][-1]['module_datas'][-1]['item_data_lists'][
'item_datas']
pdata.extend(page_data)
return pdata
except Exception as e:
print(f"Error processing episodes: {str(e)}")
return []
def josn_to_params(self, params, skip_empty=False):
query = []
for k, v in params.items():
if skip_empty and not v:
continue
query.append(f"{k}={v}")
return "&".join(query)