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 07:03:13 +00:00
parent 0ad060ce78
commit 63edf83ad4
13 changed files with 2180 additions and 1964 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tvbox/py/枫叶影院.py
+54 -1
View File
@@ -1,20 +1,73 @@
# -*- coding: utf-8 -*-
import re, urllib.parse
import json
import time
from bs4 import BeautifulSoup
import requests
from base.spider import Spider as BaseSpider
class Spider(BaseSpider):
# 缓存变量，避免频繁请求发布页
_cache_host = ""
_cache_time = 0
CACHE_DURATION = 300 # 缓存5分钟
def init(self, extend=""):
self.host = "https://www.ht10010.com"
self.headers = {
"User-Agent": "Mozilla/5.0 (iPhone; CPU iPhone OS 16_0 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/16.0 Mobile/15E148 Safari/604.1",
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
"Accept-Language": "zh-CN,zh;q=0.9",
}
# 优先读取缓存，没有或过期则重新获取
now = time.time()
if self._cache_host and now - self._cache_time < self.CACHE_DURATION:
self.host = self._cache_host
else:
publish_url = "https://www.vip1949.com/"
self.host = self.get_online_host(publish_url)
self._cache_host = self.host
self._cache_time = now
def get_online_host(self, publish_url):
"""从发布页解析JS域名列表，返回第一个在线可用主站"""
try:
resp = requests.get(publish_url, headers=self.headers, timeout=15)
resp.raise_for_status()
html = resp.text
# 正则提取JS中的domains数组
pattern = r'const domains = (\[.*?\]);'
match = re.search(pattern, html, re.S)
if not match:
return "https://www.cd-zj.com"
js_str = match.group(1)
# 修复JS对象格式为标准JSON
js_str = re.sub(r'(\w+)\s*:', r'"\1":', js_str)
js_str = re.sub(r'"+', '"', js_str)
domains = json.loads(js_str)
# 逐个检测连通性，返回第一个可用
for item in domains:
url = item["url"]
if self.check_url_online(url):
return url.rstrip("/")
return "https://www.cd-zj.com"
except Exception as e:
print(f"获取发布页异常: {e}")
return "https://www.cd-zj.com"
def check_url_online(self, url, timeout=3):
"""检测网址是否可正常访问"""
try:
r = requests.head(url, headers=self.headers, timeout=timeout, allow_redirects=True)
return 200 <= r.status_code < 400
except:
return False
def getName(self):
return '枫叶影院'
tvbox/py/红果短剧.py
+247 -103
View File
@@ -1,17 +1,21 @@
# -*- coding: utf-8 -*-
# by @嗷呜
import re
import sys
from pyquery import PyQuery as pq
sys.path.append('..')
from base.spider import Spider
import requests
import re
import urllib.parse
host = "https://mov.cenguigui.cn"
base_url = host + "/duanju/api.php"
quality_host = "https://mov.cenguigui.cn"
headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/132.0.0.0 Safari/537.36'
}
timeout = 10
class Spider(Spider):
def init(self, extend=""):
pass
def getName(self):
return "小心儿悠悠"
def init(self, extend):
pass
def isVideoFormat(self, url):
@@ -20,108 +24,248 @@ class Spider(Spider):
def manualVideoCheck(self):
pass
def destroy(self):
pass
host='https://www.hongguodj.cc'
headers = {
'Accept': '*/*',
'Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.8',
'Cache-Control': 'no-cache',
'Connection': 'keep-alive',
'DNT': '1',
'Origin': host,
'Pragma': 'no-cache',
'Sec-Fetch-Dest': 'empty',
'Sec-Fetch-Mode': 'cors',
'Sec-Fetch-Site': 'cross-site',
'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/134.0.0.0 Safari/537.36',
'sec-ch-ua': '"Not/A)Brand";v="8", "Chromium";v="134", "Google Chrome";v="134"',
'sec-ch-ua-mobile': '?0',
'sec-ch-ua-platform': '"macOS"',
}
def homeContent(self, filter):
result = {}
class_names = "推荐榜&热播榜&新剧榜&漫剧榜大唐&大秦&大明&擦边&逆袭&霸总&豪门恩怨&神豪&都市日常&大女主&都市修仙&强者回归&重生&闪婚&赘婿逆袭&追妻&萌宝&奇幻脑洞&传承觉醒&奇幻爱情&乡村&历史古代&王妃&娱乐圈&暗恋成真&系统&真假千金&穿书&女帝&团宠&年代爱情&玄幻仙侠&皇后&逆袭&霸总&现代言情&打脸虐渣&豪门恩怨&神豪&马甲&都市日常&战神归来&小人物&女性成长&大女主&穿越&都市修仙&强者回归&亲情&古装&重生&闪婚&赘婿逆袭&虐恋&追妻&天下无敌&家庭伦理&萌宝&古风权谋&职场&奇幻脑洞&异能&无敌神医&古风言情&传承觉醒&现言甜宠&奇幻爱情&乡村&历史古代&王妃&高手下山&娱乐圈&强强联合&破镜重圆&暗恋成真&民国&欢喜冤家&系统&真假千金&龙王&校园&穿书&女帝&团宠&年代爱情&玄幻仙侠&青梅竹马&悬疑推理&皇后&替身&大叔&喜剧&剧情"
class_list = class_names.split('&')
classes = []
vlist = []
data = pq(self.fetch(self.host, headers=self.headers).text)
for i in list(data('.slip li').items())[1:]:
for class_name in class_list:
classes.append({
'type_name': i.text(),
'type_id': re.findall(r'\d+', i('a').attr('href'))[0]
"type_id": class_name,
"type_name": class_name
})
for i in data('.wrap .rows').items():
vlist.extend(self.getlist(i('li')))
result['class'] = classes
result['list'] = vlist
return result
return {"class": classes}
def homeVideoContent(self):
pass
return {'list': []}
def categoryContent(self, tid, pg, filter, extend):
data=pq(self.fetch(f'{self.host}/type/{tid}-{pg}.html', headers=self.headers).text)
result = {}
result['list'] = self.getlist(data('.list ul li'))
result['page'] = pg
result['pagecount'] = 9999
result['limit'] = 90
result['total'] = 999999
return result
def categoryContent(self, cid, pg, filter, ext):
videos = []
page = int(pg) if pg else 1
params = f"page={page}&name={urllib.parse.quote(cid)}"
tab_type = "19"
if ext and 'tab_type' in ext:
tab_type = ext['tab_type']
params += f"&tab_type={tab_type}"
url = f"{base_url}?{params}"
try:
response = requests.get(url=url, headers=headers, timeout=timeout)
if response.status_code != 200:
return {'list': []}
response.encoding = "utf-8"
data = response.json()
if data.get('code') == 200 and data.get('data'):
for vod in data['data']:
vod_id = f"book_id={vod.get('book_id', '')}&actor={vod.get('author', '')}&type={vod.get('type', '')}"
videos.append({
"vod_id": vod_id,
"vod_name": vod.get('title', ''),
"vod_pic": vod.get('cover', ''),
"vod_remarks": vod.get('type', ''),
"vod_content": vod.get('intro', '')
})
except Exception:
return {'list': []}
return {
'list': videos,
'page': pg,
'pagecount': 9999,
'limit': 20,
'total': 999999
}
def detailContent(self, ids):
data=pq(self.fetch(f'{self.host}{ids[0]}', headers=self.headers).text)
v=data('.info')
p=v('p')
vod = {
'vod_name': v('h1').text(),
'type_name': p.eq(2).text(),
'vod_year': p.eq(3).text(),
'vod_area': p.eq(4).text(),
'vod_remarks': v('em').text(),
'vod_actor': p.eq(0).text(),
'vod_director': p.eq(1).text(),
'vod_content': data('#desc .text').text(),
'vod_play_from': '',
'vod_play_url': ''
}
names = [i.text() for i in data('.title.slip a').items()]
plist=[]
for i in data('.play-list ul').items():
plist.append('#'.join([f'{j("a").text()}${j("a").attr("href")}' for j in i('li').items()]))
vod['vod_play_from'] = '$$$'.join(names)
vod['vod_play_url'] = '$$$'.join(plist)
return {'list': [vod]}
def searchContent(self, key, quick, pg="1"):
data=pq(self.fetch(f'{self.host}/search/{key}----------{pg}---.html', headers=self.headers).text)
return {'list': self.getlist(data('.show.rows li')),'page':pg}
did = ids[0]
params = {}
queryString = did.split('?')[1] if '?' in did else did
pairs = queryString.split('&')
for i in range(len(pairs)):
pair = pairs[i].split('=')
if len(pair) == 2:
params[pair[0]] = pair[1]
book_id = params.get('book_id', '')
actor = params.get('actor', '')
fullType = params.get('type', '')
if not book_id:
match = re.search(r'book_id=([^&]*)', did)
if match and match[1]:
book_id = match[1]
if not book_id:
return {'list': []}
apiUrl = f"{base_url}?book_id={book_id}"
try:
response = requests.get(url=apiUrl, headers=headers, timeout=timeout)
if response.status_code != 200:
return {'list': []}
data = response.json()
if data.get('code') == 200 and data.get('data'):
vod_list = data['data']
play_from = []
play_url = []
quality_options = [
("超清", "2160p"),
("高清", "1080p"),
("标清", "720p"),
("低清", "480p"),
("流畅", "360p")
]
for quality_name, quality_value in quality_options:
urls = []
for item in vod_list:
chapterName = item.get('title', '')
videoId = item.get('video_id', '')
playUrl = f"{quality_host}/duanju/api.php?video_id={videoId}&type=json&level={quality_value}"
urls.append(f"{chapterName}${playUrl}")
play_from.append(quality_name)
play_url.append("#".join(urls))
actors = []
try:
actor_api_url = f"{base_url}?series_id={book_id}&showRawParams=false"
actor_response = requests.get(url=actor_api_url, headers=headers, timeout=timeout)
if actor_response.status_code == 200:
actor_data = actor_response.json()
if actor_data.get('code') == 200 and 'celebrities' in actor_data:
celebrities = actor_data['celebrities']
if isinstance(celebrities, list):
for celeb in celebrities:
actor_name = celeb.get('user_name') or celeb.get('name') or celeb.get('actor_name') or ''
if actor_name and actor_name.strip():
if actor_name not in actors:
actors.append(actor_name)
except Exception as e:
print(f"获取演员信息失败: {e}")
if actor:
actors = [actor]
actor_str = ", ".join(actors) if actors else (actor or "")
categories = []
if 'category_names' in data and isinstance(data['category_names'], list):
categories = data['category_names'][:3] # 只取前3个
elif 'category' in data:
categories = [data['category']][:1]
type_str = ""
if categories:
type_str = ", ".join(categories)
remarks_str = f"共{len(vod_list)}集"
content_str = data.get('desc', '')
VOD = {
"vod_id": did,
"vod_name": data.get('book_name', ''),
"vod_pic": data.get('book_pic', ''),
"vod_actor": actor_str,
"type_name": type_str or fullType,
"vod_remarks": remarks_str,
"vod_content": content_str,
"vod_play_from": "$$$".join(play_from),
"vod_play_url": "$$$".join(play_url)
}
return {'list': [VOD]}
else:
return {'list': []}
except Exception as e:
print(f"获取详情失败: {e}")
return {'list': []}
def playerContent(self, flag, id, vipFlags):
p=0
uid=f'{self.host}{id}'
data=pq(self.fetch(uid, headers=self.headers).text)
url=data('.video.ratio').attr('data-play')
if not url:
url = uid
p = 1
return {'parse': p, 'url': url, 'header': self.headers}
def localProxy(self, param):
pass
def getlist(self,data):
vlist = []
for j in data.items():
vlist.append({
'vod_id': j('a').attr('href'),
'vod_name': j('img').attr('alt'),
'vod_pic': self.host + j('img').attr('data-src'),
'vod_year': j('.bg').text(),
'vod_remarks': j('p').text()
})
return vlist
max_retries = 3
for i in range(max_retries):
try:
response = requests.get(url=id, headers=headers, timeout=timeout)
if response.status_code == 200:
data = response.json()
if data.get('code') == 200 and data.get('data'):
play_url = data['data'].get('url', '')
if play_url:
return {
"parse": 0,
"playUrl": '',
"url": play_url,
"header": headers
}
break
except:
if i < max_retries - 1:
continue
else:
break
return {
"parse": 0,
"playUrl": '',
"url": 'about:blank',
"header": headers
}
def searchContent(self, key, quick, pg=1):
try:
page = int(pg)
except:
page = 1
params = f"page={page}&name={urllib.parse.quote(key)}&tab_type=19"
search_url = f"{base_url}?{params}"
try:
response = requests.get(search_url, headers=headers, timeout=timeout)
if response.status_code != 200:
return {'list': []}
response.encoding = "utf-8"
data = response.json()
if data.get('code') != 200 or not data.get('data'):
return {'list': []}
videos = []
for vod in data['data']:
vod_id = f"book_id={vod.get('book_id', '')}&actor={vod.get('author', '')}&type={vod.get('type', '')}"
videos.append({
"vod_id": vod_id,
"vod_name": vod.get('title', ''),
"vod_pic": vod.get('cover', ''),
"vod_remarks": vod.get('type', ''),
"vod_content": vod.get('intro', '')
})
return {
'list': videos,
'page': page,
'pagecount': 9999,
'limit': len(videos),
'total': 999999
}
except Exception:
return {'list': []}