"""门户跳转集成配置:全部从环境变量(.env)读取。 对接对端门户系统时,优先只改 .env(provider=http 的声明式对接); 形态特殊时改 portal_provider.py 中的一个函数(见该文件内模板)。 本模块在每次请求时读取环境变量,改 .env 后重启进程生效。 """ from __future__ import annotations import os from dataclasses import dataclass ALLOWED_PROVIDERS = {"", "mock", "http", "custom"} ALLOWED_LOCAL_FIELDS = {"externalId", "username", "realName", "orgName"} class PortalConfigError(Exception): """门户配置缺失或非法(服务端 .env 问题,非用户输入问题)。""" @dataclass(frozen=True) class PortalConfig: provider: str # ""=关闭 / mock / http / custom token_param: str # ---- http provider:验签 ---- verify_url: str verify_method: str # GET / POST verify_token_in: str # header / body / query verify_header: str verify_header_template: str # 内含 {token} 占位 verify_success_field: str verify_success_value: str verify_user_field: str # 验签响应中用户信息对象所在字段路径(如 data) # ---- http provider:独立的用户信息接口(可空=验签已含用户信息)---- userinfo_url: str userinfo_uid_param: str # ---- 用户信息字段映射:local -> remote ---- field_map: dict[str, str] # ---- 新用户落库默认值 ---- default_org_id: int | None # ---- 其他 ---- http_timeout_seconds: int @property def enabled(self) -> bool: return bool(self.provider) def _env(name: str, default: str = "") -> str: return os.environ.get(name, default).strip() def _parse_field_map(raw: str) -> dict[str, str]: """解析 ``local:remote,local:remote`` 形式的字段映射。""" field_map: dict[str, str] = {} for chunk in raw.split(","): chunk = chunk.strip() if not chunk: continue local, sep, remote = chunk.partition(":") local = local.strip() remote = remote.strip() if not sep or local not in ALLOWED_LOCAL_FIELDS or not remote: raise PortalConfigError( f"DMS_PORTAL_FIELD_MAP格式非法({chunk})," f"合法local字段:{sorted(ALLOWED_LOCAL_FIELDS)}" ) field_map[local] = remote return field_map def load_portal_config() -> PortalConfig: provider = _env("DMS_PORTAL_PROVIDER").lower() if provider not in ALLOWED_PROVIDERS: raise PortalConfigError( "DMS_PORTAL_PROVIDER必须是 mock/http/custom 之一,留空表示关闭门户登录" ) field_map = _parse_field_map( _env( "DMS_PORTAL_FIELD_MAP", "externalId:userId,username:loginName,realName:realName,orgName:orgName", ) ) if "externalId" not in field_map: raise PortalConfigError("DMS_PORTAL_FIELD_MAP必须包含externalId映射") raw_org_id = _env("DMS_PORTAL_DEFAULT_ORG_ID") try: default_org_id = int(raw_org_id) if raw_org_id else None except ValueError as exc: raise PortalConfigError("DMS_PORTAL_DEFAULT_ORG_ID必须是整数") from exc timeout_raw = _env("DMS_PORTAL_HTTP_TIMEOUT_SECONDS", "10") try: timeout_seconds = int(timeout_raw) if timeout_raw else 10 except ValueError as exc: raise PortalConfigError("DMS_PORTAL_HTTP_TIMEOUT_SECONDS必须是整数") from exc token_in = _env("DMS_PORTAL_VERIFY_TOKEN_IN", "header").lower() if token_in not in {"header", "body", "query"}: raise PortalConfigError("DMS_PORTAL_VERIFY_TOKEN_IN必须是header/body/query") return PortalConfig( provider=provider, token_param=_env("DMS_PORTAL_TOKEN_PARAM", "token"), verify_url=_env("DMS_PORTAL_VERIFY_URL"), verify_method=_env("DMS_PORTAL_VERIFY_METHOD", "POST").upper(), verify_token_in=token_in, verify_header=_env("DMS_PORTAL_VERIFY_HEADER", "Authorization"), verify_header_template=_env("DMS_PORTAL_VERIFY_HEADER_TEMPLATE", "Bearer {token}"), verify_success_field=_env("DMS_PORTAL_VERIFY_SUCCESS_FIELD", "code"), verify_success_value=_env("DMS_PORTAL_VERIFY_SUCCESS_VALUE", "0"), verify_user_field=_env("DMS_PORTAL_VERIFY_USER_FIELD", "data"), userinfo_url=_env("DMS_PORTAL_USERINFO_URL"), userinfo_uid_param=_env("DMS_PORTAL_USERINFO_UID_PARAM", "userId"), field_map=field_map, default_org_id=default_org_id, http_timeout_seconds=timeout_seconds, )