From 5f9fadb9e6d883e554378d92cd7c09e6bdc24f8d Mon Sep 17 00:00:00 2001 From: liuwei Date: Mon, 17 Nov 2025 15:34:21 +0800 Subject: [PATCH] =?UTF-8?q?=E5=8A=A0=E5=85=A5=E4=BA=86=E6=8A=96=E9=9F=B3?= =?UTF-8?q?=E8=A7=86=E9=A2=91=E5=A4=84=E7=90=86=E9=80=BB=E8=BE=91?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- plugins/douyin_parser/main.py | 125 ++++++++++++++++++++++++++-------- 1 file changed, 98 insertions(+), 27 deletions(-) diff --git a/plugins/douyin_parser/main.py b/plugins/douyin_parser/main.py index 399641e..3d7ef16 100644 --- a/plugins/douyin_parser/main.py +++ b/plugins/douyin_parser/main.py @@ -4,6 +4,7 @@ import time import traceback import requests from typing import Dict, Any, List, Optional, Tuple +from urllib.parse import urlparse from loguru import logger from pathlib import Path @@ -194,39 +195,109 @@ class DouyinParserPlugin(MessagePluginInterface): return data def _parse_douyin(self, url: str) -> Dict[str, Any]: - """解析抖音链接""" try: - api_url = "https://api.pearktrue.cn/api/video/douyin/?url=" - pay_api_url = "https://api.pearktrue.cn/api/video/api.php" clean_url = self._clean_url(url) - # api_url = api_url + clean_url - # response = requests.get(api_url, timeout=30) - # 切换付费模式 - # 发送请求 - params = { - "url": clean_url, - "key": "f56c1fed0c6e64e7" - } - response = requests.post(pay_api_url, params=params, timeout=30) - - if response.status_code != 200: - raise DouyinParserError(f"API请求失败，状态码: {response.status_code}") - - data = response.json() - self.LOG.info(f"[抖音] API响应数据: {data}") - - if data.get("code") == 200: - result = data.get("data", {}) - self.LOG.info(f"[抖音] API响应数据result: {result}") - if result.get('url'): - return result - - else: - raise DouyinParserError(data.get("message", "未知错误")) + primary = self._parse_from_internal_api(clean_url) + if primary and primary.get('url'): + return self._clean_response_data(primary) + secondary = self._parse_from_external_api(clean_url) + if secondary and secondary.get('url'): + return self._clean_response_data(secondary) + raise DouyinParserError("两种渠道均未获取到视频地址") except Exception as e: self.LOG.error(f"[抖音] 解析过程发生未知错误: {str(e)}\n{traceback.format_exc()}") raise DouyinParserError(f"未知错误: {str(e)}") + def _build_proxies(self) -> Optional[Dict[str, str]]: + if self.http_proxy: + return {"http": self.http_proxy, "https": self.http_proxy} + return None + + def _parse_from_internal_api(self, clean_url: str) -> Optional[Dict[str, Any]]: + try: + endpoint = "http://192.168.2.32:8999/api/hybrid/video_data" + headers = {"accept": "application/json"} + params = {"url": clean_url, "minimal": "false"} + response = requests.get(endpoint, headers=headers, params=params, timeout=10, proxies=self._build_proxies()) + if response.status_code != 200: + return None + body = response.json() or {} + if body.get("code") != 200: + return None + data = body.get("data") or {} + video = data.get("video") or {} + bit_rates = video.get("bit_rate") or [] + chosen_url = "" + mp4_sorted = sorted([br for br in bit_rates if br.get("format") == "mp4"], key=lambda x: x.get("bit_rate") or 0, reverse=True) + for br in mp4_sorted: + play_addr = br.get("play_addr") or {} + urls = play_addr.get("url_list") or [] + selected = self._prefer_v3_v10(urls) + if selected: + chosen_url = selected + break + if not chosen_url: + play_addr = video.get("play_addr") or {} + urls = play_addr.get("url_list") or [] + selected = self._prefer_v3_v10(urls) + if selected: + chosen_url = selected + cover = (video.get("cover") or {}).get("url_list") or [] + cover_url = cover[0] if cover else "" + caption = data.get("caption") or "无标题" + author = (data.get("author") or {}) + nickname = author.get("nickname") or author.get("unique_id") or "未知作者" + result = {"url": chosen_url or "", "title": caption, "author": nickname, "cover": cover_url} + if result.get("url"): + return result + return None + except Exception: + return None + + def _parse_from_external_api(self, clean_url: str) -> Optional[Dict[str, Any]]: + try: + pay_api_url = "https://api.pearktrue.cn/api/video/api.php" + params = {"url": clean_url, "key": "f56c1fed0c6e64e7"} + response = requests.post(pay_api_url, params=params, timeout=10, proxies=self._build_proxies()) + if response.status_code != 200: + return None + data = response.json() or {} + if data.get("code") == 200: + result = data.get("data", {}) + if result.get("url"): + return result + return None + except Exception: + return None + + def _prefer_v3_v10(self, urls: List[str]) -> Optional[str]: + try: + if not urls: + return None + cleaned = [(u or "").strip().strip("`") for u in urls if u] + def is_vx(n: str) -> bool: + return bool(re.match(r"^v(3|4|5|6|7|8|9|10)(?:[\-.]|$)", n, re.I)) + def is_douyinvod(n: str) -> bool: + return "douyinvod.com" in n.lower() + first = None + for s in cleaned: + netloc = urlparse(s).netloc + if is_vx(netloc) and is_douyinvod(netloc): + return s + if first is None: + first = s + for s in cleaned: + netloc = urlparse(s).netloc + if is_vx(netloc): + return s + for s in cleaned: + netloc = urlparse(s).netloc + if is_douyinvod(netloc): + return s + return first + except Exception: + return urls[0] if urls else None + def _download_stream(self, url, save_path): """ 从指定URL读取视频流并保存到本地