上传文件至「py」

This commit is contained in:
2026-08-12 18:35:22 +02:00
parent 560542b354
commit eb84f08e05
3 changed files with 1041 additions and 0 deletions
+522
View File
@@ -0,0 +1,522 @@
#基于嗷呜大佬修复搜索
# coding=utf-8
# !/usr/bin/python
import sys
sys.path.append('..')
from base.spider import Spider
import json
import re
try:
import ujson
except ImportError:
ujson = json
try:
from pyquery import PyQuery as pq
except ImportError:
pq = None
try:
from cachetools import TTLCache
except ImportError:
class TTLCache:
def __init__(self, maxsize=100, ttl=600):
self.cache = {}
self.maxsize = maxsize
def __contains__(self, key):
return key in self.cache
def __getitem__(self, key):
return self.cache[key]
def __setitem__(self, key, value):
if len(self.cache) >= self.maxsize:
first_key = next(iter(self.cache))
del self.cache[first_key]
self.cache[key] = value
def __len__(self):
return len(self.cache)
class Spider(Spider):
def __init__(self):
self.cache = TTLCache(maxsize=100, ttl=600)
def getName(self):
return "Libvio"
def init(self, extend=""):
print("============{0}============".format(extend))
if not hasattr(self, 'cache'):
self.cache = TTLCache(maxsize=100, ttl=600)
pass
def _fetch_with_cache(self, url, headers=None):
cache_key = f"{url}_{hash(str(headers))}"
if cache_key in self.cache:
return self.cache[cache_key]
try:
response = self.fetch(url, headers=headers or self.header)
except Exception as e:
print(f"Fetch failed for {url}: {e}")
response = None # Fallback to None on error
if response:
self.cache[cache_key] = response
return response
def _parse_html_fast(self, html_text):
if not html_text:
return None
if pq is not None:
try:
return pq(html_text)
except:
pass
return self.html(self.cleanText(html_text))
def homeContent(self, filter):
result = {}
cateManual = {"电影": "1", "电视剧": "2", "动漫": "4", "日韩剧": "15", "欧美剧": "16"}
classes = []
for k in cateManual:
classes.append({'type_name': k, 'type_id': cateManual[k]})
result['class'] = classes
if (filter):
result['filters'] = self._generate_filters()
return result
def homeVideoContent(self):
rsp = self._fetch_with_cache("https://www.libvio.site")
if not rsp:
return {'list': []}
doc = self._parse_html_fast(rsp.text)
videos = []
if pq is not None and hasattr(doc, '__call__'):
try:
thumb_links = doc('a.stui-vodlist__thumb.lazyload')
for i in range(thumb_links.length):
try:
thumb = thumb_links.eq(i)
href = thumb.attr('href')
if not href: continue
sid_match = re.search(r'/detail/(\d+)\.html', href)
if not sid_match: continue
sid = sid_match.group(1)
name = thumb.attr('title')
if not name: continue
pic = thumb.attr('data-original') or ""
mark = thumb.text().strip()
videos.append({"vod_id": sid, "vod_name": name.strip(), "vod_pic": pic, "vod_remarks": mark})
except Exception as e: continue
except: pass
if not videos:
try:
thumb_links = doc.xpath("//a[@class='stui-vodlist__thumb lazyload']")
for thumb in thumb_links:
try:
href = thumb.xpath("./@href")[0]
sid_match = re.search(r'/detail/(\d+)\.html', href)
if not sid_match: continue
sid = sid_match.group(1)
name = thumb.xpath("./@title")[0].strip()
if not name: continue
pic_list = thumb.xpath("./@data-original")
pic = pic_list[0] if pic_list else ""
mark_list = thumb.xpath("./text()")
mark = mark_list[0].strip() if mark_list else ""
videos.append({"vod_id": sid, "vod_name": name, "vod_pic": pic, "vod_remarks": mark})
except Exception as e: continue
except Exception as e: print(f"Homepage parse failed: {e}")
result = {'list': videos}
return result
def categoryContent(self, tid, pg, filter, extend):
result = {}
url = 'https://www.libvio.site/type/{0}-{1}.html'.format(tid, pg)
print(url)
rsp = self._fetch_with_cache(url)
if not rsp:
return result
doc = self._parse_html_fast(rsp.text)
videos = []
if pq is not None and hasattr(doc, '__call__'):
try:
thumb_links = doc('a.stui-vodlist__thumb.lazyload')
for i in range(thumb_links.length):
try:
thumb = thumb_links.eq(i)
href = thumb.attr('href')
if not href: continue
sid_match = re.search(r'/detail/(\d+)\.html', href)
if not sid_match: continue
sid = sid_match.group(1)
name = thumb.attr('title')
if not name: continue
pic = thumb.attr('data-original') or ""
mark = thumb.text().strip()
videos.append({"vod_id": sid, "vod_name": name.strip(), "vod_pic": pic, "vod_remarks": mark})
except Exception as e: continue
except: pass
if not videos:
try:
thumb_links = doc.xpath("//a[@class='stui-vodlist__thumb lazyload']")
for thumb in thumb_links:
try:
href = thumb.xpath("./@href")[0]
sid_match = re.search(r'/detail/(\d+)\.html', href)
if not sid_match: continue
sid = sid_match.group(1)
name = thumb.xpath("./@title")[0].strip()
if not name: continue
pic_list = thumb.xpath("./@data-original")
pic = pic_list[0] if pic_list else ""
mark_list = thumb.xpath("./text()")
mark = mark_list[0].strip() if mark_list else ""
videos.append({"vod_id": sid, "vod_name": name, "vod_pic": pic, "vod_remarks": mark})
except Exception as e: continue
except Exception as e: print(f"Category parse failed: {e}")
result['list'] = videos
result['page'] = pg
result['pagecount'] = 9999
result['limit'] = 90
result['total'] = 999999
return result
def detailContent(self, array):
tid = array[0]
url = 'https://www.libvio.site/detail/{0}.html'.format(tid)
rsp = self._fetch_with_cache(url)
if not rsp:
return {'list': []}
doc = self._parse_html_fast(rsp.text)
title = doc('h1').text().strip() or ""
pic = doc('img').attr('data-original') or doc('img').attr('src') or ""
detail = ""
try:
detail_content = doc('.detail-content').text().strip()
if detail_content: detail = detail_content
else:
detail_text = doc('*:contains("简介:")').text()
if detail_text and '简介:' in detail_text:
detail_part = detail_text.split('简介:')[1]
if '详情' in detail_part: detail_part = detail_part.replace('详情', '')
detail = detail_part.strip()
except: pass
douban = "0.0"
score_text = doc('.detail-info *:contains("")').text() or ""
score_match = re.search(r'(\d+\.?\d*)\s*分', score_text)
if score_match: douban = score_match.group(1)
vod = {"vod_id": tid, "vod_name": title, "vod_pic": pic, "type_name": "", "vod_year": "", "vod_area": "", "vod_remarks": "", "vod_actor": "", "vod_director": "", "vod_douban_score": douban, "vod_content": detail}
info_text = doc('p').text()
if '类型:' in info_text:
type_match = re.search(r'类型:([^/]+)', info_text)
if type_match: vod['type_name'] = type_match.group(1).strip()
if '主演:' in info_text:
actor_match = re.search(r'主演:([^/]+)', info_text)
if actor_match: vod['vod_actor'] = actor_match.group(1).strip()
if '导演:' in info_text:
director_match = re.search(r'导演:([^/]+)', info_text)
if director_match: vod['vod_director'] = director_match.group(1).strip()
playFrom = []
playList = []
# 改进的播放线路提取逻辑
vodlist_heads = doc('.stui-vodlist__head')
for i in range(vodlist_heads.length):
head = vodlist_heads.eq(i)
h3_elem = head.find('h3')
if h3_elem.length == 0:
continue
header_text = h3_elem.text().strip()
if not any(keyword in header_text for keyword in ['播放', '下载', 'BD5', 'UC', '夸克']):
continue
playFrom.append(header_text)
vodItems = []
# 提取当前播放线路下的所有播放链接
play_links = head.find('a[href*="/play/"]')
for j in range(play_links.length):
try:
link = play_links.eq(j)
href = link.attr('href')
name = link.text().strip()
if not href or not name:
continue
tId_match = re.search(r'/play/([^.]+)\.html', href)
if not tId_match:
continue
tId = tId_match.group(1)
vodItems.append(name + "$" + tId)
except:
continue
playList.append('#'.join(vodItems) if vodItems else "")
vod['vod_play_from'] = '$$$'.join(playFrom) if playFrom else ""
vod['vod_play_url'] = '$$$'.join(playList) if playList else ""
result = {'list': [vod]}
return result
def searchContent(self, key, quick, page=None):
url = 'https://www.libvio.site/index.php/ajax/suggest?mid=1&wd={0}'.format(key)
rsp = self._fetch_with_cache(url, headers=self.header)
if not rsp:
return {'list': []}
try:
jo = ujson.loads(rsp.text)
except:
jo = json.loads(rsp.text)
result = {}
jArray = []
if jo.get('total', 0) > 0:
for j in jo.get('list', []):
jArray.append({
"vod_id": j.get('id', ''),
"vod_name": j.get('name', ''),
"vod_pic": j.get('pic', ''),
"vod_remarks": ""
})
result = {'list': jArray}
return result
def _generate_filters(self):
years = [{"n": "全部", "v": ""}]
for year in range(2025, 1999, -1):
years.append({"n": str(year), "v": str(year)})
movie_filters = [
{
"key": "class", "name": "剧情",
"value": [
{"n": "全部", "v": ""}, {"n": "爱情", "v": "爱情"}, {"n": "恐怖", "v": "恐怖"},
{"n": "动作", "v": "动作"}, {"n": "科幻", "v": "科幻"}, {"n": "剧情", "v": "剧情"},
{"n": "战争", "v": "战争"}, {"n": "警匪", "v": "警匪"}, {"n": "犯罪", "v": "犯罪"},
{"n": "动画", "v": "动画"}, {"n": "奇幻", "v": "奇幻"}, {"n": "武侠", "v": "武侠"},
{"n": "冒险", "v": "冒险"}, {"n": "枪战", "v": "枪战"}, {"n": "悬疑", "v": "悬疑"},
{"n": "惊悚", "v": "惊悚"}, {"n": "经典", "v": "经典"}, {"n": "青春", "v": "青春"},
{"n": "文艺", "v": "文艺"}, {"n": "微电影", "v": "微电影"}, {"n": "古装", "v": "古装"},
{"n": "历史", "v": "历史"}, {"n": "运动", "v": "运动"}, {"n": "农村", "v": "农村"},
{"n": "儿童", "v": "儿童"}, {"n": "网络电影", "v": "网络电影"}
]
},
{
"key": "area", "name": "地区",
"value": [
{"n": "全部", "v": ""}, {"n": "大陆", "v": "中国大陆"}, {"n": "香港", "v": "中国香港"},
{"n": "台湾", "v": "中国台湾"}, {"n": "美国", "v": "美国"}, {"n": "法国", "v": "法国"},
{"n": "英国", "v": "英国"}, {"n": "日本", "v": "日本"}, {"n": "韩国", "v": "韩国"},
{"n": "德国", "v": "德国"}, {"n": "泰国", "v": "泰国"}, {"n": "印度", "v": "印度"},
{"n": "意大利", "v": "意大利"}, {"n": "西班牙", "v": "西班牙"},
{"n": "加拿大", "v": "加拿大"}, {"n": "其他", "v": "其他"}
]
},
{"key": "year", "name": "年份", "value": years}
]
tv_filters = [
{
"key": "class", "name": "剧情",
"value": [
{"n": "全部", "v": ""}, {"n": "战争", "v": "战争"}, {"n": "青春偶像", "v": "青春偶像"},
{"n": "喜剧", "v": "喜剧"}, {"n": "家庭", "v": "家庭"}, {"n": "犯罪", "v": "犯罪"},
{"n": "动作", "v": "动作"}, {"n": "奇幻", "v": "奇幻"}, {"n": "剧情", "v": "剧情"},
{"n": "历史", "v": "历史"}, {"n": "经典", "v": "经典"}, {"n": "乡村", "v": "乡村"},
{"n": "情景", "v": "情景"}, {"n": "商战", "v": "商战"}, {"n": "网剧", "v": "网剧"},
{"n": "其他", "v": "其他"}
]
},
{
"key": "area", "name": "地区",
"value": [
{"n": "全部", "v": ""}, {"n": "大陆", "v": "中国大陆"}, {"n": "台湾", "v": "中国台湾"},
{"n": "香港", "v": "中国香港"}, {"n": "韩国", "v": "韩国"}, {"n": "日本", "v": "日本"},
{"n": "美国", "v": "美国"}, {"n": "泰国", "v": "泰国"}, {"n": "英国", "v": "英国"},
{"n": "新加坡", "v": "新加坡"}, {"n": "其他", "v": "其他"}
]
},
{"key": "year", "name": "年份", "value": years}
]
anime_filters = [
{
"key": "class", "name": "剧情",
"value": [
{"n": "全部", "v": ""}, {"n": "科幻", "v": "科幻"}, {"n": "热血", "v": "热血"},
{"n": "推理", "v": "推理"}, {"n": "搞笑", "v": "搞笑"}, {"n": "冒险", "v": "冒险"},
{"n": "萝莉", "v": "萝莉"}, {"n": "校园", "v": "校园"}, {"n": "动作", "v": "动作"},
{"n": "机战", "v": "机战"}, {"n": "运动", "v": "运动"}, {"n": "战争", "v": "战争"},
{"n": "少年", "v": "少年"}, {"n": "少女", "v": "少女"}, {"n": "社会", "v": "社会"},
{"n": "原创", "v": "原创"}, {"n": "亲子", "v": "亲子"}, {"n": "益智", "v": "益智"},
{"n": "励志", "v": "励志"}, {"n": "其他", "v": "其他"}
]
},
{
"key": "area", "name": "地区",
"value": [
{"n": "全部", "v": ""}, {"n": "中国", "v": "中国"}, {"n": "日本", "v": "日本"},
{"n": "欧美", "v": "欧美"}, {"n": "其他", "v": "其他"}
]
},
{"key": "year", "name": "年份", "value": years}
]
asian_filters = [
{
"key": "class", "name": "剧情",
"value": [
{"n": "全部", "v": ""}, {"n": "剧情", "v": "剧情"}, {"n": "喜剧", "v": "喜剧"},
{"n": "爱情", "v": "爱情"}, {"n": "动作", "v": "动作"}, {"n": "悬疑", "v": "悬疑"},
{"n": "惊悚", "v": "惊悚"}, {"n": "恐怖", "v": "恐怖"}, {"n": "犯罪", "v": "犯罪"}
]
},
{
"key": "area", "name": "地区",
"value": [
{"n": "全部", "v": ""}, {"n": "韩国", "v": "韩国"}, {"n": "日本", "v": "日本"},
{"n": "泰国", "v": "泰国"}
]
},
{"key": "year", "name": "年份", "value": years[:25]}
]
western_filters = [
{
"key": "class", "name": "剧情",
"value": [
{"n": "全部", "v": ""}, {"n": "剧情", "v": "剧情"}, {"n": "喜剧", "v": "喜剧"},
{"n": "爱情", "v": "爱情"}, {"n": "动作", "v": "动作"}, {"n": "科幻", "v": "科幻"},
{"n": "悬疑", "v": "悬疑"}, {"n": "惊悚", "v": "惊悚"}, {"n": "恐怖", "v": "恐怖"},
{"n": "犯罪", "v": "犯罪"}
]
},
{
"key": "area", "name": "地区",
"value": [
{"n": "全部", "v": ""}, {"n": "美国", "v": "美国"}, {"n": "英国", "v": "英国"},
{"n": "加拿大", "v": "加拿大"}, {"n": "其他", "v": "其他"}
]
},
{"key": "year", "name": "年份", "value": years[:25]}
]
return {
"1": movie_filters, # 电影
"2": tv_filters, # 电视剧
"4": anime_filters, # 动漫
"15": asian_filters, # 日韩剧
"16": western_filters # 欧美剧
}
header = {"Referer": "https://www.libvio.site", "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.114 Safari/537.36"}
def playerContent(self, flag, id, vipFlags):
# 如果已经是push链接,直接返回
if id.startswith('push://'):
return {"parse": 0, "playUrl": "", "url": id, "header": ""}
result = {}
url = 'https://www.libvio.site/play/{0}.html'.format(id)
try:
rsp = self._fetch_with_cache(url, headers=self.header)
if not rsp:
return {"parse": 1, "playUrl": "", "url": url, "header": ujson.dumps(self.header)}
return self._handle_cloud_drive(url, rsp, id)
except Exception as e:
print(f"Player parse error: {e}")
return {"parse": 1, "playUrl": "", "url": url, "header": ujson.dumps(self.header)}
def _handle_cloud_drive(self, url, rsp, id):
try:
page_text = rsp.text
# 首先尝试从JavaScript变量中提取网盘链接
script_pattern = r'var player_[^=]*=\s*({[^}]+})'
matches = re.findall(script_pattern, page_text)
for match in matches:
try:
player_data = ujson.loads(match)
from_value = player_data.get('from', '')
url_value = player_data.get('url', '')
if from_value == 'kuake' and url_value:
# 夸克网盘
drive_url = url_value.replace('\\/', '/')
return {"parse": 0, "playUrl": "", "url": f"push://{drive_url}", "header": ""}
elif from_value == 'uc' and url_value:
# UC网盘
drive_url = url_value.replace('\\/', '/')
return {"parse": 0, "playUrl": "", "url": f"push://{drive_url}", "header": ""}
except:
continue
except Exception as e:
print(f"Cloud drive parse error: {e}")
# 如果所有网盘解析都失败,尝试BD5播放源
return self._handle_bd5_player(url, rsp, id)
def _handle_bd5_player(self, url, rsp, id):
try:
doc = self._parse_html_fast(rsp.text)
page_text = rsp.text
api_match = re.search(r'https://www\.libvio\.site/vid/plyr/vr2\.php\?url=([^&"\s]+)', page_text)
if api_match:
return {"parse": 0, "playUrl": "", "url": api_match.group(1), "header": ujson.dumps
({"User-Agent": self.header["User-Agent"], "Referer": "https://www.libvio.site/"})}
iframe_src = doc('iframe').attr('src')
if iframe_src:
try:
iframe_content = self._fetch_with_cache(iframe_src, headers=self.header)
if not iframe_content: raise Exception("Iframe fetch failed")
video_match = re.search(r'https://[^"\s]+\.mp4', iframe_content.text)
if video_match: return {"parse": 0, "playUrl": "", "url": video_match.group(0), "header": ujson.dumps({"User-Agent": self.header["User-Agent"], "Referer": "https://www.libvio.site/"})}
except Exception as e: print(f"iframe视频解析失败: {e}")
script_match = re.search(r'var player_[^=]*=\s*({[^}]+})', page_text)
if script_match:
try:
jo = ujson.loads(script_match.group(1))
if jo:
nid = str(jo.get('nid', ''))
player_from = jo.get('from', '')
if player_from:
scriptUrl = f'https://www.libvio.site/static/player/{player_from}.js'
scriptRsp = self._fetch_with_cache(scriptUrl)
if not scriptRsp: raise Exception("Script fetch failed")
parse_match = re.search(r'src="([^"]+url=)', scriptRsp.text)
if parse_match:
parseUrl = parse_match.group(1)
path = f"{jo.get('url', '')}&next={jo.get('link_next', '')}&id={jo.get('id', '')}&nid={nid}"
parseRsp = self._fetch_with_cache(parseUrl + path, headers=self.header)
if not parseRsp: raise Exception("Parse fetch failed")
url_match = re.search(r"urls\s*=\s*'([^']+)'", parseRsp.text)
if url_match: return {"parse": 0, "playUrl": "", "url": url_match.group(1), "header": ""}
except Exception as e: print(f"JavaScript播放器解析失败: {e}")
except Exception as e: print(f"BD5播放源解析错误: {e}")
return {"parse": 1, "playUrl": "", "url": url, "header": ujson.dumps(self.header)}
def isVideoFormat(self, url):
return False
def manualVideoCheck(self):
pass
def localProxy(self, param):
action = b''
try:
header_dict = json.loads(param.get('header', '{}')) if param.get('header') else {}
resp = self.fetch(param['url'], headers=header_dict)
action = resp.content
except Exception as e:
print(f"Local proxy error: {e}")
return [200, "video/MP2T", action, param.get('header', '')]
+368
View File
@@ -0,0 +1,368 @@
# coding=utf-8
# !/usr/bin/python
"""
作者 丢丢喵 内容均从互联网收集而来 仅供交流学习使用 严禁用于商业用途 请于24小时内删除
====================Diudiumiao====================
"""
from Crypto.Util.Padding import unpad
from Crypto.Util.Padding import pad
from urllib.parse import unquote
from Crypto.Cipher import ARC4
from urllib.parse import quote
from base.spider import Spider
from Crypto.Cipher import AES
from datetime import datetime
from bs4 import BeautifulSoup
from base64 import b64decode
from base64 import b64encode
import urllib.request
import urllib.parse
import datetime
import binascii
import requests
import base64
import json
import time
import sys
import re
import os
sys.path.append('..')
xurl = "http://110.42.67.221:8006"
headerx = {
'User-Agent': 'Mozilla/5.0 (Windows NT 6.1; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/50.0.2661.87 Safari/537.36'
}
headers = {
'Content-Type': 'application/x-www-form-urlencoded',
'Host': '110.42.67.221:8006',
'Connection': 'Keep-Alive',
'User-Agent': 'okhttp/3.10.0',
'Accept-Encoding': 'gzip, deflate'
}
class Spider(Spider):
def getName(self):
return "丢丢喵"
def init(self, extend):
pass
def isVideoFormat(self, url):
pass
def manualVideoCheck(self):
pass
def decrypt_aes_cbc(self,ciphertext_base64):
key_hex = "31323334353637383961626364656667"
iv_hex = "31323334353637383961626364656667"
key = bytes.fromhex(key_hex)
iv = bytes.fromhex(iv_hex)
ciphertext = b64decode(ciphertext_base64)
cipher = AES.new(key, AES.MODE_CBC, iv)
decrypted_padded = cipher.decrypt(ciphertext)
padding_length = decrypted_padded[-1]
decrypted = decrypted_padded[:-padding_length]
return decrypted.decode('utf-8')
def encrypt_aes_cbc(self,plaintext):
key_hex = "31323334353637383961626364656667"
iv_hex = "31323334353637383961626364656667"
key = bytes.fromhex(key_hex)
iv = bytes.fromhex(iv_hex)
plaintext_bytes = plaintext.encode('utf-8')
padded_plaintext = pad(plaintext_bytes, AES.block_size)
cipher = AES.new(key, AES.MODE_CBC, iv)
encrypted = cipher.encrypt(padded_plaintext)
return b64encode(encrypted).decode('utf-8')
def homeContent(self, filter):
result = {"class": []}
data = self.fetch_home_data()
self.process_type_list(data, result)
return result
def fetch_home_data(self):
url = f"{xurl}/api.php/qijiappapi.index/initV122"
detail = requests.get(url=url, headers=headers)
detail.encoding = "utf-8"
data = detail.json()
data = data['data']
data = self.decrypt_aes_cbc(data)
return json.loads(data)
def process_type_list(self, data, result):
for vod in data['type_list']:
name = vod['type_name']
if self.should_skip_type(name):
continue
id = vod['type_id']
result["class"].append({"type_id": id, "type_name": name})
def should_skip_type(self, name):
skip_names = ["全部"]
return name in skip_names
def homeVideoContent(self):
videos = []
data = self.fetch_home_video_data()
self.process_video_list(data, videos)
return {'list': videos}
def fetch_home_video_data(self):
url = f"{xurl}/api.php/qijiappapi.index/initV122"
detail = requests.get(url=url, headers=headers)
detail.encoding = "utf-8"
data = detail.json()
data = data['data']
data = self.decrypt_aes_cbc(data)
return json.loads(data)
def process_video_list(self, data, videos):
for vods in data['type_list']:
for vod in vods['recommend_list']:
video = self.create_video_item(vod)
videos.append(video)
def create_video_item(self, vod):
name = vod['vod_name']
id = vod['vod_id']
pic = vod['vod_pic']
remark = vod.get('vod_remarks', '暂无备注')
return {"vod_id": id,"vod_name": name,"vod_pic": pic,"vod_remarks": remark}
def categoryContent(self, cid, pg, filter, ext):
videos = []
page = self.parse_page_number(pg)
data = self.build_category_request_data(cid, page)
response_data = self.fetch_category_data(data)
self.process_category_videos(response_data, videos)
return self.build_category_result(videos, pg)
def parse_page_number(self, pg):
return int(pg) if pg else 1
def build_category_request_data(self, cid, page):
return {'area': '全部','year': '全部','type_id': cid,'page': page,'sort': '最新','lang': '全部','class': '全部'}
def fetch_category_data(self, data):
url = f"{xurl}/api.php/qijiappapi.index/typeFilterVodList"
response = requests.post(url=url, headers=headers, data=data)
response_data = response.json()
data = response_data['data']
data = self.decrypt_aes_cbc(data)
return json.loads(data)
def process_category_videos(self, data, videos):
for vod in data['recommend_list']:
video = self.create_category_video_item(vod)
videos.append(video)
def create_category_video_item(self, vod):
name = vod['vod_name']
id = vod['vod_id']
pic = vod['vod_pic']
remark = vod.get('vod_remarks', '暂无备注')
return {"vod_id": id,"vod_name": name,"vod_pic": pic,"vod_remarks": remark}
def build_category_result(self, videos, pg):
return {'list': videos,'page': pg,'pagecount': 9999,'limit': 90,'total': 999999}
def detailContent(self, ids):
did = ids[0]
result = {}
videos = []
data = self.build_detail_request_data(did)
response_data = self.fetch_detail_data(data)
vod_data = self.parse_detail_response(response_data)
content = self.extract_detail_content(vod_data)
director = self.extract_detail_director(vod_data)
actor = self.extract_detail_actor(vod_data)
remarks = self.extract_detail_remarks(vod_data)
year = self.extract_detail_year(vod_data)
area = self.extract_detail_area(vod_data)
xianlu = self.extract_detail_xianlu(vod_data)
bofang = self.extract_detail_play_urls(vod_data)
videos.append({
"vod_id": did,
"vod_director": director,
"vod_actor": actor,
"vod_remarks": remarks,
"vod_year": year,
"vod_area": area,
"vod_content": content,
"vod_play_from": xianlu,
"vod_play_url": bofang
})
result['list'] = videos
return result
def build_detail_request_data(self, did):
return {
'vod_id': did
}
def fetch_detail_data(self, data):
url = f"{xurl}/api.php/qijiappapi.index/vodDetail3"
response = requests.post(url=url, headers=headers, data=data)
response_data = response.json()
data = response_data['data']
data = self.decrypt_aes_cbc(data)
return json.loads(data)
def parse_detail_response(self, response_data):
return response_data
def extract_detail_content(self, vod_data):
vod_blurb = vod_data.get('vod', {}).get('vod_blurb', '').replace('\u3000', '')
return '集多为您介绍剧情📢' + vod_blurb
def extract_detail_director(self, vod_data):
return vod_data.get('vod', {}).get('vod_director', '')
def extract_detail_actor(self, vod_data):
return vod_data.get('vod', {}).get('vod_actor', '')
def extract_detail_remarks(self, vod_data):
return vod_data.get('vod', {}).get('vod_remarks', '')
def extract_detail_year(self, vod_data):
return vod_data.get('vod', {}).get('vod_year', '')
def extract_detail_area(self, vod_data):
return vod_data.get('vod', {}).get('vod_area', '')
def extract_detail_xianlu(self, vod_data):
xianlu = ''
for vod in vod_data['vod_play_list']:
name = vod['player_info']['show']
xianlu = xianlu + name + '$$$'
return xianlu[:-3]
def extract_detail_play_urls(self, vod_data):
bofang = ''
for vods in vod_data['vod_play_list']:
for vod in vods['urls']:
name = vod['name']
id = vod['parse_api_url']
bofang = bofang + name + '$' + id + '#'
bofang = bofang[:-1] + '$$$'
return bofang[:-3]
def playerContent(self, flag, id, vipFlags):
url_value = self.process_player_url(id)
return self.build_player_result(url_value)
def process_player_url(self, id):
if 'YYNB-' in id:
return self.handle_special_url(id, 'YYNB-')
elif 'https://v.qq.com' in id:
return self.handle_special_url(id, 'https://v.qq.com')
elif 'https://vip.ffzy' in id:
return self.handle_special_url(id, 'https://vip.ffzy')
elif 'https://v.lzcdn' in id:
return self.handle_special_url(id, 'https://v.lzcdn')
elif 'https://cdn.yzzy' in id:
return self.handle_special_url(id, 'https://cdn.yzzy')
elif 'https://www.iqiyi.com' in id:
return self.handle_special_url(id, 'https://www.iqiyi.com')
elif 'https://www.mgtv.com' in id:
return self.handle_special_url(id, 'https://www.mgtv.com')
elif 'https://v.youku.com' in id:
return self.handle_special_url(id, 'https://v.youku.com')
elif 'https://www.bilibili.com' in id:
return self.handle_special_url(id, 'https://www.bilibili.com')
elif 'NBY-' in id or 'Ace_Net' in id or 'Ace_JP' in id:
return self.handle_direct_url(id)
return ''
def handle_special_url(self, id, prefix):
fenge = id.split(prefix)
url2 = f"{prefix}{fenge[1]}"
url2 = self.encrypt_aes_cbc(url2)
params = {'parse_api': fenge[0],'url': url2,'player_parse_type': 1,'token': ''}
url = f"{xurl}/api.php/qijiappapi.index/vodParse"
response = requests.post(url, data=params, headers=headers)
response_data = response.json()
data_value = response_data['data']
data = self.decrypt_aes_cbc(data_value)
data = json.loads(data)
inner_json = json.loads(data['json'])
return inner_json['url']
def handle_direct_url(self, id):
detail = requests.get(url=id, headers=headers)
detail.encoding = "utf-8"
data = detail.json()
return data['url']
def build_player_result(self, url_value):
return {"parse": 0,"playUrl": '',"url": url_value,"header": headerx}
def searchContentPage(self, key, quick, pg):
videos = []
page = self.parse_search_page(pg)
data = self.build_search_request_data(key, page)
response_data = self.fetch_search_data(data)
self.process_search_results(response_data, videos)
return self.build_search_result(videos, pg)
def parse_search_page(self, pg):
return int(pg) if pg else 1
def build_search_request_data(self, key, page):
return {'keywords': key,'type_id': 0,'page': page}
def fetch_search_data(self, data):
url = "http://110.42.67.221:8006/api.php/qijiappapi.index/searchList7"
response = requests.post(url=url, headers=headers, data=data)
response_data = response.json()
data = response_data['data']
data = self.decrypt_aes_cbc(data)
return json.loads(data)
def process_search_results(self, data, videos):
for vod in data['search_list']:
video = self.create_search_video_item(vod)
videos.append(video)
def create_search_video_item(self, vod):
name = vod['vod_name']
id = vod['vod_id']
pic = vod['vod_pic']
remark = vod.get('vod_remarks', '暂无备注')
return {"vod_id": id,"vod_name": name,"vod_pic": pic,"vod_remarks": remark}
def build_search_result(self, videos, pg):
return {'list': videos,'page': pg,'pagecount': 9999,'limit': 90,'total': 999999}
def searchContent(self, key, quick, pg="1"):
return self.searchContentPage(key, quick, '1')
def localProxy(self, params):
if params['type'] == "m3u8":
return self.proxyM3u8(params)
elif params['type'] == "media":
return self.proxyMedia(params)
elif params['type'] == "ts":
return self.proxyTs(params)
return None
+151
View File
@@ -0,0 +1,151 @@
# coding=utf-8
# !/usr/bin/python
import requests
from bs4 import BeautifulSoup
import re
from base.spider import Spider
import sys
import json
import os
import base64
sys.path.append('..')
xurl='https://panyq.com'
headerx = {
'User-Agent': 'Mozilla/5.0 (Windows NT 6.1; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/50.0.2661.87 Safari/537.36'
}
class Spider(Spider):
global xurl2
global xurl
global headerx
def getName(self):
return "首页"
def init(self, extend):
pass
def isVideoFormat(self, url):
pass
def manualVideoCheck(self):
pass
def homeContent(self, filter):
pass
return result
def homeVideoContent(self):
pass
def categoryContent(self, cid, pg, filter, ext):
pass
def detailContent(self, ids):
try:
data = json.loads(bytes.fromhex(ids[0]).decode())
verify = requests.post(f'{xurl}/search/{data["hash"]}',
headers=self.getheader(-1),
data=json.dumps(data['data'], separators=(",", ":")).encode(),
)
if verify.status_code == 200:
eid = data['data'][0]['eid']
rdata = json.dumps([{"eid": eid}], separators=(",", ":")).encode()
res = requests.post(f'{xurl}/go/{eid}', headers=self.getheader(1), data=rdata)
purl = json.loads(res.text.strip().split('\n')[-1].split(":", 1)[-1])['data']['link']
if not re.search(r'pwd=|码', purl) and data['password']:
purl = f"{purl}{'&' if '?' in purl else '?'}pwd={data['password']}"
print("获取盘链接为:", purl)
else:
raise Exception('验证失败')
vod = {
'vod_id': '',
'vod_name': '',
'vod_pic': '',
'type_name': '',
'vod_year': '',
'vod_area': '',
'vod_remarks': '',
'vod_actor': '',
'vod_director': '',
'vod_content': '',
'vod_play_from': '集多网盘',
'vod_play_url': purl
}
params = {
"do": "push",
"url": purl
}
response = requests.post("http://127.0.0.1:9978/action", data=params, headers={
"Content-Type": "application/x-www-form-urlencoded"
})
return {'list': [vod]}
except Exception as e:
print(e)
return {'list': []}
def playerContent(self, flag, id, vipFlags):
pass
def searchContentPage(self, key, quick, page='1'):
sign, sha, hash = self.getsign(key, page)
headers = self.getheader()
res = requests.get(f'{xurl}/api/search', params={'sign': sign}, headers=headers).json()
videos = []
for i in res['data']['hits']:
ccc = [{"eid": i.get("eid"), "sha": sha, "page_num": page}]
ddd = (json.dumps({'sign': sign, 'hash': hash, 'data': ccc, 'password': i.get('password')})).encode().hex()
if i.get('group')=='quark':
pic='https://android-artworks.25pp.com/fs08/2024/12/27/7/125_d45d9de77c805e17ede25e4a2d9d3444_con.png'
elif i.get('group')=='baidu':
pic='https://is4-ssl.mzstatic.com/image/thumb/Purple126/v4/dd/45/eb/dd45eb77-d21d-92f2-c46d-979797a6be4a/AppIcon-0-0-1x_U007emarketing-0-0-0-7-0-0-sRGB-0-0-0-GLES2_U002c0-512MB-85-220-0-0.png/1024x1024bb.jpg'
else:
pic='https://gimg2.baidu.com/image_search/src=http%3A%2F%2Fimg.alicdn.com%2Fbao%2Fuploaded%2Fi4%2F2213060290763%2FO1CN01joakK61HVUwob2JIJ_%21%212213060290763.jpg&refer=http%3A%2F%2Fimg.alicdn.com&app=2002&size=f9999,10000&q=a80&n=0&g=0n&fmt=auto?sec=1757745912&t=e7b98fced3a4f092c8ef26490997b004'
videos.append({
'vod_id': ddd,
'vod_name': i.get('desc').split('<mark>')[0].replace('<mark>', ""),
'vod_pic': pic,
'vod_remarks': i.get('group'),
})
return {'list': videos, 'page': page}
def searchContent(self, key, quick):
return self.searchContentPage(key, quick, '1')
def searchContent(self, key, quick, pg):
return self.searchContentPage(key, quick, pg)
def getsign(self,key,pg):
headers=self.getheader()
data=json.dumps([{"cat":"all","query":key,"pageNum":int(pg),"enableSearchMusic":False,"enableSearchGame":False,"enableSearchEbook":False}],separators=(",", ":"),ensure_ascii= False).encode()
res = requests.post(xurl, headers=headers, data=data).text
hash=re.search(r'"hash",\s*"([^"]+)"', res).group(1)
sign = re.search(r'"sign":\s*"([^"]+)"', res).group(1)
sha= re.search(r'"sha":\s*"([^"]+)"', res).group(1)
return sign,sha,hash
def getheader(self,k=0):
kes=['ecce0904d756da58b9ea5dd03da3cacea9fa29c6','4c5c1ef8a225004ce229e9afa4cc7189eed3e6fe','c4ed62e2b5a8e3212b334619f0cdbaa77fa842ff']
headers = {
'origin': xurl,
'referer': f'{xurl}/',
'next-action': kes[k],
'sec-ch-ua': '"Not/A)Brand";v="8", "Chromium";v="136", "Google Chrome";v="136"',
'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/136.0.7103.48 Safari/537.36',
}
return headers
def localProxy(self, params):
if params['type'] == "m3u8":
return self.proxyM3u8(params)
elif params['type'] == "media":
return self.proxyMedia(params)
elif params['type'] == "ts":
return self.proxyTs(params)
return None