diff --git a/.github/workflows/main.yml b/.github/workflows/main.yml index b343d6d..67ee7e4 100644 --- a/.github/workflows/main.yml +++ b/.github/workflows/main.yml @@ -1,6 +1,4 @@ -name: Auto Deploy NeoBot (Full Env Secrets) - -# 触发条件:推送到main分支 或 手动触发 +name: Auto Deploy NeoBot (FRP + SSH 密码登录) on: push: branches: [ main ] @@ -8,50 +6,61 @@ on: jobs: deploy-to-server: - # 关联你的仓库环境(ENV) environment: ENV runs-on: ubuntu-latest + timeout-minutes: 10 # 防超时堵塞 steps: + # ========== 1. 检查环境密钥配置 ========== - name: 检查环境密钥配置 run: | echo "✅ 已关联环境: ${{ github.environment }}" - echo "✅ API_URL 密钥是否存在: ${{ secrets.API_URL != '' }}" - echo "✅ API_TOKEN 密钥是否存在: ${{ secrets.NEOBOT_DEPLOY_TOKEN != '' }}" + # 仅检查密码登录必需的3个密钥 + echo "✅ PROD_SERVER_HOST 密钥是否存在: ${{ secrets.PROD_SERVER_HOST != '' }}" + echo "✅ PROD_SERVER_USER 密钥是否存在: ${{ secrets.PROD_SERVER_USER != '' }}" + echo "✅ PROD_SERVER_PASS 密钥是否存在: ${{ secrets.PROD_SERVER_PASS != '' }}" - - name: 调用部署API - env: - # 从环境密钥中读取API地址和Token(均为密文) - API_URL: ${{ secrets.API_URL }} - API_TOKEN: ${{ secrets.NEOBOT_DEPLOY_TOKEN }} + # ========== 2. 安装 sshpass(密码登录必需) ========== + - name: 安装 sshpass 工具 run: | - # 安装jq用于解析JSON - sudo apt-get update && sudo apt-get install -y jq - - # 打印关键信息(脱敏,仅验证是否读取到值) - echo "📌 调用的API地址(脱敏): $(echo $API_URL | sed 's/http:\/\///; s/\/deploy//')" - - # 发送POST请求到部署API(所有配置均来自密钥) - RESPONSE=$(curl -s -X POST \ - $API_URL \ - -H "Content-Type: application/json" \ - -H "X-API-Token: $API_TOKEN" \ - -d '{"script_name":"deploy.sh"}') - - # 打印完整响应(便于调试) - echo "📝 API响应详情:" - echo $RESPONSE | jq . - - # 解析status字段判断部署结果 - STATUS=$(echo $RESPONSE | jq -r '.status') - if [ "$STATUS" = "success" ]; then - echo "✅ 部署成功!" + sudo apt-get update && sudo apt-get install -y sshpass + + # ========== 3. 密码登录服务器 + 执行部署 ========== + - name: 执行FRP穿透部署(用户名+密码登录) + id: ssh_deploy_step + continue-on-error: true + run: | + # 核心:sshpass 实现密码登录,-p 8000 是FRP转发端口 + sshpass -p "${{ secrets.PROD_SERVER_PASS }}" \ + ssh -o StrictHostKeyChecking=no -p 8000 ${{ secrets.PROD_SERVER_USER }}@${{ secrets.PROD_SERVER_HOST }} << 'EOF' + set -e + # 适配NeoBot项目:更新依赖+重启systemd服务 + cd /home/k/NeoBot + git pull + pip install -r requirements.txt --upgrade --timeout 300 --only-binary=:all: + sudo systemctl daemon-reload + sudo systemctl restart neobot + # 验证服务状态 + if ! sudo systemctl is-active --quiet neobot; then + echo "❌ NeoBot服务启动失败,最后10行日志:" + sudo journalctl -u neobot -n 10 --no-pager + exit 1 + fi + echo "✅ NeoBot服务重启成功" + EOF + + # ========== 4. 判定最终部署结果 ========== + - name: 判定最终部署结果 + run: | + if [ ${{ steps.ssh_deploy_step.outcome }} = 'success' ]; then + echo "✅ 最终部署成功!已更新依赖并重启NeoBot systemd服务" exit 0 else - echo "❌ 部署失败!错误信息:$(echo $RESPONSE | jq -r '.message')" + echo "❌ 最终部署失败!核心SSH部署步骤执行出错" exit 1 fi + # ========== 5. 部署失败通知(可选) ========== - name: 部署失败通知(可选) if: failure() run: | - echo "⚠️ 部署失败,可在此添加通知逻辑" + echo "⚠️ 部署失败,可在此添加钉钉/企业微信通知逻辑" diff --git a/plugins/web_parser/parsers/douyin.py b/plugins/web_parser/parsers/douyin.py index 72cd12b..933a4db 100644 --- a/plugins/web_parser/parsers/douyin.py +++ b/plugins/web_parser/parsers/douyin.py @@ -1,6 +1,7 @@ # -*- coding: utf-8 -*- import re import aiohttp +import asyncio from typing import Optional, Dict, Any, List from core.utils.logger import logger @@ -24,9 +25,9 @@ class DouyinParser(BaseParser): # 消息去重缓存 self.processed_messages: TTLCache[int, bool] = TTLCache(maxsize=100, ttl=10) - async def parse(self, url: str) -> Optional[Dict[str, Any]]: + async def _parse_api_xhus(self, url: str) -> Optional[Dict[str, Any]]: """ - 解析抖音视频信息 + 使用 xhus API 解析抖音视频 Args: url (str): 抖音视频URL @@ -35,31 +36,29 @@ class DouyinParser(BaseParser): 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}") + logger.error(f"[{self.name}] xhus API请求失败,状态码: {response.status}") return None response_data = await response.json() if not isinstance(response_data, dict): - logger.error(f"[{self.name}] API返回格式错误: {response_data}") + logger.error(f"[{self.name}] xhus API返回格式错误: {response_data}") return None if response_data.get("code") != 200: - logger.error(f"[{self.name}] API返回错误: {response_data}") + logger.error(f"[{self.name}] xhus API返回错误: {response_data}") return None data = response_data.get("data", {}) if not data: - logger.error(f"[{self.name}] API返回数据为空") + logger.error(f"[{self.name}] xhus 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", ""), @@ -74,13 +73,92 @@ class DouyinParser(BaseParser): "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}") + logger.error(f"[{self.name}] xhus API解析失败: {e}") + return None + + async def _parse_api_mmp(self, url: str) -> Optional[Dict[str, Any]]: + """ + 使用 mmp API 解析抖音视频 + Args: + url (str): 抖音视频URL + + Returns: + Optional[Dict[str, Any]]: 视频信息字典,如果失败则返回None + """ + try: + api_url = f"https://api.mmp.cc/api/Jiexi?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}] mmp API请求失败,状态码: {response.status}") + return None + + response_data = await response.json() + + if not isinstance(response_data, dict): + logger.error(f"[{self.name}] mmp API返回格式错误: {response_data}") + return None + + if response_data.get("code") != 200: + logger.error(f"[{self.name}] mmp API返回错误: {response_data}") + return None + + data = response_data.get("data", {}) + if not data: + logger.error(f"[{self.name}] mmp API返回数据为空") + return None + + return { + "type": data.get("type", "video"), + "video_url": data.get("video_url", ""), + "video_url_HQ": data.get("video_url_HQ", ""), + "nickname": data.get("nickname", "未知作者"), + "desc": data.get("desc", "无描述"), + "aweme_id": data.get("aweme_id", ""), + "like": data.get("like", 0), + "cover": data.get("cover", ""), + "time": data.get("time", 0), + "author_avatar": data.get("author_avatar", ""), + "music": data.get("music", {}), + } + + except Exception as e: + logger.error(f"[{self.name}] mmp API解析失败: {e}") + return None + + async def parse(self, url: str) -> Optional[Dict[str, Any]]: + """ + 解析抖音视频信息(并发请求多个API,取最快返回的结果) + + Args: + url (str): 抖音视频URL + + Returns: + Optional[Dict[str, Any]]: 视频信息字典,如果失败则返回None + """ + async def try_api(coro, api_name: str) -> tuple: + try: + result = await coro + return (result, api_name) + except Exception as e: + logger.error(f"[{self.name}] {api_name} API异常: {e}") + return (None, api_name) + + tasks = [ + try_api(self._parse_api_xhus(url), "xhus"), + try_api(self._parse_api_mmp(url), "mmp"), + ] + + for coro in asyncio.as_completed(tasks): + result, api_name = await coro + if result: + logger.info(f"[{self.name}] 使用 {api_name} API 成功解析") + return result + + logger.error(f"[{self.name}] 所有API解析均失败") return None async def get_real_url(self, short_url: str) -> Optional[str]: