From 54a3c5003567dde8eb0c03004d785b528f84dfdd Mon Sep 17 00:00:00 2001 From: CI Bot Date: Thu, 17 Sep 2026 08:48:26 +0000 Subject: [PATCH] style: auto-format with black + isort + ruff + prettier [skip ci-format-check] --- apps/api/app/api/routes/scripts_ai.py | 9 +++++---- packages/douyin_parser/__init__.py | 22 +++++++++++++++++----- 2 files changed, 22 insertions(+), 9 deletions(-) diff --git a/apps/api/app/api/routes/scripts_ai.py b/apps/api/app/api/routes/scripts_ai.py index 929560698..f8217e740 100644 --- a/apps/api/app/api/routes/scripts_ai.py +++ b/apps/api/app/api/routes/scripts_ai.py @@ -356,7 +356,7 @@ def _fetch_ttwid(): # 方式一:直接调 ttwid 注册接口 try: with httpx.Client(timeout=8, follow_redirects=True, verify=False) as http: - r = http.post( + http.post( "https://ttwid.bytedance.com/ttwid/union/register/", json={ "region": "cn", @@ -769,7 +769,7 @@ def _ytdlp_download_and_local_asr(page_url, temp_dir, cookiefile=None): cookie_files_to_try.append(cookiefile) # 2. 合成 cookie 重试(复用 _ytdlp_extract 的 cookie 生成逻辑,8次重试) - import secrets, time as _time + import time as _time ttwid = _fetch_ttwid() for i in range(8): if i % 3 == 0: @@ -945,9 +945,10 @@ def _html_scrape_direct_url(page_url, timeout=15): @router.get("/douyin/__debug_diag") def douyin_diag(): """[Staging only] 容器内网络/yt-dlp诊断""" - import httpx as _httpx - import time as _t import tempfile as _tf + import time as _t + + import httpx as _httpx import yt_dlp as _ytdlp results = {} diff --git a/packages/douyin_parser/__init__.py b/packages/douyin_parser/__init__.py index dd3242b7b..d486e827f 100644 --- a/packages/douyin_parser/__init__.py +++ b/packages/douyin_parser/__init__.py @@ -4,6 +4,7 @@ 接口直接返回 aweme_list 包含视频元信息和 play_addr 无水印直链。 此接口不需要任何签名算法、不需要 cookies、不需要特殊 TLS 指纹,稳定性 >95%。 """ + from __future__ import annotations import logging @@ -39,7 +40,9 @@ def _extract_aweme_id(url: str) -> Optional[str]: return None -def fetch_douyin_video_url(page_url: str, timeout: int = 15, max_retries: int = 3) -> Tuple[Optional[str], Optional[str]]: +def fetch_douyin_video_url( + page_url: str, timeout: int = 15, max_retries: int = 3 +) -> Tuple[Optional[str], Optional[str]]: """通过抖音 App Feed API 获取视频无水印直链和文案。 Args: @@ -90,7 +93,9 @@ def fetch_douyin_video_url(page_url: str, timeout: int = 15, max_retries: int = continue if len(resp.text) < 100: - logger.warning("Feed API 第%d次: 返回内容过短 len=%d body=%s", attempt + 1, len(resp.text), resp.text[:100]) + logger.warning( + "Feed API 第%d次: 返回内容过短 len=%d body=%s", attempt + 1, len(resp.text), resp.text[:100] + ) time.sleep(0.5 * (attempt + 1)) continue @@ -101,7 +106,9 @@ def fetch_douyin_video_url(page_url: str, timeout: int = 15, max_retries: int = status_msg = data.get("status_msg", "") logger.warning( "Feed API 第%d次: aweme_list 为空 status_code=%s msg=%s", - attempt + 1, status_code, status_msg, + attempt + 1, + status_code, + status_msg, ) time.sleep(0.5 * (attempt + 1)) continue @@ -131,7 +138,11 @@ def fetch_douyin_video_url(page_url: str, timeout: int = 15, max_retries: int = duration = video.get("duration", 0) / 1000 if video.get("duration") else 0 logger.info( "抖音 Feed API 成功(第%d次): aweme_id=%s play_len=%d desc_len=%d dur=%.1f", - attempt + 1, aweme_id, len(play_url), len(desc), duration, + attempt + 1, + aweme_id, + len(play_url), + len(desc), + duration, ) return play_url, desc @@ -148,11 +159,12 @@ def fetch_douyin_video_url(page_url: str, timeout: int = 15, max_retries: int = if __name__ == "__main__": import sys + logging.basicConfig(level=logging.INFO) test_url = sys.argv[1] if len(sys.argv) > 1 else "https://www.douyin.com/video/7661819662322649065" url, desc = fetch_douyin_video_url(test_url) if url: - print(f"\n✅ SUCCESS!") + print("\n✅ SUCCESS!") print(f"desc: {desc}") print(f"play_url: {url[:200]}") else: