| 12345678910111213141516171819202122232425262728293031323334 |
- from dataclasses import dataclass, field
- from urllib.parse import urlsplit
- import config
- @dataclass(frozen=True)
- class GatewaySettings:
- base_url: str
- bearer: str = field(repr=False)
- file_base_url: str = ""
- upload_path: str = "ai-agent/ventilation"
- timeout: float = 30
- max_file_bytes: int = 50 * 1024 * 1024
- def __post_init__(self):
- for value in (self.base_url, self.file_base_url):
- if value:
- parts = urlsplit(value)
- if parts.scheme not in {"http", "https"} or not parts.netloc or parts.username or parts.password or parts.query or parts.fragment:
- raise ValueError("网关及文件系统地址必须为不含凭证或查询串的 HTTP(S) 根地址")
- if not self.base_url or not self.bearer.strip():
- raise ValueError("启用网关需要 SHANXI_GATEWAY_URL 和 SHANXI_GATEWAY_TOKEN")
- if self.timeout <= 0 or self.max_file_bytes <= 0:
- raise ValueError("网关超时及文件大小上限必须大于零")
- @classmethod
- def from_config(cls):
- return cls(config.get_str("SHANXI_GATEWAY_URL").rstrip("/"),
- config.get_str("SHANXI_GATEWAY_TOKEN"),
- config.get_str("SHANXI_FILE_BASE_URL").rstrip("/"),
- config.get_str("SHANXI_UPLOAD_PATH", "ai-agent/ventilation"),
- config.get_float("SHANXI_HTTP_TIMEOUT", 30),
- config.get_int("SHANXI_MAX_FILE_MB", 50) * 1024 * 1024)
|