XC-Pro/sg
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-14 11:25:26 +02:00
parent 73e705eb40
commit 5baa1f5c7f
5 changed files with 1354 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
py/企鹅.py
+519
View File
@@ -0,0 +1,519 @@
# coding=utf-8
#!/usr/bin/python
# 精彩
import json
import sys
import uuid
import copy
import traceback
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):
return "腾讯视频"
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']): item['type_id'] for item in classes}
for future in as_completed(futures):
cid = futures[future]
try:
_, data = future.result()
if not data.get('data', {}).get('module_list_datas'):
continue
filter_dict = {}
try:
# 安全获取嵌套数据
module_list = data['data']['module_list_datas']
if not module_list:
continue
last_module = module_list[-1]
if not last_module.get('module_datas'):
continue
module_datas = last_module['module_datas']
if not module_datas:
continue
item_lists = module_datas[-1].get('item_data_lists', {}).get('item_datas', [])
for item in item_lists:
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.get('index_name', ''),
'value': []
}
filter_dict[filter_key]['value'].append({
'n': params.get('option_name', ''),
'v': params.get('option_value', '')
})
except (IndexError, KeyError, TypeError) as e:
print(f"处理分类 {cid} 筛选数据时出错: {str(e)}")
continue
filters[cid] = list(filter_dict.values())
except Exception as e:
print(f"获取分类 {cid} 筛选失败: {str(e)}")
continue
result['class'] = classes
result['filters'] = filters
return result
def homeVideoContent(self):
vlist = []
try:
data = self.gethtml(self.host)
its = data('script')
s = None
for it in its.items():
text = it.text()
if text and 'window.__INITIAL_STATE__' in text:
s = text
break
if s:
index = s.find('=')
if index != -1:
json_str = s[index + 1:].strip()
try:
sd = json.loads(json_str)
except json.JSONDecodeError as e:
print(f"JSON解析错误: {str(e)}")
return {'list': []}
channels_map = sd.get('storeModulesData', {}).get('channelsModulesMap', {})
choice_data = channels_map.get('choice', {})
if choice_data and choice_data.get('cardListData'):
for its in choice_data['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']
# 安全解析JSON标签
tag = {}
try:
tag_str = p.get('uni_imgtag') or p.get('imgtag', '{}')
if tag_str:
tag = json.loads(tag_str)
except (json.JSONDecodeError, TypeError):
tag = {}
id = it.get('id') or p.get('cid')
name = p.get('mz_title') or p.get('title')
if name and id and 'http' not in str(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', '')
})
except Exception as e:
print(f"首页内容获取失败: {str(e)}")
traceback.print_exc()
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)
try:
response = 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)
data = response.json()
except Exception as e:
print(f"分类请求失败: {str(e)}")
return {'list': [], 'page': pg, 'pagecount': 0, 'limit': 90, 'total': 0}
ndata = data.get('data', {})
if not ndata:
return {'list': [], 'page': pg, 'pagecount': 0, 'limit': 90, 'total': 0}
if ndata.get('has_next_page'):
result['pagecount'] = 9999
self.body['page_context'] = ndata.get('next_page_context', '')
else:
result['pagecount'] = int(pg)
vlist = []
try:
# 安全获取列表数据
module_list = ndata.get('module_list_datas', [])
if module_list and module_list[-1].get('module_datas'):
item_datas = module_list[-1]['module_datas'][-1].get('item_data_lists', {}).get('item_datas', [])
for its in item_datas:
id = its.get('item_params', {}).get('cid')
if id:
p = its['item_params']
tag = {}
try:
tag_str = p.get('uni_imgtag') or p.get('imgtag', '{}')
if tag_str:
tag = json.loads(tag_str)
except (json.JSONDecodeError, TypeError):
tag = {}
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', '')
})
except (IndexError, KeyError, TypeError) as e:
print(f"解析分类数据失败: {str(e)}")
result['list'] = vlist
result['page'] = pg
result['limit'] = 90
result['total'] = 999999
return result
def detailContent(self, ids):
if not ids:
return self.handle_exception(None, "Empty 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
}
vdata = {}
data = {}
# 修复：添加异常处理
with ThreadPoolExecutor(max_workers=2) as executor:
future_detail = executor.submit(self.get_vdata, vbody)
future_episodes = executor.submit(self.get_vdata, body)
try:
vdata = future_detail.result()
except Exception as e:
print(f"获取详情失败: {str(e)}")
try:
data = future_episodes.result()
except Exception as e:
print(f"获取剧集失败: {str(e)}")
pdata = self.process_tabs(data, body, ids)
if not pdata:
return self.handle_exception(None, "No pdata available")
try:
# 安全获取演员列表
actors = []
try:
star_list = vdata.get('data', {}).get('module_list_datas', [{}])[0].get('module_datas', [{}])[0].get('item_data_lists', {}).get('item_datas', [{}])[0].get('sub_items', {}).get('star_list', {}).get('item_datas', [])
actors = [star.get('item_params', {}).get('name', '') for star in star_list if star.get('item_params', {}).get('name')]
except (IndexError, KeyError, AttributeError):
pass
names = ['腾讯视频', '预告片']
plist, ylist = self.process_pdata(pdata, ids)
if not plist:
names = [n for n in names if n != '腾讯视频']
if not ylist:
names = [n for n in names if n != '预告片']
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"}
}
try:
response = self.post(
f'{self.apihost}/trpc.videosearch.mobile_search.MultiTerminalSearch/MbSearch?vplatform=2',
json=body, headers=self.headers)
data = response.json()
except Exception as e:
print(f"搜索请求失败: {str(e)}")
return {'list': [], 'page': pg}
vlist = []
try:
area_box_list = data.get('data', {}).get('areaBoxList', [])
if area_box_list:
for k in area_box_list[-1].get('itemList', []):
if k.get('doc', {}).get('id'):
img_tag = k.get('videoInfo', {}).get('imgTag', '{}')
tag = {}
if isinstance(img_tag, str):
try:
tag = json.loads(img_tag)
except json.JSONDecodeError:
tag = {}
pic = k.get('videoInfo', {}).get('imgUrl', '')
vlist.append({
'vod_id': k['doc']['id'],
'vod_name': k.get('videoInfo', {}).get('title', ''),
'vod_pic': pic,
'vod_year': tag.get('tag_2', {}).get('text', ''),
'vod_remarks': tag.get('tag_4', {}).get('text', '')
})
except (IndexError, KeyError, TypeError) as e:
print(f"解析搜索结果失败: {str(e)}")
return {'list': vlist, 'page': pg}
def playerContent(self, flag, id, vipFlags):
ids = id.split('@')
if len(ids) < 2:
return {'parse': 0, 'url': '', 'header': ''}
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):
try:
rsp = self.fetch(url, headers=self.headers)
text = self.cleanText(rsp.text)
# 修复：确保传入字符串给PyQuery
return pq(text)
except Exception as e:
print(f"获取HTML失败 {url}: {str(e)}")
# 返回空的PyQuery对象避免崩溃
return pq('<html></html>')
def get_filter_data(self, cid):
try:
hbody = self.dbody.copy()
hbody['page_params']['channel_id'] = cid
response = 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)
return cid, response.json()
except Exception as e:
print(f"获取筛选数据失败 {cid}: {str(e)}")
return cid, {}
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()
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 = []
if not pdata:
return plist, ylist
for k in pdata:
if k.get('item_id'):
try:
title = k.get('item_params', {}).get('union_title', '')
pid = f"{title}${ids[0]}@{k['item_id']}"
if '预告' in title:
ylist.append(pid)
else:
plist.append(pid)
except Exception as e:
continue
return plist, ylist
def build_vod(self, vdata, actors, plist, ylist, names):
try:
d = vdata['data']['module_list_datas'][0]['module_datas'][0]['item_data_lists']['item_datas'][0]['item_params']
except (KeyError, IndexError):
d = {}
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) if actors else '',
'vod_content': d.get('cover_description', ''),
'vod_play_from': '$$$'.join(names) if names else '',
'vod_play_url': '$$$'.join(urls) if urls else ''
}
return vod
def handle_exception(self, e, message):
if e:
print(f"{message}: {str(e)}")
traceback.print_exc()
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:
try:
tabs = json.loads(tabs)
except json.JSONDecodeError:
tabs = []
if len(tabs) > 1:
remaining_tabs = tabs[1:]
task_queue = []
for tab in remaining_tabs:
nbody = copy.deepcopy(body)
nbody['page_params']['page_context'] = tab.get('page_context', '')
task_queue.append(nbody)
with ThreadPoolExecutor(max_workers=min(10, len(task_queue))) 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]
try:
results[idx] = future.result()
except Exception as e:
print(f"获取标签页 {idx} 失败: {str(e)}")
for result in results:
if result and isinstance(result, dict):
try:
page_data = result['data']['module_list_datas'][-1]['module_datas'][-1]['item_data_lists']['item_datas']
pdata.extend(page_data)
except (KeyError, IndexError, TypeError):
continue
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)
py/农民影视.py
+278
View File
@@ -0,0 +1,278 @@
# -*- coding: utf-8 -*-
# by @嗷呜
import concurrent.futures
import json
import re
import sys
import time
from base64 import b64decode, b64encode
import requests
from pyquery import PyQuery as pq
sys.path.append('..')
from base.spider import Spider
class Spider(Spider):
def init(self, extend=""):
self.host = "https://vip.wwgz.cn:5200"
self.headers = {
'User-Agent': 'Mozilla/5.0 (Linux; Android 10; K) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/130.0.0.0 Safari/537.36',
'Referer': self.host + '/',
'Accept': 'text/html'
}
self.cateConfig = {
"12": [{"key": "cateId", "name": "类型", "value": [{"n": "国产剧", "v": "12"}]}],
"4": [{"key": "cateId", "name": "类型", "value": [{"n": "动漫", "v": "4"}]}],
"1": [{"key": "cateId", "name": "类型", "value": [{"n": "电影", "v": "1"}]}],
"2": [{"key": "cateId", "name": "类型", "value": [{"n": "电视剧", "v": "2"}]}],
"3": [{"key": "cateId", "name": "类型", "value": [{"n": "综艺", "v": "3"}]}],
"26": [{"key": "cateId", "name": "类型", "value": [{"n": "短剧", "v": "26"}]}]
}
self.filterConfig = {}
def getName(self):
return "农民影视"
def isVideoFormat(self, url):
pass
def manualVideoCheck(self):
pass
def destroy(self):
pass
def homeContent(self, filter):
result = {}
classes = [
{'type_name': '国产剧', 'type_id': '12'},
{'type_name': '动漫', 'type_id': '4'},
{'type_name': '电影', 'type_id': '1'},
{'type_name': '电视剧', 'type_id': '2'},
{'type_name': '综艺', 'type_id': '3'},
{'type_name': '短剧', 'type_id': '26'}
]
try:
data = self.fetch(self.host, headers=self.headers).text
doc = pq(data)
videos = []
# 修改选择器并添加去重逻辑
seen_ids = set() # 用于记录已处理的影片ID
for item in doc('.globalPicList li:has(img)').items():
vod_id = self.host + item('a').attr('href')
if vod_id not in seen_ids: # 检查是否已处理过
seen_ids.add(vod_id) # 记录已处理的ID
pic_url = item('img').attr('data-echo') or item('img').attr('data-src') or item('img').attr('src')
# 替换图片域名
if pic_url and 'pic.lzzypic.com' in pic_url:
pic_url = pic_url.replace('https://pic.lzzypic.com', 'https://img.lzzyimg.com')
videos.append({
'vod_id': vod_id,
'vod_name': item('.sTit').text(),
'vod_pic': pic_url,
'vod_remarks': item('.sBottom').text()
})
result['class'] = classes
result['filters'] = self.cateConfig
result['list'] = videos
except Exception as e:
print(f"首页数据获取失败: {str(e)}")
result['class'] = classes
result['filters'] = self.cateConfig
result['list'] = []
return result
def homeVideoContent(self):
pass
def categoryContent(self, tid, pg, filter, extend):
result = {}
try:
if tid == "4-dm":
# 处理大陆人气动漫分类
url = "https://www.wwgz.cn/vod-list-id-4-pg-{}-order--by-hits-class-0-year-0-letter--area-大陆-lang-.html".format(pg)
else:
cateId = tid
url = f"{self.host}/vod-list-id-{cateId}-pg-{pg}.html"
data = self.fetch(url, headers=self.headers).text
doc = pq(data)
videos = []
for item in doc('.globalPicList li').items():
pic_url = item('img').attr('data-echo') or item('img').attr('data-src') or item('img').attr('src')
# 替换图片域名
if pic_url and 'pic.lzzypic.com' in pic_url:
pic_url = pic_url.replace('https://pic.lzzypic.com', 'https://img.lzzyimg.com')
videos.append({
'vod_id': self.host + item('a').attr('href'),
'vod_name': item('.sTit').text(),
'vod_pic': pic_url,
'vod_remarks': item('.sBottom').text()
})
result['list'] = videos
result['page'] = pg
result['pagecount'] = 9999
result['limit'] = 90
result['total'] = 999999
except Exception as e:
print(f"分类数据获取失败: {str(e)}")
result['list'] = []
result['page'] = pg
result['pagecount'] = 1
result['limit'] = 90
result['total'] = 0
return result
def detailContent(self, ids):
result = {}
try:
url = ids[0]
data = self.fetch(url, headers=self.headers).text
doc = pq(data)
# 获取播放线路和剧集
play_from = []
play_url = []
tab_box = doc('#leftTabBox')
if tab_box:
for tab in tab_box('ul li').items():
play_from.append(tab.text())
play_lists = []
for num_list in tab_box('.numList').items():
episodes = []
# 修改这里：将items()转换为列表后反转顺序
for ep in list(num_list('li').items())[::-1]: # 反转列表顺序
episodes.append(f"{ep('a').text()}${self.host}{ep('a').attr('href')}")
play_lists.append('#'.join(episodes))
play_url = play_lists
# 获取详情信息
vod = {
'vod_name': doc('h1 a').text(),
'vod_year': doc('span:contains("年代：")').text().replace('年代：', ''),
'vod_area': '',
'vod_actor': doc('.sDes:contains("主演:")').text().replace('主演:', ''),
'vod_director': '',
'vod_content': doc('.detail-con p').text().replace('简介:', ''),
'vod_play_from': '$$$'.join(play_from),
'vod_play_url': '$$$'.join(play_url)
}
result['list'] = [vod]
except Exception as e:
print(f"详情数据获取失败: {str(e)}")
result['list'] = []
return result
def searchContent(self, key, quick, pg="1"):
result = {}
try:
url = f"{self.host}/index.php?m=vod-search"
data = {'wd': key}
headers = {
'User-Agent': self.headers['User-Agent'],
'Referer': self.host + '/'
}
html = self.post(url, data=data, headers=headers).text
doc = pq(html)
videos = []
for item in doc('#data_list li').items():
pic_url = item('.lazyload').attr('data-src')
# 替换图片域名
if pic_url and 'pic.lzzypic.com' in pic_url:
pic_url = pic_url.replace('https://pic.lzzypic.com', 'https://img.lzzyimg.com')
videos.append({
'vod_id': self.host + item('a').attr('href'),
'vod_name': item('.sTit').text(),
'vod_pic': pic_url,
'vod_remarks': item('.sDes').eq(-1).text()
})
result['list'] = videos
result['page'] = pg
except Exception as e:
print(f"搜索数据获取失败: {str(e)}")
result['list'] = []
result['page'] = pg
return result
def playerContent(self, flag, id, vipFlags):
result = {}
try:
if '@' in id:
ids = id.split('@')
if not ids[0]:
raise Exception('未找到播放地址')
js_url = f"{self.host}/player/{ids[0]}.js"
js_data = self.fetch(js_url, headers=self.headers).text
jxurl = re.search(r'http.*?url=', js_data).group()
data = self.fetch(f"{jxurl}{ids[1]}", headers=self.headers).text
matches = re.findall(r'http.*?url=', data)
if matches:
url = []
for i, x in enumerate(matches):
js = {'jx': x, 'id': ids[1]}
purl = f"{self.getProxyUrl()}&wdict={self.e64(json.dumps(js))}"
url.extend([f'线路{i + 1}', purl])
else:
url = re.search(r"url='(.*?)'", data).group(1)
if not url:
raise Exception('未找到播放地址')
p = 0
else:
p, url = 1, id
result['parse'] = p
result['url'] = url
result['header'] = self.headers
except Exception as e:
print(f"播放数据获取失败: {str(e)}")
result['parse'] = 1
result['url'] = id
result['header'] = self.headers
return result
def localProxy(self, param):
try:
wdict = json.loads(self.d64(param['wdict']))
url = f"{wdict['jx']}{wdict['id']}"
data = self.fetch(url, headers=self.headers).text
doc = pq(data)
html = doc('script').eq(-1).text()
url = re.search(r'src="(.*?)"', html).group(1)
return [302, 'text/html', None, {'Location': url}]
except Exception as e:
print(f"代理处理失败: {str(e)}")
return [500, 'text/plain', str(e).encode('utf-8')]
def liveContent(self, url):
pass
def e64(self, text):
try:
text_bytes = text.encode('utf-8')
encoded_bytes = b64encode(text_bytes)
return encoded_bytes.decode('utf-8')
except Exception as e:
print(f"Base64编码错误: {str(e)}")
return ""
def d64(self, encoded_text):
try:
encoded_bytes = encoded_text.encode('utf-8')
decoded_bytes = b64decode(encoded_bytes)
return decoded_bytes.decode('utf-8')
except Exception as e:
print(f"Base64解码错误: {str(e)}")
return ""
py/奈飞.py
+373
View File
@@ -0,0 +1,373 @@
# -*- coding: utf-8 -*-
"""
奈飞影视 - naifei.im
"""
import re
import json
import sys
import time
from urllib.parse import quote, urljoin
from base.spider import Spider
class Spider(Spider):
def __init__(self):
super(Spider, self).__init__()
self.host = "https://naifei.im"
self.name = "奈飞影视"
self.headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/131.0.0.0 Safari/537.36',
'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8',
'Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.8',
'Accept-Encoding': 'gzip, deflate, br',
'Connection': 'keep-alive',
'Upgrade-Insecure-Requests': '1',
'Sec-Fetch-Dest': 'document',
'Sec-Fetch-Mode': 'navigate',
'Sec-Fetch-Site': 'none',
'Sec-Fetch-User': '?1',
'Cache-Control': 'max-age=0',
'Referer': self.host
}
self.categories = {
'1': '电影',
'2': '剧集',
'3': '综艺',
'4': '动漫',
'5': '短剧'
}
self._detail_cache = {}
def getName(self):
return "奈飞影视"
def init(self, extend=""):
pass
def homeContent(self, filter):
classes = [
{"type_id": "1", "type_name": "电影"},
{"type_id": "2", "type_name": "剧集"},
{"type_id": "3", "type_name": "综艺"},
{"type_id": "4", "type_name": "动漫"},
{"type_id": "5", "type_name": "短剧"},
]
return {'class': classes, 'filters': {}, 'list': []}
def homeVideoContent(self):
try:
videos = self._fetch_home()
return {'list': videos}
except Exception as e:
print(f'[{self.name}] 首页爬取失败: {e}')
return {'list': []}
def categoryContent(self, tid, pg, filter, extend):
try:
page = int(pg) if pg and str(pg).isdigit() else 1
videos = self._fetch_category(tid, page)
return {
'page': page,
'pagecount': 9999,
'limit': 20,
'total': 99999,
'list': videos
}
except Exception as e:
print(f'[{self.name}] 分类爬取失败: {e}')
return {'page': int(pg), 'pagecount': 0, 'limit': 20, 'total': 0, 'list': []}
def detailContent(self, ids):
try:
vod_id = ids[0] if isinstance(ids, list) else ids
detail = self._fetch_detail(vod_id)
if detail:
return {'list': [detail]}
return {'list': []}
except Exception as e:
print(f'[{self.name}] 详情爬取失败: {e}')
return {'list': []}
def playerContent(self, flag, id, vipFlags):
try:
play_url = ''
if id and id.startswith('http'):
play_url = id
elif '$' in str(id):
parts = str(id).split('$', 1)
if len(parts) == 2:
play_url = parts[1]
else:
play_url = id
# 如果是网页链接，需要解析获取真实播放地址
if play_url and 'naifei.im' in play_url:
real_url = self._parse_play_url(play_url)
if real_url:
play_url = real_url
return {
'parse': 0,
'playUrl': '',
'url': play_url,
}
except Exception as e:
print(f'[{self.name}] 播放失败: {e}')
return {
'parse': 1,
'playUrl': '',
'url': str(id),
}
def _parse_play_url(self, url):
"""解析播放页面获取真实播放地址"""
import re
html = self._fetch_page(url)
if not html:
return None
# 直接提取url字段
url_match = re.search(r'"url"\s*:\s*"(https?:[^"]+)"', html)
if url_match:
video_url = url_match.group(1).replace('\\/', '/')
if video_url and video_url.startswith('http'):
return video_url
# 备用：直接匹配m3u8地址
m3u8_match = re.search(r'(https?://[^\s"\'\\]+\.m3u8[^\s"\'\\]*)', html)
if m3u8_match:
return m3u8_match.group(1).replace('\\/', '/')
return None
def searchContent(self, key, quick, pg="1"):
try:
page = int(pg) if pg and str(pg).isdigit() else 1
videos = self._fetch_search(key, page)
return {'list': videos}
except Exception as e:
print(f'[{self.name}] 搜索失败: {e}')
return {'list': []}
def _fetch_page(self, url, retries=2):
"""获取页面内容"""
import requests
session = requests.Session()
session.headers.update(self.headers)
for attempt in range(retries + 1):
try:
resp = session.get(url, timeout=15)
if resp.status_code == 403:
redirect_match = re.search(r'window\.location\.href\s*=\s*"([^"]+)"', resp.text)
if redirect_match:
redirect_path = redirect_match.group(1)
if redirect_path.startswith('/'):
new_url = self.host + redirect_path
else:
new_url = redirect_path
resp = session.get(new_url, timeout=15)
elif attempt < retries:
time.sleep(1)
session.get(self.host, timeout=10)
continue
resp.raise_for_status()
resp.encoding = 'utf-8'
return resp.text
except Exception as e:
if attempt < retries:
time.sleep(1)
continue
print(f'[{self.name}] 请求失败: {url}, 错误: {e}')
return ''
return ''
def _fetch_home(self):
"""获取首页视频"""
from bs4 import BeautifulSoup
html = self._fetch_page(self.host)
if not html:
return []
soup = BeautifulSoup(html, 'html.parser')
videos = []
items_containers = soup.find_all('div', class_='module-items')
for container in items_containers:
items = container.find_all('a', class_='module-poster-item')
for item in items[:20]:
vod = self._parse_video_item(item)
if vod:
videos.append(vod)
return videos[:50]
def _fetch_category(self, tid, page=1):
"""获取分类视频"""
from bs4 import BeautifulSoup
if page <= 1:
url = f"{self.host}/vodtype/{tid}.html"
else:
url = f"{self.host}/vodtype/{tid}-{page}.html"
html = self._fetch_page(url)
if not html:
return []
soup = BeautifulSoup(html, 'html.parser')
videos = []
items = soup.find_all('a', class_='module-poster-item')
for item in items:
vod = self._parse_video_item(item)
if vod:
videos.append(vod)
return videos
def _fetch_detail(self, vid):
"""获取视频详情"""
from bs4 import BeautifulSoup
if vid in self._detail_cache:
return self._detail_cache[vid]
url = f"{self.host}/voddetail/{vid}.html"
html = self._fetch_page(url)
if not html:
return None
soup = BeautifulSoup(html, 'html.parser')
result = {"vod_id": vid}
title = soup.find('h1', class_='video-info-heading')
result['vod_name'] = title.text.strip() if title else ''
cover = soup.find('img', class_='lazy lazyload')
if cover:
pic = cover.get('data-original', '') or cover.get('src', '')
if pic and pic.startswith('//'):
pic = 'https:' + pic
result['vod_pic'] = pic
else:
result['vod_pic'] = ''
info_items = soup.find_all('li', class_='list-item')
for item in info_items:
text = item.text.strip()
if '主演' in text:
result['vod_actor'] = text.split('：', 1)[-1] if '：' in text else ''
elif '导演' in text:
result['vod_director'] = text.split('：', 1)[-1] if '：' in text else ''
elif '地区' in text or '语言' in text:
result['vod_area'] = text.split('：', 1)[-1] if '：' in text else ''
elif '年份' in text:
result['vod_year'] = text.split('：', 1)[-1] if '：' in text else ''
elif '更新' in text or '集数' in text:
result['vod_remarks'] = text.split('：', 1)[-1] if '：' in text else ''
desc = soup.find('div', class_='video-info-content')
result['vod_content'] = desc.text.strip() if desc else ''
episodes = []
episode_list = soup.find('div', class_='module-play-list')
if episode_list:
ep_items = episode_list.find_all('a')
for ep in ep_items:
ep_link = ep.get('href', '')
ep_title = ep.text.strip()
if ep_title and ep_link:
full_url = urljoin(self.host, ep_link) if ep_link.startswith('/') else ep_link
episodes.append(f'{ep_title}${full_url}')
if episodes:
result['vod_play_from'] = '奈飞影视'
result['vod_play_url'] = '#'.join(episodes)
else:
result['vod_play_from'] = ''
result['vod_play_url'] = ''
self._detail_cache[vid] = result
return result
def _fetch_search(self, keyword, page=1):
"""搜索视频"""
import json
url = f"{self.host}/index.php/ajax/suggest?mid=1&limit=20&wd={quote(keyword)}"
html = self._fetch_page(url)
if not html:
return []
videos = []
try:
data = json.loads(html)
if data.get('code') == 1 and data.get('list'):
for item in data['list']:
vod = self._parse_search_item(item)
if vod:
videos.append(vod)
except Exception as e:
print(f'[{self.name}] 解析搜索结果失败: {e}')
return videos
def _parse_search_item(self, item):
"""解析搜索结果项"""
try:
vid = str(item.get('id', ''))
name = item.get('name', '')
if not vid or not name:
return None
pic = item.get('pic', '')
if pic and pic.startswith('//'):
pic = 'https:' + pic
return {
'vod_id': vid,
'vod_name': name,
'vod_pic': pic,
'vod_remarks': '',
}
except Exception as e:
return None
def _parse_video_item(self, item):
"""解析视频项"""
try:
link = item.get('href', '')
title = item.get('title', '')
img = item.find('img')
cover = ''
if img:
cover = img.get('data-original', '') or img.get('src', '')
if cover and cover.startswith('//'):
cover = 'https:' + cover
note = item.find('div', class_='module-item-note')
quality = note.text.strip() if note else ''
vid = ''
match = re.search(r'/voddetail/(\d+)\.html', link)
if match:
vid = match.group(1)
if not vid or not title:
return None
return {
'vod_id': vid,
'vod_name': title,
'vod_pic': cover,
'vod_remarks': quality,
}
except Exception as e:
return None
py/漫闪.jar
BIN
View File
Binary file not shown.
py/热播.py
+184
View File
@@ -0,0 +1,184 @@
# -*- coding: utf-8 -*-
# by @嗷呜
import json
import sys
import time
import requests
from base64 import b64decode, b64encode
from Crypto.Hash import MD5
from pyquery import PyQuery as pq
sys.path.append('..')
from base.spider import Spider
class Spider(Spider):
def init(self, extend=""):
pass
def getName(self):
pass
def isVideoFormat(self, url):
pass
def manualVideoCheck(self):
pass
def destroy(self):
pass
host='http://v.rbotv.cn'
headers = {
'User-Agent': 'okhttp-okgo/jeasonlzy',
'Accept-Language': 'zh-CN,zh;q=0.8'
}
def homeContent(self, filter):
data=requests.post(f'{self.host}/v3/type/top_type',headers=self.headers,files=self.getfiles({'': (None, '')})).json()
result = {}
classes = []
filters = {}
for k in data['data']['list']:
classes.append({
'type_name': k['type_name'],
'type_id': k['type_id']
})
fts = []
for i,x in k.items():
if isinstance(x, list) and len(x)>2:
fts.append({
'name': i,
'key': i,
'value': [{'n': j, 'v': j} for j in x if j and j!= '全部']
})
if len(fts):filters[k['type_id']] = fts
result['class'] = classes
result['filters'] = filters
return result
def homeVideoContent(self):
data=requests.post(f'{self.host}/v3/type/tj_vod',headers=self.headers,files=self.getfiles({'': (None, '')})).json()
return {'list':self.getv(data['data']['cai']+data['data']['loop'])}
def categoryContent(self, tid, pg, filter, extend):
files = {
'type_id': (None, tid),
'limit': (None, '12'),
'page': (None, pg)
}
for k,v in extend.items():
if k=='extend':k='class'
files[k] = (None, v)
data=requests.post(f'{self.host}/v3/home/type_search',headers=self.headers,files=self.getfiles(files)).json()
result = {}
result['list'] = self.getv(data['data']['list'])
result['page'] = pg
result['pagecount'] = 9999
result['limit'] = 90
result['total'] = 999999
return result
def detailContent(self, ids):
data=requests.post(f'{self.host}/v3/home/vod_details',headers=self.headers,files=self.getfiles({'vod_id': (None, ids[0])})).json()
v=data['data']
vod = {
'vod_name': v.get('vod_name'),
'type_name': v.get('type_name'),
'vod_year': v.get('vod_year'),
'vod_area': v.get('vod_area'),
'vod_remarks': v.get('vod_remarks'),
'vod_actor': v.get('vod_actor'),
'vod_director': v.get('vod_director'),
'vod_content': pq(pq(v.get('vod_content','无') or '无').text()).text()
}
n,p=[],[]
for o,i in enumerate(v['vod_play_list']):
n.append(f"线路{o+1}({i.get('flag')})")
c=[]
for j in i.get('urls'):
d={'url':j.get('url'),'p':i.get('parse_urls'),'r':i.get('referer'),'u':i.get('ua')}
c.append(f"{j.get('name')}${self.e64(json.dumps(d))}")
p.append('#'.join(c))
vod.update({'vod_play_from':'$$$'.join(n),'vod_play_url':'$$$'.join(p)})
return {'list':[vod]}
def searchContent(self, key, quick, pg="1"):
files = {
'limit': (None, '12'),
'page': (None, pg),
'keyword': (None, key),
}
data=requests.post(f'{self.host}/v3/home/search',headers=self.headers,files=self.getfiles(files)).json()
return {'list':self.getv(data['data']['list']),'page':pg}
def playerContent(self, flag, id, vipFlags):
ids=json.loads(self.d64(id))
url=ids['url']
if isinstance(ids['p'],list) and len(ids['p']):
url=[]
for i,x in enumerate(ids['p']):
up={'url':ids['url'],'p':x,'r':ids['r'],'u':ids['u']}
url.extend([f"解析{i+1}",f"{self.getProxyUrl()}&data={self.e64(json.dumps(up))}"])
h={}
if ids.get('r'):
h['Referer'] = ids['r']
if ids.get('u'):
h['User-Agent'] = ids['u']
return {'parse': 0, 'url': url, 'header': h}
def localProxy(self, param):
data=json.loads(self.d64(param['data']))
h = {}
if data.get('r'):
h['Referer'] = data['r']
if data.get('u'):
h['User-Agent'] = data['u']
res=self.fetch(f"{data['p']}{data['url']}",headers=h).json()
url=res.get('url') or res['data'].get('url')
return [302,'video/MP2T',None,{'Location':url}]
def liveContent(self, url):
pass
def getfiles(self, p=None):
if p is None:p = {}
t=str(int(time.time()))
h = MD5.new()
h.update(f"7gp0bnd2sr85ydii2j32pcypscoc4w6c7g5spl{t}".encode('utf-8'))
s = h.hexdigest()
files = {
'sign': (None, s),
'timestamp': (None, t)
}
p.update(files)
return p
def getv(self,data):
videos = []
for i in data:
if i.get('vod_id') and str(i['vod_id']) != '0':
videos.append({
'vod_id': i['vod_id'],
'vod_name': i.get('vod_name'),
'vod_pic': i.get('vod_pic') or i.get('vod_pic_thumb'),
'vod_year': i.get('tag'),
'vod_remarks': i.get('vod_remarks')
})
return videos
def e64(self, text):
try:
text_bytes = text.encode('utf-8')
encoded_bytes = b64encode(text_bytes)
return encoded_bytes.decode('utf-8')
except Exception as e:
return ""
def d64(self,encoded_text):
try:
encoded_bytes = encoded_text.encode('utf-8')
decoded_bytes = b64decode(encoded_bytes)
return decoded_bytes.decode('utf-8')
except Exception as e:
return ""