mirror of
https://github.com/qist/tvbox.git
synced 2026-06-21 00:53:15 +00:00
255 lines
9.7 KiB
Python
255 lines
9.7 KiB
Python
# -*- coding: utf-8 -*-
|
|
# by @Qist
|
|
"""
|
|
ITalkBB TV - 海外华人影视
|
|
"""
|
|
import time
|
|
import uuid
|
|
import requests
|
|
from base.spider import Spider
|
|
|
|
|
|
class Spider(Spider):
|
|
def getName(self):
|
|
return 'ITalkBB TV'
|
|
|
|
def init(self, extend=""):
|
|
self._login()
|
|
|
|
def isVideoFormat(self, url):
|
|
pass
|
|
|
|
def manualVideoCheck(self):
|
|
pass
|
|
|
|
def destroy(self):
|
|
pass
|
|
|
|
def __init__(self):
|
|
self.name = 'ITalkBB TV'
|
|
self.host = 'https://www.italkbbtv.com'
|
|
self.api = 'https://api.italkbbtv.com/classictv'
|
|
self.token = ''
|
|
self.header = {
|
|
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/136.0.0.0 Safari/537.36 Edg/136.0.0.0',
|
|
'Referer': 'https://www.italkbbtv.com/',
|
|
'Origin': 'https://www.italkbbtv.com',
|
|
}
|
|
self.timeout = 20
|
|
self._token = ''
|
|
self._expire = 0
|
|
self._device_id = str(uuid.uuid4())
|
|
|
|
def _login(self):
|
|
now = int(time.time() * 1000)
|
|
if self._token and now < self._expire - 60000:
|
|
return
|
|
data = {
|
|
'login_type': 'password', 'grant_type': 'tv_login_in',
|
|
'device_id': self._device_id,
|
|
'login_name': f'{self._device_id}@web.visitor.italkbb.com',
|
|
'password': 'visitor_secret',
|
|
'device_type': 'web', 'device_model': 'Chrome148', 'app_version': '305013',
|
|
}
|
|
try:
|
|
r = requests.post('https://api.italkbbtv.com/auth/v1/token', headers=self.header, data=data, timeout=self.timeout)
|
|
if r.status_code == 200:
|
|
self._set_token(r.json().get('access', {}))
|
|
except:
|
|
pass
|
|
|
|
def _set_token(self, acc):
|
|
self._token = acc.get('token', '')
|
|
self._expire = acc.get('expire_time', 0)
|
|
if self._token:
|
|
self.header['Authorization'] = 'Bearer ' + self._token
|
|
|
|
self.cats = {
|
|
'drama': ('62ac4df64beefe53586474ff', '62c670dc1dca2d424404499c'),
|
|
'movie': ('62ac4e644beefe535864785c', '62ac4ef36e0b5a13ed291544'),
|
|
'variety': ('62ac4e1f4beefe5358647642', '62ce7417c7daaa4a5d3fea14'),
|
|
'cartoon': ('62ac4e6a4beefe53586478a5', '62ac4e6e4beefe53586478ca'),
|
|
'shorts': ('66a9e3e49f7e8378f2152312', '66b1d25cf2dde82c215f9b59'),
|
|
}
|
|
self.class_names = '电视剧&电影&综艺&动画&短剧&直播频道'.split('&')
|
|
self.class_urls = 'drama&movie&variety&cartoon&shorts&live'.split('&')
|
|
|
|
def _api_get(self, path, params=None):
|
|
self._login()
|
|
try:
|
|
r = requests.get(self.api + path, headers=self.header, params=params, timeout=self.timeout)
|
|
if r.status_code == 200:
|
|
return r.json()
|
|
except:
|
|
pass
|
|
return None
|
|
|
|
def _get_live_stream(self, ch_id):
|
|
self._login()
|
|
try:
|
|
r = requests.get('https://api.italkbbtv.com/playauth/v1/live', headers=self.header, params={'series_id': ch_id, 'hl': 'zh_CN'}, timeout=self.timeout)
|
|
if r.status_code == 200:
|
|
return r.json().get('manifest', '')
|
|
except:
|
|
pass
|
|
return ''
|
|
|
|
def _make_vod(self, s):
|
|
sid = s.get('id', '')
|
|
name = s.get('name', '')
|
|
images = s.get('images', {}) or {}
|
|
poster = (images.get('poster') or [''])[0] if images.get('poster') else ''
|
|
landscape = (images.get('landscape') or [''])[0] if images.get('landscape') else ''
|
|
pic = poster or landscape
|
|
ep_count = s.get('episode_count', 0) or 0
|
|
latest = s.get('latest_episode_shortname', '') or s.get('latest_episode_name', '')
|
|
if ep_count and ep_count > 1:
|
|
remarks = f'更新至{latest}' if latest else f'{ep_count}集'
|
|
elif ep_count == 1:
|
|
remarks = '全1集'
|
|
else:
|
|
remarks = latest
|
|
return {'vod_id': sid, 'vod_name': name, 'vod_pic': pic, 'vod_remarks': remarks}
|
|
|
|
def homeContent(self, filter):
|
|
result = {'class': [], 'list': []}
|
|
for name, cid in zip(self.class_names, self.class_urls):
|
|
result['class'].append({'type_name': name, 'type_id': cid})
|
|
data = self._api_get('/vod/v1/series', {'root_id': self.cats['drama'][0], 'category_id': self.cats['drama'][1], 'page': 1, 'size': 24, 'hl': 'zh_CN'})
|
|
if data:
|
|
result['list'] = [self._make_vod(s) for s in data.get('series', [])]
|
|
return result
|
|
|
|
def homeVideoContent(self):
|
|
return {}
|
|
|
|
def categoryContent(self, tid, pg, filter, extend):
|
|
pg = int(pg)
|
|
result = {'list': [], 'page': pg, 'pagecount': 999, 'limit': 24, 'total': 0}
|
|
|
|
# 直播频道
|
|
if tid == 'live':
|
|
data = self._api_get('/live/v1/lives', {'root_id': '62ac4e2e4beefe535864769d', 'category_id': '62ac4e314beefe53586476c2', 'page': 1, 'size': 100, 'hl': 'zh_CN'})
|
|
if data:
|
|
vods = []
|
|
for ch in data.get('lives', []):
|
|
ch_id = ch.get('id', '')
|
|
ch_name = ch.get('name', '')
|
|
images = ch.get('images', {}) or {}
|
|
icon = (images.get('icon') or [''])[0] if images.get('icon') else ''
|
|
vods.append({'vod_id': 'live@' + ch_id, 'vod_name': ch_name, 'vod_pic': icon, 'vod_remarks': '直播'})
|
|
result['list'] = vods
|
|
result['total'] = len(vods)
|
|
result['pagecount'] = 1
|
|
return result
|
|
|
|
# 点播
|
|
cat = self.cats.get(tid)
|
|
if not cat:
|
|
return result
|
|
root_id, cat_id = cat
|
|
data = self._api_get('/vod/v1/series', {'root_id': root_id, 'category_id': cat_id, 'page': pg, 'size': 24, 'hl': 'zh_CN'})
|
|
if data:
|
|
result['list'] = [self._make_vod(s) for s in data.get('series', [])]
|
|
result['total'] = data.get('total', 0)
|
|
result['pagecount'] = (result['total'] + 23) // 24
|
|
return result
|
|
|
|
def detailContent(self, ids):
|
|
if not ids or not ids[0]:
|
|
return {'list': []}
|
|
vid = ids[0]
|
|
|
|
# 直播频道
|
|
if vid.startswith('live@'):
|
|
ch_id = vid.replace('live@', '')
|
|
stream = self._get_live_stream(ch_id)
|
|
return {'list': [{
|
|
'vod_id': vid, 'vod_name': ch_id, 'vod_pic': '',
|
|
'vod_play_from': 'ITalkBB直播',
|
|
'vod_play_url': f'直播${stream}' if stream else '',
|
|
}]}
|
|
|
|
# 点播
|
|
series_data = self._api_get(f'/vod/v1/series/{vid}', {'hl': 'zh_CN'})
|
|
s = (series_data or {}).get('series', {})
|
|
name = s.get('name', '')
|
|
desc = s.get('description', '')
|
|
images = s.get('images', {}) or {}
|
|
pic = (images.get('poster') or [''])[0] if images.get('poster') else ''
|
|
stars = s.get('stars', {}) or {}
|
|
actor = '/'.join([a.get('name', '') for a in (stars.get('actor') or [])[:5]])
|
|
director = '/'.join([d.get('name', '') for d in (stars.get('director') or [])])
|
|
|
|
eps_data = self._api_get(f'/vod/v1/series/{vid}/episodes', {'hl': 'zh_CN'})
|
|
eps = eps_data if isinstance(eps_data, list) else (eps_data or {}).get('episodes', [])
|
|
|
|
play_urls = []
|
|
for ep in eps:
|
|
ep_name = ep.get('shortname', '') or ep.get('name', '') or ep.get('id', '')[-4:]
|
|
ep_id = ep.get('id', '')
|
|
play_urls.append(f'{ep_name}$play@{vid}@{ep_id}')
|
|
|
|
if not play_urls:
|
|
play_urls.append(f'播放$play@{vid}@')
|
|
|
|
return {'list': [{
|
|
'vod_id': vid, 'vod_name': name, 'vod_pic': pic,
|
|
'vod_remarks': '', 'vod_year': '', 'type_name': '',
|
|
'vod_content': desc, 'vod_actor': actor, 'vod_director': director,
|
|
'vod_play_from': 'ITalkBB', 'vod_play_url': '#'.join(play_urls),
|
|
}]}
|
|
|
|
def searchContent(self, key, quick, pg="1"):
|
|
result = {'list': []}
|
|
for tid in self.cats:
|
|
cat = self.cats[tid]
|
|
data = self._api_get('/vod/v1/series', {'root_id': cat[0], 'category_id': cat[1], 'page': 1, 'size': 50, 'hl': 'zh_CN'})
|
|
if data:
|
|
for s in data.get('series', []):
|
|
if key in s.get('name', ''):
|
|
result['list'].append(self._make_vod(s))
|
|
seen = set()
|
|
unique = []
|
|
for v in result['list']:
|
|
if v['vod_id'] not in seen:
|
|
seen.add(v['vod_id'])
|
|
unique.append(v)
|
|
result['list'] = unique
|
|
return result
|
|
|
|
def playerContent(self, flag, id, vipFlags):
|
|
h = {'User-Agent': self.header['User-Agent']}
|
|
# 直播: 直接用m3u8地址
|
|
if id.startswith('live@'):
|
|
m3u8 = id.replace('live@', '')
|
|
if m3u8.startswith('http'):
|
|
return {'parse': 0, 'url': m3u8, 'header': h, 'playUrl': ''}
|
|
stream = self._get_live_stream(m3u8)
|
|
return {'parse': 0, 'url': stream, 'header': h, 'playUrl': ''}
|
|
# m3u8直链
|
|
if id.startswith('http'):
|
|
return {'parse': 0, 'url': id, 'header': h, 'playUrl': ''}
|
|
# 点播
|
|
parts = id.split('@')
|
|
route = parts[0] if len(parts) > 1 else 'play'
|
|
sid = parts[1] if len(parts) > 1 else ''
|
|
eid = parts[2] if len(parts) > 2 else ''
|
|
url = f'{self.host}/{route}/{sid}'
|
|
if eid:
|
|
url += f'?ep={eid}'
|
|
return {'parse': 1, 'url': url, 'header': self.header, 'playUrl': ''}
|
|
|
|
def fetch(self, url):
|
|
try:
|
|
resp = requests.get(url, headers=self.header, timeout=self.timeout)
|
|
resp.encoding = 'utf-8'
|
|
if resp.status_code == 200:
|
|
return resp.text
|
|
return None
|
|
except:
|
|
return None
|
|
|
|
def localProxy(self, param):
|
|
return None
|