vent_agent_3/service/auth.py

import requests
from typing import Optional

from config.settings import BASE_URL, VENT_SYS_USERNAME, VENT_SYS_PASSWORD

_LOGIN_PATH = "/jeecg-system/sys/login"
_REQUEST_TIMEOUT = 10


class AuthService:
    """Manages authentication token for the ventilation system API.

    Uses lazy initialization: the token is fetched on first access and
    cached in a class-level variable for reuse across all instances.
    """

    _token: Optional[str] = None

    @classmethod
    def get_token(cls) -> str:
        """Return the current auth token, logging in if necessary."""
        if cls._token is None:
            cls._authenticate()
        return cls._token

    @classmethod
    def _authenticate(cls) -> None:
        login_url = f"{BASE_URL}{_LOGIN_PATH}"
        credentials = {"username": VENT_SYS_USERNAME, "password": VENT_SYS_PASSWORD}
        response = requests.post(login_url, json=credentials, timeout=_REQUEST_TIMEOUT)
        response.raise_for_status()
        payload = response.json()
        if not payload.get("success"):
            raise RuntimeError(f"登录失败: {payload.get('message')}")
        cls._token = payload["result"]["token"]
        print("✅ 登录成功，token 已获取")

    @classmethod
    def refresh_token(cls) -> None:
        """Force re-authentication, discarding any cached token."""
        cls._token = None
        cls.get_token()