from dataclasses import dataclass, field import math from urllib.parse import urlsplit import config @dataclass(frozen=True) class ResultPlatformSettings: # 完整的接收接口 URL,不假设正式平台的路径。 url: str token: str = field(default="", repr=False) auth_header: str = "Authorization" auth_scheme: str = "Bearer" timeout: float = 30 success_field: str = "code" success_values: tuple[str, ...] = ("0", "200") def __post_init__(self): parts = urlsplit(self.url) 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("RESULT_PLATFORM_URL 必须是无内嵌凭证、查询串和片段的 HTTP(S) 接收接口地址") if self.auth_header and not self.token.strip(): raise ValueError("结果平台鉴权需要 RESULT_PLATFORM_TOKEN;无鉴权接口应设置 RESULT_PLATFORM_AUTH_HEADER=none") if any("\r" in value or "\n" in value for value in (self.token, self.auth_header, self.auth_scheme)): raise ValueError("结果平台鉴权配置不能包含换行") if not math.isfinite(self.timeout) or self.timeout <= 0: raise ValueError("结果平台超时必须大于零") if not self.success_field or not self.success_values: raise ValueError("结果平台需要配置成功回执字段及成功值") @classmethod def from_config(cls): header = config.get_str("RESULT_PLATFORM_AUTH_HEADER", "Authorization").strip() scheme = config.get_str("RESULT_PLATFORM_AUTH_SCHEME", "Bearer").strip() return cls( url=config.get_str("RESULT_PLATFORM_URL").strip(), token=config.get_str("RESULT_PLATFORM_TOKEN"), auth_header="" if header.lower() == "none" else header, auth_scheme="" if scheme.lower() == "none" else scheme, timeout=config.get_float("RESULT_PLATFORM_TIMEOUT", 30), success_field=config.get_str("RESULT_PLATFORM_SUCCESS_FIELD", "code"), success_values=tuple(s.strip() for s in config.get_str("RESULT_PLATFORM_SUCCESS_VALUES", "0,200").split(",") if s.strip()), )