From 1c9d5f81bbee366c4d0f654ef72adeb65308e134 Mon Sep 17 00:00:00 2001 From: CI Bot Date: Thu, 17 Sep 2026 09:18:52 +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 | 40 +++++++++++++++++++++------ 2 files changed, 36 insertions(+), 13 deletions(-) diff --git a/apps/api/app/api/routes/scripts_ai.py b/apps/api/app/api/routes/scripts_ai.py index c37575728..7bf1d3f0b 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 1c762b0ba..c4fb1e700 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 @@ -41,7 +42,12 @@ def _pick_best_url(url_list): return None for host_hint in _CDN_HOSTS: for u in url_list: - if isinstance(u, str) and host_hint in u and u.endswith(".mp4") or (isinstance(u, str) and host_hint in u and "/mp4/" in u): + if ( + isinstance(u, str) + and host_hint in u + and u.endswith(".mp4") + or (isinstance(u, str) and host_hint in u and "/mp4/" in u) + ): return u for host_hint in _CDN_HOSTS: for u in url_list: @@ -66,7 +72,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: @@ -117,7 +125,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 @@ -128,7 +138,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 @@ -147,8 +159,12 @@ def fetch_douyin_video_url(page_url: str, timeout: int = 15, max_retries: int = # 此时没有可用视频直链，返回 (None, desc) 让调用方仅使用文案。 images = item.get("images") or [] if images and not video.get("play_addr_h264", {}).get("url_list"): - logger.info("Feed API 返回图文视频: aweme_id=%s images=%d desc_len=%d（无视频直链，返回文案）", - aweme_id, len(images), len(desc)) + logger.info( + "Feed API 返回图文视频: aweme_id=%s images=%d desc_len=%d（无视频直链，返回文案）", + aweme_id, + len(images), + len(desc), + ) return None, desc # 提取无水印直链：优先 download_addr（含 logo 但 CDN 直链稳定），再 play_addr_h264/play_addr @@ -167,7 +183,7 @@ def fetch_douyin_video_url(page_url: str, timeout: int = 15, max_retries: int = # bit_rate 里的多码率地址作为最后兜底 if not play_url: - for br_entry in (video.get("bit_rate") or []): + for br_entry in video.get("bit_rate") or []: for addr_key in addr_keys: addr = br_entry.get(addr_key) or {} url_list = addr.get("url_list") or [] @@ -183,7 +199,12 @@ 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 type=%d", - attempt + 1, aweme_id, len(play_url), len(desc), duration, aweme_type, + attempt + 1, + aweme_id, + len(play_url), + len(desc), + duration, + aweme_type, ) return play_url, desc @@ -200,11 +221,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: