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-05 04:58:38 +00:00
parent 286d42c940
commit a45a9bde55
651 changed files with 337816 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tbapi2/py/app/xiaocao.py
+366
View File
@@ -0,0 +1,366 @@
# -*- coding: utf-8 -*-
# 本资源来源于互联网公开渠道，仅可用于个人学习爬虫技术。
# 严禁将其用于任何商业用途，下载后请于 24 小时内删除，搜索结果均来自源站，本人不承担任何责任。
from base.spider import Spider
from Crypto.Cipher import AES
from urllib.parse import quote, unquote, urljoin, urlparse
import re, sys, time, json, random, base64, hashlib, urllib3
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
sys.path.append('..')
class Spider(Spider):
headers,play_header,host,play_domain,encrypt_domain,proxyurl = {
'User-Agent': "okhttp/4.9.0",
'Connection': "Keep-Alive",
'Accept-Encoding': "gzip",
'log-header': "I am the log request header.",
'app_id': "",
'channel_code': "",
'cur_time': "",
'device_id': "",
'mob_mfr': "huawei",
'mobmodel': "DCO-AL00",
'package_name': "",
'sign': "",
'sys_platform': "2",
'sysrelease': "12",
'token': "",
'version': "40000"
},{'User-Agent': 'Mozi'},'','','',''
def init(self, extend=''):
try:
ext = json.loads(extend.strip())
host = ext.get('host').rstrip('/')
if not re.match(r'^https?://[a-zA-Z0-9-]+(\.[a-zA-Z0-9-]+)*(:\d+)?/?$', host): return None
appid = ext.get('app_id').strip()
deviceid = ext.get('deviceid').strip()
version_code = ext.get('versionCode').strip()
channel = ext.get('UMENG_CHANNEL').strip()
package_name = ext.get('package_name').strip()
self.encrypt_domain = ext.get('EncryptDomain').strip()
if not(appid and deviceid and version_code and channel and package_name and self.encrypt_domain): return None
if not self.is_valid_android_id(deviceid): return None
self.headers['app_id'] = appid
self.headers['channel_code'] = channel
self.headers['device_id'] = deviceid
self.headers['version'] = version_code
self.headers['package_name'] = package_name
try:
self.proxyurl = f'{self.getProxyUrl(True)}&type=xiaocao'
except Exception:
self.proxyurl = 'http://127.0.0.1:9978/proxy?do=py&type=xiaocao'
self.login(host)
except Exception:
return
def homeContent(self, filter):
if not self.host: return None
headers = self.headers.copy()
headers['cur_time'] = self.timestamp()
headers['sign'] = self.sign(headers['cur_time'])
response = self.post(f'{self.host}/api/type/get_list', headers=headers, verify=False).text
data = json.loads(self.decrypt(response))
classes = [{'type_id': i['id'], 'type_name': i['name']} for i in data.get('result', [])]
if not any(i['type_id'] == 31 for i in classes):
classes.append({'type_id': 31, 'type_name': '短剧'})
return {'class': classes}
def homeVideoContent(self):
if not self.host: return None
headers = self.headers.copy()
headers['cur_time'] = self.timestamp()
headers['sign'] = self.sign(headers['cur_time'])
try:
response = self.post(f'{self.host}/api/channel/get_list', headers=headers,verify=False).text
data = self.decrypt(response)
result = json.loads(data)['result']
classes, videos, recommend_id = [], [], ''
for i in result:
if isinstance(i, dict):
if i.get('channel_name') == '推荐':
recommend_id = i.get('id')
break
headers['cur_time'] = self.timestamp()
headers['sign'] = self.sign(headers['cur_time'])
except Exception:
recommend_id = '324'
payload2 = {
'psize': "100",
'channel_id': recommend_id,
'pn': "1"
}
response2 = self.post(f'{self.host}/api/channel/get_info',data=payload2, headers=headers, verify=False).text
data2 = self.decrypt(response2)
result2 = json.loads(data2)['result']
for item in result2:
if not isinstance(item, dict):
continue
for block in item.get('block_list', []):
if not isinstance(block, dict):
continue
for vod in block.get('vod_list', []):
if vod['type_pid'] == 1:
remark = f"评分：{vod['vod_douban_score']}"
elif vod['type_pid'] != 2 and vod['type_pid'] != 4 and vod['type_pid'] != 31:
remark = vod['collection_new_title']
if vod['type_pid'] == 3:
remark = self.remove_invalid_episode(remark)
else:
remark = f"{vod['vod_serial']}集全" if vod.get('vod_isend') == 1 else f"更新至{vod['vod_serial']}集"
videos.append({
'vod_id': vod['id'],
'vod_name': vod['vod_name'],
'vod_pic': vod['vod_pic'],
'vod_remarks': remark,
'vod_year': vod['vod_year']
})
return {'list': videos}
def categoryContent(self, tid, pg, filter, extend):
if not self.host: return None
headers = self.headers.copy()
headers['cur_time'] = self.timestamp()
headers['sign'] = self.sign(headers['cur_time'])
payload = {
'area': "",
'year': "",
'type_id': tid,
'sort': "",
'type': "",
'pn': pg
}
response = self.post(f'{self.host}/api/search/screen', data=payload, headers=headers, verify=False).text
data = json.loads(self.decrypt(response))
videos = []
for i in data.get('result',[]):
if i['type_pid'] == 1:
remark = f"评分：{i['vod_douban_score']}"
elif i['type_pid'] != 2 and i['type_pid'] != 4 and i['type_pid'] != 31:
remark = i['collection_new_title']
if i['type_pid'] == 3:
remark = self.remove_invalid_episode(remark)
else:
remark = f"{i['vod_serial']}集全" if i.get('vod_isend') == 1 else f"更新至{i['vod_serial']}集"
videos.append({
'vod_id': i['id'],
'vod_name': i['vod_name'],
'vod_pic': i['vod_pic'],
'vod_remarks': remark,
'vod_year': i['vod_year']
})
return {'list': videos, 'page': pg}
def remove_invalid_episode(self, content):
pattern = r'第(.*?)集'
def replace_match(match):
middle_content = match.group(1)
return middle_content
processed_content = re.sub(pattern, replace_match, content)
return processed_content
def searchContent(self, key, quick, pg='1'):
if not self.host: return None
headers = self.headers.copy()
headers['cur_time'] = self.timestamp()
headers['sign'] = self.sign(headers['cur_time'])
payload = {'kw': key,'pn': pg}
response = self.post(f'{self.host}/api/search/result', data=payload, headers=headers, verify=False).text
data = json.loads(self.decrypt(response))
videos = []
for i in data.get('result',[]):
if i['type_pid'] == 1:
remark = f"评分：{i['vod_douban_score']}"
elif i['type_pid'] != 2 and i['type_pid'] != 4 and i['type_pid'] != 31:
remark = i['collection_new_title']
if i['type_pid'] == 3:
remark = self.remove_invalid_episode(remark)
else:
vod_serial = i['vod_serial']
remark = f"{i['vod_tag']},短剧" if i['type_pid'] == 31 else f"{vod_serial}集全" if i.get('vod_isend') == 1 else f"更新至{vod_serial}集"
videos.append({
'vod_id': i['id'],
'vod_name': i['vod_name'],
'vod_pic': i['vod_pic'],
'vod_remarks': remark,
'vod_year': i['vod_year']
})
return {'list': videos, 'page': pg}
def detailContent(self, ids):
if not self.host: return None
headers = self.headers.copy()
headers['cur_time'] = self.timestamp()
headers['sign'] = self.sign(headers['cur_time'])
payload = {
'sig': '',
'nc_token': '',
'code': '',
'phone': '',
'vod_id': ids[0],
'session_id': ''
}
response = self.post(f'{self.host}/api/video/result', data=payload, headers=headers, verify=False).text
data = self.decrypt(response)
result = json.loads(data)['result']
play_urls = []
for i in result['vod_collection']:
if isinstance(i, dict):
play_urls.append(f"{i['title']}${i['vod_id']}@{i['id']}@{i['cur_time']}@{i['vod_token']}")
video = {
'vod_id': result['id'],
'vod_name': result['vod_name'],
'vod_pic': result['vod_pic'],
'vod_remarks': result['remarks'],
'vod_year': result['vod_year'],
'vod_area': result['vod_area'],
'vod_actor': result['vod_actor'],
'vod_director': result['vod_director'],
'vod_content': result['vod_blurb'],
'vod_play_from': '小草',
'vod_play_url': '#'.join(play_urls),
'type_name': result['vod_tag']
}
return {'list': [video]}
def playerContent(self, flag, id, vipflags):
vod_id, collection_id, cur_time, vod_token = id.split('@', 3)
headers = self.headers.copy()
headers['cur_time'] = self.timestamp()
headers['sign'] = self.sign(headers['cur_time'])
payload = {
'collection_id': collection_id,
'sig': "",
'nc_token': "",
'code': "",
'phone': "",
'vod_id': vod_id,
'session_id': "",
'vod_token': vod_token,
'cur_time': cur_time
}
response = self.post(f'{self.host}/api/video/collection', data=payload, headers=headers, verify=False).text
data = self.decrypt(response)
result = json.loads(data)['result']
try:
ck = base64.b64decode(result['ck']).decode('utf-8')
except Exception:
ck = result['ck']
vod_url = result['vod_url']
check_url = result.get('check_page_url')
url = check_url or self.proxyurl + '&url=' + quote(f"{vod_url}?{ck}", safe='')
return {'jx': '0', 'parse': '0', 'url': url, 'header': self.play_header}
def localProxy(self, params):
if params['type'] == "xiaocao":
return self.xiaocao_m3u8_proxy(params)
return None
def login(self, host):
if self.headers['token'] and self.host: return
headers = self.headers.copy()
headers['cur_time'] = self.timestamp()
headers['sign'] = self.sign(headers['cur_time'])
payload = {
'invited_by': '',
'ua': 'Mozilla/5.0 (Linux; Android 12; 23116PN5BC Build/V417IR; wv) AppleWebKit/537.36 (KHTML, like Gecko) Version/4.0 Chrome/101.0.4951.61 Safari/537.36',
'is_install': '0'
}
response = self.post(f'{host}/api/public/init', data=payload, headers=headers, verify=False).text
data = self.decrypt(response)
result = json.loads(data)['result']
self.headers['token'] = result['user_info']['token']
self.host = result['sys_conf']['api_url'].rstrip('/')
if not (self.headers['token'] or self.play_domain or self.play_domain):
self.host = ''
def sign(self, cur_time):
return self.md5(f"zD9[bM4~sF4~uY2){self.headers['device_id']}{cur_time}").upper()
def md5(self, str):
md5_hash = hashlib.md5()
md5_hash.update(str.encode('utf-8'))
return md5_hash.hexdigest()
def xiaocao_m3u8_proxy(self, params):
url = unquote(params['url'])
if params.get('format') == "ts":
data = {"Location": url + self.hls_sign(url), "Content-Length": "0"}
return [302, "text/html; charset=utf-8", None, data]
else:
data = self.xiaocao_modify_m3u8(url)
return [200, "application/vnd.apple.mpegurl", data]
def xiaocao_modify_m3u8(self, raw_url):
ck = raw_url.split('?')[1]
m3u8_url = raw_url + self.hls_sign(raw_url)
m3u8_content = self.fetch(m3u8_url, headers=self.play_header, verify=False).text
content = m3u8_content if m3u8_content is not None else self.m3u8_content
if content is None: raise ValueError("M3U8为空")
parsed = urlparse(raw_url)
base = f"{parsed.scheme}://{parsed.netloc}{parsed.path.rsplit('/', 1)[0]}/"
output_lines = []
for line in content.splitlines():
stripped_line = line.strip()
if stripped_line and not stripped_line.startswith('#'):
if not stripped_line.startswith(('http://', 'https://')):
full_url = urljoin(base, stripped_line)
else:
full_url = stripped_line
signed_url = self.proxyurl + f'&format=ts&url=' + quote(full_url + '&' + ck,safe='')
output_lines.append(signed_url)
else:
output_lines.append(line)
return '\n'.join(output_lines)
def decrypt(self, base64_ciphertext):
try:
ciphertext = base64.b64decode(base64_ciphertext)
cipher = AES.new('aZ9$kU5%qI7=yC2=zH2#gM0@pX7^wF3a'.encode('utf-8'), AES.MODE_CBC, 'hY2&tN3]kF7,dL7='.encode('utf-8'))
plaintext_bytes = cipher.decrypt(ciphertext)
padding_len = plaintext_bytes[-1]
plaintext_bytes = plaintext_bytes[:-padding_len]
return plaintext_bytes.decode('utf-8')
except Exception as e:
raise Exception(f"解密失败: {str(e)}")
def get_url_path(self, url):
parsed_url = urlparse(url)
return parsed_url.path
def hls_sign(self, url):
hex_time = self.hex_time()
if '?' in url: url = url.split('?')[0]
data = self.encrypt_domain + self.get_url_path(url) + hex_time
return f"&wsSecret={self.md5(data)}&wsTime={hex_time}"
def is_valid_android_id(self, android_id):
if not isinstance(android_id, str):
return False
pattern = r'^[0-9a-f]{16}$'
return bool(re.fullmatch(pattern, android_id))
def random_device_id(self):
chars = '0123456789abcdef'
android_id = ''.join(random.choice(chars) for _ in range(16))
return android_id
def timestamp(self):
return str(int(time.time() * 1000))
def hex_time(self):
return hex(int(time.time()))[2:]
def getName(self):
pass
def isVideoFormat(self, url):
pass
def manualVideoCheck(self):
pass
def destroy(self):
pass