| 123456789101112131415161718192021222324252627282930313233343536373839404142434445 |
- 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()),
- )
|
