from __future__ import annotations from datetime import datetime, timezone from typing import Any from pydantic import BaseModel, Field def utcnow_iso() -> str: return datetime.now(timezone.utc).isoformat() class BackendRecord(BaseModel): backend_id: str name: str base_url: str frontend_base_url: str | None = None status: str = "active" created_at: str = Field(default_factory=utcnow_iso) updated_at: str = Field(default_factory=utcnow_iso) class BackendCredential(BaseModel): backend_id: str client_id: str client_secret_hash: str created_at: str = Field(default_factory=utcnow_iso) rotated_at: str | None = None class UserRecord(BaseModel): username: str email: str | None = None default_backend_id: str | None = None created_at: str = Field(default_factory=utcnow_iso) updated_at: str = Field(default_factory=utcnow_iso) class PermissionsModel(BaseModel): permissions: dict[str, Any] = Field(default_factory=dict) class ChannelSettings(BaseModel): configured: bool = True config: dict[str, Any] = Field(default_factory=dict) secrets: dict[str, Any] = Field(default_factory=dict) updated_at: str = Field(default_factory=utcnow_iso) def masked_dict(self) -> dict[str, Any]: data = self.model_dump(mode="json") secrets = data.pop("secrets", {}) data["secrets_masked"] = bool(secrets) data["secret_keys"] = sorted( str(key).strip() for key in (secrets.keys() if isinstance(secrets, dict) else []) if str(key).strip() ) return data class OutlookSettings(BaseModel): configured: bool = True email: str username: str domain: str | None = None service_endpoint: str | None = None server: str | None = None autodiscover: bool = False default_timezone: str = "Asia/Shanghai" password: str updated_at: str = Field(default_factory=utcnow_iso) def masked_dict(self) -> dict[str, Any]: data = self.model_dump() data.pop("password", None) data["password_masked"] = True return data class BackendRoutingPayload(BaseModel): name: str | None = None backend_id: str | None = None base_url: str | None = None frontend_base_url: str | None = None class RegisterBackendRequest(BaseModel): name: str base_url: str backend_id: str | None = None frontend_base_url: str | None = None class UpdateBackendRequest(BaseModel): name: str | None = None base_url: str | None = None frontend_base_url: str | None = None class RegisterBackendResponse(BaseModel): backend_id: str client_id: str client_secret: str created_at: str frontend_base_url: str | None = None class RegisterUserRequest(BaseModel): username: str password: str email: str | None = None name: str | None = None backend_name: str | None = None backend_id: str | None = None base_url: str | None = None public_base_url: str | None = None frontend_base_url: str | None = None backend: BackendRoutingPayload | None = None class RegisterUserBackendResult(BaseModel): backend_id: str client_id: str client_secret: str | None = None name: str base_url: str frontend_base_url: str | None = None status: str = "active" created_at: str class RegisterUserResponse(BaseModel): user: UserRecord backend: RegisterUserBackendResult class RotateSecretResponse(BaseModel): backend_id: str client_id: str client_secret: str rotated_at: str class OAuthTokenRequest(BaseModel): grant_type: str = "client_credentials" client_id: str client_secret: str aud: str scopes: list[str] = Field(default_factory=list) class OAuthTokenResponse(BaseModel): access_token: str token_type: str = "bearer" expires_in: int class IntrospectRequest(BaseModel): token: str class IntrospectResponse(BaseModel): active: bool client_id: str | None = None backend_id: str | None = None aud: str | None = None scp: list[str] = Field(default_factory=list) exp: int | None = None