254 lines
9.8 KiB
Python
254 lines
9.8 KiB
Python
# -*- coding: utf-8 -*-
|
||
import re
|
||
import aiohttp
|
||
from typing import Optional, Dict, Any, List
|
||
|
||
from core.utils.logger import logger
|
||
from models import MessageEvent, MessageSegment
|
||
from ..base import BaseParser
|
||
from ..utils import extract_original_text
|
||
from cachetools import TTLCache
|
||
|
||
|
||
class DouyinParser(BaseParser):
|
||
"""
|
||
抖音视频解析器
|
||
"""
|
||
|
||
def __init__(self):
|
||
super().__init__()
|
||
self.name = "抖音解析器"
|
||
self.url_pattern = re.compile(r"https?://v\.douyin\.com/[a-zA-Z0-9_-]+/?", re.IGNORECASE)
|
||
self.short_pattern = re.compile(r"(?:https?://)?v\.douyin\.com/[a-zA-Z0-9_-]+/?", re.IGNORECASE)
|
||
self.nickname = "抖音视频解析"
|
||
# 消息去重缓存
|
||
self.processed_messages: TTLCache[int, bool] = TTLCache(maxsize=100, ttl=10)
|
||
|
||
async def parse(self, url: str) -> Optional[Dict[str, Any]]:
|
||
"""
|
||
解析抖音视频信息
|
||
|
||
Args:
|
||
url (str): 抖音视频URL
|
||
|
||
Returns:
|
||
Optional[Dict[str, Any]]: 视频信息字典，如果失败则返回None
|
||
"""
|
||
try:
|
||
# 使用第三方API解析抖音视频
|
||
api_url = f"http://api.xhus.cn/api/douyin?url={url}"
|
||
|
||
session = self.get_session()
|
||
async with session.get(api_url, headers=self.HEADERS, timeout=aiohttp.ClientTimeout(total=10)) as response:
|
||
if response.status != 200:
|
||
logger.error(f"[{self.name}] API请求失败，状态码: {response.status}")
|
||
return None
|
||
|
||
response_data = await response.json()
|
||
|
||
if not isinstance(response_data, dict):
|
||
logger.error(f"[{self.name}] API返回格式错误: {response_data}")
|
||
return None
|
||
|
||
if response_data.get("code") != 200:
|
||
logger.error(f"[{self.name}] API返回错误: {response_data}")
|
||
return None
|
||
|
||
data = response_data.get("data", {})
|
||
if not data:
|
||
logger.error(f"[{self.name}] API返回数据为空")
|
||
return None
|
||
|
||
# 转换API响应格式
|
||
return {
|
||
"type": "video" if not data.get("images") or not isinstance(data.get("images"), list) else "image",
|
||
"video_url": data.get("url", ""),
|
||
"video_url_HQ": data.get("url", ""),
|
||
"nickname": data.get("author", "未知作者"),
|
||
"desc": data.get("title", "无描述"),
|
||
"aweme_id": data.get("uid", ""),
|
||
"like": data.get("like", 0),
|
||
"cover": data.get("cover", ""),
|
||
"time": data.get("time", 0),
|
||
"author_avatar": data.get("avatar", ""),
|
||
"music": data.get("music", {}),
|
||
}
|
||
|
||
except (aiohttp.ClientError, KeyError, AttributeError, ValueError) as e:
|
||
logger.error(f"[{self.name}] 解析抖音视频信息失败: {e}")
|
||
logger.debug(f"失败的URL: {url}")
|
||
except Exception as e:
|
||
logger.error(f"[{self.name}] 解析抖音视频时发生未知错误: {e}")
|
||
logger.debug(f"失败的URL: {url}")
|
||
|
||
return None
|
||
|
||
async def get_real_url(self, short_url: str) -> Optional[str]:
|
||
"""
|
||
获取抖音短链接的真实URL
|
||
|
||
Args:
|
||
short_url (str): 抖音短链接
|
||
|
||
Returns:
|
||
Optional[str]: 真实URL，如果失败则返回None
|
||
"""
|
||
try:
|
||
session = self.get_session()
|
||
async with session.get(short_url, allow_redirects=True, timeout=aiohttp.ClientTimeout(total=10)) as response:
|
||
redirected_url = str(response.url)
|
||
|
||
# 检查重定向后的URL是否是有效的视频或图文页
|
||
if 'douyin.com/video/' in redirected_url or 'douyin.com/note/' in redirected_url:
|
||
logger.info(f"[{self.name}] 成功获取真实URL: {redirected_url}")
|
||
return redirected_url
|
||
else:
|
||
logger.warning(f"[{self.name}] 短链接 {short_url} 重定向到了非预期的页面: {redirected_url}")
|
||
return None
|
||
|
||
except Exception as e:
|
||
logger.error(f"[{self.name}] 获取真实URL失败: {e}")
|
||
return None
|
||
|
||
async def format_response(self, event: MessageEvent, data: Dict[str, Any]) -> List[Any]:
|
||
"""
|
||
格式化抖音视频响应消息
|
||
|
||
Args:
|
||
event (MessageEvent): 消息事件对象
|
||
data (Dict[str, Any]): 视频信息
|
||
|
||
Returns:
|
||
List[Any]: 消息段列表
|
||
"""
|
||
# 构建回复消息，包含原分享中的文本内容（如果有）
|
||
original_text = extract_original_text(event.message, self.url_pattern)
|
||
|
||
# 构建回复消息
|
||
text_parts = ["抖音视频解析"]
|
||
text_parts.append("--------------------")
|
||
|
||
if original_text:
|
||
text_parts.append(f" 分享内容: {original_text}")
|
||
text_parts.append("--------------------")
|
||
|
||
text_parts.append(f" 作者: {data['nickname']}")
|
||
text_parts.append(f" 抖音号: {data['aweme_id']}")
|
||
text_parts.append(f" 标题: {data['desc']}")
|
||
text_parts.append(f" 点赞: {self.format_count(data['like'])}")
|
||
text_parts.append(f" 类型: {data['type']}")
|
||
|
||
# 如果是音乐，添加音乐信息
|
||
if data.get('music'):
|
||
music_info = data['music']
|
||
text_parts.append("--------------------")
|
||
text_parts.append(" 背景音乐:")
|
||
text_parts.append(f" 标题: {music_info.get('title', '')}")
|
||
text_parts.append(f" 作者: {music_info.get('author', '')}")
|
||
|
||
text_parts.append("--------------------")
|
||
|
||
text_message = "\n".join(text_parts)
|
||
|
||
# 准备转发消息节点
|
||
nodes = []
|
||
|
||
# 添加文本信息节点
|
||
text_node = event.bot.build_forward_node(
|
||
user_id=event.self_id,
|
||
nickname=self.nickname,
|
||
message=text_message
|
||
)
|
||
nodes.append(text_node)
|
||
|
||
# 添加封面图片节点（如果有）
|
||
if data.get('cover'):
|
||
try:
|
||
cover_node = event.bot.build_forward_node(
|
||
user_id=event.self_id,
|
||
nickname=self.nickname,
|
||
message=[
|
||
MessageSegment.text("抖音视频封面：\n"),
|
||
MessageSegment.image(data['cover'])
|
||
]
|
||
)
|
||
nodes.append(cover_node)
|
||
except Exception as e:
|
||
logger.warning(f"[{self.name}] 无法添加封面图片: {e}")
|
||
|
||
# 添加作者头像节点（如果有）
|
||
if data.get('author_avatar'):
|
||
try:
|
||
avatar_node = event.bot.build_forward_node(
|
||
user_id=event.self_id,
|
||
nickname=self.nickname,
|
||
message=[
|
||
MessageSegment.text("作者头像：\n"),
|
||
MessageSegment.image(data['author_avatar'])
|
||
]
|
||
)
|
||
nodes.append(avatar_node)
|
||
except Exception as e:
|
||
logger.warning(f"[{self.name}] 无法添加作者头像: {e}")
|
||
|
||
# 尝试添加视频直链（单独节点）
|
||
video_success = False
|
||
direct_message = None
|
||
try:
|
||
if data.get('video_url'):
|
||
video_url = data.get('video_url', '')
|
||
# 检查视频类型
|
||
if data.get('type') == 'video':
|
||
video_message = MessageSegment.video(video_url)
|
||
direct_message = video_message
|
||
video_type_text = "视频直链："
|
||
else: # image类型
|
||
video_message = MessageSegment.image(video_url) # 单个图片
|
||
direct_message = video_message
|
||
video_type_text = "图集首图："
|
||
|
||
# 构建视频/图片节点
|
||
video_node = event.bot.build_forward_node(
|
||
user_id=event.self_id,
|
||
nickname=self.nickname,
|
||
message=[
|
||
MessageSegment.text(video_type_text + "\n"),
|
||
video_message
|
||
]
|
||
)
|
||
nodes.append(video_node)
|
||
video_success = True
|
||
except Exception as e:
|
||
logger.error(f"[{self.name}] 无法添加视频/图片: {e}")
|
||
|
||
# 如果无法添加视频，添加提示信息
|
||
if not video_success:
|
||
no_video_node = event.bot.build_forward_node(
|
||
user_id=event.self_id,
|
||
nickname=self.nickname,
|
||
message="视频解析成功，但无法获取直链或播放视频。"
|
||
)
|
||
nodes.append(no_video_node)
|
||
|
||
# 同时直接发送视频/图片（如果获取到直链）
|
||
if direct_message:
|
||
try:
|
||
await event.reply(direct_message)
|
||
except Exception as e:
|
||
logger.error(f"[{self.name}] 直接发送视频/图片失败: {e}")
|
||
|
||
return nodes
|
||
|
||
def should_handle_url(self, url: str) -> bool:
|
||
"""
|
||
判断是否应该处理该URL
|
||
|
||
Args:
|
||
url (str): URL
|
||
|
||
Returns:
|
||
bool: 是否应该处理
|
||
"""
|
||
# 检查是否是抖音相关域名
|
||
return ('douyin.com' in url or bool(self.url_pattern.search(url)) or bool(self.short_pattern.search(url)))
|