shihao/JieXi
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: 新增平台

Browse Source
This commit is contained in:
shihao
2025-11-30 19:49:25 +08:00
parent c3e56a954d
commit fbd2c491b2
41 changed files with 4293 additions and 76 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
parsers/base.py
View File

@@ -44,11 +44,12 @@ class BaseParser(ABC):
except requests.RequestException as e:
raise Exception(f"请求失败: {str(e)}")
def _normalize_response(self, cover: str, video_url: str, title: str, description: str) -> Dict:
def _normalize_response(self, cover: str, video_url: str, title: str, description: str, author: str = "") -> Dict:
"""标准化返回数据"""
return {
"cover": cover or "",
"video_url": video_url or "",
"title": title or "",
"description": description or ""
"description": description or "",
"author": author or ""
}

9
parsers/bilibili.py
View File

@@ -32,8 +32,9 @@ class BilibiliMirParser(BaseParser):
video_url = video_data.get("url", "") or video_data.get("video_url", "")
title = video_data.get("title", "")
description = video_data.get("desc", "") or video_data.get("description", "")
author = video_data.get("author", "") or video_data.get("owner", {}).get("name", "")
return self._normalize_response(cover, video_url, title, description)
return self._normalize_response(cover, video_url, title, description, author)
else:
raise Exception(f"解析失败: {data.get('msg', '未知错误')}")
except Exception as e:
@@ -70,8 +71,9 @@ class BilibiliBugPKParser(BaseParser):
video_url = video_data.get("url", "") or video_data.get("video_url", "")
title = video_data.get("title", "")
description = video_data.get("desc", "") or video_data.get("description", "")
author = video_data.get("author", "") or video_data.get("owner", {}).get("name", "")
return self._normalize_response(cover, video_url, title, description)
return self._normalize_response(cover, video_url, title, description, author)
else:
raise Exception(f"解析失败: {data.get('msg', '未知错误')}")
except Exception as e:
@@ -105,6 +107,7 @@ class BilibiliYaohuParser(BaseParser):
cover = basic.get("cover", "")
title = basic.get("title", "")
description = basic.get("description", "")
author = basic.get("author", "") or basic.get("owner", {}).get("name", "")
# 提取视频URL - 优先使用data.video_url，其次使用videos[0].url
video_url = video_data.get("video_url", "")
@@ -113,7 +116,7 @@ class BilibiliYaohuParser(BaseParser):
if isinstance(videos, list) and videos:
video_url = videos[0].get("url", "")
return self._normalize_response(cover, video_url, title, description)
return self._normalize_response(cover, video_url, title, description, author)
else:
raise Exception(f"解析失败: 不支持的类型 {data.get('parse_type')}")
except Exception as e:

16
parsers/douyin.py
View File

@@ -57,15 +57,15 @@ class DouyinParser(BaseParser):
# 提取标题（描述）
title = video_info.get("desc", "")
# 提取作者信息作为简介
author = video_info.get("author", {})
author_name = author.get("nickname", "")
author_signature = author.get("signature", "")
description = f"作者: {author_name}"
if author_signature:
description += f" | {author_signature}"
# 提取作者信息
author_info = video_info.get("author", {})
author_name = author_info.get("nickname", "")
author_signature = author_info.get("signature", "")
return self._normalize_response(cover, video_url, title, description)
# 简介使用作者签名
description = author_signature or ""
return self._normalize_response(cover, video_url, title, description, author_name)
except Exception as e:
raise Exception(f"数据提取失败: {str(e)}")

66
parsers/factory.py
View File

