151 lines
4.9 KiB
Python
151 lines
4.9 KiB
Python
"""YouTube 开播前网络预检（参照 d2ypp2 network_preflight，Windows/Electron 适配）。"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import os
|
||
from typing import Any, Optional
|
||
|
||
import httpx
|
||
|
||
from web2_network import (
|
||
CONNECT_TIMEOUT,
|
||
DOUYIN_CHECK_URL,
|
||
GOOGLE_CHECK_URL,
|
||
READ_TIMEOUT,
|
||
_latency_get,
|
||
)
|
||
from web2_proxy_config import read_proxy_config
|
||
|
||
YOUTUBE_CHECK_URL = "https://www.youtube.com/generate_204"
|
||
|
||
_PROBE_TARGETS: tuple[dict[str, str], ...] = (
|
||
{"id": "google", "label": "Google", "url": GOOGLE_CHECK_URL, "role": "global"},
|
||
{"id": "youtube", "label": "YouTube", "url": YOUTUBE_CHECK_URL, "role": "global"},
|
||
{"id": "douyin", "label": "抖音源站", "url": DOUYIN_CHECK_URL, "role": "source"},
|
||
)
|
||
|
||
|
||
def _preflight_disabled() -> bool:
|
||
raw = os.environ.get("YOUTUBE_NETWORK_PREFLIGHT", "1").strip().lower()
|
||
return raw in {"0", "false", "no", "off"}
|
||
|
||
|
||
def _global_ok(probes: list[dict[str, Any]]) -> bool:
|
||
return any(p.get("ok") for p in probes if p.get("role") == "global")
|
||
|
||
|
||
def _douyin_ok(probes: list[dict[str, Any]]) -> bool:
|
||
return any(p.get("ok") for p in probes if p.get("id") == "douyin")
|
||
|
||
|
||
async def _probe_all(client: httpx.AsyncClient) -> list[dict[str, Any]]:
|
||
rows: list[dict[str, Any]] = []
|
||
for target in _PROBE_TARGETS:
|
||
ok, ms, err = await _latency_get(client, target["url"])
|
||
rows.append(
|
||
{
|
||
**target,
|
||
"ok": ok,
|
||
"latency_ms": round(ms, 1) if ms >= 0 else None,
|
||
"error": err,
|
||
}
|
||
)
|
||
return rows
|
||
|
||
|
||
async def run_youtube_preflight() -> dict[str, Any]:
|
||
"""返回完整预检报告，供 /youtube/preflight 与开播拦截共用。"""
|
||
if _preflight_disabled():
|
||
return {
|
||
"ok": True,
|
||
"mode": "disabled",
|
||
"message": "网络预检已关闭（YOUTUBE_NETWORK_PREFLIGHT=0）",
|
||
"checks": {},
|
||
"direct": [],
|
||
"proxy": [],
|
||
"proxy_config": read_proxy_config(),
|
||
}
|
||
|
||
timeout = httpx.Timeout(READ_TIMEOUT, connect=CONNECT_TIMEOUT)
|
||
proxy_cfg = read_proxy_config()
|
||
|
||
async with httpx.AsyncClient(timeout=timeout, follow_redirects=True) as client:
|
||
direct = await _probe_all(client)
|
||
|
||
proxy_rows: list[dict[str, Any]] = []
|
||
proxy_http = (proxy_cfg.get("http") or "").strip()
|
||
if proxy_cfg.get("enabled") and proxy_http:
|
||
try:
|
||
async with httpx.AsyncClient(
|
||
timeout=timeout,
|
||
follow_redirects=True,
|
||
proxy=proxy_http,
|
||
) as client:
|
||
proxy_rows = await _probe_all(client)
|
||
except Exception as e:
|
||
proxy_rows = [
|
||
{**t, "ok": False, "latency_ms": None, "error": str(e)}
|
||
for t in _PROBE_TARGETS
|
||
]
|
||
|
||
global_direct = _global_ok(direct)
|
||
global_proxy = _global_ok(proxy_rows)
|
||
douyin_direct = _douyin_ok(direct)
|
||
|
||
ok = bool(global_direct or global_proxy)
|
||
mode = "direct" if global_direct else ("proxy" if global_proxy else "failed")
|
||
|
||
checks = {
|
||
"google": next((p for p in direct if p.get("id") == "google"), {}).get("ok", False),
|
||
"youtube": next((p for p in direct if p.get("id") == "youtube"), {}).get("ok", False),
|
||
"douyin": douyin_direct,
|
||
"global_direct": global_direct,
|
||
"global_proxy": global_proxy,
|
||
"proxy_configured": bool(proxy_cfg.get("enabled") and proxy_http),
|
||
}
|
||
|
||
if ok:
|
||
if global_direct:
|
||
message = "直连可达 Google/YouTube，可尝试开播"
|
||
else:
|
||
message = f"直连失败，但录制代理 {proxy_http} 可达 Google/YouTube"
|
||
if not douyin_direct:
|
||
message += "；抖音源站当前不可达，可能无法拉流"
|
||
else:
|
||
parts = []
|
||
if not global_direct:
|
||
parts.append("Google/YouTube 直连不可达")
|
||
if proxy_cfg.get("enabled"):
|
||
if not global_proxy:
|
||
parts.append("录制代理也未打通 Google/YouTube")
|
||
else:
|
||
parts.append("未启用录制代理")
|
||
if not douyin_direct:
|
||
parts.append("抖音源站不可达")
|
||
message = "；".join(parts)
|
||
|
||
blocking_message: Optional[str] = None
|
||
if not ok:
|
||
blocking_message = (
|
||
f"{message}。请先在「网络」页检测并配置代理，或在下方「录制代理」中填写 Clash 混合端口后再开播。"
|
||
)
|
||
|
||
return {
|
||
"ok": ok,
|
||
"mode": mode,
|
||
"message": message,
|
||
"blocking_message": blocking_message,
|
||
"checks": checks,
|
||
"direct": direct,
|
||
"proxy": proxy_rows,
|
||
"proxy_config": proxy_cfg,
|
||
"douyin_direct_ok": douyin_direct,
|
||
}
|
||
|
||
|
||
async def youtube_start_preflight_message() -> str | None:
|
||
result = await run_youtube_preflight()
|
||
if result.get("ok"):
|
||
return None
|
||
return str(result.get("blocking_message") or result.get("message") or "网络预检未通过")
|