41 lines
1.4 KiB
Python
41 lines
1.4 KiB
Python
"""services/settings.py —— 中枢级键值设置(hub_id / hub_name)。"""
|
|
from __future__ import annotations
|
|
|
|
import uuid
|
|
|
|
from ..db import Database
|
|
|
|
_HUB_ID_KEY = "hub_id"
|
|
_HUB_NAME_KEY = "hub_name"
|
|
DEFAULT_HUB_NAME = "蜂群中枢"
|
|
|
|
|
|
async def get_setting(db: Database, key: str, default: str | None = None) -> str | None:
|
|
row = await db.fetchone("SELECT value FROM settings WHERE key = ?", (key,))
|
|
return row["value"] if row else default
|
|
|
|
|
|
async def set_setting(db: Database, key: str, value: str) -> None:
|
|
row = await db.fetchone("SELECT 1 FROM settings WHERE key = ?", (key,))
|
|
if row is not None:
|
|
await db.execute("UPDATE settings SET value = ? WHERE key = ?", (value, key))
|
|
else:
|
|
await db.execute("INSERT INTO settings (key, value) VALUES (?, ?)", (key, value))
|
|
|
|
|
|
async def get_hub_settings(db: Database) -> dict:
|
|
hub_id = await get_setting(db, _HUB_ID_KEY)
|
|
if not hub_id:
|
|
hub_id = str(uuid.uuid4())
|
|
await set_setting(db, _HUB_ID_KEY, hub_id)
|
|
hub_name = (await get_setting(db, _HUB_NAME_KEY, DEFAULT_HUB_NAME)) or DEFAULT_HUB_NAME
|
|
return {"hub_id": hub_id, "hub_name": hub_name}
|
|
|
|
|
|
async def set_hub_settings(db: Database, hub_id: str | None, hub_name: str | None) -> dict:
|
|
if hub_id and hub_id.strip():
|
|
await set_setting(db, _HUB_ID_KEY, hub_id.strip())
|
|
if hub_name and hub_name.strip():
|
|
await set_setting(db, _HUB_NAME_KEY, hub_name.strip())
|
|
return await get_hub_settings(db)
|