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-06-29 13:15:54 +00:00
parent 81d9f40fa6
commit f4a0a125d7
49 changed files with 12094 additions and 13649 deletions
Download Patch File Download Diff File Expand all files Collapse all files
jaychouqq/tvbox/file.json
+1 -1
View File
File diff suppressed because one or more lines are too long
jaychouqq/tvbox/file1.json
+1 -1
View File
File diff suppressed because one or more lines are too long
jaychouqq/tvbox/file2.json
+1 -1
View File
File diff suppressed because one or more lines are too long
jaychouqq/tvbox/jaychouqq
+1 -1
View File
File diff suppressed because one or more lines are too long
jaychouqq/yingshi/py8/007吃瓜.py
+389
View File
@@ -0,0 +1,389 @@
import json
import re
import sys
import hashlib
from base64 import b64decode, b64encode
from urllib.parse import urlparse
import requests
from Crypto.Cipher import AES
from Crypto.Util.Padding import unpad
from pyquery import PyQuery as pq
sys.path.append('..')
from base.spider import Spider as BaseSpider
img_cache = {}
class Spider(BaseSpider):
def init(self, extend=""):
try:
self.proxies = json.loads(extend)
except:
self.proxies = {}
self.headers = {
'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.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',
'Connection': 'keep-alive',
'Cache-Control': 'no-cache',
}
self.host = self.get_working_host()
self.headers.update({'Origin': self.host, 'Referer': f"{self.host}/"})
print(f"使用站点: {self.host}")
def getName(self):
return "🌈 91吃瓜中心|终极完美版"
def isVideoFormat(self, url):
return any(ext in (url or '') for ext in ['.m3u8', '.mp4', '.ts'])
def manualVideoCheck(self):
return False
def destroy(self):
global img_cache
img_cache.clear()
def get_working_host(self):
dynamic_urls = [
'https://but.vncchqw.cc/'
]
for url in dynamic_urls:
try:
response = requests.get(url, headers=self.headers, proxies=self.proxies, timeout=10)
if response.status_code == 200:
return url
except Exception:
continue
return dynamic_urls[0]
def homeContent(self, filter):
try:
response = requests.get(self.host, headers=self.headers, proxies=self.proxies, timeout=15)
if response.status_code != 200: return {'class': [], 'list': []}
data = self.getpq(response.text)
classes = []
category_selectors = ['.category-list ul li', '.nav-menu li', '.menu li', 'nav ul li']
for selector in category_selectors:
for k in data(selector).items():
link = k('a')
href = (link.attr('href') or '').strip()
name = (link.text() or '').strip()
if not href or href == '#' or not name: continue
classes.append({'type_name': name, 'type_id': href})
if classes: break
if not classes:
classes = [{'type_name': '最新', 'type_id': '/latest/'}, {'type_name': '热门', 'type_id': '/hot/'}]
return {'class': classes, 'list': self.getlist(data('#index article, article'))}
except Exception as e:
return {'class': [], 'list': []}
def homeVideoContent(self):
try:
response = requests.get(self.host, headers=self.headers, proxies=self.proxies, timeout=15)
if response.status_code != 200: return {'list': []}
data = self.getpq(response.text)
return {'list': self.getlist(data('#index article, article'))}
except Exception as e:
return {'list': []}
def categoryContent(self, tid, pg, filter, extend):
try:
if '@folder' in tid:
v = self.getfod(tid.replace('@folder', ''))
return {'list': v, 'page': 1, 'pagecount': 1, 'limit': 90, 'total': len(v)}
pg = int(pg) if pg else 1
if tid.startswith('http'):
base_url = tid.rstrip('/')
else:
path = tid if tid.startswith('/') else f"/{tid}"
base_url = f"{self.host}{path}".rstrip('/')
if pg == 1:
url = f"{base_url}/"
else:
url = f"{base_url}/{pg}/"
response = requests.get(url, headers=self.headers, proxies=self.proxies, timeout=15)
if response.status_code != 200: return {'list': [], 'page': pg, 'pagecount': 9999, 'limit': 90, 'total': 0}
data = self.getpq(response.text)
videos = self.getlist(data('#archive article, #index article, article'), tid)
return {'list': videos, 'page': pg, 'pagecount': 9999, 'limit': 90, 'total': 999999}
except Exception as e:
return {'list': [], 'page': pg, 'pagecount': 9999, 'limit': 90, 'total': 0}
def detailContent(self, ids):
try:
url = ids[0] if ids[0].startswith('http') else f"{self.host}{ids[0]}"
response = requests.get(url, headers=self.headers, proxies=self.proxies, timeout=15)
data = self.getpq(response.text)
plist = []
used_names = set()
if data('.dplayer'):
for c, k in enumerate(data('.dplayer').items(), start=1):
try:
config_attr = k.attr('data-config')
if config_attr:
config = json.loads(config_attr)
video_url = config.get('video', {}).get('url', '')
if video_url:
ep_name = ''
parent = k.parents().eq(0)
for _ in range(4):
if not parent: break
heading = parent.find('h2, h3, h4').eq(0).text().strip()
if heading:
ep_name = heading
break
parent = parent.parents().eq(0)
base_name = ep_name if ep_name else f"视频{c}"
name = base_name
count = 2
while name in used_names:
name = f"{base_name} {count}"
count += 1
used_names.add(name)
plist.append(f"{name}${video_url}")
except: continue
if not plist:
content_area = data('.post-content, article')
for i, link in enumerate(content_area('a').items(), start=1):
link_text = link.text().strip()
link_href = link.attr('href')
if link_href and any(kw in link_text for kw in ['点击观看', '观看', '播放', '视频', '第一弹', '第二弹', '第三弹', '第四弹', '第五弹', '第六弹', '第七弹', '第八弹', '第九弹', '第十弹']):
ep_name = link_text.replace('点击观看：', '').replace('点击观看', '').strip()
if not ep_name: ep_name = f"视频{i}"
if not link_href.startswith('http'):
link_href = f"{self.host}{link_href}" if link_href.startswith('/') else f"{self.host}/{link_href}"
plist.append(f"{ep_name}${link_href}")
play_url = '#'.join(plist) if plist else f"未找到视频源${url}"
vod_content = ''
try:
tags = []
seen_names = set()
seen_ids = set()
tag_links = data('.tags a, .keywords a, .post-tags a')
candidates = []
for k in tag_links.items():
title = k.text().strip()
href = k.attr('href')
if title and href:
candidates.append({'name': title, 'id': href})
candidates.sort(key=lambda x: len(x['name']), reverse=True)
for item in candidates:
name = item['name']
id_ = item['id']
if id_ in seen_ids: continue
is_duplicate = False
for seen in seen_names:
if name in seen:
is_duplicate = True
break
if not is_duplicate:
target = json.dumps({'id': id_, 'name': name})
tags.append(f'[a=cr:{target}/]{name}[/a]')
seen_names.add(name)
seen_ids.add(id_)
if tags:
vod_content = ' '.join(tags)
else:
vod_content = data('.post-title').text()
except Exception:
vod_content = '获取标签失败'
if not vod_content:
vod_content = data('h1').text() or '91吃瓜中心'
return {'list': [{'vod_play_from': '91吃瓜中心', 'vod_play_url': play_url, 'vod_content': vod_content}]}
except:
return {'list': [{'vod_play_from': '91吃瓜中心', 'vod_play_url': '获取失败'}]}
def searchContent(self, key, quick, pg="1"):
try:
pg = int(pg) if pg else 1
if pg == 1:
url = f"{self.host}/search/{key}/"
else:
url = f"{self.host}/search/{key}/{pg}/"
response = requests.get(url, headers=self.headers, proxies=self.proxies, timeout=15)
return {'list': self.getlist(self.getpq(response.text)('article')), 'page': pg, 'pagecount': 9999}
except:
return {'list': [], 'page': pg, 'pagecount': 9999}
def playerContent(self, flag, id, vipFlags):
parse = 0 if self.isVideoFormat(id) else 1
url = self.proxy(id) if '.m3u8' in id else id
return {'parse': parse, 'url': url, 'header': self.headers}
def localProxy(self, param):
try:
type_ = param.get('type')
url = param.get('url')
if type_ == 'cache':
key = param.get('key')
if content := img_cache.get(key):
return [200, 'image/jpeg', content]
return [404, 'text/plain', b'Expired']
elif type_ == 'img':
real_url = self.d64(url) if not url.startswith('http') else url
res = requests.get(real_url, headers=self.headers, proxies=self.proxies, timeout=10)
content = self.aesimg(res.content)
return [200, 'image/jpeg', content]
elif type_ == 'm3u8':
return self.m3Proxy(url)
else:
return self.tsProxy(url)
except:
return [404, 'text/plain', b'']
def proxy(self, data, type='m3u8'):
if data and self.proxies: return f"{self.getProxyUrl()}&url={self.e64(data)}&type={type}"
return data
def m3Proxy(self, url):
url = self.d64(url)
res = requests.get(url, headers=self.headers, proxies=self.proxies)
data = res.text
base = res.url.rsplit('/', 1)[0]
lines = []
for line in data.split('\n'):
if '#EXT' not in line and line.strip():
if not line.startswith('http'):
line = f"{base}/{line}"
lines.append(self.proxy(line, 'ts'))
else:
lines.append(line)
return [200, "application/vnd.apple.mpegurl", '\n'.join(lines)]
def tsProxy(self, url):
return [200, 'video/mp2t', requests.get(self.d64(url), headers=self.headers, proxies=self.proxies).content]
def e64(self, text):
return b64encode(str(text).encode()).decode()
def d64(self, text):
return b64decode(str(text).encode()).decode()
def aesimg(self, data):
if len(data) < 16: return data
keys = [(b'f5d965df75336270', b'97b60394abc2fbe1'), (b'75336270f5d965df', b'abc2fbe197b60394')]
for k, v in keys:
try:
dec = unpad(AES.new(k, AES.MODE_CBC, v).decrypt(data), 16)
if dec.startswith(b'\xff\xd8') or dec.startswith(b'\x89PNG'): return dec
except: pass
try:
dec = unpad(AES.new(k, AES.MODE_ECB).decrypt(data), 16)
if dec.startswith(b'\xff\xd8'): return dec
except: pass
return data
def getlist(self, data, tid=''):
videos = []
is_folder = '/mrdg' in (tid or '')
for k in data.items():
card_html = k.outer_html() if hasattr(k, 'outer_html') else str(k)
a = k if k.is_('a') else k('a').eq(0)
href = a.attr('href')
title = k('h2').text() or k('.entry-title').text() or k('.post-title').text()
if not title and k.is_('a'): title = k.text()
if href and title:
img = self.getimg(k('script').text(), k, card_html)
videos.append({
'vod_id': f"{href}{'@folder' if is_folder else ''}",
'vod_name': title.strip(),
'vod_pic': img,
'vod_remarks': k('time').text() or '',
'vod_tag': 'folder' if is_folder else '',
'style': {"type": "rect", "ratio": 1.33}
})
return videos
def getfod(self, id):
url = f"{self.host}{id}"
data = self.getpq(requests.get(url, headers=self.headers, proxies=self.proxies).text)
videos = []
for i, h2 in enumerate(data('.post-content h2').items()):
p_txt = data('.post-content p').eq(i * 2)
p_img = data('.post-content p').eq(i * 2 + 1)
p_html = p_img.outer_html() if hasattr(p_img, 'outer_html') else str(p_img)
videos.append({
'vod_id': p_txt('a').attr('href'),
'vod_name': p_txt.text().strip(),
'vod_pic': self.getimg('', p_img, p_html),
'vod_remarks': h2.text().strip()
})
return videos
def getimg(self, text, elem=None, html_content=None):
if m := re.search(r"loadBannerDirect\('([^']+)'", text or ''):
return self._proc_url(m.group(1))
if html_content is None and elem is not None:
html_content = elem.outer_html() if hasattr(elem, 'outer_html') else str(elem)
if not html_content: return ''
html_content = html_content.replace('&quot;', '"').replace('&apos;', "'").replace('&amp;', '&')
if 'data:image' in html_content:
m = re.search(r'(data:image/[a-zA-Z0-9+/=;,]+)', html_content)
if m: return self._proc_url(m.group(1))
m = re.search(r'(https?://[^"\'\s)]+\.(?:jpg|png|jpeg|webp))', html_content, re.I)
if m: return self._proc_url(m.group(1))
if 'url(' in html_content:
m = re.search(r'url\s*\(\s*[\'"]?([^"\'\)]+)[\'"]?\s*\)', html_content, re.I)
if m: return self._proc_url(m.group(1))
return ''
def _proc_url(self, url):
if not url: return ''
url = url.strip('\'" ')
if url.startswith('data:'):
try:
_, b64_str = url.split(',', 1)
raw = b64decode(b64_str)
if not (raw.startswith(b'\xff\xd8') or raw.startswith(b'\x89PNG') or raw.startswith(b'GIF8')):
raw = self.aesimg(raw)
key = hashlib.md5(raw).hexdigest()
img_cache[key] = raw
return f"{self.getProxyUrl()}&type=cache&key={key}"
except: return ""
if not url.startswith('http'):
url = f"{self.host}{url}" if url.startswith('/') else f"{self.host}/{url}"
return f"{self.getProxyUrl()}&url={self.e64(url)}&type=img"
def getpq(self, data):
try: return pq(data)
except: return pq(data.encode('utf-8'))
jaychouqq/yingshi/py8/4K影院.py
+473
View File
@@ -0,0 +1,473 @@
# coding=utf-8
# !/usr/bin/python
"""
作者 丢丢喵推荐 🚓 内容均从互联网收集而来 仅供交流学习使用 版权归原创者所有 如侵犯了您的权益 请通知作者 将及时删除侵权内容
====================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
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 = "https://www.4kvm.net"
headerx = {
'User-Agent': 'Mozilla/5.0 (Linux; U; Android 8.0.0; zh-cn; Mi Note 2 Build/OPR1.170623.032) AppleWebKit/537.36 (KHTML, like Gecko) Version/4.0 Chrome/61.0.3163.128 Mobile Safari/537.36 XiaoMi/MiuiBrowser/10.1.1'
}
class Spider(Spider):
global xurl
global headerx
def getName(self):
return "首页"
def init(self, extend):
pass
def isVideoFormat(self, url):
pass
def manualVideoCheck(self):
pass
def extract_middle_text(self, text, start_str, end_str, pl, start_index1: str = '', end_index2: str = ''):
if pl == 3:
plx = []
while True:
start_index = text.find(start_str)
if start_index == -1:
break
end_index = text.find(end_str, start_index + len(start_str))
if end_index == -1:
break
middle_text = text[start_index + len(start_str):end_index]
plx.append(middle_text)
text = text.replace(start_str + middle_text + end_str, '')
if len(plx) > 0:
purl = ''
for i in range(len(plx)):
matches = re.findall(start_index1, plx[i])
output = ""
for match in matches:
match3 = re.search(r'(?:^|[^0-9])(\d+)(?:[^0-9]|$)', match[1])
if match3:
number = match3.group(1)
else:
number = 0
if 'http' not in match[0]:
output += f"#{match[1]}${number}{xurl}{match[0]}"
else:
output += f"#{match[1]}${number}{match[0]}"
output = output[1:]
purl = purl + output + "$$$"
purl = purl[:-3]
return purl
else:
return ""
else:
start_index = text.find(start_str)
if start_index == -1:
return ""
end_index = text.find(end_str, start_index + len(start_str))
if end_index == -1:
return ""
if pl == 0:
middle_text = text[start_index + len(start_str):end_index]
return middle_text.replace("\\", "")
if pl == 1:
middle_text = text[start_index + len(start_str):end_index]
matches = re.findall(start_index1, middle_text)
if matches:
jg = ' '.join(matches)
return jg
if pl == 2:
middle_text = text[start_index + len(start_str):end_index]
matches = re.findall(start_index1, middle_text)
if matches:
new_list = [f'{item}' for item in matches]
jg = '$$$'.join(new_list)
return jg
def homeContent(self, filter):
result = {"class": []}
detail = requests.get(url=xurl, headers=headerx)
detail.encoding = "utf-8"
res = detail.text
doc = BeautifulSoup(res, "lxml")
soups = doc.find_all('ul', class_="main-header")
for soup in soups:
vods = soup.find_all('li')
for vod in vods:
name = vod.text.strip()
if any(keyword in name for keyword in ["首页", "电视剧", "高分电影", "影片下载", "热门播放"]):
continue
id = vod.find('a')['href']
if 'http' not in id:
id = xurl + id
result["class"].append({"type_id": id, "type_name": name})
return result
def homeVideoContent(self):
videos = []
detail = requests.get(url=xurl, headers=headerx)
detail.encoding = "utf-8"
res = detail.text
doc = BeautifulSoup(res, "lxml")
soups = doc.find_all('article', class_="item movies")
for vod in soups:
name = vod.find('img')['alt']
ids = vod.find('div', class_="poster")
id = ids.find('a')['href']
pic = vod.find('img')['src']
remarks = vod.find('div', class_="rating")
remark = remarks.text.strip()
video = {
"vod_id": id,
"vod_name": name,
"vod_pic": pic,
"vod_remarks": remark
}
videos.append(video)
result = {'list': videos}
return result
def categoryContent(self, cid, pg, filter, ext):
result = {}
videos = []
if 'movies' not in cid:
if '@' in cid:
fenge = cid.split("@")
detail = requests.get(url=fenge[0], headers=headerx)
detail.encoding = "utf-8"
res = detail.text
doc = BeautifulSoup(res, "lxml")
soups = doc.find_all('div', class_="se-c")
for vod in soups:
name = vod.text.strip()
id = vod.find('a')['href']
pic = self.extract_middle_text(str(res), '<meta property="og:image" content="', '"', 0).replace('#038;', '')
remark = "推荐"
video = {
"vod_id": id,
"vod_name": name,
"vod_pic": pic,
"vod_remarks": remark
}
videos.append(video)
else:
if pg:
page = int(pg)
else:
page = 1
url = f'{cid}/page/{str(page)}'
detail = requests.get(url=url, headers=headerx)
detail.encoding = "utf-8"
res = detail.text
doc = BeautifulSoup(res, "lxml")
soups = doc.find_all('article', class_="item tvshows")
for vod in soups:
name = vod.find('img')['alt']
ids = vod.find('div', class_="poster")
id = ids.find('a')['href']
pic = vod.find('img')['src']
remarks = vod.find('div', class_="update")
remark = remarks.text.strip()
video = {
"vod_id": id + '@' + name,
"vod_name": name,
"vod_pic": pic,
"vod_tag": "folder",
"vod_remarks": remark
}
videos.append(video)
else:
if pg:
page = int(pg)
else:
page = 1
url = f'{cid}/page/{str(page)}'
detail = requests.get(url=url, headers=headerx)
detail.encoding = "utf-8"
res = detail.text
doc = BeautifulSoup(res, "lxml")
soups = doc.find_all('div', class_="animation-2")
for item in soups:
vods = item.find_all('article')
for vod in vods:
name = vod.find('img')['alt']
ids = vod.find('div', class_="poster")
id = ids.find('a')['href']
pic = vod.find('img')['src']
remarks = vod.find('div', class_="rating")
remark = remarks.text.strip()
video = {
"vod_id": id,
"vod_name": name,
"vod_pic": pic,
"vod_remarks": remark
}
videos.append(video)
if len(videos)<30:
pagecount=1
else:
pagecount = 9999
result = {'list': videos}
result['page'] = pg
result['pagecount'] = pagecount
result['total'] = 999
result['limit'] = len(videos)
return result
def detailContent(self, ids):
did = ids[0]
result = {}
videos = []
xianlu = ''
bofang = ''
if 'movies' not in did:
res = requests.get(url=did, headers=headerx)
res.encoding = "utf-8"
res = res.text
doc = BeautifulSoup(res, "lxml")
content = '剧情介绍📢' + self.extract_middle_text(res,'<meta name="description" content="','"', 0)
postid = self.extract_middle_text(res, 'postid:', ',', 0)
res1 = self.extract_middle_text(res,'videourls:[','],', 0)
data = json.loads(res1)
for vod in data:
name = str(vod['name'])
id = f"{vod['url']}@{postid}"
bofang = bofang + name + '$' + id + '#'
bofang = bofang[:-1]
xianlu = '4K影院'
else:
res = requests.get(url=did, headers=headerx)
res.encoding = "utf-8"
res = res.text
doc = BeautifulSoup(res, "lxml")
content = '剧情介绍📢' + self.extract_middle_text(res, '<meta name="description" content="', '"', 0)
bofang = self.extract_middle_text(res, "data-postid='", "'", 0)
xianlu = '4K影院'
videos.append({
"vod_id": did,
"vod_content": content,
"vod_play_from": xianlu,
"vod_play_url": bofang
})
result['list'] = videos
return result
def playerContent(self, flag, id, vipFlags):
if '@' in id:
fenge = id.split("@")
url = f'{xurl}/artplayer?id={fenge[1]}&source=0&ep={fenge[0]}'
detail = requests.get(url=url, headers=headerx)
detail.encoding = "utf-8"
res = detail.text
expires = self.extract_middle_text(res, "expires: '", "'", 0)
client = self.extract_middle_text(res, "client: '", "'", 0)
nonce = self.extract_middle_text(res, "nonce: '", "'", 0)
token = self.extract_middle_text(res, "token: '", "'", 0)
source = self.extract_middle_text(res, "source: '", "'", 0)
payload = {
"expires": expires,
"client": client,
"nonce": nonce,
"token": token,
"source": source
}
response = requests.post(url=source, headers=headerx, json=payload)
response_data = json.loads(response.text)
url = response_data['url']
else:
url = f'{xurl}/artplayer?mvsource=0&id={id}&type=hls'
detail = requests.get(url=url, headers=headerx)
detail.encoding = "utf-8"
res = detail.text
expires = self.extract_middle_text(res, "expires: '", "'", 0)
client = self.extract_middle_text(res, "client: '", "'", 0)
nonce = self.extract_middle_text(res, "nonce: '", "'", 0)
token = self.extract_middle_text(res, "token: '", "'", 0)
source = self.extract_middle_text(res, "source: '", "'", 0)
payload = {
"expires": expires,
"client": client,
"nonce": nonce,
"token": token,
"source": source
}
response = requests.post(url=source, headers=headerx, json=payload)
response_data = json.loads(response.text)
url = response_data['url']
result = {}
result["parse"] = 0
result["playUrl"] = ''
result["url"] = url
result["header"] = headerx
return result
def searchContentPage(self, key, quick, pg):
result = {}
videos = []
url = f'{xurl}/xssearch?s={key}'
detail = requests.get(url=url, headers=headerx)
detail.encoding = "utf-8"
res = detail.text
doc = BeautifulSoup(res, "lxml")
soups = doc.find_all('div', class_="result-item")
for vod in soups:
ids = vod.find('div', class_="title")
id = ids.find('a')['href']
if 'movies' not in id:
name = vod.find('img')['alt']
pic = vod.find('img')['src']
remark = "推荐"
video = {
"vod_id": id + '@' + name,
"vod_name": name,
"vod_pic": pic,
"vod_tag": "folder",
"vod_remarks": remark
}
videos.append(video)
else:
name = vod.find('img')['alt']
pic = vod.find('img')['src']
remark = "推荐"
video = {
"vod_id": id,
"vod_name": name,
"vod_pic": pic,
"vod_remarks": remark
}
videos.append(video)
result['list'] = videos
result['page'] = pg
result['pagecount'] = 1
result['limit'] = 90
result['total'] = 999999
return result
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
jaychouqq/yingshi/py8/TOPTV.py
+124
View File
@@ -0,0 +1,124 @@
import sys
import re
import json
import requests
from requests.adapters import HTTPAdapter
from requests.packages.urllib3.util.retry import Retry
from base.spider import Spider
class Spider(Spider):
def getName(self):
return "TOPTV"
def init(self, extend=""):
super().init(extend)
self.site_url = "https://toptv15.cyou"
self.headers = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
"Referer": self.site_url
}
self.sess = requests.Session()
self.sess.mount("https://", HTTPAdapter(max_retries=Retry(total=3, backoff_factor=1)))
def fetch(self, url, timeout=10):
try:
res = self.sess.get(url, headers=self.headers, timeout=timeout, verify=False)
res.encoding = "utf-8"
return res
except:
return None
def homeContent(self, filter):
cate_list = [
{"type_name": "国产自拍", "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": "美乳巨乳", "type_id": "6"},
{"type_name": "强制侵犯", "type_id": "7"},
{"type_name": "制服诱惑", "type_id": "8"},
{"type_name": "绝色佳人", "type_id": "9"},
{"type_name": "家庭乱伦", "type_id": "10"},
{"type_name": "绝顶潮吹", "type_id": "11"},
{"type_name": "网红主播", "type_id": "12"}
]
return {"class": cate_list}
def categoryContent(self, tid, pg, filter, extend):
if not hasattr(self, 'site_url'): self.init()
pg = int(pg) if str(pg).isdigit() else 1
list_url = f"{self.site_url}/index.php/vod/type/id/{tid}/page/{pg}.html"
res = self.fetch(list_url)
video_list = []
if res:
pattern = r'href="(/index.php/vod/detail/id/(\d+).html)".*?data-original="(.*?)".*?vod-name.*?>(.*?)<'
matches = re.findall(pattern, res.text, re.S)
for href, v_id, pic, name in matches:
video_list.append({
"vod_id": v_id,
"vod_name": name.strip(),
"vod_pic": pic if pic.startswith("http") else self.site_url + pic,
"vod_remarks": ""
})
return {'list': video_list, 'page': pg, 'pagecount': 999, 'limit': 20, 'total': 9999}
def detailContent(self, ids):
if not hasattr(self, 'site_url'): self.init()
vod_id = ids[0]
res = self.fetch(f"{self.site_url}/index.php/vod/detail/id/{vod_id}.html")
if not res: return {}
html = res.text
name_match = re.search(r'vod-name.*?>(.*?)<', html) or re.search(r'title-box.*?>(.*?)<', html)
pic_match = re.search(r'detail-pic.*?src="(.*?)"', html) or re.search(r'data-original="(.*?)"', html)
play_matches = re.findall(r'href="(/index.php/vod/play/id/(\d+)/sid/(\d+)/nid/(\d+).html)">(.*?)<', html)
play_urls = []
for m in play_matches:
play_urls.append(f"{m[4]}${m[1]}-{m[2]}-{m[3]}")
if not play_urls:
play_urls.append(f"立即播放${vod_id}-1-1")
vod = {
"vod_id": vod_id,
"vod_name": name_match.group(1).strip() if name_match else "视频详情",
"vod_pic": pic_match.group(1) if pic_match else "",
"vod_play_from": "TOP-TV",
"vod_play_url": "#".join(play_urls)
}
return {"list": [vod]}
def playerContent(self, flag, id, vipFlags):
if not hasattr(self, 'site_url'): self.init()
parts = id.split('-')
if len(parts) == 3:
v_id, s_id, n_id = parts
play_url = f"{self.site_url}/index.php/vod/play/id/{v_id}/sid/{s_id}/nid/{n_id}.html"
else:
play_url = f"{self.site_url}/index.php/vod/play/id/{id}.html"
res = self.fetch(play_url)
if res:
data_json = re.search(r'var player_aaaa=(.*?)</script>', res.text)
if data_json:
try:
url = json.loads(data_json.group(1)).get("url", "")
return {"parse": 0, "url": url, "header": self.headers}
except:
pass
return {"parse": 1, "url": play_url}
def searchContent(self, key, quick, pg=1):
if not hasattr(self, 'site_url'): self.init()
res = self.fetch(f"{self.site_url}/index.php/vod/search/page/{pg}/wd/{key}.html")
video_list = []
if res:
pattern = r'href="(/index.php/vod/detail/id/(\d+).html)".*?data-original="(.*?)".*?vod-name.*?>(.*?)<'
matches = re.findall(pattern, res.text, re.S)
for href, v_id, pic, name in matches:
video_list.append({
"vod_id": v_id,
"vod_name": name.strip(),
"vod_pic": pic if pic.startswith("http") else self.site_url + pic
})
return {"list": video_list}
jaychouqq/yingshi/py8/av (1).py
+259
View File
@@ -0,0 +1,259 @@
# -*- coding: utf-8 -*-
# 123AV短视频 - Fongmi影视App适配爬虫
# 优化为短视频模式，支持滑动切换
import sys
import re
import json
import urllib.parse
sys.path.append('..')
from base.spider import Spider
class Spider(Spider):
def getName(self):
return "123AV"
def init(self, extend=''):
self.home_url = 'https://123av.fun'
self.ua = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/126.0.0.0 Safari/537.36"
def getDependence(self):
return []
def isVideoFormat(self, url):
return False
def manualVideoCheck(self):
return False
def homeContent(self, filter):
return {
'class': [
{'type_id': 'publish-time/sort-desc', 'type_name': '最新发布'},
{'type_id': 'view-count/sort-desc', 'type_name': '最多播放'},
{'type_id': 'comment-count/sort-desc', 'type_name': '最多评论'},
{'type_id': 'favorite-count/sort-desc', 'type_name': '最多收藏'},
{'type_id': 'explore', 'type_name': '探索发现'},
{'type_id': 'list', 'type_name': '排行榜'},
],
'filters': {}
}
def homeVideoContent(self):
return self.categoryContent('publish-time/sort-desc', 1, {}, {})
def _fetch_html(self, url):
try:
rsp = self.fetch(url, headers={
"User-Agent": self.ua,
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
"Accept-Language": "zh-CN,zh;q=0.9",
}, timeout=15)
if rsp and hasattr(rsp, 'text') and rsp.text:
return rsp.text
except Exception as e:
print(f'fetch error: {e}')
return ''
def _extract_video_list(self, html):
videos = []
if not html:
return videos
# 匹配视频卡片
card_pattern = re.compile(
r'<a\s+([^>]*data-src="https://static\.123av\.fun/[^"]+\.m3u8"[^>]*)>(.*?)</a>',
re.S
)
cards = card_pattern.findall(html)
for attrs, content in cards:
try:
src_match = re.search(r'data-src="(https://static\.123av\.fun/[^"]+\.m3u8)"', attrs)
poster_match = re.search(r'data-poster="([^"]*)"', attrs)
id_match = re.search(r'data-id="(\d+)"', attrs)
dur_match = re.search(r'data-duration="(\d+)"', attrs)
title_match = re.search(r'<xwya-video[^>]*alt="([^"]*)"', content)
if src_match and id_match:
m3u8_url = src_match.group(1)
vid = id_match.group(1)
poster = poster_match.group(1) if poster_match else ''
duration = dur_match.group(1) if dur_match else '0'
title = title_match.group(1).strip() if title_match else f'视频{vid}'
dur = int(duration)
if dur >= 3600:
duration_str = f'{dur // 3600}:{(dur % 3600) // 60:02d}:{dur % 60:02d}'
else:
duration_str = f'{dur // 60:02d}:{dur % 60:02d}'
videos.append({
'vod_id': vid,
'vod_name': title,
'vod_pic': poster,
'vod_remarks': duration_str,
})
except Exception as e:
continue
return videos
def categoryContent(self, tid, page, filter, ext):
video_list = []
if tid in ('explore', 'list', 'subscribe'):
url = f'{self.home_url}/{tid}/page-{page}'
else:
url = f'{self.home_url}/{tid}/page-{page}'
html = self._fetch_html(url)
video_list = self._extract_video_list(html)
return {
'list': video_list,
'page': int(page),
'pagecount': 999,
'limit': 20,
'total': 999 * 20
}
def detailContent(self, did):
"""视频详情 - 关键修改：返回播放URL让playerContent处理"""
video_list = []
try:
vid = did[0]
detail_url = f'{self.home_url}/detail/{vid}'
html = self._fetch_html(detail_url)
if html:
src_match = re.search(r'data-src="(https://static\.123av\.fun/[^"]+\.m3u8)"', html)
poster_match = re.search(r'data-poster="([^"]*)"', html)
title_match = re.search(r'<h1[^>]*>([^<]+)</h1>', html)
if not title_match:
title_match = re.search(r'property="og:title"\s+content="([^"]*)"', html)
if not title_match:
title_match = re.search(r'<xwya-video[^>]*alt="([^"]*)"', html)
desc_match = re.search(r'property="og:description"\s+content="([^"]*)"', html)
dur_match = re.search(r'data-duration="(\d+)"', html)
m3u8_url = src_match.group(1) if src_match else ''
vod_pic = poster_match.group(1) if poster_match else ''
vod_name = title_match.group(1).strip() if title_match else ''
vod_content = desc_match.group(1) if desc_match else ''
duration_str = ''
if dur_match:
dur = int(dur_match.group(1))
if dur >= 3600:
duration_str = f'{dur // 3600}:{(dur % 3600) // 60:02d}:{dur % 60:02d}'
else:
duration_str = f'{dur // 60:02d}:{dur % 60:02d}'
else:
m3u8_url = ''
vod_pic = ''
vod_name = ''
vod_content = ''
duration_str = ''
# 关键修改：如果直接有m3u8，放入播放URL
# 使用特殊格式让Fongmi识别为短视频
if m3u8_url:
# 格式: 集数名称$url#集数名称$url
vod_play_url = f'正片${m3u8_url}'
else:
vod_play_url = ''
video_list.append({
'vod_id': vid,
'vod_name': vod_name,
'vod_pic': vod_pic,
'vod_remarks': duration_str,
'vod_content': vod_content,
'vod_play_from': '短视频', # 改为短视频，可能触发滑动模式
'vod_play_url': vod_play_url,
'type_name': '短视频',
'vod_year': '',
'vod_area': '',
'vod_director': '',
'vod_actor': '',
})
except Exception as e:
print(f'detailContent error: {e}')
return {
'list': video_list,
'parse': 0,
'jx': 0
}
def searchContent(self, key, quick, page='1'):
video_list = []
try:
encoded_key = urllib.parse.quote(key)
url = f'{self.home_url}/search/{encoded_key}/page-{page}'
html = self._fetch_html(url)
video_list = self._extract_video_list(html)
except Exception as e:
print(f'searchContent error: {e}')
return {
'list': video_list,
'page': int(page),
'pagecount': 99,
'limit': 20,
'total': 99 * 20
}
def playerContent(self, flag, pid, vipFlags):
"""播放器内容 - 关键修改"""
# 如果pid已经是m3u8地址，直接返回
if pid.startswith('http') and '.m3u8' in pid:
return {
'parse': 0, # 直接播放
'url': pid,
'header': {
'User-Agent': self.ua,
'Referer': self.home_url + '/'
}
}
# 如果是详情页URL，获取m3u8
if '/detail/' in pid:
html = self._fetch_html(pid)
if html:
src_match = re.search(r'data-src="(https://static\.123av\.fun/[^"]+\.m3u8)"', html)
if src_match:
return {
'parse': 0,
'url': src_match.group(1),
'header': {
'User-Agent': self.ua,
'Referer': self.home_url + '/'
}
}
# 默认返回，让外部解析
return {
'parse': 1,
'url': pid,
'header': {
'User-Agent': self.ua,
'Referer': self.home_url + '/'
}
}
def localProxy(self, params):
return {}
def destroy(self):
return '正在Destroy'
if __name__ == '__main__':
pass
jaychouqq/yingshi/py8/dy.py
+470
View File
@@ -0,0 +1,470 @@
# -*- coding: utf-8 -*-
import re
import sys
import json
import time
import random
import string
sys.path.append('..')
from base.spider import Spider
class Spider(Spider):
def init(self, extend=""):
self.extend = extend
self.cookie_cache = ""
def getName(self):
return "抖音直播"
def isVideoFormat(self, url):
return False
def manualVideoCheck(self):
return False
def destroy(self):
pass
def localProxy(self, param):
return None
host = "https://live.douyin.com"
ua = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36"
headers = {
"User-Agent": ua,
"Accept": "application/json, text/plain, */*",
"Accept-Language": "zh-CN,zh;q=0.9",
}
classes_config = [
{"type_id": "10000$3", "type_name": "娱乐天地"},
{"type_id": "10001$3", "type_name": "科技文化"},
{"type_id": "102$4", "type_name": "音乐"},
{"type_id": "103$4", "type_name": "游戏"},
{"type_id": "105$4", "type_name": "舞蹈"},
{"type_id": "101$4", "type_name": "聊天"},
{"type_id": "108$4", "type_name": "运动"},
{"type_id": "107$4", "type_name": "生活"},
{"type_id": "106$4", "type_name": "文化"},
{"type_id": "104$4", "type_name": "二次元"},
]
# ==================== 工具函数 ====================
def _generate_device_id(self):
timestamp = self._base36_encode(int(time.time() * 1000))
random_part = ''.join(random.choices(string.ascii_lowercase + string.digits, k=13))
return f"{timestamp}{random_part}"
@staticmethod
def _base36_encode(num):
alphabet = '0123456789abcdefghijklmnopqrstuvwxyz'
if num == 0:
return '0'
res = []
while num > 0:
num, rem = divmod(num, 36)
res.append(alphabet[rem])
return ''.join(reversed(res))
def _get_cookie(self):
if self.cookie_cache:
return self.cookie_cache
try:
resp = self.fetch(self.host, headers=self.headers, verify=False)
cookies = resp.headers.get('set-cookie', '')
if cookies:
match = re.search(r'ttwid=([^;]+)', cookies)
if match:
self.cookie_cache = f"ttwid={match.group(1)}"
except Exception:
pass
return self.cookie_cache
def _get_headers(self):
cookie = self._get_cookie()
hd = self.headers.copy()
hd["Referer"] = self.host
if cookie:
hd["Cookie"] = cookie
return hd
def _first_non_empty(self, *values):
for v in values:
if v is not None and v != '':
return v
return None
def _parse_raw_live_data(self, item):
if not isinstance(item, dict):
return None
candidates = [
item.get('lives', {}).get('rawdata'),
item.get('lives', {}).get('raw_data'),
item.get('live', {}).get('rawdata'),
item.get('live_info', {}).get('rawdata'),
item.get('aweme_info', {}).get('live_info', {}).get('rawdata'),
item.get('data', {}).get('rawdata'),
item.get('rawdata'),
item.get('lives'),
item.get('live'),
item.get('live_info'),
item.get('aweme_info', {}).get('live_info'),
item.get('aweme_info'),
item.get('data'),
item
]
for c in candidates:
if isinstance(c, str):
try:
parsed = json.loads(c)
if isinstance(parsed, dict):
return parsed
except Exception:
continue
elif isinstance(c, dict):
return c
return None
def _normalize_search_item(self, raw, fallback=None):
if not isinstance(raw, dict):
return None
fallback = fallback or {}
room_id = self._first_non_empty(
raw.get('id_str'),
raw.get('room_id_str'),
raw.get('room', {}).get('id_str'),
raw.get('room', {}).get('id'),
raw.get('room_id'),
raw.get('roomId')
)
if not room_id:
return None
web_rid = self._first_non_empty(
raw.get('owner', {}).get('web_rid'),
raw.get('web_rid'),
raw.get('room', {}).get('owner', {}).get('web_rid')
) or self._generate_device_id()
nickname = self._first_non_empty(
raw.get('owner', {}).get('nickname'),
raw.get('nickname'),
raw.get('room', {}).get('owner', {}).get('nickname'),
fallback.get('nickname')
) or '抖音直播'
title = self._first_non_empty(
raw.get('title'),
raw.get('room', {}).get('title'),
fallback.get('title'),
nickname
)
pic = self._first_non_empty(
raw.get('owner', {}).get('avatar_large', {}).get('url_list', [None])[0],
raw.get('room', {}).get('cover', {}).get('url_list', [None])[0],
raw.get('cover', {}).get('url_list', [None])[0],
raw.get('cover_url')
) or ''
online_text = self._first_non_empty(
raw.get('room', {}).get('stats', {}).get('user_count_str'),
raw.get('user_count_str'),
raw.get('room', {}).get('user_count_str'),
raw.get('user_count')
)
tag_text = self._first_non_empty(
raw.get('video_feed_tag'),
raw.get('room', {}).get('partition_road_map', [{}])[0].get('title'),
raw.get('partition', {}).get('title'),
fallback.get('tag')
)
remark = ' '.join(filter(None, [tag_text, online_text]))
return {
"vod_id": f"{web_rid}@@{room_id}",
"vod_name": nickname,
"vod_pic": pic,
"vod_remarks": remark,
"vod_content": title
}
def _extract_search_videos(self, payload):
if not isinstance(payload, list):
return []
results = []
seen = set()
for item in payload:
raw = self._parse_raw_live_data(item)
norm = self._normalize_search_item(raw, {
"nickname": item.get('nickname'),
"title": item.get('title') or item.get('desc'),
"tag": item.get('search_keyword')
})
if not norm:
continue
if norm['vod_id'] in seen:
continue
seen.add(norm['vod_id'])
results.append(norm)
return results
# ==================== 框架标准方法 ====================
def homeContent(self, filter):
classes = self.classes_config
return {
"class": classes,
"list": []
}
def homeVideoContent(self):
return {}
def categoryContent(self, tid, pg, filter, extend):
category_id = str(tid)
page = int(pg or 1)
offset = 15 * (page - 1)
parts = category_id.split('$')
if len(parts) < 2:
return {"list": [], "page": page, "pagecount": 0, "limit": 15, "total": 9999}
partition, ptype = parts[0], parts[1]
params = {
"aid": "6383",
"app_name": "douyin_web",
"live_id": "1",
"device_platform": "web",
"language": "zh-CN",
"browser_language": "zh-CN",
"browser_platform": "Win32",
"browser_name": "Chrome",
"browser_version": "120.0.0.0",
"partition": partition,
"partition_type": ptype,
"count": "15",
"offset": str(offset),
"web_rid": self._generate_device_id(),
"cookie_enabled": "true",
"screen_width": "1920",
"screen_height": "1080"
}
headers = self._get_headers()
urls = [
"https://live.douyin.com/webcast/web/partition/detail/room/v2/",
"https://webcast.amemv.com/webcast/web/partition/detail/room/v2/",
]
list_ = []
for url in urls:
try:
resp = self.fetch(url, headers=headers, params=params, verify=False)
data = resp.json()
if data.get('status_code') != 0:
continue
if not data.get('data', {}).get('data'):
break
items = data['data']['data']
for it in items:
web_rid = it.get('web_rid') or self._generate_device_id()
room = it['room']
list_.append({
"vod_id": f"{web_rid}@@{room['id_str']}",
"vod_name": room['title'],
"vod_pic": room['cover']['url_list'][0],
"vod_remarks": f"{room['owner']['nickname']} (🔥{room['stats']['user_count_str']})"
})
break
except Exception:
continue
return {
"list": list_,
"page": page,
"pagecount": 9999,
"limit": 15,
"total": 999999
}
def searchContent(self, key, quick, pg="1"):
kw = key.strip()
if not kw:
return {"list": [], "page": 1}
page = 1
offset = 0
headers = self._get_headers()
# 策略1 专用直播搜索
try:
params1 = {
"device_platform": "webapp",
"aid": "6383",
"channel": "channel_pc_web",
"search_channel": "aweme_live",
"search_source": "switch_tab",
"query_correct_type": "1",
"need_filter_settings": "1",
"list_type": "single",
"keyword": kw,
"offset": str(offset),
"count": "20",
"os_version": "10"
}
r = self.fetch("https://www.douyin.com/aweme/v1/web/live/search/",
params=params1, headers=headers, verify=False)
data = r.json()
list_ = self._extract_search_videos(data.get('data'))
if list_:
return {"list": list_, "page": 1}
except Exception:
pass
# 策略2 通用搜索
try:
params2 = {
"device_platform": "webapp",
"aid": "6383",
"channel": "channel_pc_web",
"search_channel": "aweme_live",
"keyword": kw,
"offset": str(offset),
"count": "20",
"os_version": "10"
}
r = self.fetch("https://www.douyin.com/aweme/v1/web/general/search/stream/",
params=params2, headers=headers, verify=False)
data = r.json()
list_ = self._extract_search_videos(data.get('data'))
if list_:
return {"list": list_, "page": 1}
except Exception:
pass
# 降级分区搜索
try:
part_url = f"https://live.douyin.com/webcast/web/partition/search/?keyword={kw}&aid=6383"
r = self.fetch(part_url, headers=self._get_headers(), verify=False)
data = r.json()
partitions = data.get('data', {}).get('SearchResult', [])
if not partitions:
return {"list": [], "page": 1}
merged = []
seen = set()
for i in range(min(3, len(partitions))):
part = partitions[i].get('partition', {})
p_id = part.get('id_str')
p_type = part.get('type')
if not p_id or p_type is None:
continue
cate_ret = self.categoryContent(f"{p_id}${p_type}", 1, None, None)
for item in cate_ret.get('list', []):
if item['vod_id'] in seen:
continue
seen.add(item['vod_id'])
item['vod_remarks'] = item['vod_remarks'] or part.get('title', kw)
merged.append(item)
if len(merged) >= 20:
break
if len(merged) >= 20:
break
return {"list": merged, "page": 1}
except Exception:
pass
return {"list": [], "page": 1}
def detailContent(self, ids):
if not ids:
return {"list": []}
raw_id = ids[0]
parts = raw_id.split('@@')
if len(parts) != 2:
return {"list": []}
web_rid, room_id = parts[0], parts[1]
url = "https://live.douyin.com/webcast/room/web/enter/"
params = {
"aid": "6383",
"app_name": "douyin_web",
"live_id": "1",
"device_platform": "web",
"enter_from": "web_live",
"browser_language": "zh-CN",
"browser_platform": "Win32",
"browser_name": "Chrome",
"browser_version": "120.0.0.0",
"web_rid": web_rid,
"room_id_str": room_id,
"enter_source": "",
"is_need_double_stream": "false"
}
headers = self._get_headers()
try:
r = self.fetch(url, params=params, headers=headers, verify=False)
data = r.json()
if not data.get('data', {}).get('data'):
return {"list": []}
info = data['data']['data'][0]
resolution_map = {
"FULL_HD1": "蓝光",
"HD1": "超清",
"ORIGION": "原画",
"SD1": "标清",
"SD2": "高清"
}
flv_pull = info.get('stream_url', {}).get('flv_pull_url', {})
flv_episodes = []
for k, v in flv_pull.items():
name = resolution_map.get(k, k)
flv_episodes.append(f"{name}${v}")
hls_pull = info.get('stream_url', {}).get('hls_pull_url_map', {})
hls_episodes = []
for k, v in hls_pull.items():
name = resolution_map.get(k, k)
hls_episodes.append(f"{name}${v}")
vod_play_from = ""
vod_play_url = ""
if flv_episodes:
vod_play_from += "FLV$$$"
vod_play_url += "#".join(flv_episodes) + "$$$"
if hls_episodes:
vod_play_from += "HLS"
vod_play_url += "#".join(hls_episodes)
vod_play_from = vod_play_from.rstrip("$$$")
vod_play_url = vod_play_url.rstrip("$$$")
vod = {
"vod_id": raw_id,
"vod_name": info['title'],
"vod_pic": info['cover']['url_list'][0],
"vod_actor": info['owner']['nickname'],
"vod_content": "【天神IY】"+info['title'],
"vod_play_from": vod_play_from,
"vod_play_url": vod_play_url
}
return {"list": [vod]}
except Exception:
return {"list": []}
def playerContent(self, flag, id, vipFlags):
if not id:
return {"parse": 0, "url": "", "header": self.headers}
return {
"parse": 0,
"url": id,
"header": {
"User-Agent": self.ua,
"Referer": self.host
}
}
jaychouqq/yingshi/py8/kw.py
+396
View File
@@ -0,0 +1,396 @@
from base.spider import Spider
import requests
import json
import re
import sys
import base64
from urllib.parse import quote
class Spider(Spider):
def getName(self):
return "小心儿悠悠"
def init(self, extend=""):
pass
def isVideoFormat(self, url):
pass
def manualVideoCheck(self):
pass
def homeContent(self, filter):
result = {}
cateId = [
{"type_name": "华语男", "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": "日韩组合", "type_id": "6"},
{"type_name": "欧美男", "type_id": "7"},
{"type_name": "欧美女", "type_id": "8"},
{"type_name": "欧美组合", "type_id": "9"},
{"type_name": "其他", "type_id": "0"}
]
result['class'] = cateId
return result
def homeVideoContent(self):
result = self.categoryContent("1", 1, False, {})
return result
def categoryContent(self, tid, pg, filter, extend):
result = {}
url = f"http://wapi.kuwo.cn/api/www/artist/artistInfo?category={tid}&prefix=&pn={pg}&rn=30"
headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36',
'Referer': 'http://www.kuwo.cn/'
}
try:
r = requests.get(url, headers=headers, timeout=10)
data = r.json()
videos = []
if data.get('data') and data['data'].get('artistList'):
for item in data['data']['artistList']:
video = {
"vod_id": str(item.get('id', '')),
"vod_name": item.get('name', ''),
"vod_pic": item.get('pic300') or item.get('pic') or item.get('pic120', ''),
"vod_remarks": f""
}
videos.append(video)
result['list'] = videos
result['page'] = pg
result['pagecount'] = 9999
result['limit'] = 90
result['total'] = 999999
except Exception as e:
result['list'] = []
return result
def detailContent(self, ids):
rid = ids[0]
result = {}
info_url = f"http://wapi.kuwo.cn/api/www/artist/artist?artistid={rid}"
headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36',
'Referer': 'http://www.kuwo.cn/'
}
try:
r = requests.get(info_url, headers=headers, timeout=10)
info_data = r.json().get('data', {})
artist_name = info_data.get('name', '')
all_songs = self._get_artist_songs(rid)
artist_info = info_data.get('info', '')
artist_info = re.sub(r'<[^>]+>', '', artist_info)
artist_info = artist_info.replace('&nbsp;', ' ')
artist_info = artist_info.replace('\r\n', '\n').replace('\r', '\n')
artist_info = artist_info.strip()
max_songs = 300
if len(all_songs) > max_songs:
all_songs = all_songs[:max_songs]
play_arr = []
for i, song in enumerate(all_songs):
name = re.sub(r'[$#]', '', song.get('name', '')).strip()
song_id = song.get('rid', '')
album = song.get('album', '')
if album:
play_arr.append(f"{name} - {album}${song_id}")
else:
play_arr.append(f"{name}${song_id}")
vod = {
"vod_id": rid,
"vod_name": artist_name,
"vod_pic": info_data.get('pic300') or info_data.get('pic', ''),
"vod_content": "【天神IY】"+artist_info if artist_info else "暂无歌手简介",
"vod_remarks": f"歌曲 : {len(all_songs)}首",
"vod_actor": artist_name,
"vod_play_from": "酷我音乐",
"vod_play_url": "#".join(play_arr)
}
result['list'] = [vod]
except Exception as e:
vod = {
"vod_id": rid,
"vod_name": "加载失败",
"vod_content": f"加载歌手信息失败: {str(e)}",
"vod_remarks": "加载失败",
"vod_actor": "未知",
"vod_play_from": "酷我音乐",
"vod_play_url": ""
}
result['list'] = [vod]
return result
def _get_artist_songs(self, rid):
headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36',
'Referer': 'http://www.kuwo.cn/'
}
songs = []
max_pages = 10
for page in range(1, max_pages + 1):
try:
url = f"http://wapi.kuwo.cn/api/www/artist/artistMusic?artistid={rid}&pn={page}&rn=30"
response = requests.get(url, headers=headers, timeout=10)
data = response.json()
if data.get('code') == 200:
music_data = data.get('data', {})
song_list = music_data.get('list', [])
if not song_list:
break
for song in song_list:
song_name = song.get('name', '').strip()
if song_name:
songs.append({
'name': song_name,
'rid': song.get('rid', ''),
'album': song.get('album', ''),
'duration': song.get('duration', '')
})
if len(songs) >= 300:
songs = songs[:300]
break
except Exception:
continue
return songs
def playerContent(self, flag, id, vipFlags):
result = {}
rid = id
qualities = []
quality_list = [
("无损FLAC", 2000, "flac"),
("高品质320K", 320, "mp3"),
("标准128K", 128, "mp3")
]
headers = {
'User-Agent': 'Mozilla/5.0 (Linux; Android 10)',
'Referer': 'https://www.kuwo.cn/'
}
for quality_name, bitrate, format_type in quality_list:
try:
api_url = f"https://nmobi.kuwo.cn/mobi.s?f=web&user=0&source=kwplayer_ar_4.4.2.7_B_nuoweida_vh.apk&type=convert_url_with_sign&rid={rid}&bitrate={bitrate}&format={format_type}"
r = requests.get(api_url, headers=headers, timeout=5)
data = r.json()
if data.get('code') == 200 and data.get('data') and data['data'].get('url'):
qualities.append((quality_name, data['data']['url']))
except Exception:
continue
if not qualities:
result["parse"] = 0
result["playUrl"] = ""
result["url"] = ""
result["header"] = {}
return result
urls = []
for quality_name, quality_url in qualities:
urls.append(quality_name)
urls.append(quality_url)
lrc = ""
pic = ""
try:
lrc_api = f"https://kuwo.cn/openapi/v1/www/lyric/getlyric?musicId={rid}"
lr = requests.get(lrc_api, timeout=5)
lj = lr.json()
if lj.get('data') and lj['data'].get('lrclist'):
lrc = "\n".join([f"[{self._format_time(float(item.get('time', 0)))}]{item.get('lineLyric', '')}"
for item in lj['data']['lrclist']])
except Exception:
pass
try:
pic_url = f"http://artistpicserver.kuwo.cn/pic.web?type=rid_pic&pictype=url&size=500&rid={rid}"
pr = requests.get(pic_url, timeout=5)
if pr.text.startswith('http'):
pic = pr.text.strip()
else:
pic = pic_url
except Exception:
pic = f"http://artistpicserver.kuwo.cn/pic.web?type=rid_pic&pictype=url&size=500&rid={rid}"
if lrc:
try:
ssa_lrc = self._create_ssa_subtitle(lrc)
ssa_base64 = base64.b64encode(ssa_lrc.encode('utf-8')).decode('utf-8')
ssa_url = f"data:text/x-ssa;base64,{ssa_base64}"
result["subs"] = [{
"name": "5行歌词",
"url": ssa_url,
"format": "text/x-ssa",
"selected": True
}]
except Exception:
pass
result["parse"] = 0
result["playUrl"] = ""
result["url"] = urls
result["header"] = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
"Referer": "https://www.kuwo.cn/"
}
return result
def _format_time(self, seconds):
m = int(seconds // 60)
s = seconds % 60
return f"{m:02d}:{s:05.2f}"
def _create_ssa_subtitle(self, lrc_text):
lines = []
pattern = r'\[(\d{2}):(\d{2})\.(\d{2})\](.*)'
for line in lrc_text.split('\n'):
match = re.match(pattern, line)
if match:
minutes = int(match.group(1))
seconds = int(match.group(2))
hundredths = int(match.group(3))
text = match.group(4).strip()
total_seconds = minutes * 60 + seconds + hundredths / 100.0
if text:
lines.append({
'start': total_seconds,
'text': text
})
if not lines:
return ""
ssa_header = """[Script Info]
ScriptType: v4.00+
Collisions: Normal
PlayResX: 1280
PlayResY: 720
Timer: 100.0000
WrapStyle: 0
[V4+ Styles]
Format: Name, Fontname, Fontsize, PrimaryColour, SecondaryColour, OutlineColour, BackColour, Bold, Italic, Underline, StrikeOut, ScaleX, ScaleY, Spacing, Angle, BorderStyle, Outline, Shadow, Alignment, MarginL, MarginR, MarginV, Encoding
Style: WAITING_TOP2,Roboto,55,&H0000FFFF,&H00808080,&H00000000,&H00000000,-1,0,0,0,100,100,0,0,1,1,1,2,0,0,180,1
Style: WAITING_TOP1,Roboto,55,&H0000FFFF,&H00808080,&H00000000,&H00000000,-1,0,0,0,100,100,0,0,1,1,1,2,0,0,260,1
Style: PLAYING_CENTER,Roboto,60,&H0000FF00,&H00808080,&H00000000,&H00000000,-1,0,0,0,100,100,0,0,1,2,2,2,0,0,340,1
Style: PLAYED_BOTTOM1,Roboto,55,&H0000FFFF,&H00808080,&H00000000,&H00000000,-1,0,0,0,100,100,0,0,1,1,1,2,0,0,420,1
Style: PLAYED_BOTTOM2,Roboto,55,&H0000FFFF,&H00808080,&H00000000,&H00000000,-1,0,0,0,100,100,0,0,1,1,1,2,0,0,500,1
[Events]
Format: Layer, Start, End, Style, Name, MarginL, MarginR, MarginV, Effect, Text
"""
def format_ssa_time(seconds):
h = int(seconds // 3600)
m = int((seconds % 3600) // 60)
s = int(seconds % 60)
cs = int((seconds * 100) % 100)
return f"{h}:{m:02d}:{s:02d}.{cs:02d}"
events = []
for i in range(len(lines)):
current = lines[i]
current_end = lines[i+1]['start'] if i+1 < len(lines) else current['start'] + 5.0
wait2 = lines[i+2] if i+2 < len(lines) else None
wait1 = lines[i+1] if i+1 < len(lines) else None
played1 = lines[i-1] if i-1 >= 0 else None
played2 = lines[i-2] if i-2 >= 0 else None
start_str = format_ssa_time(current['start'])
end_str = format_ssa_time(current_end)
if wait2:
events.append(f"Dialogue: 1,{start_str},{end_str},WAITING_TOP2,,0,0,0,,{wait2['text']}")
if wait1:
events.append(f"Dialogue: 2,{start_str},{end_str},WAITING_TOP1,,0,0,0,,{wait1['text']}")
events.append(f"Dialogue: 3,{start_str},{end_str},PLAYING_CENTER,,0,0,0,,{current['text']}")
if played1:
events.append(f"Dialogue: 4,{start_str},{end_str},PLAYED_BOTTOM1,,0,0,0,,{played1['text']}")
if played2:
events.append(f"Dialogue: 5,{start_str},{end_str},PLAYED_BOTTOM2,,0,0,0,,{played2['text']}")
return ssa_header + "\n".join(events)
def searchContent(self, key, quick, pg=1):
result = {}
wd = quote(key)
page_num = (int(pg) - 1) * 30
url = f"https://search.kuwo.cn/r.s?client=kt&pn={page_num}&rn=30&all={wd}&vipver=1&ft=artist&encoding=utf8&rformat=json&mobi=1"
headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36',
'Referer': 'http://www.kuwo.cn/'
}
try:
r = requests.get(url, headers=headers, timeout=10)
data = r.json()
videos = []
if data.get('abslist'):
base_path = data.get('BASEPICPATH', 'http://img1.kuwo.cn/star/starheads/')
for item in data['abslist']:
aid = item.get('ARTISTID') or item.get('DC_TARGETID', '')
pic = item.get('hts_PICPATH') or (base_path + item['PICPATH'] if item.get('PICPATH') else '')
video = {
"vod_id": str(aid),
"vod_name": item.get('ARTIST', ''),
"vod_pic": pic,
"vod_remarks": f"歌曲 : {item.get('SONGNUM', 0)}首"
}
videos.append(video)
result['list'] = videos
result['page'] = pg
result['pagecount'] = 9999
result['limit'] = 30
result['total'] = 999999
except Exception as e:
result['list'] = []
return result
def searchContentPage(self, key, quick, pg):
return self.searchContent(key, quick, pg)
def localProxy(self, param):
return None
jaychouqq/yingshi/py8/星芽短剧.py
+343 -343
View File
@@ -1,343 +1,343 @@
# coding = utf-8
# !/usr/bin/python
"""
"""
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 bs4 import BeautifulSoup
from base64 import b64decode
import urllib.request
import urllib.parse
import binascii
import requests
import base64
import json
import time
import sys
import re
import os
sys.path.append('..')
xurl = "https://app.whjzjx.cn"
headers = {
'User-Agent': 'Linux; Android 12; Pixel 3 XL) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/98.0.4758.101 Mobile Safari/537.36'
}
headerf = {
"platform": "1",
"user_agent": "Mozilla/5.0 (Linux; Android 9; V1938T Build/PQ3A.190705.08211809; wv) AppleWebKit/537.36 (KHTML, like Gecko) Version/4.0 Chrome/91.0.4472.114 Safari/537.36",
"content-type": "application/json; charset=utf-8"
}
times = int(time.time() * 1000)
data = {
"device": "2a50580e69d38388c94c93605241fb306",
"package_name": "com.jz.xydj",
"android_id": "ec1280db12795506",
"install_first_open": True,
"first_install_time": 1752505243345,
"last_update_time": 1752505243345,
"report_link_url": "",
"authorization": "",
"timestamp": times
}
plain_text = json.dumps(data, separators=(',', ':'), ensure_ascii=False)
key = "B@ecf920Od8A4df7"
key_bytes = key.encode('utf-8')
plain_bytes = plain_text.encode('utf-8')
cipher = AES.new(key_bytes, AES.MODE_ECB)
padded_data = pad(plain_bytes, AES.block_size)
ciphertext = cipher.encrypt(padded_data)
encrypted = base64.b64encode(ciphertext).decode('utf-8')
response = requests.post("https://u.shytkjgs.com/user/v3/account/login", headers=headerf, data=encrypted)
response_data = response.json()
Authorization = response_data['data']['token']
headerx = {
'authorization': Authorization,
'platform': '1',
'version_name': '3.8.3.1'
}
class Spider(Spider):
global xurl
global headerx
global headers
def getName(self):
return "首页"
def init(self, extend):
pass
def isVideoFormat(self, url):
pass
def manualVideoCheck(self):
pass
def extract_middle_text(self, text, start_str, end_str, pl, start_index1: str = '', end_index2: str = ''):
if pl == 3:
plx = []
while True:
start_index = text.find(start_str)
if start_index == -1:
break
end_index = text.find(end_str, start_index + len(start_str))
if end_index == -1:
break
middle_text = text[start_index + len(start_str):end_index]
plx.append(middle_text)
text = text.replace(start_str + middle_text + end_str, '')
if len(plx) > 0:
purl = ''
for i in range(len(plx)):
matches = re.findall(start_index1, plx[i])
output = ""
for match in matches:
match3 = re.search(r'(?:^|[^0-9])(\d+)(?:[^0-9]|$)', match[1])
if match3:
number = match3.group(1)
else:
number = 0
if 'http' not in match[0]:
output += f"#{match[1]}${number}{xurl}{match[0]}"
else:
output += f"#{match[1]}${number}{match[0]}"
output = output[1:]
purl = purl + output + "$$$"
purl = purl[:-3]
return purl
else:
return ""
else:
start_index = text.find(start_str)
if start_index == -1:
return ""
end_index = text.find(end_str, start_index + len(start_str))
if end_index == -1:
return ""
if pl == 0:
middle_text = text[start_index + len(start_str):end_index]
return middle_text.replace("\\", "")
if pl == 1:
middle_text = text[start_index + len(start_str):end_index]
matches = re.findall(start_index1, middle_text)
if matches:
jg = ' '.join(matches)
return jg
if pl == 2:
middle_text = text[start_index + len(start_str):end_index]
matches = re.findall(start_index1, middle_text)
if matches:
new_list = [f'{item}' for item in matches]
jg = '$$$'.join(new_list)
return jg
def homeContent(self, filter):
result = {}
result = {"class": [{"type_id": "1", "type_name": "剧场"},
{"type_id": "3", "type_name": "新剧"},
{"type_id": "2", "type_name": "热播"},
{"type_id": "7", "type_name": "星选"},
{"type_id": "5", "type_name": "阳光"}],
}
return result
def homeVideoContent(self):
videos = []
url= f'{xurl}/v1/theater/home_page?theater_class_id=1&class2_id=4&page_num=1&page_size=24'
detail = requests.get(url=url, headers=headerx)
detail.encoding = "utf-8"
if detail.status_code == 200:
data = detail.json()
for vod in data['data']['list']:
name = vod['theater']['title']
id = vod['theater']['id']
pic = vod['theater']['cover_url']
remark = vod['theater']['play_amount_str']
video = {
"vod_id": id,
"vod_name": name,
"vod_pic": pic,
"vod_remarks": remark
}
videos.append(video)
result = {'list': videos}
return result
def categoryContent(self, cid, pg, filter, ext):
result = {}
videos = []
url = f'{xurl}/v1/theater/home_page?theater_class_id={cid}&page_num={pg}&page_size=24'
detail = requests.get(url=url,headers=headerx)
detail.encoding = "utf-8"
if detail.status_code == 200:
data = detail.json()
for vod in data['data']['list']:
name = vod['theater']['title']
id = vod['theater']['id']
pic = vod['theater']['cover_url']
remark = vod['theater']['theme']
video = {
"vod_id": id,
"vod_name": name,
"vod_pic": pic,
"vod_remarks": remark
}
videos.append(video)
result = {'list': videos}
result['page'] = pg
result['pagecount'] = 9999
result['limit'] = 90
result['total'] = 999999
return result
def detailContent(self, ids):
did = ids[0]
result = {}
videos = []
xianlu = ''
bofang = ''
url = f'{xurl}/v2/theater_parent/detail?theater_parent_id={did}'
detail = requests.get(url=url, headers=headerx)
detail.encoding = "utf-8"
if detail.status_code == 200:
data = detail.json()
url = 'https://fs-im-kefu.7moor-fs1.com/ly/4d2c3f00-7d4c-11e5-af15-41bf63ae4ea0/1732707176882/jiduo.txt'
response = requests.get(url)
response.encoding = 'utf-8'
code = response.text
name = self.extract_middle_text(code, "s1='", "'", 0)
Jumps = self.extract_middle_text(code, "s2='", "'", 0)
content = '剧情：' + data['data']['introduction']
area = data['data']['desc_tags'][0]
remarks = data['data']['filing']
# 修复剧集只有一集的问题 - 检查theaters数据是否存在且不为空
if 'theaters' in data['data'] and data['data']['theaters']:
for sou in data['data']['theaters']:
id = sou['son_video_url']
name = sou['num']
bofang = bofang + str(name) + '$' + id + '#'
bofang = bofang[:-1] if bofang.endswith('#') else bofang
xianlu = '星芽'
else:
# 如果没有theaters数据，检查是否有单个视频URL
if 'video_url' in data['data'] and data['data']['video_url']:
bofang = '1$' + data['data']['video_url']
xianlu = '星芽'
else:
bofang = Jumps
xianlu = '1'
videos.append({
"vod_id": did,
"vod_content": content,
"vod_remarks": remarks,
"vod_area": area,
"vod_play_from": xianlu,
"vod_play_url": bofang
})
result['list'] = videos
return result
def playerContent(self, flag, id, vipFlags):
result = {}
result["parse"] = 0
result["playUrl"] = ''
result["url"] = id
result["header"] = headers
return result
def searchContentPage(self, key, quick, page):
result = {}
videos = []
payload = {
"text": key
}
url = f"{xurl}/v3/search"
detail = requests.post(url=url, headers=headerx, json=payload)
if detail.status_code == 200:
detail.encoding = "utf-8"
data = detail.json()
for vod in data['data']['theater']['search_data']:
name = vod['title']
id = vod['id']
pic = vod['cover_url']
remark = vod['score_str']
video = {
"vod_id": id,
"vod_name": name,
"vod_pic": pic,
"vod_remarks": remark
}
videos.append(video)
result['list'] = videos
result['page'] = page
result['pagecount'] = 9999
result['limit'] = 90
result['total'] = 999999
return result
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
# coding = utf-8
# !/usr/bin/python
"""
"""
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 bs4 import BeautifulSoup
from base64 import b64decode
import urllib.request
import urllib.parse
import binascii
import requests
import base64
import json
import time
import sys
import re
import os
sys.path.append('..')
xurl = "https://app.whjzjx.cn"
headers = {
'User-Agent': 'Linux; Android 12; Pixel 3 XL) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/98.0.4758.101 Mobile Safari/537.36'
}
headerf = {
"platform": "1",
"user_agent": "Mozilla/5.0 (Linux; Android 9; V1938T Build/PQ3A.190705.08211809; wv) AppleWebKit/537.36 (KHTML, like Gecko) Version/4.0 Chrome/91.0.4472.114 Safari/537.36",
"content-type": "application/json; charset=utf-8"
}
times = int(time.time() * 1000)
data = {
"device": "2a50580e69d38388c94c93605241fb306",
"package_name": "com.jz.xydj",
"android_id": "ec1280db12795506",
"install_first_open": True,
"first_install_time": 1752505243345,
"last_update_time": 1752505243345,
"report_link_url": "",
"authorization": "",
"timestamp": times
}
plain_text = json.dumps(data, separators=(',', ':'), ensure_ascii=False)
key = "B@ecf920Od8A4df7"
key_bytes = key.encode('utf-8')
plain_bytes = plain_text.encode('utf-8')
cipher = AES.new(key_bytes, AES.MODE_ECB)
padded_data = pad(plain_bytes, AES.block_size)
ciphertext = cipher.encrypt(padded_data)
encrypted = base64.b64encode(ciphertext).decode('utf-8')
response = requests.post("https://u.shytkjgs.com/user/v3/account/login", headers=headerf, data=encrypted)
response_data = response.json()
Authorization = response_data['data']['token']
headerx = {
'authorization': Authorization,
'platform': '1',
'version_name': '3.8.3.1'
}
class Spider(Spider):
global xurl
global headerx
global headers
def getName(self):
return "首页"
def init(self, extend):
pass
def isVideoFormat(self, url):
pass
def manualVideoCheck(self):
pass
def extract_middle_text(self, text, start_str, end_str, pl, start_index1: str = '', end_index2: str = ''):
if pl == 3:
plx = []
while True:
start_index = text.find(start_str)
if start_index == -1:
break
end_index = text.find(end_str, start_index + len(start_str))
if end_index == -1:
break
middle_text = text[start_index + len(start_str):end_index]
plx.append(middle_text)
text = text.replace(start_str + middle_text + end_str, '')
if len(plx) > 0:
purl = ''
for i in range(len(plx)):
matches = re.findall(start_index1, plx[i])
output = ""
for match in matches:
match3 = re.search(r'(?:^|[^0-9])(\d+)(?:[^0-9]|$)', match[1])
if match3:
number = match3.group(1)
else:
number = 0
if 'http' not in match[0]:
output += f"#{match[1]}${number}{xurl}{match[0]}"
else:
output += f"#{match[1]}${number}{match[0]}"
output = output[1:]
purl = purl + output + "$$$"
purl = purl[:-3]
return purl
else:
return ""
else:
start_index = text.find(start_str)
if start_index == -1:
return ""
end_index = text.find(end_str, start_index + len(start_str))
if end_index == -1:
return ""
if pl == 0:
middle_text = text[start_index + len(start_str):end_index]
return middle_text.replace("\\", "")
if pl == 1:
middle_text = text[start_index + len(start_str):end_index]
matches = re.findall(start_index1, middle_text)
if matches:
jg = ' '.join(matches)
return jg
if pl == 2:
middle_text = text[start_index + len(start_str):end_index]
matches = re.findall(start_index1, middle_text)
if matches:
new_list = [f'{item}' for item in matches]
jg = '$$$'.join(new_list)
return jg
def homeContent(self, filter):
result = {}
result = {"class": [{"type_id": "1", "type_name": "剧场"},
{"type_id": "3", "type_name": "新剧"},
{"type_id": "2", "type_name": "热播"},
{"type_id": "7", "type_name": "星选"},
{"type_id": "5", "type_name": "阳光"}],
}
return result
def homeVideoContent(self):
videos = []
url= f'{xurl}/v1/theater/home_page?theater_class_id=1&class2_id=4&page_num=1&page_size=24'
detail = requests.get(url=url, headers=headerx)
detail.encoding = "utf-8"
if detail.status_code == 200:
data = detail.json()
for vod in data['data']['list']:
name = vod['theater']['title']
id = vod['theater']['id']
pic = vod['theater']['cover_url']
remark = vod['theater']['play_amount_str']
video = {
"vod_id": id,
"vod_name": name,
"vod_pic": pic,
"vod_remarks": remark
}
videos.append(video)
result = {'list': videos}
return result
def categoryContent(self, cid, pg, filter, ext):
result = {}
videos = []
url = f'{xurl}/v1/theater/home_page?theater_class_id={cid}&page_num={pg}&page_size=24'
detail = requests.get(url=url,headers=headerx)
detail.encoding = "utf-8"
if detail.status_code == 200:
data = detail.json()
for vod in data['data']['list']:
name = vod['theater']['title']
id = vod['theater']['id']
pic = vod['theater']['cover_url']
remark = vod['theater']['theme']
video = {
"vod_id": id,
"vod_name": name,
"vod_pic": pic,
"vod_remarks": remark
}
videos.append(video)
result = {'list': videos}
result['page'] = pg
result['pagecount'] = 9999
result['limit'] = 90
result['total'] = 999999
return result
def detailContent(self, ids):
did = ids[0]
result = {}
videos = []
xianlu = ''
bofang = ''
url = f'{xurl}/v2/theater_parent/detail?theater_parent_id={did}'
detail = requests.get(url=url, headers=headerx)
detail.encoding = "utf-8"
if detail.status_code == 200:
data = detail.json()
url = 'https://fs-im-kefu.7moor-fs1.com/ly/4d2c3f00-7d4c-11e5-af15-41bf63ae4ea0/1732707176882/jiduo.txt'
response = requests.get(url)
response.encoding = 'utf-8'
code = response.text
name = self.extract_middle_text(code, "s1='", "'", 0)
Jumps = self.extract_middle_text(code, "s2='", "'", 0)
content = '剧情：' + data['data']['introduction']
area = data['data']['desc_tags'][0]
remarks = data['data']['filing']
# 修复剧集只有一集的问题 - 检查theaters数据是否存在且不为空
if 'theaters' in data['data'] and data['data']['theaters']:
for sou in data['data']['theaters']:
id = sou['son_video_url']
name = sou['num']
bofang = bofang + str(name) + '$' + id + '#'
bofang = bofang[:-1] if bofang.endswith('#') else bofang
xianlu = '星芽'
else:
# 如果没有theaters数据，检查是否有单个视频URL
if 'video_url' in data['data'] and data['data']['video_url']:
bofang = '1$' + data['data']['video_url']
xianlu = '星芽'
else:
bofang = Jumps
xianlu = '1'
videos.append({
"vod_id": did,
"vod_content": content,
"vod_remarks": remarks,
"vod_area": area,
"vod_play_from": xianlu,
"vod_play_url": bofang
})
result['list'] = videos
return result
def playerContent(self, flag, id, vipFlags):
result = {}
result["parse"] = 0
result["playUrl"] = ''
result["url"] = id
result["header"] = headers
return result
def searchContentPage(self, key, quick, page):
result = {}
videos = []
payload = {
"text": key
}
url = f"{xurl}/v3/search"
detail = requests.post(url=url, headers=headerx, json=payload)
if detail.status_code == 200:
detail.encoding = "utf-8"
data = detail.json()
for vod in data['data']['theater']['search_data']:
name = vod['title']
id = vod['id']
pic = vod['cover_url']
remark = vod['score_str']
video = {
"vod_id": id,
"vod_name": name,
"vod_pic": pic,
"vod_remarks": remark
}
videos.append(video)
result['list'] = videos
result['page'] = page
result['pagecount'] = 9999
result['limit'] = 90
result['total'] = 999999
return result
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
jaychouqq/yingshi/py8/色播聚合.py
+72
View File
@@ -0,0 +1,72 @@
#coding=utf-8
#!/usr/bin/python
import sys
sys.path.append('..')
from base.spider import Spider
class Spider(Spider):
def init(self,extend=""):
self.base_url='http://api.hclyz.com:81/mf'
def homeContent(self,filter):
classes = [{"type_name": "色播聚合","type_id":"/json.txt"}]
result = {"class": classes}
return result
def categoryContent(self,tid,pg,filter,extend):
home = self.fetch(f'{self.base_url}/json.txt').json()
data = home.get("pingtai")[1:]
videos = [
{
"vod_id": "/" + item['address'],
"vod_name": item['title'],
"vod_pic": item['xinimg'].replace("http://cdn.gcufbd.top/img/",
"https://slink.ltd/https://raw.githubusercontent.com/fish2018/lib/refs/heads/main/imgs/"),
"vod_remarks": item['Number'],
"style": {"type": "rect", "ratio": 1.33}
} for item in sorted(data, key=lambda x: int(x['Number']), reverse=True)
]
result = {
"page": pg,
"pagecount": 1,
"limit": len(videos),
"total": len(videos),
"list": videos
}
return result
def detailContent(self,array):
id = array[0]
data = self.fetch(f'{self.base_url}/{id}').json()
zhubo = data['zhubo']
playUrls = '#'.join([f"{vod['title']}${vod['address']}" for vod in zhubo])
vod = [{
"vod_play_from": 'sebo',
"vod_play_url": playUrls,
"vod_content": 'https://github.com/fish2018',
}]
result = {"list": vod}
return result
def playerContent(self,flag,id,vipFlags):
result = {
'parse': 0,
'url': id
}
return result
def getName(self):
return '色播聚合'
def homeVideoContent(self):
pass
def isVideoFormat(self,url):
pass
def manualVideoCheck(self):
pass
def searchContent(self,key,quick):
pass
def destroy(self):
pass
def localProxy(self, param):
pass