document-management/backend/dms/services/portal_provider.py

"""门户身份适配器：验签 + 拉取用户信息（返回统一中间结构 PortalIdentity）。

════════════════════════════════════════════════════════════════
对端门户系统确定后，本文件是唯一可能需要修改的后端代码：

1. 标准REST形态 → .env 配 DMS_PORTAL_PROVIDER=http + 相关键，无需改代码；
2. 特殊形态（本地公钥验JWT / HMAC共享密钥 / 怪异接口）→
   .env 配 DMS_PORTAL_PROVIDER=custom，并替换本文件底部
   _custom_provider 函数体（按其内的示例模板手写）。
════════════════════════════════════════════════════════════════
"""

from __future__ import annotations

from dataclasses import dataclass
from urllib.parse import parse_qs

import requests as http_client

from dms.services.portal_config import PortalConfigError, load_portal_config


@dataclass(frozen=True)
class PortalIdentity:
    """对端门户用户身份的统一中间结构（与对端具体格式解耦）。"""

    external_id: str
    username: str
    real_name: str
    org_name: str | None = None


class PortalTokenError(Exception):
    """对端验签未通过（token非法/过期），端点层转为401。"""


class PortalIdentityError(Exception):
    """对端返回的用户信息缺少必要字段。"""


def extract_token(query_string: str) -> str | None:
    """从跳转URL的query串中提取token（参数名来自配置）。"""
    cfg = load_portal_config()
    values = parse_qs(query_string.lstrip("?")).get(cfg.token_param, [])
    value = values[0].strip() if values else ""
    return value or None


def resolve_portal_identity(token: str) -> PortalIdentity:
    """按配置的provider类型分发验签与身份拉取。"""
    cfg = load_portal_config()
    if cfg.provider == "mock":
        return _mock_provider(token)
    if cfg.provider == "http":
        return _http_provider(token)
    if cfg.provider == "custom":
        return _custom_provider(token)
    raise PortalConfigError("DMS_PORTAL_PROVIDER未配置")


# ---------------------------------------------------------------- mock


def _mock_provider(token: str) -> PortalIdentity:
    """开发/测试用。token自描述格式：``mock:<externalId>:<username>:<realName>[:orgName]``。

    示例（浏览器直接访问验证）：
        http://localhost:9345/?token=mock:1001:portal_user:门户联调用户:机关
    """
    parts = token.split(":")
    if len(parts) != 4 and len(parts) != 5:
        raise PortalTokenError(
            "mock token格式应为 mock:<externalId>:<username>:<realName>[:orgName]"
        )
    org_name = parts[4] if len(parts) == 5 and parts[4] else None
    return PortalIdentity(
        external_id=parts[1],
        username=parts[2],
        real_name=parts[3],
        org_name=org_name,
    )


# ---------------------------------------------------------------- http


def _dig(data: object, path: str) -> object:
    """按 ``a.b`` 点路径从嵌套dict取值（仅支持dict层级，不支持数组）。"""
    current = data
    for key in path.split("."):
        if not isinstance(current, dict):
            return None
        current = current.get(key)
    return current


def _mapped_str(user_obj: dict, local_field: str, *, required: bool, fallback: str = "") -> str | None:
    cfg = load_portal_config()
    remote = cfg.field_map.get(local_field, "")
    value = user_obj.get(remote) if remote else None
    if isinstance(value, str) and value.strip():
        return value.strip()
    if isinstance(value, int):
        return str(value)
    if required and not fallback:
        raise PortalIdentityError(f"对端用户信息缺少{local_field}（远端字段名：{remote}）")
    return fallback or None


