154 lines
6.4 KiB
Python
154 lines
6.4 KiB
Python
from __future__ import annotations
|
||
|
||
import os
|
||
from dataclasses import dataclass
|
||
from pathlib import Path
|
||
|
||
from dotenv import load_dotenv
|
||
|
||
# 优先从包上级目录（wx_python 根）加载 .env，避免工作目录不在项目根时失效
|
||
_ROOT = Path(__file__).resolve().parent.parent
|
||
load_dotenv(_ROOT / ".env")
|
||
load_dotenv()
|
||
|
||
|
||
def _parse_bot_usernames(raw: str) -> tuple[str, ...]:
|
||
"""从逗号分隔字符串解析 Bot 用户名（去 @、去空）。"""
|
||
parts = [p.strip().lstrip("@") for p in raw.split(",")]
|
||
bots = tuple(p for p in parts if p)
|
||
if not bots:
|
||
raise RuntimeError(
|
||
"缺少 TELEGRAM_BOT_USERNAME：填写至少一个 Bot 用户名；多个用英文逗号分隔，不要带 @ 也可"
|
||
)
|
||
return bots
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class Settings:
|
||
api_id: int
|
||
api_hash: str
|
||
session_path: Path
|
||
"""已配置的 Bot 用户名列表（顺序：默认目标为第一个）。"""
|
||
bot_usernames: tuple[str, ...]
|
||
bridge_host: str
|
||
bridge_port: int
|
||
bridge_token: str | None
|
||
# Telethon 连接 Telegram 用的代理（与系统「全局代理」无关，需显式配置）
|
||
proxy_type: str | None
|
||
proxy_host: str | None
|
||
proxy_port: int | None
|
||
proxy_user: str | None
|
||
proxy_password: str | None
|
||
# SOCKS 上 rdns=True 会在代理侧解析域名，部分代理会卡死；默认 False 多为本机解析再连
|
||
proxy_rdns: bool
|
||
# Telethon 单次连接超时秒数（默认 10 经代理不够）
|
||
connect_timeout: int
|
||
connection_retries: int
|
||
retry_delay: int
|
||
# Telethon Connection 类：tcp_obfuscated 在受限网络 + 代理下常比 tcp_full 更稳
|
||
connection_mode: str
|
||
# /v1/forward 在 wait_reply 时等待 Bot 回复的上限（秒）
|
||
bot_reply_timeout: float
|
||
# wait_reply 时默认取 Bot 第几条连续回复（可被请求体 reply_take_nth 覆盖）
|
||
bot_reply_take_nth: int
|
||
|
||
@property
|
||
def default_bot_username(self) -> str:
|
||
return self.bot_usernames[0]
|
||
|
||
def resolve_bot_username(self, bot: str | None) -> str:
|
||
"""将请求里的 bot 解析为已配置的用户名；省略则用列表第一个。"""
|
||
if bot is None:
|
||
return self.bot_usernames[0]
|
||
key = bot.strip().lstrip("@").lower()
|
||
by_lower = {b.lower(): b for b in self.bot_usernames}
|
||
if key not in by_lower:
|
||
raise ValueError(
|
||
f"未知 Bot「{bot}」，当前允许: {', '.join(self.bot_usernames)}"
|
||
)
|
||
return by_lower[key]
|
||
|
||
@staticmethod
|
||
def load() -> "Settings":
|
||
api_id_raw = os.environ.get("TELEGRAM_API_ID", "").strip()
|
||
api_hash = os.environ.get("TELEGRAM_API_HASH", "").strip()
|
||
if not api_id_raw or not api_hash:
|
||
raise RuntimeError(
|
||
"缺少 TELEGRAM_API_ID / TELEGRAM_API_HASH。"
|
||
"请到 https://my.telegram.org 申请后写入环境变量或 .env"
|
||
)
|
||
bot_raw = os.environ.get("TELEGRAM_BOT_USERNAME", "").strip()
|
||
bots = _parse_bot_usernames(bot_raw)
|
||
|
||
# 默认可写绝对路径：避免「login 在项目目录、服务从别的工作目录启动」找不到同一份 session
|
||
_default_session = (_ROOT / "tg_bridge.session").resolve()
|
||
session_raw = os.environ.get("TELEGRAM_SESSION_PATH", "").strip()
|
||
session = (
|
||
Path(session_raw).expanduser().resolve()
|
||
if session_raw
|
||
else _default_session
|
||
)
|
||
|
||
token = os.environ.get("BRIDGE_TOKEN", "").strip() or None
|
||
host = os.environ.get("BRIDGE_HOST", "0.0.0.0").strip()
|
||
port = int(os.environ.get("BRIDGE_PORT", "18080"))
|
||
|
||
ptype = os.environ.get("TELEGRAM_PROXY_TYPE", "").strip() or None
|
||
phost = os.environ.get("TELEGRAM_PROXY_HOST", "").strip() or None
|
||
pport_raw = os.environ.get("TELEGRAM_PROXY_PORT", "").strip()
|
||
pport: int | None = int(pport_raw) if pport_raw else None
|
||
puser = os.environ.get("TELEGRAM_PROXY_USER", "").strip() or None
|
||
ppwd_raw = os.environ.get("TELEGRAM_PROXY_PASSWORD")
|
||
ppwd: str | None = None
|
||
if puser is not None:
|
||
ppwd = ppwd_raw if ppwd_raw is not None else ""
|
||
|
||
if ptype and (not phost or not pport):
|
||
raise RuntimeError(
|
||
"已设置 TELEGRAM_PROXY_TYPE 时，须同时设置 TELEGRAM_PROXY_HOST 与 TELEGRAM_PROXY_PORT"
|
||
)
|
||
if (phost or pport_raw) and not ptype:
|
||
raise RuntimeError("设置 TELEGRAM_PROXY_HOST / TELEGRAM_PROXY_PORT 时须设置 TELEGRAM_PROXY_TYPE（http|socks5|socks4）")
|
||
|
||
rdns_raw = os.environ.get("TELEGRAM_PROXY_RDNS", "").strip().lower()
|
||
if rdns_raw in ("1", "true", "yes", "on"):
|
||
proxy_rdns = True
|
||
elif rdns_raw in ("0", "false", "no", "off"):
|
||
proxy_rdns = False
|
||
elif not rdns_raw:
|
||
# 未配置：HTTP 代理常用远程 DNS；SOCKS 默认本机解析更稳
|
||
proxy_rdns = bool(ptype and str(ptype).lower() in ("http", "https"))
|
||
else:
|
||
raise RuntimeError("TELEGRAM_PROXY_RDNS 请使用 true/false")
|
||
|
||
connect_timeout = int(os.environ.get("TELEGRAM_CONNECT_TIMEOUT", "90"))
|
||
connection_retries = int(os.environ.get("TELEGRAM_CONNECTION_RETRIES", "5"))
|
||
retry_delay = int(os.environ.get("TELEGRAM_RETRY_DELAY", "3"))
|
||
connection_mode = os.environ.get("TELEGRAM_CONNECTION", "tcp_full").strip() or "tcp_full"
|
||
bot_reply_timeout = float(os.environ.get("TELEGRAM_BOT_REPLY_TIMEOUT", "120"))
|
||
bot_reply_take_nth = int(os.environ.get("BOT_REPLY_TAKE_NTH", "1"))
|
||
if bot_reply_take_nth < 1 or bot_reply_take_nth > 20:
|
||
raise RuntimeError("BOT_REPLY_TAKE_NTH 须在 1～20 之间")
|
||
|
||
return Settings(
|
||
api_id=int(api_id_raw),
|
||
api_hash=api_hash,
|
||
session_path=session,
|
||
bot_usernames=bots,
|
||
bridge_host=host,
|
||
bridge_port=port,
|
||
bridge_token=token,
|
||
proxy_type=ptype,
|
||
proxy_host=phost,
|
||
proxy_port=pport,
|
||
proxy_user=puser,
|
||
proxy_password=ppwd,
|
||
proxy_rdns=proxy_rdns,
|
||
connect_timeout=connect_timeout,
|
||
connection_retries=connection_retries,
|
||
retry_delay=retry_delay,
|
||
connection_mode=connection_mode,
|
||
bot_reply_timeout=bot_reply_timeout,
|
||
bot_reply_take_nth=bot_reply_take_nth,
|
||
)
|