File size: 3,070 Bytes
8ede856 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 | """Network error handling utilities for providers."""
import httpx
from astrbot import logger
def is_connection_error(exc: BaseException) -> bool:
"""Check if an exception is a connection/network related error.
Uses explicit exception type checking instead of brittle string matching.
Handles httpx network errors, timeouts, and common Python network exceptions.
Args:
exc: The exception to check
Returns:
True if the exception is a connection/network error
"""
# Check for httpx network errors
if isinstance(
exc,
(
httpx.ConnectError,
httpx.ConnectTimeout,
httpx.ReadTimeout,
httpx.WriteTimeout,
httpx.PoolTimeout,
httpx.NetworkError,
httpx.ProxyError,
httpx.RequestError,
),
):
return True
# Check for common Python network errors
if isinstance(exc, (TimeoutError, OSError, ConnectionError)):
return True
# Check the __cause__ chain for wrapped connection errors
cause = getattr(exc, "__cause__", None)
if cause is not None and cause is not exc:
return is_connection_error(cause)
return False
def log_connection_failure(
provider_label: str,
error: Exception,
proxy: str | None = None,
) -> None:
"""Log a connection failure with proxy information.
If proxy is not provided, will fallback to check os.environ for
http_proxy/https_proxy environment variables.
Args:
provider_label: The provider name for log prefix (e.g., "OpenAI", "Gemini")
error: The exception that occurred
proxy: The proxy address if configured, or None/empty string
"""
import os
error_type = type(error).__name__
# Fallback to environment proxy if not configured
effective_proxy = proxy
if not effective_proxy:
effective_proxy = os.environ.get(
"http_proxy", os.environ.get("https_proxy", "")
)
if effective_proxy:
logger.error(
f"[{provider_label}] 网络/代理连接失败 ({error_type})。"
f"代理地址: {effective_proxy}，错误: {error}"
)
else:
logger.error(f"[{provider_label}] 网络连接失败 ({error_type})。错误: {error}")
def create_proxy_client(
provider_label: str,
proxy: str | None = None,
) -> httpx.AsyncClient | None:
"""Create an httpx AsyncClient with proxy configuration if provided.
Note: The caller is responsible for closing the client when done.
Consider using the client as a context manager or calling aclose() explicitly.
Args:
provider_label: The provider name for log prefix (e.g., "OpenAI", "Gemini")
proxy: The proxy address (e.g., "http://127.0.0.1:7890"), or None/empty
Returns:
An httpx.AsyncClient configured with the proxy, or None if no proxy
"""
if proxy:
logger.info(f"[{provider_label}] 使用代理: {proxy}")
return httpx.AsyncClient(proxy=proxy)
return None
|