220 lines
7.1 KiB
Python
220 lines
7.1 KiB
Python
"""
|
||
System Settings Database Service
|
||
"""
|
||
from sqlalchemy import select, update, delete
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
from typing import Optional, List, Dict, Any
|
||
import json
|
||
import logging
|
||
|
||
from app.models.db import SystemSettings
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
class SystemSettingsService:
|
||
"""Service for managing system settings"""
|
||
|
||
async def get_all_settings(
|
||
self,
|
||
session: AsyncSession,
|
||
category: Optional[str] = None,
|
||
is_public: Optional[bool] = None
|
||
) -> List[SystemSettings]:
|
||
"""Get all settings, optionally filtered by category or public status"""
|
||
query = select(SystemSettings)
|
||
|
||
if category:
|
||
query = query.where(SystemSettings.category == category)
|
||
if is_public is not None:
|
||
query = query.where(SystemSettings.is_public == is_public)
|
||
|
||
result = await session.execute(query)
|
||
return result.scalars().all()
|
||
|
||
async def get_setting(
|
||
self,
|
||
key: str,
|
||
session: AsyncSession
|
||
) -> Optional[SystemSettings]:
|
||
"""Get a setting by key"""
|
||
result = await session.execute(
|
||
select(SystemSettings).where(SystemSettings.key == key)
|
||
)
|
||
return result.scalar_one_or_none()
|
||
|
||
async def get_setting_value(
|
||
self,
|
||
key: str,
|
||
session: AsyncSession,
|
||
default: Any = None
|
||
) -> Any:
|
||
"""Get setting value with type conversion"""
|
||
setting = await self.get_setting(key, session)
|
||
if not setting:
|
||
return default
|
||
|
||
# Convert value based on type
|
||
try:
|
||
if setting.value_type == "int":
|
||
return int(setting.value)
|
||
elif setting.value_type == "float":
|
||
return float(setting.value)
|
||
elif setting.value_type == "bool":
|
||
return setting.value.lower() in ("true", "1", "yes")
|
||
elif setting.value_type == "json":
|
||
return json.loads(setting.value)
|
||
else: # string
|
||
return setting.value
|
||
except Exception as e:
|
||
logger.error(f"Error converting setting {key}: {e}")
|
||
return default
|
||
|
||
async def create_setting(
|
||
self,
|
||
data: Dict[str, Any],
|
||
session: AsyncSession
|
||
) -> SystemSettings:
|
||
"""Create a new setting"""
|
||
# Convert value to string for storage
|
||
value = data.get("value")
|
||
value_type = data.get("value_type", "string")
|
||
|
||
if value_type == "json" and not isinstance(value, str):
|
||
value = json.dumps(value)
|
||
else:
|
||
value = str(value)
|
||
|
||
new_setting = SystemSettings(
|
||
key=data["key"],
|
||
value=value,
|
||
value_type=value_type,
|
||
category=data.get("category", "general"),
|
||
label=data["label"],
|
||
description=data.get("description"),
|
||
is_public=data.get("is_public", False)
|
||
)
|
||
|
||
session.add(new_setting)
|
||
await session.flush()
|
||
await session.refresh(new_setting)
|
||
return new_setting
|
||
|
||
async def update_setting(
|
||
self,
|
||
key: str,
|
||
data: Dict[str, Any],
|
||
session: AsyncSession
|
||
) -> Optional[SystemSettings]:
|
||
"""Update a setting"""
|
||
setting = await self.get_setting(key, session)
|
||
if not setting:
|
||
return None
|
||
|
||
# Convert value to string if needed
|
||
if "value" in data:
|
||
value = data["value"]
|
||
value_type = data.get("value_type", setting.value_type)
|
||
|
||
if value_type == "json" and not isinstance(value, str):
|
||
data["value"] = json.dumps(value)
|
||
else:
|
||
data["value"] = str(value)
|
||
|
||
for key, value in data.items():
|
||
if hasattr(setting, key) and value is not None:
|
||
setattr(setting, key, value)
|
||
|
||
await session.flush()
|
||
await session.refresh(setting)
|
||
return setting
|
||
|
||
async def delete_setting(
|
||
self,
|
||
key: str,
|
||
session: AsyncSession
|
||
) -> bool:
|
||
"""Delete a setting"""
|
||
result = await session.execute(
|
||
delete(SystemSettings).where(SystemSettings.key == key)
|
||
)
|
||
return result.rowcount > 0
|
||
|
||
async def initialize_default_settings(self, session: AsyncSession):
|
||
"""Initialize default system settings if they don't exist"""
|
||
defaults = [
|
||
{
|
||
"key": "default_password",
|
||
"value": "cosmo",
|
||
"value_type": "string",
|
||
"category": "security",
|
||
"label": "默认重置密码",
|
||
"description": "管理员重置用户密码时使用的默认密码",
|
||
"is_public": False
|
||
},
|
||
{
|
||
"key": "timeline_interval_days",
|
||
"value": "30",
|
||
"value_type": "int",
|
||
"category": "visualization",
|
||
"label": "时间轴播放间隔（天）",
|
||
"description": "星图时间轴播放时每次跳转的天数间隔",
|
||
"is_public": True
|
||
},
|
||
{
|
||
"key": "current_cache_ttl_hours",
|
||
"value": "1",
|
||
"value_type": "int",
|
||
"category": "cache",
|
||
"label": "当前位置缓存时间（小时）",
|
||
"description": "当前位置数据在缓存中保存的时间",
|
||
"is_public": False
|
||
},
|
||
{
|
||
"key": "historical_cache_ttl_days",
|
||
"value": "7",
|
||
"value_type": "int",
|
||
"category": "cache",
|
||
"label": "历史位置缓存时间（天）",
|
||
"description": "历史位置数据在缓存中保存的时间",
|
||
"is_public": False
|
||
},
|
||
{
|
||
"key": "page_size",
|
||
"value": "10",
|
||
"value_type": "int",
|
||
"category": "ui",
|
||
"label": "每页显示数量",
|
||
"description": "管理页面默认每页显示的条数",
|
||
"is_public": True
|
||
},
|
||
{
|
||
"key": "nasa_api_timeout",
|
||
"value": "30",
|
||
"value_type": "int",
|
||
"category": "api",
|
||
"label": "NASA API超时时间（秒）",
|
||
"description": "查询NASA Horizons API的超时时间",
|
||
"is_public": False
|
||
},
|
||
{
|
||
"key": "orbit_points",
|
||
"value": "200",
|
||
"value_type": "int",
|
||
"category": "visualization",
|
||
"label": "轨道线点数",
|
||
"description": "生成轨道线时使用的点数，越多越平滑但性能越低",
|
||
"is_public": True
|
||
},
|
||
]
|
||
|
||
for default in defaults:
|
||
existing = await self.get_setting(default["key"], session)
|
||
if not existing:
|
||
await self.create_setting(default, session)
|
||
logger.info(f"Created default setting: {default['key']}")
|
||
|
||
|
||
# Singleton instance
|
||
system_settings_service = SystemSettingsService()
|