@@ -1,12 +1,48 @@
from parsers.douyin import DouyinParser
from parsers.tiktok import TikTokParser
from parsers.bilibili import BilibiliMirParser, BilibiliBugPKParser, BilibiliYaohuParser
from parsers.kuaishou import KuaishouBugPKParser, KuaishouUctbParser
from parsers.pipixia import PipixiaBugPKParser, PipixiaUctbParser
from parsers.weibo import WeiboUctbParser, WeiboYaohuParser
from models import ParserAPI
import random
import requests
class ParserFactory:
"""解析器工厂类"""
@staticmethod
def expand_short_url(url: str) -> str:
"""展开短链接，获取真实URL"""
short_domains = ['b23.tv', 'v.douyin.com', 't.cn']
# 检查是否是短链接
is_short = any(domain in url.lower() for domain in short_domains)
if not is_short:
return url
try:
# 发送请求，不跟随重定向，获取 Location 头
headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
}
response = requests.head(url, headers=headers, allow_redirects=False, timeout=10)
if response.status_code in [301, 302, 303, 307, 308]:
location = response.headers.get('Location', '')
if location:
# 如果还是短链接，继续展开
if any(domain in location.lower() for domain in short_domains):
return ParserFactory.expand_short_url(location)
return location
# 如果 HEAD 请求不行，尝试 GET 请求跟随重定向
response = requests.get(url, headers=headers, allow_redirects=True, timeout=10)
return response.url
except Exception:
# 展开失败，返回原URL
return url
@staticmethod
def create_parser(api_config: ParserAPI):
"""根据API配置创建解析器实例"""
@@ -28,6 +64,30 @@ class ParserFactory:
return BilibiliYaohuParser(api_url, api_key)
else:
return BilibiliMirParser(api_url, api_key)
elif platform == 'kuaishou':
# 快手解析器
if 'bugpk' in api_url:
return KuaishouBugPKParser(api_url, api_key)
elif 'uctb' in api_url:
return KuaishouUctbParser(api_url, api_key)
else:
return KuaishouBugPKParser(api_url, api_key)
elif platform == 'pipixia':
# 皮皮虾解析器
if 'bugpk' in api_url:
return PipixiaBugPKParser(api_url, api_key)
elif 'uctb' in api_url:
return PipixiaUctbParser(api_url, api_key)
else:
return PipixiaBugPKParser(api_url, api_key)
elif platform == 'weibo':
# 微博解析器
if 'uctb' in api_url:
return WeiboUctbParser(api_url, api_key)
elif 'yaohud' in api_url:
return WeiboYaohuParser(api_url, api_key)
else:
return WeiboUctbParser(api_url, api_key)
else:
raise ValueError(f"不支持的平台: {platform}")
@@ -83,5 +143,11 @@ class ParserFactory:
return 'tiktok'
elif 'bilibili.com' in url_lower or 'b23.tv' in url_lower:
return 'bilibili'
elif 'kuaishou.com' in url_lower or 'gifshow.com' in url_lower:
return 'kuaishou'
elif 'pipix.com' in url_lower or 'pipixia.com' in url_lower or 'h5.pipix.com' in url_lower:
return 'pipixia'
elif 'weibo.com' in url_lower or 'weibo.cn' in url_lower:
return 'weibo'
else:
raise ValueError("无法识别的视频平台")

89
parsers/kuaishou.py Normal file
View File

