yeyin/tvbox_collect
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Sync all projects

Browse Source
This commit is contained in:
github-actions[bot]
2026-07-21 17:58:07 +00:00
parent ee7a20c91a
commit bcf9300430
24 changed files with 8773 additions and 6481 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Tomorrow/py/aiting.py
+502
View File
@@ -0,0 +1,502 @@
# -*- coding: utf-8 -*-
# 修复：歌手不显示歌手图片
# by：垃圾星河
# 代码指导：嗷呜呜呜呜
# 增加：自动人机验证绕过
import re
import sys
import time
from base64 import b64encode, b64decode
from urllib.parse import quote, unquote
from pyquery import PyQuery as pq
from requests import Session, adapters
from urllib3.util.retry import Retry
from concurrent.futures import ThreadPoolExecutor, as_completed
sys.path.append('..')
from base.spider import Spider
class Spider(Spider):
def init(self, extend=""):
self.host = "https://www.22a5.com"
self.session = Session()
adapter = adapters.HTTPAdapter(
max_retries=Retry(total=3, backoff_factor=0.5, status_forcelist=[429, 500, 502, 503, 504]),
pool_connections=20,
pool_maxsize=50
)
self.session.mount("http://", adapter)
self.session.mount("https://", adapter)
self.headers = {
"User-Agent": "Mozilla/5.0 (Linux; Android 10; K) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Mobile Safari/537.36"
}
self.session.headers.update(self.headers)
def getName(self):
return "爱听音乐"
def isVideoFormat(self, url):
return bool(re.search(r'\.(m3u8|mp4|mp3|m4a|flv)(\?|$)', url or "", re.I))
def manualVideoCheck(self):
return False
def destroy(self):
self.session.close()
# ==================== 新增人机验证绕过 ====================
def _bypass_verification(self, url, response_text):
"""若当前页面是人机验证，自动提交勾选"""
if '安全人机验证' not in response_text and 'human_check' not in response_text:
return None
# 提取 csrf_token（支持多种格式）
token_match = re.search(r'name="csrf_token"\s+value="([^"]+)"', response_text)
if not token_match:
print("[爱听音乐] 未找到 csrf_token，跳过绕过")
return None
token = token_match.group(1)
# 构造提交数据
data = {
'csrf_token': token,
'human_check': 'on'
}
try:
# 发送 POST 请求，自动跟随重定向
post_resp = self.session.post(url, data=data, allow_redirects=True, timeout=10)
print("[爱听音乐] 人机验证已绕过")
return post_resp
except Exception as e:
print(f"[爱听音乐] 人机验证提交失败: {e}")
return None
def getpq(self, url):
"""获取页面并自动处理人机验证（重试3次）"""
full_url = self._abs(url)
for attempt in range(3):
try:
resp = self.session.get(full_url, timeout=5)
# 若遇到验证页，尝试绕过
if '安全人机验证' in resp.text or 'human_check' in resp.text:
print("[爱听音乐] 检测到人机验证，尝试自动绕过...")
bypass_resp = self._bypass_verification(full_url, resp.text)
if bypass_resp:
# 验证成功后，返回最终页面（可能是重定向后的目标）
return pq(bypass_resp.text)
else:
# 绕过失败，等待后重试
time.sleep(1)
continue
else:
# 正常页面直接返回
return pq(resp.text)
except Exception as e:
print(f"[爱听音乐] 请求失败 (尝试 {attempt+1}/3): {e}")
time.sleep(0.5 * (attempt + 1))
# 全部失败，返回空文档
return pq("<html></html>")
# ==================== 原有功能（保持不变） ====================
def homeContent(self, filter):
classes = [
{"type_name": n, "type_id": i}
for n, i in [
("歌手", "/singerlist/index/index/index/index.html"),
("TOP榜单", "/list/top.html"),
("新歌榜", "/list/new.html"),
("电台", "/radiolist/index.html"),
("高清MV", "/mvlist/oumei.html"),
("专辑", "/albumlist/index.html"),
("歌单", "/playtype/index.html")
]
]
filters = {}
for p in [c["type_id"] for c in classes if "singer" not in c["type_id"]]:
d = self._fetch_filters(p)
if d:
filters[p] = d
if "/radiolist/index.html" not in filters:
filters["/radiolist/index.html"] = [{
"key": "id",
"name": "分类",
"value": [
{"n": n, "v": v}
for n, v in zip(
["最新", "最热", "有声小说", "相声", "音乐", "情感", "国漫", "影视", "脱口秀", "历史", "儿童", "教育", "八卦", "推理", "头条"],
["index", "hot", "novel", "xiangyi", "music", "emotion", "game", "yingshi", "talkshow", "history",
"children", "education", "gossip", "tuili", "headline"]
)
]
}]
filters["/singerlist/index/index/index/index.html"] = [
{
"key": "area",
"name": "地区",
"value": [
{"n": "全部", "v": "index"},
{"n": "华语", "v": "huayu"},
{"n": "欧美", "v": "oumei"},
{"n": "韩国", "v": "hanguo"},
{"n": "日本", "v": "ribrn"}
]
},
{
"key": "sex",
"name": "性别",
"value": [
{"n": "全部", "v": "index"},
{"n": "男", "v": "male"},
{"n": "女", "v": "girl"},
{"n": "组合", "v": "band"}
]
},
{
"key": "genre",
"name": "流派",
"value": [
{"n": "全部", "v": "index"},
{"n": "流行", "v": "liuxing"},
{"n": "电子", "v": "dianzi"},
{"n": "摇滚", "v": "yaogun"},
{"n": "嘻哈", "v": "xiha"},
{"n": "R&B", "v": "rb"},
{"n": "民谣", "v": "minyao"},
{"n": "爵士", "v": "jueshi"},
{"n": "古典", "v": "gudian"}
]
}
]
return {"class": classes, "filters": filters, "list": []}
def homeVideoContent(self):
return {"list": []}
def categoryContent(self, tid, pg, filter, extend):
pg = int(pg or 1)
url = tid
if "/singerlist/" in tid:
parts = tid.split('/')
if len(parts) >= 6:
url = "/".join(parts[:2] + [extend.get(k, parts[i]) for i, k in enumerate(["area", "sex", "genre"], 2)] +
[f"{extend.get('char', 'index')}.html"])
elif "id" in extend and extend["id"] not in ["index", "top"]:
url = tid.replace("index.html", f"{extend['id']}.html").replace("top.html", f"{extend['id']}.html")
if url == tid:
url = f"{tid.rsplit('/', 1)[0]}/{extend['id']}.html"
if pg > 1:
sep = "/" if any(x in url for x in ["/singerlist/", "/radiolist/", "/mvlist/", "/playtype/", "/list/"]) else "_"
url = re.sub(r'(_\d+|/\d+)?\.html$', f'{sep}{pg}.html', url)
doc = self.getpq(url)
items = doc(".play_list li, .video_list li, .pic_list li, .singer_list li, .ali li, .layui-row li, .base_l li")
return {
"list": self._parse_list(items, tid),
"page": pg,
"pagecount": 9999,
"limit": 90,
"total": 999999
}
def searchContent(self, key, quick, pg="1"):
doc = self.getpq(f"/so/{quote(key)}/{pg}.html")
items = doc(".base_l li, .play_list li")
return {
"list": self._parse_list(items, "search"),
"page": int(pg)
}
def detailContent(self, ids):
url = self._abs(ids[0])
doc = self.getpq(url)
vod = {
"vod_id": url,
"vod_name": self._clean(doc("h1").text() or doc("title").text()),
"vod_pic": self._abs(doc(".djpg img, .pic img, .djpic img").attr("src")),
"vod_play_from": "爱听音乐",
"vod_content": ""
}
if any(x in url for x in ["/playlist/", "/album/", "/list/", "/singer/", "/special/", "/radio/", "/radiolist/"]):
eps = self._get_eps(doc)
page_urls = {
self._abs(a.attr("href"))
for a in doc(".page a, .dede_pages a, .pagelist a").items()
if a.attr("href") and "javascript" not in a.attr("href")
} - {url}
if page_urls:
with ThreadPoolExecutor(max_workers=5) as ex:
futures = []
for u in sorted(page_urls, key=lambda x: int(re.search(r'[_/](\d+)\.html', x).group(1)) if re.search(
r'[_/](\d+)\.html', x) else 0):
futures.append(ex.submit(lambda uu: self._get_eps(self.getpq(uu)), u))
for f in as_completed(futures):
eps.extend(f.result() or [])
if eps:
vod.update({
"vod_play_from": "播放列表",
"vod_play_url": "#".join(eps)
})
return {"list": [vod]}
play_list = []
if mid := re.search(r'/(song|mp3|radio|radiolist|radioplay)/([^/]+)\.html', url):
lrc_url = f"{self.host}/plug/down.php?ac=music&lk=lrc&id={mid.group(2)}"
play_list = [f"播放${self.e64('0@@@@' + url + '|||' + lrc_url)}"]
elif vid := re.search(r'/(video|mp4)/([^/]+)\.html', url):
with ThreadPoolExecutor(max_workers=3) as ex:
fs = {
ex.submit(self._api, "/plug/down.php", {"ac": "vplay", "id": vid.group(2), "q": q}): n
for n, q in [("蓝光", 1080), ("超清", 720), ("高清", 480)]
}
play_list = [f"{fs[f]}${self.e64('0@@@@'+u)}" for f in as_completed(fs) if (u := f.result())]
play_list.sort(key=lambda x: {"蓝": 0, "超": 1, "高": 2}.get(x[0], 3))
vod["vod_play_url"] = "#".join(play_list) if play_list else f"解析失败${self.e64('1@@@@'+url)}"
return {"list": [vod]}
def playerContent(self, flag, id, vipFlags):
raw = self.d64(id).split("@@@@")[-1]
url, subt = raw.split("|||") if "|||" in raw else (raw, "")
url = url.replace(r"\/", "/")
if ".html" in url and not self.isVideoFormat(url):
if mid := re.search(r'/(song|mp3|radio|radiolist|radioplay)/([^/]+)\.html', url):
if r_url := self._api("/js/play.php", method="POST", data={"id": mid.group(2), "type": "music"},
headers={"Referer": url.replace("http://", "https://"),
"X-Requested-With": "XMLHttpRequest"}):
url = r_url if ".php" not in r_url else url
elif vid := re.search(r'/(video|mp4)/([^/]+)\.html', url):
with ThreadPoolExecutor(max_workers=3) as ex:
for f in as_completed(
[ex.submit(self._api, "/plug/down.php", {"ac": "vplay", "id": vid.group(2), "q": q}) for q in
[1080, 720, 480]]):
if v_url := f.result():
url = v_url
break
result = {"parse": 0, "url": url, "header": {"User-Agent": self.headers["User-Agent"]}}
if "22a5.com" in url:
result["header"]["Referer"] = self.host + "/"
# OK影视3.6.5+支持LRC格式滚动歌词
if subt:
try:
r = self.session.get(subt, headers={"Referer": self.host + "/"}, timeout=5)
lrc_content = r.text
if lrc_content:
lrc_content = self._filter_lrc_ads(lrc_content)
result["lrc"] = lrc_content
except:
pass
return result
def _filter_lrc_ads(self, lrc_text):
"""过滤LRC歌词中的广告内容"""
lines = lrc_text.splitlines()
filtered_lines = []
# 广告关键词模式
ad_patterns = [
r'欢迎来访.*',
r'本站.*',
r'.*广告.*',
r'QQ群.*',
r'.*www\..*',
r'.*http.*',
r'.*\.com.*',
r'.*\.cn.*',
r'.*\.net.*',
r'.*音乐网.*',
r'.*提供.*',
r'.*下载.*',
]
for line in lines:
if re.match(r'\[\d{2}:\d{2}', line):
is_ad = False
for pattern in ad_patterns:
if re.search(pattern, line, re.IGNORECASE):
is_ad = True
break
if not is_ad:
filtered_lines.append(line)
else:
filtered_lines.append(line)
return '\n'.join(filtered_lines)
def localProxy(self, param):
url = unquote(param.get("url", ""))
type_ = param.get("type")
if type_ == "img":
try:
headers = {
"Referer": "https://www.baidu.com/",
"User-Agent": self.headers["User-Agent"],
"Accept": "image/webp,image/apng,image/*,*/*;q=0.8",
"Accept-Language": "zh-CN,zh;q=0.9"
}
resp = self.session.get(url, headers=headers, timeout=10)
return [200, "image/jpeg", resp.content, {}]
except Exception as e:
print(f"图片代理失败: {e}")
return [404, "text/plain", b"", {}]
elif type_ == "lrc":
try:
r = self.session.get(url, headers={"Referer": self.host + "/"}, timeout=5)
lrc_content = r.text
lrc_content = self._filter_lrc_ads(lrc_content)
return [200, "application/octet-stream", lrc_content.encode('utf-8'), {}]
except:
return [404, "text/plain", "Error", {}]
return None
# ==================== 辅助方法 ====================
def _parse_list(self, items, tid=""):
"""解析列表项，修复歌手头像 - 直接返回原始图片URL"""
res = []
for li in items.items():
a = li("a").eq(0)
if not (href := a.attr("href")) or href == "/" or any(x in href for x in ["/user/", "/login/", "javascript"]):
continue
if not (name := self._clean(li(".name").text() or a.attr("title") or a.text())):
continue
is_singer = "/singer/" in href or "/singerlist" in tid
pic = ""
src = ""
if is_singer:
img = li(".pic img").eq(0)
src = img.attr("src") or ""
if not src:
img = li("img").eq(0)
src = img.attr("src") or ""
else:
img = li("img").eq(0)
src = img.attr("src") or ""
if not src:
img = li(".pic img").eq(0)
src = img.attr("src") or ""
if src:
if src.startswith('//'):
src = 'https:' + src
elif src.startswith('/'):
src = self.host + src
pic = src
res.append({
"vod_id": self._abs(href),
"vod_name": name,
"vod_pic": pic,
"style": {
"type": "oval" if is_singer else ("list" if any(
x in tid for x in ["/list/", "/playtype/", "/albumlist/"]) else "rect"),
"ratio": 1 if is_singer else 1.33
}
})
return res
def _get_eps(self, doc):
eps = []
for li in doc(".play_list li, .song_list li, .music_list li").items():
a = li("a").eq(0)
href = a.attr("href")
if not href or not re.search(r'/(song|mp3|radio|radiolist|radioplay)/([^/]+)\.html', href):
continue
full_url = self._abs(href)
lrc_part = ""
mid = re.search(r'/(song|mp3|radio|radiolist|radioplay)/([^/]+)\.html', full_url)
if mid:
lrc_url = f"{self.host}/plug/down.php?ac=music&lk=lrc&id={mid.group(2)}"
lrc_part = f"|||{lrc_url}"
eps.append(f"{self._clean(a.text() or li('.name').text())}${self.e64('0@@@@' + full_url + lrc_part)}")
return eps
def _clean(self, text):
return re.sub(
r'(爱玩音乐网|视频下载说明|视频下载地址|www\.2t58\.com|MP3免费下载|LRC歌词下载|全部歌曲|\[第\d+页\]|刷新|每日推荐|最新|热门|推荐|MV|高清|无损)',
'',
text or '',
flags=re.I
).strip()
def _fetch_filters(self, url):
doc = self.getpq(url)
filters = []
for i, group in enumerate([
doc(".ilingku_fl"),
doc(".class_list"),
doc(".screen_list"),
doc(".box_list"),
doc(".nav_list")
]):
if group:
opts = [{"n": "全部", "v": "top" if "top" in url else "index"}]
seen = set()
for a in group("a").items():
v = (a.attr("href") or "").split("?")[0].rstrip('/').split('/')[-1].replace('.html', '')
if v and v not in seen:
opts.append({"n": a.text().strip(), "v": v})
seen.add(v)
if len(opts) > 1:
filters.append({"key": f"id{i}" if i else "id", "name": "分类", "value": opts})
return filters
def _api(self, path, params=None, method="GET", headers=None, data=None):
try:
h = self.headers.copy()
if headers:
h.update(headers)
r = (self.session.post if method == "POST" else self.session.get)(
f"{self.host}{path}",
params=params,
data=data,
headers=h,
timeout=10,
allow_redirects=False
)
if loc := r.headers.get("Location"):
return self._abs(loc.strip())
return self._abs(r.json().get("url", "").replace(r"\/", "/")) or (r.text.strip() if r.text.strip().startswith(
"http") else "")
except:
return ""
def _abs(self, url):
if not url:
return ""
if url.startswith("http"):
return url
if url.startswith("//"):
return "https:" + url
return f"{self.host}{'/' if not url.startswith('/') else ''}{url}"
def e64(self, text):
return b64encode(text.encode("utf-8")).decode("utf-8")
def d64(self, text):
return b64decode(text.encode("utf-8")).decode("utf-8")
Tomorrow/py/kuwo.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": 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
Tomorrow/py/光鸭社区.py
+125
View File
@@ -0,0 +1,125 @@
import re,json,base64,requests,threading
from urllib.parse import quote
from base.spider import Spider
class Spider(Spider):
def getName(self): return "光鸭社区"
def init(self,extend=""):
self.host="https://gy.lubao2.de5.net";self.sb="https://houhvngggkkcqcjmzafx.supabase.co";self.tmdb="https://api.tmdb.org";self.tmdb_img="https://images.tmdb.org/t/p";self.tmdb_key="2894d9a1baf7812b451de03c801b0281"
self.key="eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJpc3MiOiJzdXBhYmFzZSIsInJlZiI6ImhvdWh2bmdnZ2trY3Fjam16YWZ4Iiwicm9sZSI6ImFub24iLCJpYXQiOjE3Nzc2Mjg2MjgsImV4cCI6MjA5MzIwNDYyOH0.FUUT9ETU-lP5iNqwK3i7nflwGLSBHM1b-qIs9CoVles"
self.cookie="";self.page_size=40;self.default_pic="https://miaoda-edit-image.cdn.bcebos.com/b74i02x880e9/IMG-b7c60ycpg3r4.png";self.pic_cache={};self.pic_pending=set();self.pic_sem=threading.BoundedSemaphore(3);self.cache_file="/storage/emulated/0/Download/gy_tmdb_pic_cache.json";self.list_fields="id,title,cover_url,year,region,genres,category_id"
self.headers={"User-Agent":"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 Chrome/122 Safari/537.36","Referer":self.host+"/","apikey":self.key,"Authorization":"Bearer "+self.key,"Accept":"application/json"}
self.session=requests.Session();self.session.headers.update(self.headers)
if extend:
try:
e=json.loads(extend);self.cookie=e.get("cookie","") or self.cookie;self.page_size=int(e.get("page_size",40) or 40);self.tmdb_key=e.get("tmdb_key",self.tmdb_key);self.tmdb=e.get("tmdb_api",self.tmdb);self.tmdb_img=e.get("tmdb_img",self.tmdb_img);self.default_pic=e.get("default_pic",self.default_pic);self.cache_file=e.get("cache_file",self.cache_file)
except Exception:self.cookie=extend
if self.cookie:self.headers["Cookie"]=self.cookie;self.session.headers.update({"Cookie":self.cookie})
self._load_cache()
def _load_cache(self):
try:
with open(self.cache_file,"r",encoding="utf-8") as f:self.pic_cache=json.load(f)
except Exception:self.pic_cache={}
def _save_cache(self):
try:
if len(self.pic_cache)%5==0:
with open(self.cache_file,"w",encoding="utf-8") as f:json.dump(self.pic_cache,f,ensure_ascii=False,separators=(",",":"))
except Exception:None
def _get(self,path):
try:
r=self.session.get(self.sb+path,timeout=12);return r.json() if r.text else []
except Exception:return []
def _tmget(self,path,params):
try:
params.update({"api_key":self.tmdb_key,"language":"zh-CN"});r=requests.get(self.tmdb+path,params=params,headers={"User-Agent":self.headers["User-Agent"]},timeout=5);return r.json() if r.text else {}
except Exception:return {}
def _enc(self,o): return base64.urlsafe_b64encode(json.dumps(o,ensure_ascii=False,separators=(",",":")).encode()).decode().rstrip("=")
def _dec(self,s):
try:return json.loads(base64.urlsafe_b64decode((s+"="*(-len(s)%4)).encode()).decode())
except Exception:return {"url":s}
def _clean(self,s): return re.sub(r"\s+"," ",str(s or "")).replace("#","＃").replace("$","＄").strip()
def _title(self,s):
s=re.sub(r"\[tmdb-\d+\]","",str(s or ""),flags=re.I);s=re.sub(r"\[[^\]]+\]|【[^】]+】|\([^)]*\)|（[^）]*）"," ",s);s=re.sub(r"\b(19|20)\d{2}\b.*$","",s);s=re.sub(r"\.(2160p|1080p|720p|WEB-DL|BluRay|H265|H264|AAC|SONYHD).*$","",s,flags=re.I);return self._clean(s).strip(" ._-—")
def _pkey(self,x):
name=x.get("title") or "";year=str(x.get("year") or "");mid=re.search(r"tmdb-(\d+)",name,re.I);return mid.group(1) if mid else self._title(name)+year
def _poster(self,x,search=False):
pic=x.get("cover_url") or ""
if pic:return pic
if not search:return self.default_pic
ck=self._pkey(x)
if not ck:return self.default_pic
if ck in self.pic_cache:return self.pic_cache[ck]
if ck not in self.pic_pending:
self.pic_pending.add(ck);threading.Thread(target=self._fill_poster,args=(dict(x),ck),daemon=True).start()
return self.default_pic
def _fill_poster(self,x,ck):
self.pic_sem.acquire();p="";name=x.get("title") or "";year=str(x.get("year") or "");mid=re.search(r"tmdb-(\d+)",name,re.I)
try:
if mid:
for t in ["movie","tv"]:
j=self._tmget("/3/%s/%s"%(t,mid.group(1)),{});p=j.get("poster_path") or j.get("backdrop_path") or ""
if p:break
if not p:
q=self._title(name);j=self._tmget("/3/search/multi",{"query":q,"year":year}) if q else {};rs=[i for i in j.get("results",[]) if i.get("poster_path")]
if not rs and q:j=self._tmget("/3/search/multi",{"query":q});rs=[i for i in j.get("results",[]) if i.get("poster_path")]
p=rs[0].get("poster_path","") if rs else ""
self.pic_cache[ck]=self.tmdb_img+"/w500"+p if p else self.default_pic;self._save_cache()
except Exception:self.pic_cache[ck]=self.default_pic
self.pic_pending.discard(ck);self.pic_sem.release()
def _cats_raw(self): return self._get("/rest/v1/categories?select=id,name&parent_id=is.null&order=sort_order.asc")
def _cats(self): return [{"type_id":x.get("id",""),"type_name":x.get("name","")} for x in self._cats_raw() if x.get("id") and x.get("name")]
def _cat_names(self): return {x.get("id",""):x.get("name","") for x in self._cats_raw() if x.get("id")}
def _item(self,x,search=True):
name=self._clean(x.get("title") or "光鸭资源");remark=" ".join([str(x.get("year") or ""),x.get("region") or ""]).strip()
if not remark and x.get("genres"):remark="/".join(x.get("genres")[:2])
return {"vod_id":x.get("id",""),"vod_name":name,"vod_pic":self._poster(x,search),"vod_remarks":remark}
def _folder(self,name,count,kw,cid,pic=""): return {"vod_id":"folder:"+self._enc({"kw":kw,"cid":cid,"name":name}),"vod_name":"📁 %s %s条"%(name,count),"vod_pic":pic or self.default_pic,"vod_remarks":"点击查看","vod_tag":"folder"}
def _code_url(self,u,p):
u=str(u or "").strip();p=str(p or "").strip()
return u+(("&" if "?" in u else "?")+"code="+quote(p)) if u.startswith("http") and "guangyapan.com/s/" in u and p and "code=" not in u else u
def _search_arr(self,key,limit=300,offset=0):
q=quote("*"+str(key or "")+"*")
return self._get("/rest/v1/resources?select=%s&status=eq.approved&title=ilike.%s&order=created_at.desc&limit=%s&offset=%s"%(self.list_fields,q,limit,offset)) if key else []
def _list(self,path): return [self._item(x) for x in self._get(path) if x.get("id")]
def homeContent(self,filter): return {"class":self._cats(),"filters":{},"list":self._list("/rest/v1/resources?select=%s&status=eq.approved&order=created_at.desc&limit=30"%self.list_fields)}
def homeVideoContent(self): return {"list":self._list("/rest/v1/resources?select=%s&status=eq.approved&order=created_at.desc&limit=30"%self.list_fields)}
def categoryContent(self,tid,pg,filter,extend):
page=int(pg) if str(pg).isdigit() else 1;start=(page-1)*self.page_size
if str(tid).startswith("folder:"):
o=self._dec(str(tid)[7:]);arr=[x for x in self._search_arr(o.get("kw",""),500,0) if x.get("category_id")==o.get("cid")];lst=arr[start:start+self.page_size]
return {"page":page,"pagecount":max(1,(len(arr)+self.page_size-1)//self.page_size),"limit":self.page_size,"total":len(arr),"list":[self._item(x) for x in lst if x.get("id")]}
arr=self._get("/rest/v1/resources?select=%s&status=eq.approved&order=created_at.desc&category_id=eq.%s&limit=%s&offset=%s"%(self.list_fields,quote(str(tid)),self.page_size,start));total=start+len(arr)+(1 if len(arr)>=self.page_size else 0)
return {"page":page,"pagecount":page+(1 if len(arr)>=self.page_size else 0),"limit":self.page_size,"total":total,"list":[self._item(x) for x in arr if x.get("id")]}
def searchContent(self,key,quick,pg="1"):
page=int(pg) if str(pg).isdigit() else 1;arr=self._search_arr(key,500,0);names=self._cat_names();groups={}
for x in arr:
cid=x.get("category_id") or "other";groups.setdefault(cid,[]).append(x)
if page==1 and len(groups)>1:
order=sorted(groups.items(),key=lambda kv:-len(kv[1]));return {"page":1,"pagecount":1,"limit":self.page_size,"total":len(arr),"list":[self._folder(names.get(cid,"其他"),len(v),key,cid,next((i.get("cover_url") for i in v if i.get("cover_url")),"")) for cid,v in order]}
start=(page-1)*self.page_size;lst=arr[start:start+self.page_size]
return {"page":page,"pagecount":max(1,(len(arr)+self.page_size-1)//self.page_size),"limit":self.page_size,"total":len(arr),"list":[self._item(x) for x in lst if x.get("id")]}
def detailContent(self,ids):
vid=ids[0] if ids else "";arr=self._get("/rest/v1/resources?select=*&id=eq.%s&limit=1"%quote(str(vid))) if vid and not str(vid).startswith("folder:") else []
if not arr:return {"list":[]}
x=arr[0];name=self._clean(x.get("title") or "光鸭资源");pic=self._poster(x,True);content=self._clean(x.get("description") or "");links=[]
if x.get("pan_link"):links.append(("主链接",self._code_url(x.get("pan_link"),x.get("pan_password") or ""),x.get("pan_password") or ""))
for i,u in enumerate(x.get("backup_links") or []):
if isinstance(u,dict):
p=u.get("password") or u.get("pwd") or "";links.append((u.get("name") or "备用%d"%(i+1),self._code_url(u.get("url") or u.get("link") or "",p),p))
elif isinstance(u,str):links.append(("备用%d"%(i+1),u,""))
if not links:links.append(("详情页",self.host+"/resource/"+vid,""))
eps=[self._clean(n)+(" 提取码:"+p if p else "")+"$"+self._enc({"url":u,"pwd":p}) for n,u,p in links if u]
return {"list":[{"vod_id":vid,"vod_name":name,"vod_pic":pic,"vod_year":str(x.get("year") or ""),"vod_area":x.get("region") or "","vod_content":content,"vod_play_from":"光鸭云盘","vod_play_url":"#".join(eps)}]}
def playerContent(self,flag,id,vipFlags):
o=self._dec(id);u=o.get("url",id) if isinstance(o,dict) else id
if not u:return {"parse":1,"jx":0,"url":""}
if re.search(r"\.(m3u8|mp4|mkv|flv|avi|mov)(\?|$)",u,re.I):return {"parse":0,"jx":0,"url":u,"header":json.dumps({"User-Agent":self.headers["User-Agent"],"Referer":self.host+"/"})}
return {"parse":0,"jx":0,"url":"push://"+self._code_url(u,o.get("pwd","") if isinstance(o,dict) else "") if u.startswith("http") else u}
def localProxy(self,params): return [404,"text/plain",""]
def isVideoFormat(self,url): return bool(re.search(r"\.(m3u8|mp4|mkv|flv|avi|mov)(\?|$)",str(url),re.I))
def manualVideoCheck(self): return False
def destroy(self):
try:
with open(self.cache_file,"w",encoding="utf-8") as f:json.dump(self.pic_cache,f,ensure_ascii=False,separators=(",",":"))
except Exception:None
return None