def _http_provider(token: str) -> PortalIdentity:
    """声明式HTTP对接：按.env配置调用对端验签（及可选的用户信息接口）。"""
    cfg = load_portal_config()
    if not cfg.verify_url:
        raise PortalConfigError("DMS_PORTAL_PROVIDER=http时必须配置DMS_PORTAL_VERIFY_URL")

    headers: dict[str, str] = {}
    params: dict[str, str] = {}
    body: dict[str, str] = {}
    if cfg.verify_token_in == "header":
        headers[cfg.verify_header] = cfg.verify_header_template.format(token=token)
    elif cfg.verify_token_in == "query":
        params[cfg.token_param] = token
    else:
        body[cfg.token_param] = token

    try:
        resp = http_client.request(
            cfg.verify_method,
            cfg.verify_url,
            headers=headers,
            params=params,
            json=body if body else None,
            timeout=cfg.http_timeout_seconds,
        )
        data = resp.json()
    except (http_client.RequestException, ValueError) as exc:
        raise PortalTokenError(f"无法访问对端验签接口：{exc}") from exc

    if str(_dig(data, cfg.verify_success_field)) != cfg.verify_success_value:
        raise PortalTokenError("对端验签未通过（token非法或已过期）")

    user_obj = _dig(data, cfg.verify_user_field)
    if not isinstance(user_obj, dict):
        raise PortalIdentityError(
            f"对端验签响应中未找到用户信息（字段路径：{cfg.verify_user_field}）"
        )

    if cfg.userinfo_url:
        # 验签响应未含完整用户信息时，再调一次用户信息接口（GET + uid参数 + Bearer token）
        uid = _mapped_str(user_obj, "externalId", required=True) or ""
        try:
            info_resp = http_client.get(
                cfg.userinfo_url,
                params={cfg.userinfo_uid_param: uid},
                headers={"Authorization": f"Bearer {token}"},
                timeout=cfg.http_timeout_seconds,
            )
            user_obj = info_resp.json()
        except (http_client.RequestException, ValueError) as exc:
            raise PortalTokenError(f"无法访问对端用户信息接口：{exc}") from exc

    external_id = _mapped_str(user_obj, "externalId", required=True)
    username = _mapped_str(user_obj, "username", required=False, fallback=external_id or "")
    real_name = _mapped_str(user_obj, "realName", required=False, fallback=username or "")
    if not external_id or not username:
        raise PortalIdentityError("对端用户信息缺少必要的身份字段")
    return PortalIdentity(
        external_id=external_id,
        username=username,
        real_name=real_name or username,
        org_name=_mapped_str(user_obj, "orgName", required=False),
    )


# ---------------------------------------------------------------- custom


def _custom_provider(token: str) -> PortalIdentity:
    """手写适配模板：DMS_PORTAL_PROVIDER=custom 时本函数被调用。

    ═══════════════ 手写示例（直接替换本函数体） ═══════════════

    场景：对端是「GET http://portal.internal/api/sso/verify?ticket=xxx」，
    返回 ``{"ok": true, "data": {"uid": "1001", "name": "张三", "dept": "作战部"}}``。

    import requests as http_client

    try:
        resp = http_client.get(
            "http://portal.internal/api/sso/verify",
            params={"ticket": token},
            timeout=10,
        )
        data = resp.json()
    except Exception:
        raise PortalTokenError("无法访问对端验签接口")

    if not data.get("ok"):
        raise PortalTokenError("对端验签未通过")

    info = data.get("data") or {}
    uid = str(info.get("uid") or "").strip()
    name = str(info.get("name") or "").strip()
    dept = str(info.get("dept") or "").strip() or None
    if not uid or not name:
        raise PortalIdentityError("对端用户信息缺少uid或name")

    return PortalIdentity(external_id=uid, username=uid, real_name=name, org_name=dept)

    ═══════════════ 说明 ═══════════════
    - 只需返回 PortalIdentity，其余（落库/匹配/设密码/登录）由本系统自动完成；
    - username 缺乏明确登录名时可用 external_id 代替（如上例）；
    - 验签不通过抛 PortalTokenError，字段缺失抛 PortalIdentityError。
    """
    raise PortalConfigError(
        "DMS_PORTAL_PROVIDER=custom 需要在 dms/services/portal_provider.py 的 "
        "_custom_provider 中按模板实现对端验签逻辑"
    )