@@ -0,0 +1,89 @@
from parsers.base import BaseParser
from typing import Dict
from urllib.parse import urlencode
class KuaishouBugPKParser(BaseParser):
"""快手解析器 - BugPK API"""
def parse(self, video_url: str) -> Dict:
"""解析快手视频"""
try:
url = f"{self.api_url}/api/kuaishou?{urlencode({'url': video_url})}"
response = self._make_request(url)
data = response.json()
return self._extract_data(data)
except Exception as e:
raise Exception(f"快手解析失败(BugPK API): {str(e)}")
def _extract_data(self, data: Dict) -> Dict:
"""提取并标准化数据
实际返回格式:
{
"code": 200,
"msg": "解析成功",
"data": {
"title": "视频标题",
"cover": "封面URL",
"url": "视频URL"
}
}
"""
try:
if data.get("code") == 200:
video_data = data.get("data", {})
cover = video_data.get("cover", "")
video_url = video_data.get("url", "")
title = video_data.get("title", "")
description = "" # BugPK API 不返回简介
author = "" # BugPK API 不返回作者
return self._normalize_response(cover, video_url, title, description, author)
else:
raise Exception(f"解析失败: {data.get('msg', '未知错误')}")
except Exception as e:
raise Exception(f"数据提取失败: {str(e)}")
class KuaishouUctbParser(BaseParser):
"""快手解析器 - 优创 API"""
def parse(self, video_url: str) -> Dict:
"""解析快手视频"""
try:
url = f"{self.api_url}/api/videojx?{urlencode({'url': video_url})}"
response = self._make_request(url)
data = response.json()
return self._extract_data(data)
except Exception as e:
raise Exception(f"快手解析失败(优创 API): {str(e)}")
def _extract_data(self, data: Dict) -> Dict:
"""提取并标准化数据
实际返回格式:
{
"code": 200,
"data": {
"desc": "视频描述",
"cover": "封面URL",
"playurl": "视频URL"
},
"msg": "请求成功"
}
"""
try:
if data.get("code") == 200:
video_data = data.get("data", {})
cover = video_data.get("cover", "")
video_url = video_data.get("playurl", "")
title = video_data.get("desc", "")
description = ""
author = "" # 优创API不返回作者
return self._normalize_response(cover, video_url, title, description, author)
else:
raise Exception(f"解析失败: {data.get('msg', '未知错误')}")
except Exception as e:
raise Exception(f"数据提取失败: {str(e)}")

91
parsers/pipixia.py Normal file
View File

@@ -0,0 +1,91 @@
from parsers.base import BaseParser
from typing import Dict
from urllib.parse import urlencode
class PipixiaBugPKParser(BaseParser):
"""皮皮虾解析器 - BugPK API"""
def parse(self, video_url: str) -> Dict:
"""解析皮皮虾视频"""
try:
url = f"{self.api_url}/api/pipixia?{urlencode({'url': video_url})}"
response = self._make_request(url)
data = response.json()
return self._extract_data(data)
except Exception as e:
raise Exception(f"皮皮虾解析失败(BugPK API): {str(e)}")
def _extract_data(self, data: Dict) -> Dict:
"""提取并标准化数据
实际返回格式:
{
"code": 200,
"msg": "解析成功",
"data": {
"author": "作者名",
"avatar": "头像URL",
"title": "标题",
"cover": "封面URL",
"url": "视频URL"
}
}
"""
try:
if data.get("code") == 200:
video_data = data.get("data", {})
cover = video_data.get("cover", "")
video_url = video_data.get("url", "")
title = video_data.get("title", "")
description = ""
author = video_data.get("author", "")
return self._normalize_response(cover, video_url, title, description, author)
else:
raise Exception(f"解析失败: {data.get('msg', '未知错误')}")
except Exception as e:
raise Exception(f"数据提取失败: {str(e)}")
class PipixiaUctbParser(BaseParser):
"""皮皮虾解析器 - 优创 API"""
def parse(self, video_url: str) -> Dict:
"""解析皮皮虾视频"""
try:
url = f"{self.api_url}/api/videojx?{urlencode({'url': video_url})}"
response = self._make_request(url)
data = response.json()
return self._extract_data(data)
except Exception as e:
raise Exception(f"皮皮虾解析失败(优创 API): {str(e)}")
def _extract_data(self, data: Dict) -> Dict:
"""提取并标准化数据
实际返回格式:
{
"code": 200,
"data": {
"desc": null,
"cover": "封面URL",
"playurl": "视频URL"
},
"msg": "请求成功"
}
"""
try:
if data.get("code") == 200:
video_data = data.get("data", {})
cover = video_data.get("cover", "")
video_url = video_data.get("playurl", "")
title = video_data.get("desc", "") or ""
description = ""
author = "" # 优创API不返回作者
return self._normalize_response(cover, video_url, title, description, author)
else:
raise Exception(f"解析失败: {data.get('msg', '未知错误')}")
except Exception as e:
raise Exception(f"数据提取失败: {str(e)}")

