149 lines
5.3 KiB
Python
149 lines
5.3 KiB
Python
"""Account Settings persistence service."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
from sqlalchemy import and_, select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from ..models import Integration
|
|
from ..schemas.account_settings import (
|
|
AccountAiPreferenceSettings,
|
|
AccountNotificationPreference,
|
|
AccountNotificationSettings,
|
|
AccountProfileSettings,
|
|
AccountSettingsResponse,
|
|
AccountSettingsUpdateRequest,
|
|
)
|
|
from ..utils.timezone import now_cn_naive
|
|
|
|
INTEGRATION_TYPE = "account_settings_preferences"
|
|
|
|
|
|
class AccountSettingsService:
|
|
def __init__(self, db: AsyncSession, *, tenant_id: int) -> None:
|
|
self.db = db
|
|
self.tenant_id = int(tenant_id)
|
|
|
|
async def get_settings(self, *, user: Any) -> AccountSettingsResponse:
|
|
bundle = await self._get_bundle()
|
|
user_key = self._user_key(user)
|
|
stored = self._settings_by_user(bundle).get(user_key)
|
|
if isinstance(stored, dict):
|
|
return AccountSettingsResponse.model_validate(stored)
|
|
return self._default_settings(user=user)
|
|
|
|
async def update_settings(self, *, user: Any, body: AccountSettingsUpdateRequest) -> AccountSettingsResponse:
|
|
row, bundle = await self._get_or_init_bundle()
|
|
user_key = self._user_key(user)
|
|
settings = AccountSettingsResponse(
|
|
profile=body.profile,
|
|
aiPreferences=body.aiPreferences,
|
|
notifications=body.notifications,
|
|
saveScope="account",
|
|
updatedAt=now_cn_naive().isoformat(),
|
|
)
|
|
users = self._settings_by_user(bundle)
|
|
users[user_key] = settings.model_dump(mode="json")
|
|
bundle["users"] = users
|
|
bundle["version"] = 1
|
|
row.config = bundle
|
|
await self.db.flush()
|
|
return settings
|
|
|
|
async def _get_bundle(self) -> dict[str, Any]:
|
|
row = (
|
|
await self.db.execute(
|
|
select(Integration).where(
|
|
and_(
|
|
Integration.tenant_id == self.tenant_id,
|
|
Integration.type == INTEGRATION_TYPE,
|
|
)
|
|
)
|
|
)
|
|
).scalar_one_or_none()
|
|
if row is None:
|
|
return {"version": 1, "users": {}}
|
|
return self._normalize_bundle(row.config)
|
|
|
|
async def _get_or_init_bundle(self) -> tuple[Integration, dict[str, Any]]:
|
|
row = (
|
|
await self.db.execute(
|
|
select(Integration).where(
|
|
and_(
|
|
Integration.tenant_id == self.tenant_id,
|
|
Integration.type == INTEGRATION_TYPE,
|
|
)
|
|
)
|
|
)
|
|
).scalar_one_or_none()
|
|
if row is None:
|
|
row = Integration(
|
|
tenant_id=self.tenant_id,
|
|
type=INTEGRATION_TYPE,
|
|
is_active=True,
|
|
config={"version": 1, "users": {}},
|
|
)
|
|
self.db.add(row)
|
|
await self.db.flush()
|
|
bundle = self._normalize_bundle(row.config)
|
|
if bundle != row.config:
|
|
row.config = bundle
|
|
await self.db.flush()
|
|
return row, bundle
|
|
|
|
@staticmethod
|
|
def _normalize_bundle(config: Any) -> dict[str, Any]:
|
|
bundle = dict(config or {}) if isinstance(config, dict) else {}
|
|
if not isinstance(bundle.get("users"), dict):
|
|
bundle["users"] = {}
|
|
if not isinstance(bundle.get("version"), int):
|
|
bundle["version"] = 1
|
|
return bundle
|
|
|
|
@staticmethod
|
|
def _settings_by_user(bundle: dict[str, Any]) -> dict[str, Any]:
|
|
users = bundle.get("users")
|
|
return dict(users) if isinstance(users, dict) else {}
|
|
|
|
@staticmethod
|
|
def _user_key(user: Any) -> str:
|
|
return str(getattr(user, "id", "") or "unknown")
|
|
|
|
@staticmethod
|
|
def _default_settings(*, user: Any) -> AccountSettingsResponse:
|
|
role = str(getattr(user, "role", "") or "").strip()
|
|
role_title = {
|
|
"admin": "管理员",
|
|
"manager": "项目负责人",
|
|
"ADMIN": "管理员",
|
|
"MANAGER": "项目负责人",
|
|
}.get(role, role or "产品负责人")
|
|
display_name = (
|
|
str(getattr(user, "name", "") or "").strip()
|
|
or str(getattr(user, "username", "") or "").strip()
|
|
or "PMFlow 用户"
|
|
)
|
|
email = str(getattr(user, "email", "") or "").strip()
|
|
return AccountSettingsResponse(
|
|
profile=AccountProfileSettings(
|
|
displayName=display_name,
|
|
roleTitle=role_title,
|
|
email=email,
|
|
defaultLanguage="zh-CN",
|
|
),
|
|
aiPreferences=AccountAiPreferenceSettings(answerDepth="balanced", evidenceDisplay="inline"),
|
|
notifications=AccountNotificationSettings(
|
|
pendingConfirmation=AccountNotificationPreference(enabled=True, desktop=True, emailDigest=False),
|
|
executionWriteback=AccountNotificationPreference(enabled=True, desktop=True, emailDigest=False),
|
|
releaseGate=AccountNotificationPreference(enabled=True, desktop=True, emailDigest=True),
|
|
failureAlert=AccountNotificationPreference(enabled=True, desktop=True, emailDigest=True),
|
|
),
|
|
saveScope="account",
|
|
updatedAt=now_cn_naive().isoformat(),
|
|
)
|
|
|
|
|
|
__all__ = ["AccountSettingsService", "INTEGRATION_TYPE"]
|