From 73e705eb40f6e6f18c0cb98f2467d3933b696470 Mon Sep 17 00:00:00 2001 From: XC-Pro <19840+xc-pro@noreply.example.org> Date: Fri, 14 Aug 2026 11:25:18 +0200 Subject: [PATCH] =?UTF-8?q?=E4=B8=8A=E4=BC=A0=E6=96=87=E4=BB=B6=E8=87=B3?= =?UTF-8?q?=E3=80=8Cpy=E3=80=8D?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- py/剧迷.py | 116 +++++ py/可可.py | 426 ++++++++++++++++ py/咖啡直播.py | 375 ++++++++++++++ py/聚合直播.py | 1260 ++++++++++++++++++++++++++++++++++++++++++++++++ py/酷我.py | 396 +++++++++++++++ 5 files changed, 2573 insertions(+) create mode 100644 py/剧迷.py create mode 100644 py/可可.py create mode 100644 py/咖啡直播.py create mode 100644 py/聚合直播.py create mode 100644 py/酷我.py diff --git a/py/剧迷.py b/py/剧迷.py new file mode 100644 index 0000000..ee80772 --- /dev/null +++ b/py/剧迷.py @@ -0,0 +1,116 @@ +# -*- coding: utf-8 -*- +import json +import re +from urllib.parse import quote, urljoin + +import requests +from lxml import etree +from base.spider import Spider + + +class Spider(Spider): + def getName(self): return "Gimy影视" + + def init(self, extend=""): + self.host = "https://gimytv.biz" + self.headers = {"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36", "Referer": self.host + "/"} + self.classes = [{"type_id": "2", "type_name": "电视剧"}, {"type_id": "4", "type_name": "动漫"}, {"type_id": "3", "type_name": "综艺"}, {"type_id": "1", "type_name": "电影"}, {"type_id": "25", "type_name": "短剧"}] + subtypes = { + "1": [("全部", "1"), ("动作片", "6"), ("喜剧片", "7"), ("爱情片", "8"), ("科幻片", "9"), ("恐怖片", "10"), ("剧情片", "11"), ("战争片", "12"), ("动画电影", "24")], + "2": [("全部", "2"), ("陆剧", "13"), ("短剧", "25"), ("韩剧", "15"), ("美剧", "16"), ("日剧", "20"), ("台剧", "14"), ("海外剧", "21"), ("港剧", "22"), ("纪录片", "23")], + "3": [("全部", "3")], "4": [("全部", "4")], "25": [("全部", "25")] + } + years = [{"n": "全部", "v": ""}] + [{"n": str(x), "v": str(x)} for x in range(2026, 2015, -1)] + sorts = [{"n": "最新更新", "v": "time"}, {"n": "最新上架", "v": "time_add"}, {"n": "周人气", "v": "hits_week"}, {"n": "总人气", "v": "hits"}] + self.filters = {tid: [{"key": "type", "name": "类型", "value": [{"n": n, "v": v} for n, v in values]}, {"key": "year", "name": "年份", "value": years}, {"key": "by", "name": "排序", "value": sorts}] for tid, values in subtypes.items()} + + def _get(self, url, sid=""): + try: + headers = dict(self.headers) + if sid: headers["Cookie"] = "sid=" + str(sid) + response = requests.get(url, headers=headers, timeout=15) + response.raise_for_status() + response.encoding = response.apparent_encoding or "utf-8" + return response.text + except Exception: + return "" + + def _fix(self, url): return urljoin(self.host + "/", url or "") + + def _parse_list(self, html): + if not html: return [] + tree, result, seen = etree.HTML(html), [], set() + for node in tree.xpath('//a[contains(concat(" ",normalize-space(@class)," ")," video-pic ") and contains(@href,"/voddetail/")]'): + match = re.search(r"/voddetail/(\d+)\.html", node.get("href", "")) + if not match or match.group(1) in seen: continue + seen.add(match.group(1)) + name = node.get("title") or "".join(node.xpath('.//img/@alt')).strip() + pic = node.get("data-original") or node.get("data-src") or node.get("data-lazyload") or "".join(node.xpath('.//img/@data-original | .//img/@data-src | .//img/@src')) + remark = "".join(node.xpath('.//*[contains(@class,"note")]//text()')).strip() + result.append({"vod_id": match.group(1), "vod_name": name.strip(), "vod_pic": self._fix(pic), "vod_remarks": remark}) + return result + + def _pagecount(self, tree, page): + values = [int(x) for x in tree.xpath('//a[contains(@href,"/vodshow/")]/@href') for x in re.findall(r"-(\d+)---", x)] + return max(values + [page]) + + def homeContent(self, filter): return {"class": self.classes, "list": self._parse_list(self._get(self.host + "/")), "filters": self.filters} + + def categoryContent(self, tid, pg, filter, extend): + page, ext = max(1, int(pg or 1)), extend if isinstance(extend, dict) else {} + selected, year, by = str(ext.get("type") or tid), str(ext.get("year") or ""), str(ext.get("by") or "time") + url = f"{self.host}/vodshow/{selected}--{by}------{page}---{year}.html" + html = self._get(url) + tree = etree.HTML(html) if html else etree.HTML("") + videos = self._parse_list(html) + return {"page": page, "pagecount": self._pagecount(tree, page), "limit": len(videos), "total": self._pagecount(tree, page) * max(len(videos), 1), "list": videos} + + def detailContent(self, ids): + result = [] + for vid in ids: + html = self._get(f"{self.host}/voddetail/{vid}.html") + if not html: continue + tree = etree.HTML(html) + name = "".join(tree.xpath('//h1/text()')).strip() or "".join(tree.xpath('//h2/text()')).strip() + pic = "".join(tree.xpath('//meta[@property="og:image"]/@content | //div[contains(@class,"detail-pic")]//img/@data-original | //div[contains(@class,"detail-pic")]//img/@src')) + content = " ".join(x.strip() for x in tree.xpath('//span[contains(@class,"detail-intro")]//text() | //div[contains(@class,"details-content-all")]//text()') if x.strip()) + sources, playlists = [], [] + panels = tree.xpath('//div[contains(concat(" ",normalize-space(@class)," ")," playlist-mobile ")]') + for panel in panels: + episodes = panel.xpath('.//ul//a[contains(@href,"/video/")]') + if not episodes: continue + source = "".join(panel.xpath('./li[1]//text()')).strip() or "".join(panel.xpath('./span[1]//text()')).strip() or f"线路{len(sources) + 1}" + plays = [] + for episode in episodes: + href = episode.get("href", "") + sid_match = re.search(r"sid=(\d+)", href) + onclick_match = re.search(r"changeSid\(['\"]?(\d+)", episode.get("onclick", "")) + sid = (onclick_match or sid_match).group(1) if onclick_match or sid_match else "1" + path = href.split("#", 1)[0] + label = "".join(episode.xpath(".//text()")).strip() or str(len(plays) + 1) + plays.append(f"{label}${path}@@{sid}") + sources.append(source) + playlists.append("#".join(plays)) + result.append({"vod_id": str(vid), "vod_name": name, "vod_pic": self._fix(pic), "vod_content": content, "vod_play_from": "$$$".join(sources), "vod_play_url": "$$$".join(playlists)}) + return {"list": result} + + def searchContent(self, key, quick, pg="1"): + page = max(1, int(pg or 1)) + url = f"{self.host}/vodsearch/-------------.html?wd={quote(key)}&page={page}" + return {"page": page, "pagecount": page, "list": self._parse_list(self._get(url))} + + def playerContent(self, flag, id, vipFlags): + path, sid = (id.rsplit("@@", 1) + ["1"])[:2] if "@@" in id else (id, "1") + url = self._fix(path) + html = self._get(url, sid) + marker = "var player_aaaa=" + if marker in html: + try: + data = json.JSONDecoder().raw_decode(html.split(marker, 1)[1])[0] + play_url = data.get("url", "") + if int(data.get("encrypt", 0)) == 1: play_url = quote(play_url, safe=":/?&=%") + if play_url and any(x in play_url.lower() for x in (".m3u8", ".mp4", ".flv")): + return {"parse": 0, "url": play_url, "header": {"User-Agent": self.headers["User-Agent"], "Referer": url}} + except Exception: + pass + return {"parse": 1, "url": url, "header": {**self.headers, "Cookie": "sid=" + sid}} diff --git a/py/可可.py b/py/可可.py new file mode 100644 index 0000000..26ff1cf --- /dev/null +++ b/py/可可.py @@ -0,0 +1,426 @@ +#coding=utf-8 +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +""" +TVBox / 影视仓 Python源脚本 +站点: 可可影视 (103.51.147.112:51120) +""" + +import sys +import re +import json +import requests +import urllib3 +from urllib.parse import quote +from pyquery import PyQuery as pq +sys.path.append('..') +from base.spider import Spider + +# 禁用SSL警告 +urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) + +class Spider(Spider): + + def __init__(self): + super().__init__() + self.site = 'https://103.51.147.112:51120' + self.session = requests.Session() + self.session.headers.update({ + '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': 'https://103.51.147.112:51120/' + }) + self.cateManual = { + '电影': '1', + '连续剧': '2', + '动漫': '3', + '综艺纪录': '4', + '短剧': '6' + } + # 隐晦的站点标识 + self._mark = chr(26143) + chr(27827) + + def init(self, extend=""): + pass + + def getName(self): + return "可可影视" + + def isVideoFormat(self, url): + pass + + def manualVideoCheck(self): + pass + + def homeContent(self, filter): + result = {'class': [], 'filters': {}, 'list': [], 'parse': 0, 'jx': 0} + for k, v in self.cateManual.items(): + result['class'].append({ + 'type_id': str(v), + 'type_name': k + }) + return result + + def homeVideoContent(self): + videos = [] + try: + url = f'{self.site}/channel/1.html' + r = self.session.get(url, timeout=15, verify=False) + r.encoding = 'utf-8' + doc = pq(r.text) + items = doc('.module-item') + seen = set() + for item in items.items(): + a = item.find('.v-item') + href = a.attr('href') or '' + vid = self.getVid(href) + if not vid or vid in seen: + continue + seen.add(vid) + + # 标题 + titles = item.find('.v-item-title') + title = '' + for j in range(len(titles)): + t = titles.eq(j).text().strip() + if t and t != '可可影视-kekys.com': + title = t + break + + # 图片 + pic = '' + imgs = item.find('img') + for j in range(len(imgs)): + img = imgs.eq(j) + src = img.attr('data-original') or '' + if src and 'placeholder' not in src and 'logo_placeholder' not in src: + pic = src + break + if pic and pic.startswith('/'): + pic = 'https://vres.zyxpedu.com' + pic + + # 备注 + note = '' + bottom = item.find('.v-item-bottom span') + if bottom.length: + note = bottom.text().strip() + + if title: + videos.append({ + 'vod_id': vid, + 'vod_name': title, + 'vod_pic': pic, + 'vod_remarks': note + }) + except Exception as e: + print(f'homeVideoContent error: {e}') + return {'list': videos[:24], 'parse': 0, 'jx': 0} + + def categoryContent(self, tid, pg, filter, extend): + result = {'list': [], 'parse': 0, 'jx': 0} + page = int(pg) if pg else 1 + try: + url = f'{self.site}/channel/{tid}.html?page={page}' + r = self.session.get(url, timeout=15, verify=False) + r.encoding = 'utf-8' + doc = pq(r.text) + items = doc('.module-item') + for item in items.items(): + a = item.find('.v-item') + href = a.attr('href') or '' + vid = self.getVid(href) + if not vid: + continue + + # 标题 + titles = item.find('.v-item-title') + title = '' + for j in range(len(titles)): + t = titles.eq(j).text().strip() + if t and t != '可可影视-kekys.com': + title = t + break + + # 图片 + pic = '' + imgs = item.find('img') + for j in range(len(imgs)): + img = imgs.eq(j) + src = img.attr('data-original') or '' + if src and 'placeholder' not in src and 'logo_placeholder' not in src: + pic = src + break + if pic and pic.startswith('/'): + pic = 'https://vres.zyxpedu.com' + pic + + # 备注 + note = '' + bottom = item.find('.v-item-bottom span') + if bottom.length: + note = bottom.text().strip() + + if title: + result['list'].append({ + 'vod_id': vid, + 'vod_name': title, + 'vod_pic': pic, + 'vod_remarks': note + }) + except Exception as e: + print(f'categoryContent error: {e}') + + result['page'] = page + # 修复：满页24条说明还有下一页，否则已到尾页 + result['pagecount'] = page + 1 if len(result['list']) >= 24 else page + result['limit'] = len(result['list']) + result['total'] = len(result['list']) + return result + + def detailContent(self, ids): + result = {'list': [], 'parse': 0, 'jx': 0} + vid = ids[0] if ids else '' + if not vid: + return result + try: + url = f'{self.site}/detail/{vid}.html' + r = self.session.get(url, timeout=15, verify=False) + r.encoding = 'utf-8' + html = r.text + + # 标题：从title标签提取，最可靠 + title = '' + title_match = re.search(r'(.+?)', html) + if title_match: + title = title_match.group(1).split('-')[0].strip() + # 去掉特殊字符水印 + title = re.sub(r'[𝕜𝕜𝕪𝕤𝟘𝟙𝕔𝕠𝕞.\s]+', ' ', title).strip() + title = re.sub(r'\s+', ' ', title).strip() + + # 图片：从og:image提取 + pic = '' + og_img = re.search(r']+href="(/play/\d+-(\d+)-(\d+)\.html)"[^>]+class="episode-item"[^>]*>(.*?)', html, re.DOTALL) + # 如果没匹配到，试试class在href前面的情况 + if not all_play: + all_play = re.findall(r']+class="episode-item"[^>]+href="(/play/\d+-(\d+)-(\d+)\.html)"[^>]*>(.*?)', html, re.DOTALL) + for href, sid, nid, link_html in all_play: + text = re.sub(r'<[^>]+>', '', link_html).strip() + if not text: + continue + if sid not in episodes_by_sid: + episodes_by_sid[sid] = [] + episodes_by_sid[sid].append(f'{text}${href}') + if sid not in seen_sids: + seen_sids.add(sid) + sids_in_order.append(sid) + + # 线路名称 + source_labels = [] + all_labels = re.findall(r'class="source-item-label"[^>]*>([^<]+)]+\.(?:m3u8|mp4)[^\s"\'<>]*', r.text) + if all_urls: + for u in all_urls: + if 'index.m3u8' in u or 'video.m3u8' in u or '.mp4' in u: + video_url = u + break + if not video_url: + video_url = all_urls[0] + + # 修复：处理相对路径视频URL + if video_url and not video_url.startswith('http'): + if video_url.startswith('/'): + video_url = self.site + video_url + else: + video_url = self.site + '/' + video_url + + if video_url: + result['parse'] = 0 + result['url'] = video_url + result['jx'] = 0 + result['header'] = { + '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 + '/' + } + else: + result['parse'] = 1 + result['url'] = play_url + result['jx'] = 0 + result['header'] = { + '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 + '/' + } + except Exception as e: + print(f'playerContent error: {e}') + result['parse'] = 1 + result['url'] = id if id.startswith('http') else self.site + id + result['jx'] = 0 + result['header'] = { + '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 + '/' + } + return result + + def searchContent(self, key, quick, pg='1'): + result = {'list': [], 'parse': 0, 'jx': 0} + page = int(pg) if pg else 1 + try: + # 先访问搜索页获取token + search_url = f'{self.site}/search?k={quote(key)}' + r = self.session.get(search_url, timeout=15, verify=False) + r.encoding = 'utf-8' + t_match = re.search(r'name="t" value="([^"]+)"', r.text) + t = t_match.group(1) if t_match else '' + + if t: + url = f'{self.site}/search?k={quote(key)}&t={quote(t)}' + if page > 1: + url += f'&page={page}' + r = self.session.get(url, timeout=15, verify=False) + r.encoding = 'utf-8' + + doc = pq(r.text) + items = doc('.search-result-item') + for item in items.items(): + # 修复：兼容 href 在元素本身或子级 标签上的情况 + href = item.attr('href') or '' + if not href: + a_tag = item.find('a') + if a_tag.length: + href = a_tag.attr('href') or '' + + vid = self.getVid(href) + if not vid: + continue + + # 标题 + title = '' + title_elem = item.find('.title') + if title_elem.length: + title = title_elem.text().strip() + if not title: + img = item.find('img') + if img.length: + title = img.attr('alt') or img.attr('title') or '' + title = title.strip() + + # 图片 + pic = '' + imgs = item.find('img') + for j in range(len(imgs)): + img = imgs.eq(j) + src = img.attr('data-original') or img.attr('src') or '' + if src and 'placeholder' not in src and 'logo_placeholder' not in src: + pic = src + break + if pic and pic.startswith('/'): + pic = 'https://vres.zyxpedu.com' + pic + + if title: + result['list'].append({ + 'vod_id': vid, + 'vod_name': title, + 'vod_pic': pic, + 'vod_remarks': '' + }) + except Exception as e: + print(f'searchContent error: {e}') + + result['page'] = page + result['pagecount'] = page + 1 if len(result['list']) > 0 else page + result['limit'] = len(result['list']) + result['total'] = len(result['list']) + return result + + def localProxy(self, params): + return [200, "video/MP2T", {}, ""] + + def getVid(self, url): + if not url: + return '' + m = re.search(r'/detail/(\d+)\.html', url) + if m: + return m.group(1) + m = re.search(r'/play/(\d+)-', url) + if m: + return m.group(1) + return '' diff --git a/py/咖啡直播.py b/py/咖啡直播.py new file mode 100644 index 0000000..b0c181b --- /dev/null +++ b/py/咖啡直播.py @@ -0,0 +1,375 @@ +# coding=utf-8 +#!/usr/bin/env python3 +# @name 咖啡直播 +# @author 转自 OmniBox JS +# @description 体育赛事录像回放 + 直播（足球/篮球/NBA） +# @version 2.0.0 + +import json +import requests + + +class Spider: + + def getName(self): + return "咖啡直播" + + def getDependence(self): + return [] + + def init(self, extend=""): + self.host = "https://kafeizhibo.cc" + self.headers = { + "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/134.0.0.0 Safari/537.36", + "Accept": "application/json, text/plain, */*", + "Referer": "https://kafeizhibo.com/live/all" + } + + def log(self, msg): + print("[咖啡直播] " + str(msg)) + + # ========================================================= + # 工具 + # ========================================================= + + def _normalize_url(self, path): + if not path: + return "" + if path.startswith("http"): + return path + if path.startswith("//"): + return "https:" + path + return self.host + ("" if path.startswith("/") else "/") + path + + def _get(self, path, params=None, referer=None): + h = dict(self.headers) + if referer: + h["Referer"] = referer + resp = requests.get(self.host + path, headers=h, params=params, timeout=10) + return resp.json() + + # ========================================================= + # 直播部分 + # ========================================================= + + def _fetch_live_all(self): + """GET /api/v1/archor — 返回所有正在直播的频道""" + return self._get("/api/v1/archor", referer=self.host + "/live/all") + + def _parse_live_list(self, items, category_filter=None): + """ + category: 1=足球, 2=篮球, None=全部 + 每个 archor 代表一个独立直播频道（同一场球可能有多个频道） + 合并同 match_id 的频道到一个 vod，多线路在 detail 里处理 + """ + # 按 room_id 去重（同一 room_id 只取第一个，避免重复） + seen_rooms = set() + result = [] + for item in items: + if category_filter and item.get("category") != category_filter: + continue + room_id = str(item.get("room_id", "")) + if room_id in seen_rooms: + continue + seen_rooms.add(room_id) + + home = item.get("home_team", "") + away = item.get("away_team", "") + league = item.get("league_name", "") + h_score = item.get("home_score", 0) + a_score = item.get("away_score", 0) + title = "{} vs {} ({})".format(home, away, league) + + pic = self._normalize_url(item.get("screenshot", "")) + if not pic or "default" in pic: + mi = item.get("match_info") or {} + pic = mi.get("home_team_logo", "") + + result.append({ + "vod_id": "live_{}".format(room_id), + "vod_name": "🔴 " + title, + "vod_pic": pic, + "vod_remarks": "{} - {} | {}".format(h_score, a_score, item.get("name", "")), + }) + return result + + def _detail_live(self, room_id): + """GET /api/v1/room/{room_id} — 获取直播间多线路""" + try: + data = self._get( + "/api/v1/room/{}".format(room_id), + referer=self.host + "/room/{}".format(room_id) + ) + if data.get("code") != 200 or not data.get("data"): + return {"list": []} + + d = data["data"] + room_info = d.get("room_info", {}) + signals = d.get("signals", []) + + home = room_info.get("home_team", "") + away = room_info.get("away_team", "") + league = room_info.get("league", "") + h_score = room_info.get("home_score", 0) + a_score = room_info.get("away_score", 0) + title = "{} vs {} ({})".format(home, away, league) + + teams = d.get("teams", {}) + pic = (teams.get("home") or {}).get("logo", "") + + # 每条 signal 是一个线路（官方直播/原声直播） + episodes = [] + for sig in signals: + url = sig.get("stream_url", "") + if url: + name = sig.get("name", "线路") + episodes.append("{}${}".format(name, url)) + + # 如果 signals 为空，fallback 到 archor + if not episodes: + archor = d.get("archor", {}) + url = archor.get("stream_url", "") + if url: + episodes.append("{}${}".format(archor.get("name", "直播"), url)) + + vod = { + "vod_id": "live_{}".format(room_id), + "vod_name": "🔴 " + title, + "vod_pic": pic, + "vod_content": "{} {} vs {}，比分 {} - {}".format( + league, home, away, h_score, a_score + ), + "vod_play_from": "直播线路", + "vod_play_url": "#".join(episodes), + } + return {"list": [vod]} + except Exception as e: + self.log("直播详情失败: " + str(e)) + return {"list": []} + + # ========================================================= + # 录像部分 + # ========================================================= + + def _fetch_recordings(self, page=1, size=30, league=None, type_id=None): + params = {"page": page, "size": size} + if league: + params["league"] = league + elif type_id and type_id not in ("all", "nba", "live_all", "live_1", "live_2"): + params["type"] = type_id + h = dict(self.headers) + h["Referer"] = self.host + "/pc/replay" + resp = requests.get(self.host + "/api/v1/recordings", headers=h, params=params, timeout=10) + return resp.json() + + def _parse_video_list(self, items): + result = [] + for item in items: + title = "{} vs {} ({})".format( + item["home_team"], item["away_team"], item["league_name"] + ) + score = "{} - {}".format(item["home_score"], item["away_score"]) + pic = item.get("cover_image", "") + if pic and not pic.startswith("http"): + pic = self._normalize_url(pic) + if not pic or "default_cover" in pic: + pic = item.get("home_team_logo", "") + remarks = "{} | {} | {}个录像".format( + score, item["start_time"], item.get("recording_count", 0) + ) + result.append({ + "vod_id": str(item["match_id"]), + "vod_name": title, + "vod_pic": pic, + "vod_remarks": remarks, + }) + return result + + def _detail_recording(self, vid): + try: + h = dict(self.headers) + h["Referer"] = self.host + "/pc/replay" + resp = requests.get( + "{}/api/v1/match/{}/recordings".format(self.host, vid), + headers=h, + timeout=10 + ) + data = resp.json() + if data.get("code") != 200 or not data.get("data"): + return {"list": []} + + match = data["data"]["match"] + replays = data["data"].get("replays", []) + highlights = data["data"].get("highlights", []) + + title = "{} vs {} ({})".format( + match["home_team"], match["away_team"], match["league_name"] + ) + pic = match.get("home_team_logo") or match.get("away_team_logo") or "" + + episodes = [] + for idx, rec in enumerate(replays): + if rec.get("video_url"): + name = rec.get("title") or "录像{}".format(idx + 1) + episodes.append("{}${}".format(name, rec["video_url"])) + for idx, rec in enumerate(highlights): + if rec.get("video_url"): + name = rec.get("title") or "集锦{}".format(idx + 1) + episodes.append("{}${}".format(name, rec["video_url"])) + + vod = { + "vod_id": str(vid), + "vod_name": title, + "vod_pic": pic, + "vod_content": "{} {} {} vs {}，比分 {} - {}，比赛时间：{}".format( + match["league_name"], match.get("match_round", ""), + match["home_team"], match["away_team"], + match["home_score"], match["away_score"], + match["start_time"] + ), + "vod_play_from": "录像源", + "vod_play_url": "#".join(episodes), + } + return {"list": [vod]} + except Exception as e: + self.log("录像详情失败: " + str(e)) + return {"list": []} + + # ========================================================= + # FongMi 接口 + # ========================================================= + + def homeContent(self, filter): + categories = [ + # 直播分类 + {"type_id": "live_all", "type_name": "🔴 直播全部"}, + {"type_id": "live_1", "type_name": "🔴 直播足球"}, + {"type_id": "live_2", "type_name": "🔴 直播篮球"}, + # 录像分类 + {"type_id": "all", "type_name": "录像全部"}, + {"type_id": "1", "type_name": "录像足球"}, + {"type_id": "2", "type_name": "录像篮球"}, + {"type_id": "nba", "type_name": "录像NBA"}, + ] + # 首页展示直播列表 + try: + data = self._fetch_live_all() + vod_list = self._parse_live_list(data.get("data", [])) if data.get("code") == 200 else [] + except Exception as e: + self.log("首页失败: " + str(e)) + vod_list = [] + return {"class": categories, "list": vod_list} + + def homeVideoContent(self): + return {"list": []} + + def categoryContent(self, tid, pg, filter, extend): + pg = int(pg) if pg else 1 + + # ---- 直播分类 ---- + if tid in ("live_all", "live_1", "live_2"): + try: + data = self._fetch_live_all() + if data.get("code") == 200: + cat = None if tid == "live_all" else int(tid.split("_")[1]) + vod_list = self._parse_live_list(data.get("data", []), category_filter=cat) + else: + vod_list = [] + except Exception as e: + self.log("直播分类失败: " + str(e)) + vod_list = [] + return {"list": vod_list, "page": 1, "pagecount": 1, "limit": 100, "total": len(vod_list)} + + # ---- 录像分类 ---- + try: + if tid == "nba": + data = self._fetch_recordings(pg, 20, league="NBA") + size = 20 + elif tid == "all": + data = self._fetch_recordings(pg, 30) + size = 30 + else: + data = self._fetch_recordings(pg, 30, type_id=tid) + size = 30 + + vod_list = [] + pagecount = 1 + if data.get("code") == 200 and data.get("data"): + vod_list = self._parse_video_list(data["data"]) + pagecount = pg + 1 if len(data["data"]) == size else pg + except Exception as e: + self.log("录像分类失败: " + str(e)) + vod_list = [] + pagecount = 1 + + return {"list": vod_list, "page": pg, "pagecount": pagecount, "limit": 30, "total": len(vod_list)} + + def detailContent(self, ids): + vid = ids[0] if isinstance(ids, list) and ids else str(ids) + if vid.startswith("live_"): + room_id = vid[5:] # 去掉 "live_" 前缀 + return self._detail_live(room_id) + else: + return self._detail_recording(vid) + + def searchContent(self, key, quick, pg=1): + if not key: + return {"list": []} + keyword = key.lower() + result = [] + + # 搜索直播 + try: + data = self._fetch_live_all() + if data.get("code") == 200: + for item in data.get("data", []): + if (keyword in item.get("home_team", "").lower() + or keyword in item.get("away_team", "").lower() + or keyword in item.get("league_name", "").lower() + or keyword in item.get("title", "").lower()): + room_id = str(item.get("room_id", "")) + home = item.get("home_team", "") + away = item.get("away_team", "") + league = item.get("league_name", "") + result.append({ + "vod_id": "live_{}".format(room_id), + "vod_name": "🔴 {} vs {} ({})".format(home, away, league), + "vod_pic": "", + "vod_remarks": "直播中", + }) + except Exception as e: + self.log("搜索直播失败: " + str(e)) + + # 搜索录像 + try: + data = self._fetch_recordings(1, 100) + if data.get("code") == 200: + for item in data["data"]: + if (keyword in item["home_team"].lower() + or keyword in item["away_team"].lower() + or keyword in item["league_name"].lower()): + title = "{} vs {} ({})".format( + item["home_team"], item["away_team"], item["league_name"] + ) + result.append({ + "vod_id": str(item["match_id"]), + "vod_name": title, + "vod_pic": "", + "vod_remarks": "{} - {}".format(item["home_score"], item["away_score"]), + }) + except Exception as e: + self.log("搜索录像失败: " + str(e)) + + return {"list": result, "page": 1, "pagecount": 1} + + def playerContent(self, flag, id, vipFlags): + return { + "parse": 0, + "playUrl": "", + "url": id, + "header": json.dumps({ + "User-Agent": self.headers["User-Agent"], + "Referer": self.host, + "Origin": self.host, + }) + } diff --git a/py/聚合直播.py b/py/聚合直播.py new file mode 100644 index 0000000..a24d511 --- /dev/null +++ b/py/聚合直播.py @@ -0,0 +1,1260 @@ +# -*- coding: utf-8 -*- +# by @ +# 修复版本 - 参考最新三合一.js重构虎牙、斗鱼、B站直播逻辑 +# 修复：虎牙清晰度选择，确保ratio参数正确传递码率值 +# 修复：斗鱼切换分辨率只能播放1秒的问题（每次重新获取安全密钥和签名） +# 修复：B站使用特殊UA和WBI签名绕过-352风控 [^90^][^30^] +import json +import re +import sys +import time +import hashlib +import random +import urllib.parse +from base64 import b64decode, b64encode +from urllib.parse import parse_qs +import requests +from pyquery import PyQuery as pq +sys.path.append('..') +from base.spider import Spider +from concurrent.futures import ThreadPoolExecutor + + +class Spider(Spider): + + def init(self, extend=""): + # 初始化B站WBI密钥 + self.bili_wbi_keys = None + self.bili_wbi_expire = 0 + pass + + def getName(self): + return "直播" + + def isVideoFormat(self, url): + pass + + def manualVideoCheck(self): + pass + + def destroy(self): + pass + + headers = [ + { + # 特殊UA绕过B站风控 [^90^] + "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/126.0.0.0 Safari/537.36 Edg/126.0.0.0" + }, + { + "User-Agent": "Dart/3.4 (dart:io)" + } + ] + + excepturl = 'https://www.baidu.com' + + hosts = { + "huya": ["https://www.huya.com", "https://mp.huya.com"], + "douyu": "https://www.douyu.com", + "wangyi": "https://cc.163.com", + "bili": ["https://api.live.bilibili.com", "https://api.bilibili.com"] + } + + referers = { + "huya": "https://live.cdn.huya.com", + "douyu": "https://m.douyu.com", + "bili": "https://live.bilibili.com" + } + + playheaders = { + "wangyi": { + "User-Agent": "ExoPlayer", + "Connection": "Keep-Alive", + "Icy-MetaData": "1" + }, + "bili": { + 'Accept': '*/*', + 'Icy-MetaData': '1', + 'referer': 'https://live.bilibili.com', + 'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/126.0.0.0 Safari/537.36' + }, + 'huya': { + 'User-Agent': 'ExoPlayer', + 'Connection': 'Keep-Alive', + 'Icy-MetaData': '1' + }, + 'douyu': { + 'User-Agent': 'libmpv', + 'Icy-MetaData': '1' + } + } + + # WBI签名相关常量 [^30^] + MIXIN_KEY_ENC_TAB = [ + 46, 47, 18, 2, 53, 8, 23, 32, 15, 50, 10, 31, 58, 3, 45, 35, 27, 43, 5, 49, + 33, 9, 42, 19, 29, 28, 14, 39, 12, 38, 41, 13, 37, 48, 7, 16, 24, 55, 40, + 61, 26, 17, 0, 1, 60, 51, 30, 4, 22, 25, 54, 21, 56, 59, 6, 63, 57, 62, 11, + 36, 20, 34, 44, 52 + ] + + def _get_bili_wbi_keys(self): + """获取B站WBI密钥 [^30^]""" + try: + # 检查缓存 + if self.bili_wbi_keys and time.time() < self.bili_wbi_expire: + return self.bili_wbi_keys + + # 从导航接口获取 - 使用特殊UA [^90^] + resp = self.fetch( + 'https://api.bilibili.com/x/web-interface/nav', + headers={ + 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/126.0.0.0 Safari/537.36', + 'Referer': 'https://www.bilibili.com/' + } + ).json() + + if resp.get('code') != 0: + return None + + img_url = resp['data']['wbi_img']['img_url'] + sub_url = resp['data']['wbi_img']['sub_url'] + + # 提取文件名作为key + img_key = img_url.rsplit('/', 1)[1].split('.')[0] + sub_key = sub_url.rsplit('/', 1)[1].split('.')[0] + + self.bili_wbi_keys = (img_key, sub_key) + self.bili_wbi_expire = time.time() + 86400 # 24小时过期 + + return self.bili_wbi_keys + except Exception as e: + print(f"获取B站WBI密钥失败: {e}") + return None + + def _get_mixin_key(self, orig: str): + """生成mixin_key [^30^]""" + return ''.join([orig[i] for i in self.MIXIN_KEY_ENC_TAB])[:32] + + def _enc_wbi(self, params: dict): + """WBI签名 [^30^]""" + keys = self._get_bili_wbi_keys() + if not keys: + return params + + img_key, sub_key = keys + mixin_key = self._get_mixin_key(img_key + sub_key) + + # 添加时间戳 + params['wts'] = round(time.time()) + + # 排序参数 + params = dict(sorted(params.items())) + + # 过滤特殊字符 + params = { + k: ''.join(filter(lambda c: c not in "!'()*", str(v))) + for k, v in params.items() + } + + # 计算签名 + query = urllib.parse.urlencode(params) + w_rid = hashlib.md5((query + mixin_key).encode()).hexdigest() + + params['w_rid'] = w_rid + return params + + def process_bili(self): + """获取B站分类列表 - 使用WBI签名 [^30^]""" + try: + # 尝试获取分类列表 - 使用特殊UA和WBI签名 + params = {'need_entrance': 1, 'parent_id': 0} + signed_params = self._enc_wbi(params) + + data = self.fetch( + f'{self.hosts["bili"][0]}/room/v1/Area/getList', + params=signed_params, + headers=self.headers[0] + ).json() + + if data.get('code') == 0 and data.get('data'): + # 保存分类数据供后续使用 + self.bili_areas = data['data'] + return ('bili', [{'key': 'cate', 'name': '分类', + 'value': [{'n': i['name'], 'v': str(i['id'])} + for i in data['data']]}]) + return 'bili', None + except Exception as e: + print(f"bili处理错误: {e}") + # 使用默认分类 + return 'bili', [{'key': 'cate', 'name': '分类', + 'value': [{'n': '网游', 'v': '2'}, {'n': '手游', 'v': '3'}, + {'n': '单机', 'v': '6'}, {'n': '娱乐', 'v': '1'}, + {'n': '电台', 'v': '5'}, {'n': '虚拟主播', 'v': '9'}, + {'n': '生活', 'v': '10'}, {'n': '知识', 'v': '11'}, + {'n': '赛事', 'v': '13'}]}] + + def process_douyu(self): + try: + self.dyufdata = self.fetch( + f'{self.referers["douyu"]}/api/cate/list', + headers=self.headers[1] + ).json() + return ('douyu', [{'key': 'cate', 'name': '分类', + 'value': [{'n': i['cate1Name'], 'v': str(i['cate1Id'])} + for i in self.dyufdata['data']['cate1Info']]}]) + except Exception as e: + print(f"douyu错误: {e}") + return 'douyu', None + + def homeContent(self, filter): + result = {} + cateManual = { + "虎牙": "huya", + "斗鱼": "douyu", + "网易": "wangyi" + + } + classes = [] + filters = { + 'huya': [{'key': 'cate', 'name': '分类', + 'value': [{'n': '网游', 'v': '1'}, {'n': '单机', 'v': '2'}, + {'n': '娱乐', 'v': '8'}, {'n': '手游', 'v': '3'}]}] + } + + with ThreadPoolExecutor(max_workers=2) as executor: + futures = { + executor.submit(self.process_bili): 'bili', + executor.submit(self.process_douyu): 'douyu' + } + + for future in futures: + platform, filter_data = future.result() + if filter_data: + filters[platform] = filter_data + + for k in cateManual: + classes.append({ + 'type_name': k, + 'type_id': cateManual[k] + }) + + result['class'] = classes + result['filters'] = filters + return result + + def homeVideoContent(self): + pass + + def categoryContent(self, tid, pg, filter, extend): + vdata = [] + result = {} + pagecount = 9999 + result['page'] = pg + result['limit'] = 90 + result['total'] = 999999 + if tid == 'wangyi': + vdata, pagecount = self.wyccContent(tid, pg, filter, extend, vdata) + elif 'bili' in tid: + vdata, pagecount = self.biliContent(tid, pg, filter, extend, vdata) + elif 'huya' in tid: + vdata, pagecount = self.huyaContent(tid, pg, filter, extend, vdata) + elif 'douyu' in tid: + vdata, pagecount = self.douyuContent(tid, pg, filter, extend, vdata) + result['list'] = vdata + result['pagecount'] = pagecount + return result + + def wyccContent(self, tid, pg, filter, extend, vdata): + params = { + 'format': 'json', + 'start': (int(pg) - 1) * 20, + 'size': '20', + } + response = self.fetch(f'{self.hosts[tid]}/api/category/live/', params=params, headers=self.headers[0]).json() + for i in response['lives']: + if i.get('cuteid'): + bvdata = self.buildvod( + vod_id=f"{tid}@@{i['cuteid']}", + vod_name=i.get('title'), + vod_pic=i.get('cover'), + vod_remarks=i.get('nickname'), + style={"type": "rect", "ratio": 1.33} + ) + vdata.append(bvdata) + return vdata, 9999 + + def biliContent(self, tid, pg, filter, extend, vdata): + """B站分类内容 - 使用WBI签名绕过风控 [^30^][^90^]""" + try: + # 分类列表 - 显示子分类 + if extend.get('cate') and pg == '1' and 'click' not in tid: + # 从已保存的分类数据中找到对应分类的子分类 + if hasattr(self, 'bili_areas'): + for area in self.bili_areas: + if str(area['id']) == extend['cate']: + for sub_area in area.get('list', []): + v = self.buildvod( + vod_id=f"click_{tid}@@{extend['cate']}@@{sub_area['id']}", + vod_name=sub_area.get('name'), + vod_pic=sub_area.get('pic'), + vod_tag=1, + style={"type": "oval", "ratio": 1} + ) + vdata.append(v) + return vdata, 1 + # 如果没有找到子分类，直接返回空，让用户进入房间列表 + return vdata, 1 + + # 房间列表 - 使用getList接口并添加WBI签名 [^30^] + if 'click' in tid: + # 子分类房间 + ids = tid.split('_')[1].split('@@') + tid = ids[0] + parent_area_id = ids[1] + area_id = ids[2] + else: + # 默认使用分类ID作为parent_area_id，area_id为0表示该分类下所有 + parent_area_id = extend.get('cate', '2') # 默认网游 + area_id = 0 + + # 构建请求参数并添加WBI签名 [^30^] + params = { + 'parent_area_id': parent_area_id, + 'area_id': area_id, + 'page': pg, + 'platform': 'web', + 'sort_type': 'online' # 按热度排序 + } + signed_params = self._enc_wbi(params) + + # 调用getList接口 + api_url = f'{self.hosts[tid][0]}/xlive/web-interface/v1/second/getList' + data = self.fetch(api_url, params=signed_params, headers=self.headers[0]).json() + + # 如果WBI签名失败，尝试不带签名 + if data.get('code') == -352: + print("WBI签名失败，尝试无签名请求...") + params = { + 'parent_area_id': parent_area_id, + 'area_id': area_id, + 'page': pg, + 'platform': 'web', + 'sort_type': 'online' + } + data = self.fetch(api_url, params=params, headers=self.headers[0]).json() + + if data.get('code') == 0: + room_list = data.get('data', {}).get('list', []) + for room in room_list: + if room.get('roomid'): + # 处理在线人数显示 + online = room.get('online', 0) + if online > 10000: + online_str = f"{online / 10000:.1f}万" + else: + online_str = str(online) + + v = self.buildvod( + f"{tid}@@{room['roomid']}", + room.get('title', '未知标题'), + room.get('cover') or room.get('system_cover'), + f"{online_str}人", + 0, + room.get('uname', ''), + style={"type": "rect", "ratio": 1.33} + ) + vdata.append(v) + + # 检查是否有更多数据 + has_more = data.get('data', {}).get('has_more', 0) + if not has_more: + pagecount = int(pg) + else: + pagecount = 9999 + else: + print(f"B站API返回错误: {data.get('message', '未知错误')} (code: {data.get('code')})") + pagecount = 1 + + return vdata, pagecount + + except Exception as e: + print(f"B站内容获取错误: {e}") + return vdata, 1 + + def huyaContent(self, tid, pg, filter, extend, vdata): + if extend.get('cate') and pg == '1' and 'click' not in tid: + id = extend.get('cate') + data = self.fetch(f'{self.referers[tid]}/liveconfig/game/bussLive?bussType={id}', + headers=self.headers[1]).json() + for i in data['data']: + v = self.buildvod( + vod_id=f"click_{tid}@@{int(i['gid'])}", + vod_name=i.get('gameFullName'), + vod_pic=f'https://huyaimg.msstatic.com/cdnimage/game/{int(i["gid"])}-MS.jpg', + vod_tag=1, + style={"type": "oval", "ratio": 1} + ) + vdata.append(v) + return vdata, 1 + else: + gid = '' + if 'click' in tid: + ids = tid.split('_')[1].split('@@') + tid = ids[0] + gid = f'&gameId={ids[1]}' + data = self.fetch(f'{self.hosts[tid][0]}/cache.php?m=LiveList&do=getLiveListByPage&tagAll=0{gid}&page={pg}', + headers=self.headers[1]).json() + for i in data['data']['datas']: + if i.get('profileRoom'): + v = self.buildvod( + f"{tid}@@{i['profileRoom']}", + i.get('introduction'), + i.get('screenshot'), + str(int(i.get('totalCount', '1')) / 10000) + '万', + 0, + i.get('nick'), + style={"type": "rect", "ratio": 1.33} + + ) + vdata.append(v) + return vdata, 9999 + + def douyuContent(self, tid, pg, filter, extend, vdata): + if extend.get('cate') and pg == '1' and 'click' not in tid: + for i in self.dyufdata['data']['cate2Info']: + if str(i['cate1Id']) == extend['cate']: + v = self.buildvod( + vod_id=f"click_{tid}@@{i['cate2Id']}", + vod_name=i.get('cate2Name'), + vod_pic=i.get('icon'), + vod_remarks=i.get('count'), + vod_tag=1, + style={"type": "oval", "ratio": 1} + ) + vdata.append(v) + return vdata, 1 + else: + path = f'/japi/weblist/apinc/allpage/6/{pg}' + if 'click' in tid: + ids = tid.split('_')[1].split('@@') + tid = ids[0] + path = f'/gapi/rkc/directory/mixList/2_{ids[1]}/{pg}' + url = f'{self.hosts[tid]}{path}' + data = self.fetch(url, headers=self.headers[1]).json() + for i in data['data']['rl']: + v = self.buildvod( + vod_id=f"{tid}@@{i['rid']}", + vod_name=i.get('rn'), + vod_pic=i.get('rs16'), + vod_year=str(int(i.get('ol', 1)) / 10000) + '万', + vod_remarks=i.get('nn'), + style={"type": "rect", "ratio": 1.33} + ) + vdata.append(v) + return vdata, 9999 + + def detailContent(self, ids): + ids = ids[0].split('@@') + if ids[0] == 'wangyi': + vod = self.wyccDetail(ids) + elif ids[0] == 'bili': + vod = self.biliDetail(ids) + elif ids[0] == 'huya': + vod = self.huyaDetail(ids) + elif ids[0] == 'douyu': + vod = self.douyuDetail(ids) + return {'list': [vod]} + + def wyccDetail(self, ids): + try: + vdata = self.getpq(f'{self.hosts[ids[0]]}/{ids[1]}', self.headers[0])('script').eq(-1).text() + + def get_quality_name(vbr): + if vbr <= 600: + return "标清" + elif vbr <= 1000: + return "高清" + elif vbr <= 2000: + return "超清" + else: + return "蓝光" + + data = json.loads(vdata)['props']['pageProps']['roomInfoInitData'] + name = data['live'].get('title', ids[0]) + vod = self.buildvod(vod_name=data.get('keywords_suffix'), vod_remarks=data['live'].get('title'), + vod_content=data.get('description_suffix')) + resolution_data = data['live']['quickplay']['resolution'] + all_streams = {} + sorted_qualities = sorted(resolution_data.items(), + key=lambda x: x[1]['vbr'], + reverse=True) + for quality, data in sorted_qualities: + vbr = data['vbr'] + quality_name = get_quality_name(vbr) + for cdn_name, url in data['cdn'].items(): + if cdn_name not in all_streams and type(url) == str and url.startswith('http'): + all_streams[cdn_name] = [] + if isinstance(url, str) and url.startswith('http'): + all_streams[cdn_name].extend([quality_name, url]) + plists = [] + names = [] + for i, (cdn_name, stream_list) in enumerate(all_streams.items(), 1): + names.append(f'线路{i}') + pstr = f"{name}${ids[0]}@@{self.e64(json.dumps(stream_list))}" + plists.append(pstr) + vod['vod_play_from'] = "$$$".join(names) + vod['vod_play_url'] = "$$$".join(plists) + return vod + except Exception as e: + return self.handle_exception(e) + + def biliDetail(self, ids): + """ + B站直播详情 - 使用playUrl接口获取多清晰度 + """ + try: + room_id = ids[1] + + # 获取房间信息 + info_res = self.fetch( + f'{self.hosts["bili"][0]}/room/v1/Room/get_info?room_id={room_id}', + headers=self.headers[0] + ).json() + + if info_res.get('code') != 0: + return self.handle_exception(Exception("获取房间信息失败")) + + room_info = info_res['data'] + title = room_info.get('title', 'B站直播') + + vod = self.buildvod( + vod_name=title, + type_name=f"{room_info.get('parent_area_name', '')}/{room_info.get('area_name', '')}", + vod_director=room_info.get('uname', ''), + vod_remarks=f"在线{room_info.get('online', 0)}人" + ) + + # 获取播放地址信息 + play_res = self.fetch( + f'{self.hosts["bili"][0]}/room/v1/Room/playUrl?cid={room_id}&qn=10000&platform=web', + headers={ + **self.headers[0], + 'Referer': 'https://live.bilibili.com/', + 'Origin': 'https://live.bilibili.com' + } + ).json() + + if play_res.get('code') != 0: + return self.handle_exception(Exception("获取播放地址失败")) + + play_data = play_res['data'] + accept_quality = play_data.get('accept_quality', ['10000', '400', '250', '150']) + quality_desc = {item['qn']: item['desc'] for item in play_data.get('quality_description', [])} + + # 构建清晰度列表 + qualities = [] + for qn in sorted([int(q) for q in accept_quality], reverse=True): + desc = quality_desc.get(qn, f'清晰度{qn}') + qualities.append(f"{desc}$bili@@{room_id}@@{qn}") + + vod['vod_play_from'] = 'B站直播' + vod['vod_play_url'] = '#'.join(qualities) + return vod + + except Exception as e: + print(f"B站详情错误: {e}") + return self.handle_exception(e) + + def huyaDetail(self, ids): + """ + 虎牙播放详情 - 参考最新三合一.js重构 + 支持多线路多清晰度选择 + 核心算法：通过房间信息API获取uid、streamName和rateArray，为每个清晰度生成签名URL + 清晰度说明： + - 蓝光8M/6M/4M/10M = 8000/6000/4000/10000 kbps = 1080P+ + - 蓝光 = 3000 kbps = 1080P + - 超清 = 2000 kbps = 1080P (官方标准) + - 高清 = 1200 kbps = 720P + - 标清/流畅 = 500-800 kbps = 480P/540P + """ + try: + room_id = ids[1] + + # 1. 获取房间信息 + api_url = f'{self.hosts[ids[0]][1]}/cache.php?m=Live&do=profileRoom&roomid={room_id}' + res = self.fetch(api_url, headers=self.headers[0]) + + if res.status_code != 200: + return self.handle_exception(Exception(f"API请求失败: {res.status_code}")) + + data = res.json() + if not data or not data.get('data'): + return self.handle_exception(Exception("房间数据为空")) + + room_data = data['data'] + + # 2. 提取关键信息 + uid = room_data.get('profileInfo', {}).get('uid') + stream_info = room_data.get('stream', {}) + live_data = room_data.get('liveData', {}) + + if not uid: + return self.handle_exception(Exception("缺少uid")) + + # 3. 获取streamName和码率信息 + base_stream_list = stream_info.get('baseSteamInfoList', []) + if not base_stream_list: + return self.handle_exception(Exception("无直播流信息")) + + # 获取第一个CDN的streamName作为基准 + base_stream = base_stream_list[0] + stream_name = base_stream.get('sStreamName') + if not stream_name: + return self.handle_exception(Exception("无法获取streamName")) + + # 4. 构建VOD对象 + vod = self.buildvod( + vod_name=live_data.get('introduction', '虎牙直播'), + type_name=live_data.get('gameFullName', ''), + vod_director=live_data.get('nick', ''), + vod_remarks=live_data.get('contentIntro', ''), + ) + + # 5. 获取所有CDN线路 + cdn_list = [] + for stream in base_stream_list: + cdn_type = stream.get('sCdnType', 'AL') + flv_url = stream.get('sFlvUrl', '') + hls_url = stream.get('sHlsUrl', '') + stream_name_cdn = stream.get('sStreamName', stream_name) + + if flv_url: + cdn_list.append({ + 'cdn': cdn_type, + 'flv_base': flv_url, + 'hls_base': hls_url, + 'stream_name': stream_name_cdn, + 'priority': stream.get('iWebPriorityRate', 0) + }) + + # 按优先级排序 + cdn_list.sort(key=lambda x: x['priority'], reverse=True) + + # 6. 获取清晰度列表 (rateArray) + rate_array = stream_info.get('rateArray', []) + + # 如果没有rateArray，尝试从vMultiStreamInfo获取 + if not rate_array and 'vMultiStreamInfo' in room_data: + rate_array = room_data['vMultiStreamInfo'] + + # 如果仍然没有，使用默认清晰度（按虎牙官方标准） + if not rate_array: + rate_array = [ + {'sDisplayName': '蓝光4M', 'iBitRate': 4000}, + {'sDisplayName': '蓝光', 'iBitRate': 3000}, + {'sDisplayName': '超清', 'iBitRate': 2000}, # 2000kbps = 1080P + {'sDisplayName': '高清', 'iBitRate': 1200}, # 1200kbps = 720P + {'sDisplayName': '流畅', 'iBitRate': 500} + ] + + # 过滤和排序清晰度 + # 虎牙的rateArray中，iBitRate就是码率值，sDisplayName是显示名称 + # 需要确保：超清=2000kbps(1080P)，高清=1200kbps(720P) + filtered_rates = [] + seen_bitrates = set() + + for rate in rate_array: + bit_rate = rate.get('iBitRate', 0) + name = rate.get('sDisplayName', '') + + # 跳过重复的码率 + if bit_rate in seen_bitrates: + continue + + # 修正清晰度名称，确保符合虎牙标准 + # 2000kbps应该是超清(1080P)，不是高清 + if bit_rate == 2000 and ('高清' in name or '720' in name): + name = '超清' # 强制修正为超清 + elif bit_rate == 1200 and ('标清' in name or '480' in name): + name = '高清' # 1200kbps对应高清 + elif bit_rate == 2000 and name == '原画': + name = '超清' # 修正原画为超清 + + seen_bitrates.add(bit_rate) + filtered_rates.append({ + 'sDisplayName': name, + 'iBitRate': bit_rate + }) + + # 按码率从高到低排序 + sorted_rates = sorted(filtered_rates, key=lambda x: x['iBitRate'], reverse=True) + + # 7. 为每个CDN生成各清晰度的播放URL + play_lines = [] + line_names = [] + + for cdn_idx, cdn in enumerate(cdn_list[:3]): # 最多取3个CDN + cdn_name = cdn['cdn'] + line_names.append(f"线路{cdn_idx + 1}({cdn_name})") + + qualities = [] + for rate in sorted_rates: + quality_name = rate['sDisplayName'] + bit_rate = rate['iBitRate'] + + # 生成该清晰度的URL + quality_url = self._generate_huya_play_url( + cdn, uid, stream_name, bit_rate + ) + + qualities.extend([quality_name, quality_url]) + + # 编码该线路的所有清晰度 + encoded_qualities = self.e64(json.dumps(qualities)) + play_lines.append(f"{live_data.get('introduction', '直播')}${ids[0]}@@{encoded_qualities}") + + # 8. 构建播放数据 + vod['vod_play_from'] = "$$$".join(line_names) + vod['vod_play_url'] = "$$$".join(play_lines) + + return vod + + except Exception as e: + return self.handle_exception(e) + + def _generate_huya_play_url(self, cdn, uid, stream_name, bit_rate): + """ + 生成虎牙播放URL，参考最新三合一.js算法 + 关键：ratio参数必须正确设置为iBitRate值（如2000、4000等） + """ + # 基础URL构建 + flv_base = cdn['flv_base'] + stream = cdn['stream_name'] + + # 生成时间戳和签名参数 + timestamp = int(time.time()) + seqid = f"{uid}{timestamp}" + ss = hashlib.md5(f"{seqid}|huya_adr|102".encode()).hexdigest() + ws_time = hex(timestamp + 21600)[2:] # 16进制，有效期6小时 + + # 计算wsSecret + ws_secret = hashlib.md5( + f"DWq8BcJ3h6DJt6TY_{uid}_{stream_name}_{ss}_{ws_time}".encode() + ).hexdigest() + + # 构建基础URL + base_url = f"{flv_base}/{stream}.flv" + + # 关键修复：ratio参数直接使用iBitRate值 + # 超清=2000，高清=1200，蓝光=3000/4000/6000/8000 + if bit_rate > 0: + ratio_param = f"ratio={bit_rate}" + else: + # 原画/0码率时，使用默认2000或从URL推断 + ratio_param = "ratio=2000" + + # 构建完整URL + play_url = ( + f"{base_url}?{ratio_param}&wsSecret={ws_secret}&wsTime={ws_time}" + f"&ctype=huya_adr&seqid={seqid}&uid={uid}" + f"&fs=bgct&ver=1&t=102" + ) + + return play_url + + def douyuDetail(self, ids): + """ + 斗鱼播放详情 - 参考最新三合一.js重构 + 核心算法：设备ID生成 -> 获取加密密钥 -> 计算签名 -> 获取播放地址 + 修复：切换分辨率只能播放1秒的问题 + 方案：存储房间号和码率信息，在playerContent中实时获取对应码率的URL + """ + try: + channel = ids[1] + headers = self.gethr(0, zr=f'{self.hosts[ids[0]]}/{channel}') + + # 1. 初始化会话和设备ID (参考JS中的initialize和setupDeviceId) + session = {} + + # 请求首页获取Cookie + try: + home_res = self.fetch(f'{self.hosts[ids[0]]}/{channel}', headers=headers) + if home_res.headers.get('Set-Cookie'): + cookie_str = home_res.headers.get('Set-Cookie') + # 解析dy_did + did_match = re.search(r'dy_did=([a-f0-9]{32})', cookie_str) + if did_match: + device_id = did_match.group(1) + else: + device_id = self._generate_random_hex(32) + else: + device_id = self._generate_random_hex(32) + except: + device_id = self._generate_random_hex(32) + + session['dy_did'] = device_id + session['mantine-color-scheme-value'] = 'light' + + # 2. 获取房间基本信息 + betard_res = self.fetch(f'{self.hosts[ids[0]]}/betard/{channel}', headers=headers).json() + if not betard_res or not betard_res.get('room'): + return self.handle_exception(Exception("获取房间信息失败")) + + room_info = betard_res['room'] + vname = room_info.get('room_name', '斗鱼直播') + + vod = self.buildvod( + vod_name=vname, + vod_remarks=room_info.get('second_lvl_name', ''), + vod_director=room_info.get('nickname', ''), + ) + + # 3. 获取安全密钥 (参考JS中的getSecurityKey) + sec_url = f"{self.hosts[ids[0]]}/wgapi/livenc/liveweb/websec/getEncryption?did={device_id}" + sec_res = self.fetch(sec_url, headers=headers).json() + + if not sec_res or sec_res.get('error') != 0: + return self.handle_exception(Exception("获取加密密钥失败")) + + security_data = sec_res['data'] + secret_key = security_data.get('key') + random_str = security_data.get('rand_str') + enc_time = security_data.get('enc_time', 1) + enc_data = security_data.get('enc_data') + + # 4. 计算签名 (参考JS中的computeSignature) + current_time = int(time.time()) + + # 迭代计算MD5 + current = random_str + for _ in range(enc_time): + current = hashlib.md5(f"{current}{secret_key}".encode()).hexdigest() + + signature = hashlib.md5(f"{current}{secret_key}{channel}{current_time}".encode()).hexdigest() + + # 5. 请求播放地址 (参考JS中的requestStreamData) + play_payload = { + 'enc_data': enc_data, + 'tt': str(current_time), + 'did': device_id, + 'auth': signature, + 'cdn': '', + 'rate': '', + 'hevc': '0', + 'fa': '0', + 'ive': '0' + } + + play_api = f"{self.hosts[ids[0]]}/lapi/live/getH5PlayV1/{channel}" + + # 构建请求头带Cookie + play_headers = headers.copy() + cookie_str = '; '.join([f"{k}={v}" for k, v in session.items()]) + play_headers['Cookie'] = cookie_str + play_headers['Content-Type'] = 'application/x-www-form-urlencoded' + + play_res = requests.post(play_api, data=play_payload, headers=play_headers, timeout=10).json() + + if not play_res or play_res.get('error') != 0: + # 尝试旧版API + play_res = self._try_legacy_douyu_api(channel, device_id, signature, current_time, play_headers) + if not play_res: + return self.handle_exception(Exception("获取播放地址失败")) + + stream_info = play_res.get('data', {}) + + # 6. 检查并更新设备ID (参考JS中的checkAndUpdateDeviceId) + rtmp_live = stream_info.get('rtmp_live', '') + if rtmp_live: + did_match = re.search(r'did=([a-f0-9]{32})', rtmp_live) + if did_match and did_match.group(1) != device_id: + device_id = did_match.group(1) + session['dy_did'] = device_id + # 重新请求 + play_payload['did'] = device_id + play_res = requests.post(play_api, data=play_payload, headers=play_headers, timeout=10).json() + if play_res and play_res.get('error') == 0: + stream_info = play_res.get('data', {}) + + # 7. 提取播放URL和多码率信息 + stream_url = None + if stream_info.get('rtmp_url') and stream_info.get('rtmp_live'): + stream_url = f"{stream_info['rtmp_url']}/{stream_info['rtmp_live']}" + elif stream_info.get('hls_url'): + stream_url = stream_info['hls_url'] + + if not stream_url: + return self.handle_exception(Exception("无法获取播放地址")) + + # 8. 构建多码率选项 + multirates = stream_info.get('multirates', []) + + # 关键修复：存储房间号和码率信息，而不是直接存储URL + # 这样在切换清晰度时可以重新获取对应码率的签名URL + qualities = [] + + if multirates: + # 按码率排序 + sorted_rates = sorted(multirates, key=lambda x: x.get('bit', 0), reverse=True) + for rate in sorted_rates: + bit_rate = rate.get('rate', -1) + name = rate.get('name', f"{bit_rate}P") + + # 存储格式：码率值，用于playerContent中重新获取URL + # 使用特殊标记#来区分这是码率值而不是URL + qualities.extend([name, f"#{bit_rate}"]) + else: + # 只有原画 + qualities = ['原画', '#-1'] + + # 同时存储房间号和设备信息，用于重新获取URL + # 格式：房间号|设备ID|签名信息（base64编码） + session_info = { + 'channel': channel, + 'device_id': device_id, + 'secret_key': secret_key, + 'random_str': random_str, + 'enc_time': enc_time, + 'enc_data': enc_data + } + encoded_session = self.e64(json.dumps(session_info)) + + # 9. 构建播放数据 + # vod_play_url格式：房间名$平台@@base64(清晰度列表)@@base64(会话信息) + encoded_qualities = self.e64(json.dumps(qualities)) + vod['vod_play_from'] = '斗鱼直播' + vod['vod_play_url'] = f"{vname}${ids[0]}@@{encoded_qualities}@@{encoded_session}" + + return vod + + except Exception as e: + return self.handle_exception(e) + + def _generate_random_hex(self, length): + """生成随机十六进制字符串""" + hex_chars = '0123456789abcdef' + return ''.join(random.choice(hex_chars) for _ in range(length)) + + def _try_legacy_douyu_api(self, channel, device_id, signature, timestamp, headers): + """尝试使用旧版API获取播放地址""" + try: + legacy_payload = { + 'did': device_id, + 'tt': str(timestamp), + 'sign': signature, + 'cdn': '', + 'rate': '-1', + 'ver': 'Douyu_223061205', + 'iar': '1', + 'ive': '1', + 'hevc': '0', + 'fa': '0' + } + legacy_api = f"https://www.douyu.com/lapi/live/getH5Play/{channel}" + res = requests.post(legacy_api, data=legacy_payload, headers=headers, timeout=10) + return res.json() if res.status_code == 200 else None + except: + return None + + def _get_douyu_play_url(self, channel, device_id, secret_key, random_str, enc_time, enc_data, rate): + """ + 获取斗鱼指定码率的播放URL（带签名） + 用于切换清晰度时重新获取URL + """ + try: + current_time = int(time.time()) + + # 重新计算签名 + current = random_str + for _ in range(enc_time): + current = hashlib.md5(f"{current}{secret_key}".encode()).hexdigest() + + signature = hashlib.md5(f"{current}{secret_key}{channel}{current_time}".encode()).hexdigest() + + # 构建请求 + play_payload = { + 'enc_data': enc_data, + 'tt': str(current_time), + 'did': device_id, + 'auth': signature, + 'cdn': '', + 'rate': str(rate) if rate > 0 else '', + 'hevc': '0', + 'fa': '0', + 'ive': '0' + } + + play_api = f"https://www.douyu.com/lapi/live/getH5PlayV1/{channel}" + + headers = { + 'User-Agent': self.headers[0]['User-Agent'], + 'Referer': f'https://www.douyu.com/{channel}', + 'Origin': 'https://www.douyu.com', + 'Cookie': f'dy_did={device_id}; mantine-color-scheme-value=light', + 'Content-Type': 'application/x-www-form-urlencoded' + } + + play_res = requests.post(play_api, data=play_payload, headers=headers, timeout=10).json() + + if not play_res or play_res.get('error') != 0: + # 尝试旧版API + return self._get_douyu_play_url_legacy(channel, device_id, signature, current_time, rate) + + stream_info = play_res.get('data', {}) + + # 检查设备ID是否匹配 + if stream_info.get('rtmp_live'): + did_match = re.search(r'did=([a-f0-9]{32})', stream_info['rtmp_live']) + if did_match and did_match.group(1) != device_id: + # 设备ID不匹配，使用新设备ID重新获取 + return self._get_douyu_play_url(channel, did_match.group(1), secret_key, random_str, enc_time, enc_data, rate) + + if stream_info.get('rtmp_url') and stream_info.get('rtmp_live'): + return f"{stream_info['rtmp_url']}/{stream_info['rtmp_live']}" + elif stream_info.get('hls_url'): + return stream_info['hls_url'] + + return None + except Exception as e: + print(f"获取斗鱼播放URL失败: {e}") + return None + + def _get_douyu_play_url_legacy(self, channel, device_id, signature, timestamp, rate): + """使用旧版API获取斗鱼播放URL""" + try: + legacy_payload = { + 'did': device_id, + 'tt': str(timestamp), + 'sign': signature, + 'cdn': '', + 'rate': str(rate) if rate > 0 else '-1', + 'ver': 'Douyu_223061205', + 'iar': '1', + 'ive': '1', + 'hevc': '0', + 'fa': '0' + } + legacy_api = f"https://www.douyu.com/lapi/live/getH5Play/{channel}" + + headers = { + 'User-Agent': self.headers[0]['User-Agent'], + 'Referer': f'https://www.douyu.com/{channel}', + 'Cookie': f'dy_did={device_id}', + 'Content-Type': 'application/x-www-form-urlencoded' + } + + res = requests.post(legacy_api, data=legacy_payload, headers=headers, timeout=10) + if res.status_code == 200: + data = res.json() + if data.get('error') == 0: + stream_info = data.get('data', {}) + if stream_info.get('rtmp_url') and stream_info.get('rtmp_live'): + return f"{stream_info['rtmp_url']}/{stream_info['rtmp_live']}" + return None + except: + return None + + def searchContent(self, key, quick, pg="1"): + pass + + def playerContent(self, flag, id, vipFlags): + try: + ids = id.split('@@') + p = 1 + if ids[0] in ['wangyi']: + p, url = 0, json.loads(self.d64(ids[1])) + elif ids[0] == 'bili': + p, url = self.biliplay(ids) + elif ids[0] == 'huya': + p, url = self.huyaplay(ids) + elif ids[0] == 'douyu': + p, url = self.douyuplay(ids) + return {'parse': p, 'url': url, 'header': self.playheaders[ids[0]]} + except Exception as e: + return {'parse': 1, 'url': self.excepturl, 'header': self.headers[0]} + + def biliplay(self, ids): + """ + B站播放解析 - 使用playUrl接口获取指定清晰度直播流 + ids: [平台, 房间号, 清晰度qn] + 支持多线路返回 + """ + try: + room_id = ids[1] + qn = ids[2] if len(ids) > 2 else '10000' + + # 使用playUrl接口获取直播流 + play_url = f'{self.hosts["bili"][0]}/room/v1/Room/playUrl?cid={room_id}&qn={qn}&platform=web' + data = self.fetch(play_url, headers={ + **self.headers[0], + 'Referer': 'https://live.bilibili.com/', + 'Origin': 'https://live.bilibili.com' + }).json() + + if data.get('code') != 0: + return 1, self.excepturl + + play_data = data['data'] + durl_list = play_data.get('durl', []) + + if not durl_list: + return 1, self.excepturl + + # 构建多线路结果 [线路1, URL1, 线路2, URL2, ...] + urls = [] + for idx, item in enumerate(durl_list, 1): + url = item.get('url') + if url: + urls.extend([f'线路{idx}', url]) + + # 如果只有一条线路，直接返回URL + if len(urls) == 2: + return 0, urls[1] # 直接返回URL字符串 + + return 0, urls + + except Exception as e: + print(f"B站播放错误: {e}") + return 1, self.excepturl + + def huyaplay(self, ids): + """ + 虎牙播放解析 - 返回所有清晰度选项供用户选择 + ids[1] 格式: base64编码的 [清晰度名称1, URL1, 清晰度名称2, URL2, ...] + """ + try: + # ids[1] 是编码后的播放地址列表 [名称1, URL1, 名称2, URL2, ...] + decoded = json.loads(self.d64(ids[1])) + # decoded 是一个列表，奇数索引是名称，偶数索引是URL + return 0, decoded + except Exception as e: + print(f"虎牙播放解析错误: {e}") + return 1, self.excepturl + + def douyuplay(self, ids): + """ + 斗鱼播放解析 - 实时获取对应码率的播放URL + ids格式: [平台, base64(清晰度列表), base64(会话信息)] + 清晰度列表: [名称1, #码率1, 名称2, #码率2, ...] + #表示这是码率值，需要重新获取URL + """ + try: + if len(ids) < 3: + # 兼容旧格式 + decoded = json.loads(self.d64(ids[1])) + return 0, decoded + + # 解析清晰度列表和会话信息 + qualities = json.loads(self.d64(ids[1])) + session_info = json.loads(self.d64(ids[2])) + + channel = session_info['channel'] + device_id = session_info['device_id'] + secret_key = session_info['secret_key'] + random_str = session_info['random_str'] + enc_time = session_info['enc_time'] + enc_data = session_info['enc_data'] + + # 为每个清晰度实时获取播放URL + result = [] + for i in range(0, len(qualities), 2): + name = qualities[i] + rate_marker = qualities[i + 1] + + # 解析码率值（去掉#前缀） + if rate_marker.startswith('#'): + rate = int(rate_marker[1:]) + else: + rate = -1 + + # 实时获取对应码率的URL + play_url = self._get_douyu_play_url( + channel, device_id, secret_key, random_str, + enc_time, enc_data, rate + ) + + if play_url: + result.extend([name, play_url]) + + if not result: + return 1, self.excepturl + + return 0, result + except Exception as e: + print(f"斗鱼播放解析错误: {e}") + return 1, self.excepturl + + def localProxy(self, param): + pass + + def e64(self, text): + try: + text_bytes = text.encode('utf-8') + encoded_bytes = b64encode(text_bytes) + return encoded_bytes.decode('utf-8') + except Exception as e: + print(f"Base64编码错误: {str(e)}") + return "" + + def d64(self, encoded_text): + try: + encoded_bytes = encoded_text.encode('utf-8') + decoded_bytes = b64decode(encoded_bytes) + return decoded_bytes.decode('utf-8') + except Exception as e: + print(f"Base64解码错误: {str(e)}") + return "" + + def josn_to_params(self, params, skip_empty=False): + query = [] + for k, v in params.items(): + if skip_empty and not v: + continue + query.append(f"{k}={v}") + return "&".join(query) + + def params_to_json(self, query_string): + parsed_data = parse_qs(query_string) + result = {key: value[0] for key, value in parsed_data.items()} + return result + + def buildvod(self, vod_id='', vod_name='', vod_pic='', vod_year='', vod_tag='', vod_remarks='', style='', + type_name='', vod_area='', vod_actor='', vod_director='', + vod_content='', vod_play_from='', vod_play_url=''): + vod = { + 'vod_id': vod_id, + 'vod_name': vod_name, + 'vod_pic': vod_pic, + 'vod_year': vod_year, + 'vod_tag': 'folder' if vod_tag else '', + 'vod_remarks': vod_remarks, + 'style': style, + 'type_name': type_name, + 'vod_area': vod_area, + 'vod_actor': vod_actor, + 'vod_director': vod_director, + 'vod_content': vod_content, + 'vod_play_from': vod_play_from, + 'vod_play_url': vod_play_url + } + vod = {key: value for key, value in vod.items() if value} + return vod + + def getpq(self, url, headers=None, cookies=None): + data = self.fetch(url, headers=headers, cookies=cookies).text + try: + return pq(data) + except Exception as e: + print(f"解析页面错误: {str(e)}") + return pq(data.encode('utf-8')) + + def gethr(self, index, rf='', zr=''): + headers = self.headers[index] + if zr: + headers['referer'] = zr + else: + headers['referer'] = f"{self.referers[rf]}/" + return headers + + def handle_exception(self, e): + print(f"报错: {str(e)}") + return {'vod_play_from': '哎呀翻车啦', 'vod_play_url': f'翻车啦${self.excepturl}'} \ No newline at end of file diff --git a/py/酷我.py b/py/酷我.py new file mode 100644 index 0000000..a6c67a9 --- /dev/null +++ b/py/酷我.py @@ -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(' ', ' ') + 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 \ No newline at end of file