b77a8cadb2
Some checks failed
Upstream Sync / Sync latest commits from upstream repo (push) Has been cancelled
97 lines
3.0 KiB
Python
97 lines
3.0 KiB
Python
"""
|
||
仅给「录制子进程」使用的代理配置（FFmpeg / youtube2 等），不修改 Windows 系统代理。
|
||
用户在本机运行 Clash Verge 等时，通常填 http://127.0.0.1:7890 与 socks5://127.0.0.1:7890。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import os
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
from configparser import ConfigParser
|
||
|
||
BASE_DIR = Path(__file__).resolve().parent
|
||
PROXY_INI = BASE_DIR / "config" / "ffmpeg_proxy.ini"
|
||
|
||
|
||
def _default_ini_text() -> str:
|
||
return (
|
||
"[proxy]\n"
|
||
"# 开启后仅作用于本软件拉起的录制/推流进程，不会改系统代理\n"
|
||
"enabled = no\n"
|
||
"# 与 Clash Verge「混合端口」一致即可（示例）\n"
|
||
"http = http://127.0.0.1:7890\n"
|
||
"socks = socks5://127.0.0.1:7890\n"
|
||
)
|
||
|
||
|
||
def ensure_proxy_config_file() -> None:
|
||
PROXY_INI.parent.mkdir(parents=True, exist_ok=True)
|
||
if not PROXY_INI.exists():
|
||
PROXY_INI.write_text(_default_ini_text(), encoding="utf-8")
|
||
|
||
|
||
def read_proxy_config() -> dict[str, Any]:
|
||
ensure_proxy_config_file()
|
||
cp = ConfigParser()
|
||
cp.read(PROXY_INI, encoding="utf-8")
|
||
if not cp.has_section("proxy"):
|
||
cp.add_section("proxy")
|
||
enabled = cp.get("proxy", "enabled", fallback="no").strip().lower() in (
|
||
"1",
|
||
"true",
|
||
"yes",
|
||
"on",
|
||
"是",
|
||
"开",
|
||
)
|
||
http = (cp.get("proxy", "http", fallback="") or "").strip()
|
||
socks = (cp.get("proxy", "socks", fallback="") or "").strip()
|
||
return {"enabled": enabled, "http": http, "socks": socks}
|
||
|
||
|
||
def write_proxy_config(data: dict[str, Any]) -> None:
|
||
ensure_proxy_config_file()
|
||
cp = ConfigParser()
|
||
cp.read(PROXY_INI, encoding="utf-8")
|
||
if not cp.has_section("proxy"):
|
||
cp.add_section("proxy")
|
||
cp.set("proxy", "enabled", "yes" if data.get("enabled") else "no")
|
||
cp.set("proxy", "http", str(data.get("http") or "").strip())
|
||
cp.set("proxy", "socks", str(data.get("socks") or "").strip())
|
||
with open(PROXY_INI, "w", encoding="utf-8") as f:
|
||
cp.write(f)
|
||
|
||
|
||
def subprocess_env_overlay() -> dict[str, str]:
|
||
"""
|
||
合并进子进程 environ 的键值（不含整个 os.environ）。
|
||
youtube2 / tiktok2 会读 STREAM_HTTP_PROXY / STREAM_SOCKS_PROXY。
|
||
"""
|
||
cfg = read_proxy_config()
|
||
out: dict[str, str] = {}
|
||
if not cfg["enabled"]:
|
||
return out
|
||
http = cfg["http"]
|
||
socks = cfg["socks"]
|
||
if http:
|
||
out["STREAM_HTTP_PROXY"] = http
|
||
out["HTTP_PROXY"] = http
|
||
out["HTTPS_PROXY"] = http
|
||
out["http_proxy"] = http
|
||
out["https_proxy"] = http
|
||
if socks:
|
||
out["STREAM_SOCKS_PROXY"] = socks
|
||
out["ALL_PROXY"] = socks
|
||
out["all_proxy"] = socks
|
||
return out
|
||
|
||
|
||
def merge_child_env(base: dict[str, str] | None = None) -> dict[str, str]:
|
||
"""供 asyncio subprocess / Popen 使用（完整继承系统环境后再叠加代理）。"""
|
||
env = dict(os.environ)
|
||
if base:
|
||
env.update(base)
|
||
env.update(subprocess_env_overlay())
|
||
return env
|