130 lines
4.8 KiB
Python
130 lines
4.8 KiB
Python
"""运行时环境配置：住宅代理 + 浏览器显示。
|
|
|
|
用于解决「部署到云服务器后」两类常见问题：
|
|
1. 机房 IP 触发抖音风控（7911）—— 通过 KEFU_DOUYIN_PROXY 让抖音请求走住宅代理。
|
|
2. 无图形界面的 Linux 起不来有头浏览器 —— 自动拉起 Xvfb 虚拟显示。
|
|
|
|
全部通过环境变量控制，无需改代码：
|
|
|
|
KEFU_DOUYIN_PROXY 抖音 IM HTTP 请求与浏览器登录走的代理，绕开机房 IP 风控。
|
|
形如 http://user:pass@host:port 或 socks5://host:port
|
|
KEFU_BROWSER_HEADLESS 是否使用无头浏览器（1/true 开启）。默认 false：抖音安全 SDK
|
|
对 headless 判定严格，无头易生成无效 ts_sign，反而刷新无效。
|
|
"""
|
|
import asyncio
|
|
import logging
|
|
import os
|
|
from typing import Optional
|
|
from urllib.parse import urlparse
|
|
|
|
logger = logging.getLogger("rpa_engine.runtime")
|
|
|
|
_TRUE = {"1", "true", "yes", "on"}
|
|
_FALSE = {"0", "false", "no", "off"}
|
|
|
|
_NO_DISPLAY_HINT = (
|
|
"当前是无图形界面的 Linux 服务器，且无法启动虚拟显示来运行有头浏览器。"
|
|
"抖音安全 SDK 对 headless 判定严格，扫码登录 / 刷新凭证需要有头 Chromium。请任选其一：\n"
|
|
" 1) 安装 Xvfb + pyvirtualdisplay，让程序自动拉起虚拟显示：\n"
|
|
" Debian/Ubuntu: apt install -y xvfb && pip install pyvirtualdisplay\n"
|
|
" CentOS/Rocky : yum install -y xorg-x11-server-Xvfb && pip install pyvirtualdisplay\n"
|
|
" 2) 或用 xvfb-run 启动后端：xvfb-run -a ./start_web.sh\n"
|
|
" 3) 或设置 KEFU_BROWSER_HEADLESS=1 强制无头（更易触发抖音风控，不推荐）。"
|
|
)
|
|
|
|
|
|
def get_douyin_proxy() -> Optional[str]:
|
|
"""读取抖音请求代理 URL（未配置返回 None）。"""
|
|
val = (os.getenv("KEFU_DOUYIN_PROXY") or "").strip()
|
|
return val or None
|
|
|
|
|
|
def httpx_proxy() -> Optional[str]:
|
|
"""供 httpx.AsyncClient(proxy=...) 使用的代理 URL。"""
|
|
return get_douyin_proxy()
|
|
|
|
|
|
def requests_proxies() -> Optional[dict]:
|
|
"""供 requests.get(proxies=...) 使用的代理字典。"""
|
|
url = get_douyin_proxy()
|
|
if not url:
|
|
return None
|
|
return {"http": url, "https": url}
|
|
|
|
|
|
def playwright_proxy() -> Optional[dict]:
|
|
"""转成 Playwright launch(proxy=...) 所需结构（未配置或无法解析返回 None）。"""
|
|
url = get_douyin_proxy()
|
|
if not url:
|
|
return None
|
|
parsed = urlparse(url)
|
|
if not parsed.hostname:
|
|
logger.warning("KEFU_DOUYIN_PROXY 格式无法解析，已忽略：%s", url)
|
|
return None
|
|
server = f"{parsed.scheme or 'http'}://{parsed.hostname}"
|
|
if parsed.port:
|
|
server += f":{parsed.port}"
|
|
proxy: dict[str, str] = {"server": server}
|
|
if parsed.username:
|
|
proxy["username"] = parsed.username
|
|
if parsed.password:
|
|
proxy["password"] = parsed.password
|
|
return proxy
|
|
|
|
|
|
def resolve_headless(default: bool = False) -> bool:
|
|
"""根据 KEFU_BROWSER_HEADLESS 决定是否无头；未设置时用 default。"""
|
|
val = (os.getenv("KEFU_BROWSER_HEADLESS") or "").strip().lower()
|
|
if val in _TRUE:
|
|
return True
|
|
if val in _FALSE:
|
|
return False
|
|
return default
|
|
|
|
|
|
# 进程内仅启动一次的虚拟显示（Xvfb）句柄
|
|
_virtual_display = None
|
|
_virtual_display_failed = False
|
|
|
|
|
|
def _start_virtual_display_sync() -> Optional[str]:
|
|
"""在无 DISPLAY 的 Linux 上启动一次 Xvfb 虚拟显示（阻塞，需放线程执行）。"""
|
|
global _virtual_display, _virtual_display_failed
|
|
|
|
# 仅 Linux 且无 DISPLAY 时才需要虚拟显示；Windows/macOS 有桌面，直接返回。
|
|
if os.name != "posix":
|
|
return os.environ.get("DISPLAY")
|
|
if os.environ.get("DISPLAY"):
|
|
return os.environ["DISPLAY"]
|
|
if _virtual_display is not None:
|
|
return os.environ.get("DISPLAY")
|
|
if _virtual_display_failed:
|
|
raise RuntimeError(_NO_DISPLAY_HINT)
|
|
|
|
try:
|
|
from pyvirtualdisplay import Display
|
|
except ImportError as e:
|
|
_virtual_display_failed = True
|
|
raise RuntimeError(_NO_DISPLAY_HINT) from e
|
|
|
|
try:
|
|
disp = Display(visible=False, size=(1280, 800))
|
|
disp.start() # 设置 os.environ['DISPLAY']
|
|
except Exception as e:
|
|
_virtual_display_failed = True
|
|
raise RuntimeError(_NO_DISPLAY_HINT) from e
|
|
|
|
_virtual_display = disp
|
|
logger.info("已启动 Xvfb 虚拟显示 DISPLAY=%s 供有头浏览器使用", os.environ.get("DISPLAY"))
|
|
return os.environ.get("DISPLAY")
|
|
|
|
|
|
async def ensure_browser_display(headless: bool) -> None:
|
|
"""有头模式在无 DISPLAY 的 Linux 上自动拉起 Xvfb 虚拟显示。
|
|
|
|
headless=True 时无需显示，直接返回；启动失败抛出带操作指引的 RuntimeError。
|
|
"""
|
|
if headless:
|
|
return
|
|
await asyncio.to_thread(_start_virtual_display_sync)
|