liuwei_x_x/abot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

测试抖音处理

Browse Source
This commit is contained in:
liuwei
2025-08-13 09:58:53 +08:00
parent db1496b195
commit d52667bb86
2 changed files with 23 additions and 96 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
plugins/douyin_parser/config.toml
View File

@@ -2,7 +2,7 @@
enable = false
# 发送模式: card(发送卡片) 或 file(下载并发送文件)
download_mode = "card"
download_mode = "file"
# Http代理设置（用于获取真实链接发送卡片，如果家里有ipv6，可以设置为空）
# 格式: http://用户名:密码@代理地址:代理端口

117
plugins/douyin_parser/main.py
View File

@@ -6,12 +6,14 @@ import requests
from typing import Dict, Any, List, Optional, Tuple
from loguru import logger
from pathlib import Path
from base.plugin_common.message_plugin_interface import MessagePluginInterface
from base.plugin_common.plugin_interface import PluginStatus
from utils.decorator.plugin_decorators import plugin_stats_decorator
from utils.robot_cmd.robot_command import Feature, PermissionStatus, GroupBotManager
from wechat_ipad import WechatAPIClient
from wechat_ipad.models.appmsg_xml import LINK_XML_NORMAL
class DouyinParserError(Exception):
@@ -77,7 +79,7 @@ class DouyinParserPlugin(MessagePluginInterface):
# 保存上下文对象
self.event_system = context.get("event_system")
self.gbm = context.get("gbm")
self.download_dir = str(Path(Path(__file__).parent, "down_load_dir"))
# 从配置中获取参数
douyin_config = self._config.get("Douyin", {})
self.enable = douyin_config.get("enable", True)
@@ -109,7 +111,7 @@ class DouyinParserPlugin(MessagePluginInterface):
return match is not None
@plugin_stats_decorator(plugin_name="抖音解析")
def process_message(self, message: Dict[str, Any]) -> Tuple[bool, Optional[str]]:
async def process_message(self, message: Dict[str, Any]) -> Tuple[bool, Optional[str]]:
"""处理消息"""
content = str(message.get("content", "")).strip()
self.LOG.debug(f"插件执行： {self.name}：{content}")
@@ -136,7 +138,7 @@ class DouyinParserPlugin(MessagePluginInterface):
print(f"❌无法解析抖音视频信息")
return False, "解析失败"
video_url = video_info.get('video', '')
video_url = video_info.get('url', '')
title = video_info.get('title', '无标题')
author = video_info.get('name', '未知作者')
cover = video_info.get('cover', '')
@@ -148,24 +150,24 @@ class DouyinParserPlugin(MessagePluginInterface):
# 根据模式选择发送方式
if self.download_mode == "file":
# 下载并发送文件
mp4_path = self._download_stream(video_url, os.path.join(self.download_dir, "douyin.mp4"))
video_filename = f"video_{int(time.time())}.mp4"
save_path = os.path.join(self.download_dir, video_filename)
self.LOG.info(f"开始下载视频到: {save_path}")
mp4_path = self._download_stream(video_url, os.path.join(self.download_dir, save_path))
if mp4_path:
bot.send_video_message((roomid if roomid else sender),mp4_path)
await self.bot.send_video_message((roomid if roomid else sender), mp4_path)
return True, "发送视频文件成功"
else:
print(f"❌下载视频失败")
return False, "下载视频失败"
else:
# 发送卡片
self.message_util.send_rich_text(
"BOT-PC直接查看",
"gh_11",
title[:30],
f"PC直接查看-{title[:20]} - {author[:10]}",
video_url,
cover,
(roomid if roomid else sender)
)
xml_content = f"{LINK_XML_NORMAL}".format(title=title,
des=author,
url=video_url,
thumburl=cover
)
await self.bot.send_link_xml_message(xml_content, (roomid if roomid else sender))
return True, "发送卡片成功"
except DouyinParserError as e:
@@ -191,88 +193,13 @@ class DouyinParserPlugin(MessagePluginInterface):
'cover'] = "https://is1-ssl.mzstatic.com/image/thumb/Purple221/v4/7c/49/e1/7c49e1af-ce92-d1c4-9a93-0a316e47ba94/AppIcon_TikTok-0-0-1x_U007epad-0-1-0-0-85-220.png/512x512bb.jpg"
return data
def _get_real_video_url(self, video_url: str) -> str:
"""获取真实视频链接"""
max_retries = 3 # 最大重试次数
retry_delay = 2 # 重试延迟秒数
max_redirects = 10 # 最大重定向次数，防止死循环
proxies = {"http": self.http_proxy, "https": self.http_proxy} if self.http_proxy else None
redirect_history = []
for retry in range(max_retries):
try:
self.LOG.info(f"[抖音] 开始获取真实视频链接: {video_url} (第{retry + 1}次尝试)")
headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36',
'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8',
'Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.8',
'Range': 'bytes=0-'
}
# 默认使用 allow_redirects=True 获取历史记录
response = requests.get(video_url, headers=headers, proxies=proxies, allow_redirects=True, timeout=60)
if response.history:
redirect_history = [resp.url for resp in response.history]
real_url = response.url
else:
# response.history 为空，手动解析重定向
current_url = video_url
for _ in range(max_redirects): # 限制最大重定向次数
resp = requests.get(current_url, headers=headers, proxies=proxies, allow_redirects=False,
timeout=60)
new_url = resp.headers.get('Location')
if not new_url:
break # 没有新的 Location，停止
if not new_url.startswith("http"):
from urllib.parse import urljoin
new_url = urljoin(current_url, new_url) # 处理相对路径重定向
if new_url in redirect_history:
self.LOG.info(f"[抖音] 检测到循环重定向: {new_url}")
break # 避免死循环
redirect_history.append(new_url)
self.LOG.info(f"[抖音] 发现重定向: {current_url} -> {new_url}")
current_url = new_url
real_url = current_url
if redirect_history:
self.LOG.info(f"[抖音] 重定向历史: {redirect_history}")
if real_url != video_url and ('v3-' in real_url.lower() or 'douyinvod.com' in real_url.lower()):
self.LOG.info(f"[抖音] 成功获取真实链接: {real_url}")
return real_url
else:
self.LOG.info("[抖音] 未能获取到符合预期的视频链接，准备重试")
if retry < max_retries - 1:
time.sleep(retry_delay)
continue
return video_url
except Exception as e:
self.LOG.error(f"[抖音] 获取真实链接失败: {str(e)} (第{retry + 1}次尝试)")
if retry < max_retries - 1:
time.sleep(retry_delay)
continue
return video_url
self.LOG.error("[抖音] 获取真实链接失败，已达到最大重试次数")
return video_url
def _parse_douyin(self, url: str) -> Dict[str, Any]:
"""解析抖音链接"""
try:
api_url = "http://192.168.2.240:9081/api/hybrid/video_data"
api_url = "https://api.52vmy.cn/api/video/jx?url="
clean_url = self._clean_url(url)
params = {'url': clean_url, 'minimal': True}
self.LOG.info(f"[抖音] 请求API: {api_url}, 参数: {repr(params)}")
proxy = {"http": self.http_proxy, "https": self.http_proxy} if self.http_proxy else None
response = requests.get(api_url, params=params, timeout=30, proxies=proxy)
api_url = api_url + clean_url
response = requests.get(api_url, timeout=30)
if response.status_code != 200:
raise DouyinParserError(f"API请求失败，状态码: {response.status_code}")
@@ -283,9 +210,9 @@ class DouyinParserPlugin(MessagePluginInterface):
if data.get("code") == 200:
result = data.get("data", {})
self.LOG.info(f"[抖音] API响应数据result: {result}")
if result.get('video'):
result['video'] = self._get_real_video_url(result['video'])
return self._clean_response_data(result)
if result.get('url'):
return result
else:
raise DouyinParserError(data.get("message", "未知错误"))
except Exception as e: