45 lines
1.2 KiB
Python
45 lines
1.2 KiB
Python
from __future__ import annotations
|
|
|
|
from aiohttp import BasicAuth
|
|
from aiogram.client.session.aiohttp import AiohttpSession
|
|
from decouple import config
|
|
|
|
|
|
SUPPORTED_PROXY_PROTOCOLS = {"http", "socks5"}
|
|
|
|
|
|
def build_bot_session() -> AiohttpSession | None:
|
|
proxy_raw = config("BOT_PROXY", default="").strip()
|
|
if not proxy_raw:
|
|
return None
|
|
|
|
proxy = parse_proxy_value(proxy_raw)
|
|
return AiohttpSession(proxy=proxy)
|
|
|
|
|
|
def parse_proxy_value(proxy_raw: str) -> str | tuple[str, BasicAuth]:
|
|
if "://" in proxy_raw:
|
|
return proxy_raw
|
|
|
|
parts = proxy_raw.split(":")
|
|
if len(parts) not in {3, 5}:
|
|
raise ValueError(
|
|
"BOT_PROXY must be in format protocol:ip:port or protocol:ip:port:user:pass"
|
|
)
|
|
|
|
protocol, host, port = parts[:3]
|
|
protocol = protocol.lower()
|
|
|
|
if protocol not in SUPPORTED_PROXY_PROTOCOLS:
|
|
raise ValueError(
|
|
f"Unsupported proxy protocol '{protocol}'. Supported: http, socks5"
|
|
)
|
|
|
|
proxy_url = f"{protocol}://{host}:{port}"
|
|
|
|
if len(parts) == 3:
|
|
return proxy_url
|
|
|
|
username, password = parts[3], parts[4]
|
|
return proxy_url, BasicAuth(login=username, password=password)
|