15
parsers/tiktok.py
View File

@@ -58,13 +58,14 @@ class TikTokParser(BaseParser):
title = video_info.get("desc", "")
author = video_info.get("author", {})
author_name = author.get("nickname", "")
author_signature = author.get("signature", "")
description = f"Author: {author_name}"
if author_signature:
description += f" | {author_signature}"
# 提取作者信息
author_info = video_info.get("author", {})
author_name = author_info.get("nickname", "")
author_signature = author_info.get("signature", "")
return self._normalize_response(cover, video_url, title, description)
# 简介使用作者签名
description = author_signature or ""
return self._normalize_response(cover, video_url, title, description, author_name)
except Exception as e:
raise Exception(f"数据提取失败: {str(e)}")

105
parsers/weibo.py Normal file
View File

@@ -0,0 +1,105 @@
from parsers.base import BaseParser
from typing import Dict
from urllib.parse import urlencode
class WeiboUctbParser(BaseParser):
"""微博解析器 - 优创 API"""
def parse(self, video_url: str) -> Dict:
"""解析微博视频"""
try:
url = f"{self.api_url}/api/videojx?{urlencode({'url': video_url})}"
response = self._make_request(url)
data = response.json()
return self._extract_data(data)
except Exception as e:
raise Exception(f"微博解析失败(优创 API): {str(e)}")
def _extract_data(self, data: Dict) -> Dict:
"""提取并标准化数据
实际返回格式:
{
"code": 200,
"data": {
"desc": "视频描述",
"cover": "封面URL",
"playurl": "视频URL"
},
"msg": "请求成功"
}
"""
try:
if data.get("code") == 200:
video_data = data.get("data", {})
cover = video_data.get("cover", "")
video_url = video_data.get("playurl", "")
title = video_data.get("desc", "") or ""
description = ""
author = "" # 优创API不返回作者
return self._normalize_response(cover, video_url, title, description, author)
else:
raise Exception(f"解析失败: {data.get('msg', '未知错误')}")
except Exception as e:
raise Exception(f"数据提取失败: {str(e)}")
class WeiboYaohuParser(BaseParser):
"""微博解析器 - 妖狐 API"""
def parse(self, video_url: str) -> Dict:
"""解析微博视频"""
try:
url = f"{self.api_url}/api/v6/video/weibo?{urlencode({'key': self.api_key, 'url': video_url})}"
response = self._make_request(url, verify=False)
data = response.json()
return self._extract_data(data)
except Exception as e:
raise Exception(f"微博解析失败(妖狐 API): {str(e)}")
def _extract_data(self, data: Dict) -> Dict:
"""提取并标准化数据
实际返回格式:
{
"code": 200,
"msg": "解析成功",
"data": {
"author": "作者名",
"title": "标题",
"cover": "封面URL",
"quality_urls": {
"超清 2K60": "视频URL",
"高清 1080P": "视频URL",
...
}
}
}
"""
try:
if data.get("code") == 200:
video_data = data.get("data", {})
cover = video_data.get("cover", "")
title = video_data.get("title", "")
description = ""
author = video_data.get("author", "")
# 从 quality_urls 中获取最高质量的视频URL
quality_urls = video_data.get("quality_urls", {})
video_url = ""
# 按优先级选择：超清 > 高清1080P > 高清720P > 其他
for quality in ["超清 2K60", "高清 1080P", "高清 720P"]:
if quality in quality_urls:
video_url = quality_urls[quality]
break
# 如果没找到，取第一个可用的
if not video_url and quality_urls:
video_url = list(quality_urls.values())[0]
return self._normalize_response(cover, video_url, title, description, author)
else:
raise Exception(f"解析失败: {data.get('msg', '未知错误')}")
except Exception as e:
raise Exception(f"数据提取失败: {str(e)}")