xiaobai-review/backend/features/accounts/http.py

from __future__ import annotations

import json
import secrets
from http import HTTPStatus

from backend.features.accounts.security import token_hash


class AccountHttpMixin:
    def _device_hash(self) -> str:
        raw = self.device_token()
        return token_hash(raw) if raw else ""

    def _ensure_device_token(self) -> str:
        return self.device_token() or secrets.token_urlsafe(32)

    def _auth_success_headers(self, session_token: str, device_raw: str) -> list[tuple[str, str]]:
        return [
            ("Set-Cookie", self.session_cookie(session_token)),
            ("Set-Cookie", self.device_cookie(device_raw)),
        ]

    def _send_authenticated_session(self, result: dict, status: HTTPStatus, device_raw: str) -> None:
        self.send_json(
            {
                "ok": True,
                "authenticated": True,
                "user": result["user"],
                "csrf_token": result["csrf_token"],
            },
            status,
            self._auth_success_headers(result["session_token"], device_raw),
        )

    def auth_register(self) -> None:
        try:
            body = self.read_json_body()
            device_raw = self._ensure_device_token()
            result = self.application_service.register_account(
                str(body.get("username") or ""),
                str(body.get("password") or ""),
                token_hash(device_raw),
            )
            self._send_authenticated_session(result, HTTPStatus.CREATED, device_raw)
        except (ValueError, json.JSONDecodeError) as exc:
            self.send_json({"error": str(exc)}, HTTPStatus.BAD_REQUEST)

    def auth_login(self) -> None:
        try:
            body = self.read_json_body()
            device_raw = self._ensure_device_token()
            result = self.application_service.login_account(
                str(body.get("username") or ""),
                str(body.get("password") or ""),
                token_hash(device_raw),
            )
            self._send_authenticated_session(result, HTTPStatus.OK, device_raw)
        except (ValueError, json.JSONDecodeError) as exc:
            self.send_json({"error": str(exc)}, HTTPStatus.UNAUTHORIZED)

    def auth_accounts(self) -> None:
        current_user_id = None
        if self.require_auth(send_error=False):
            current_user_id = int(self.auth_user["id"])
        payload = self.application_service.accounts.list_device_accounts(
            self._device_hash(),
            current_user_id,
        )
        self.send_json({"ok": True, **payload})

    def auth_switch(self) -> None:
        try:
            body = self.read_json_body()
            try:
                user_id = int(body.get("user_id"))
            except (TypeError, ValueError):
                user_id = 0
            device_raw = self.device_token()
            result = self.application_service.accounts.switch_account(
                token_hash(device_raw) if device_raw else "",
                user_id,
            )
            self._send_authenticated_session(
                result,
                HTTPStatus.OK,
                device_raw or self._ensure_device_token(),
            )
        except PermissionError as exc:
            self.send_json({"error": str(exc)}, HTTPStatus.UNAUTHORIZED)
        except (ValueError, json.JSONDecodeError) as exc:
            self.send_json({"error": str(exc)}, HTTPStatus.BAD_REQUEST)

    def auth_forget(self) -> None:
        try:
            body = self.read_json_body()
            try:
                user_id = int(body.get("user_id"))
            except (TypeError, ValueError):
                user_id = 0
            self.application_service.accounts.forget_account(self._device_hash(), user_id)
        except (ValueError, json.JSONDecodeError):
            pass
        self.send_json({"ok": True})

    def auth_me(self) -> None:
        service = self.application_service
        if not self.require_auth(send_error=False):
            self.send_json(
                {
                    "ok": True,
                    "authenticated": False,
                    "registration_required": service.database.count_users() == 0,
                }
            )
            return
        headers = None
        if not self.device_token():
            device_raw = secrets.token_urlsafe(32)
            service.accounts.remember_account(
                token_hash(device_raw),
                int(self.auth_user["id"]),
                fresh=True,
            )
            headers = [("Set-Cookie", self.device_cookie(device_raw))]
        self.send_json(
            {
                "ok": True,
                "authenticated": True,
                "user": {
                    "id": int(self.auth_user["id"]),
                    "username": str(self.auth_user["username"]),
                    "role": str(self.auth_user.get("role") or "user"),
                    "membership": service.membership(),
                },
                "csrf_token": str(self.auth_user["csrf_token"]),
            },
            headers=headers,
        )

    def auth_logout(self) -> None:
        raw_token = self.session_token()
        user_id = int(getattr(self, "auth_user", {}).get("id") or 0)
        if raw_token:
            self.application_service.database.delete_session(token_hash(raw_token))
        self.application_service.accounts.revoke_current_device_grant(
            self._device_hash(),
            user_id,
        )
        self.send_json(
            {"ok": True},
            headers={"Set-Cookie": self.session_cookie("", clear=True)},
        )

    def save_birth_profile(self) -> None:
        try:
            body = self.read_json_body()
            personal = self.application_service.save_birth_profile(body)
            self.send_json({"ok": True, "personal": personal})
        except (ValueError, json.JSONDecodeError) as exc:
            self.send_json({"error": str(exc)}, HTTPStatus.BAD_REQUEST)

    def change_password(self) -> None:
        try:
            body = self.read_json_body()
            current = str(body.get("current_password") or "")
            new = str(body.get("new_password") or "")
            confirmation = str(body.get("confirm_password") or "")
            if new != confirmation:
                raise ValueError("两次输入的新密码不一致。")
            self.application_service.change_password(current, new)
            self.send_json({"ok": True})
        except (ValueError, json.JSONDecodeError) as exc:
            self.send_json({"error": str(exc)}, HTTPStatus.BAD_REQUEST)

    def save_membership(self) -> None:
        try:
            service = self.application_service
            service.update_membership(self.read_json_body())
            self.send_json({"ok": True, "users": service.admin_users()})
        except (ValueError, json.JSONDecodeError) as exc:
            self.send_json({"error": str(exc)}, HTTPStatus.BAD_REQUEST)