style: auto-format with black + isort + ruff + prettier [skip ci-format-check]
CI/CD Pipeline / Check if frontend-only change (push) Has been skipped
CI/CD Pipeline / Dedup Check - skip PR tests when covered by push pipeline (push) Successful in 2s
CI/CD Pipeline / Frontend Lint (push) Has been skipped
CI/CD Pipeline / PR Build API Image (push) Has been skipped
CI/CD Pipeline / PR Build Web Image (push) Has been skipped
CI/CD Pipeline / PR Build Worker Image (push) Has been skipped
CI/CD Pipeline / Frontend Unit Tests (push) Successful in 2m14s
CI/CD Pipeline / Validate - Style (push) Failing after 4m4s
CI/CD Pipeline / Validate - Python (mypy + alembic) (push) Successful in 4m29s
CI/CD Pipeline / Validate - Security (push) Successful in 11m23s
CI/CD Pipeline / Check push changed paths (push) Successful in 12m8s
CI/CD Pipeline / Build Staging Web Image (push) Has been skipped
CI/CD Pipeline / Build Staging Worker Image (push) Successful in 23s
CI/CD Pipeline / Build Staging API Image (push) Successful in 29s
CI/CD Pipeline / Retag skipped Staging API Image (push) Has been skipped
CI/CD Pipeline / Retag skipped Staging Worker Image (push) Has been skipped
CI/CD Pipeline / Retag skipped Staging Web Image (push) Successful in 10s
CI/CD Pipeline / Deploy Staging (Watchtower auto-deploy) (push) Successful in 42s
CI/CD Pipeline / Staging E2E Tests (push) Successful in 1m5s
CI/CD Pipeline / ACR Image Cleanup (push) Successful in 1m37s
CI/CD Pipeline / Staging API Integration Tests (push) Successful in 3m17s
CI/CD Pipeline / Integration Tests (push) Successful in 17m1s
CI/CD Pipeline / Unit Tests (push) Failing after 18m24s
CI/CD Pipeline / Build Production API Image (push) Has been skipped
CI/CD Pipeline / Build Production Web Image (push) Has been skipped
CI/CD Pipeline / Build Production Worker Image (push) Has been skipped
CI/CD Pipeline / CI Gate (push) Has been skipped
CI/CD Pipeline / Deploy Production (push) Has been skipped
CI/CD Pipeline / Canary Release to Production (push) Has been skipped
CI/CD Pipeline / Production Browser E2E (push) Has been skipped

This commit is contained in:
CI Bot
2026-09-17 09:18:52 +00:00
parent 62720a4c70
commit 1c9d5f81bb
2 changed files with 36 additions and 13 deletions
+5 -4
View File
@@ -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 = {}
+31 -9
View File
@@ -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: