resume-agent/backend/app/offerpai_auth.py

"""OfferPai account authentication and profile lookup.

The landing token is deliberately used only at the HTTP boundary.  It is never
returned to callers or persisted in the resume-agent database.
"""

from __future__ import annotations

from dataclasses import dataclass
import re
from typing import Any, Protocol

import httpx

from .validators import strict_phone


_TOKEN_PATTERN = re.compile(r"^[A-Za-z0-9._~-]{16,4096}$")


@dataclass(frozen=True, slots=True)
class OfferPaiIdentity:
    user_id: str
    mobile_number: str
    nick: str | None = None
    invite_code: str | None = None
    create_time: int | None = None

    def profile_value(self) -> dict[str, Any]:
        return {
            "provider": "offerpai",
            "user_id": self.user_id,
            "mobile_number": self.mobile_number,
            "nick": self.nick,
            "invite_code": self.invite_code,
            "create_time": self.create_time,
        }


class OfferPaiIdentityProvider(Protocol):
    def authenticate(self, token: str) -> OfferPaiIdentity: ...


class OfferPaiAuthError(RuntimeError):
    def __init__(self, code: str, public_message: str, *, status_code: int) -> None:
        super().__init__(public_message)
        self.code = code
        self.public_message = public_message
        self.status_code = status_code


class OfferPaiAuthClient:
    """Validate an OfferPai token and load the associated account profile."""

    def __init__(
        self,
        base_url: str,
        *,
        timeout_seconds: float = 8.0,
        client: Any | None = None,
    ) -> None:
        self.base_url = base_url.rstrip("/")
        self.timeout_seconds = timeout_seconds
        self._client = client

    def authenticate(self, token: str) -> OfferPaiIdentity:
        normalized = token.strip()
        if not _TOKEN_PATTERN.fullmatch(normalized):
            raise OfferPaiAuthError(
                "external_auth_invalid",
                "登录凭证无效或已过期，请重新从 OfferPai 进入。",
                status_code=401,
            )

        owned_client = self._client is None
        client = self._client or httpx.Client(
            base_url=self.base_url,
            timeout=self.timeout_seconds,
            follow_redirects=False,
            headers={"Accept": "application/json"},
        )
        try:
            login = self._get_json(client, "/api/public/checkLogin", normalized)
            if str(login.get("code")) != "0" or login.get("data") is not True:
                raise OfferPaiAuthError(
                    "external_auth_invalid",
                    "登录凭证无效或已过期，请重新从 OfferPai 进入。",
                    status_code=401,
                )

            profile = self._get_json(client, "/api/user/manage/info", normalized)
            data = profile.get("data")
            if str(profile.get("code")) != "0" or not isinstance(data, dict):
                raise OfferPaiAuthError(
                    "external_profile_unavailable",
                    "暂时无法读取 OfferPai 账号信息，请稍后重试。",
                    status_code=502,
                )
        finally:
            if owned_client:
                client.close()

        user_id = str(data.get("id") or "").strip()
        mobile_number = str(data.get("mobileNumber") or "").strip()
        if not user_id:
            raise OfferPaiAuthError(
                "external_profile_invalid",
                "OfferPai 账号缺少用户标识，请联系管理员。",
                status_code=502,
            )
        if not strict_phone(mobile_number):
            raise OfferPaiAuthError(
                "external_mobile_unavailable",
                "OfferPai 账号未配置有效手机号，请先完善账号手机号。",
                status_code=422,
            )

        create_time = data.get("createTime")
        return OfferPaiIdentity(
            user_id=user_id,
            mobile_number=mobile_number,
            nick=str(data.get("nick") or "").strip() or None,
            invite_code=str(data.get("inviteCode") or "").strip() or None,
            create_time=create_time if isinstance(create_time, int) else None,
        )

    @staticmethod
    def _get_json(client: Any, path: str, token: str) -> dict[str, Any]:
        try:
            response = client.get(path, headers={"Cookie": f"Token={token}"})
            response.raise_for_status()
            payload = response.json()
        except httpx.TimeoutException as exc:
            raise OfferPaiAuthError(
                "external_auth_timeout",
                "OfferPai 账号服务响应超时，请稍后重试。",
                status_code=504,
            ) from exc
        except httpx.HTTPStatusError as exc:
            if exc.response.status_code in {401, 403}:
                raise OfferPaiAuthError(
                    "external_auth_invalid",
                    "登录凭证无效或已过期，请重新从 OfferPai 进入。",
                    status_code=401,
                ) from exc
            raise OfferPaiAuthError(
                "external_auth_unavailable",
                "OfferPai 账号服务暂时不可用，请稍后重试。",
                status_code=502,
            ) from exc
        except (httpx.HTTPError, ValueError, TypeError) as exc:
            raise OfferPaiAuthError(
                "external_auth_unavailable",
                "OfferPai 账号服务暂时不可用，请稍后重试。",
                status_code=502,
            ) from exc
        if not isinstance(payload, dict):
            raise OfferPaiAuthError(
                "external_auth_unavailable",
                "OfferPai 账号服务返回了无效数据，请稍后重试。",
                status_code=502,
            )
        return payload