refactor: 主要功能实现

目前的工作已经实现的功能:
- 基本 FastAPI 路由;
- 基本 AI 聊天和创作功能;
- 用户信息管理、权限验证、JWT 令牌签发和验证、端点保护;
- HTML 验证码邮件发送和验证码验证。
This commit is contained in:
2026-05-24 13:58:51 +08:00
parent f06de85257
commit 21f0d7725e
98 changed files with 6483 additions and 116 deletions
+94
View File
@@ -0,0 +1,94 @@
import json
import logging
from pathlib import Path
from typing import Any, TypeVar
import aiofiles
from .config import Config
logger = logging.getLogger(__name__)
CONFIG_PATH = Path.cwd() / ".nyahome" / "config.json"
T = TypeVar("T")
class ConfigManager:
def __init__(self) -> None:
CONFIG_PATH.parent.mkdir(exist_ok=True)
self._config = Config()
def _parse(self, config: dict) -> None:
"""
解析给定的字典作为配置。
Args:
config: 配置字典
"""
for key, value in config.items():
setattr(self._config, key, value)
def _dumps(self) -> str:
"""
将配置项序列化为 json 字符串,包含格式化缩进。
Returns:
json 字符串。
"""
config = {}
for attr in dir(self._config):
if not attr.startswith("_"):
value = getattr(self._config, attr)
config[attr] = value
return json.dumps(config, ensure_ascii=False, indent=2)
async def async_load_config(self) -> None:
async with aiofiles.open(CONFIG_PATH, "r", encoding="utf-8") as f:
self._parse(json.loads(await f.read()))
logger.info("异步从 config.json 读取设置完成。")
def sync_load_config(self) -> None:
with open(CONFIG_PATH, "r", encoding="utf-8") as f:
self._parse(json.load(f))
logger.info("同步从 config.json 读取设置完成。")
async def async_save_config(self) -> None:
async with aiofiles.open(CONFIG_PATH, "w", encoding="utf-8") as f:
await f.write(self._dumps())
logger.info("异步保存设置到 config.json 完成。")
def sync_save_config(self) -> None:
with open(CONFIG_PATH, "w", encoding="utf-8") as f:
f.write(self._dumps())
logger.info("同步保存设置到 config.json 完成。")
def get(self, key: str, default: T | None = None) -> T:
"""
获取配置项
Args:
key: 配置键
default: 默认值,如果不提供则会在获取配置项失败时报错
Returns:
返回配置值,返回类型根据提供的默认值进行推断。
"""
return getattr(self._config, key, default) # type: ignore[return-value]
def get_config(self) -> dict[str, Any]:
config = {}
for attr in dir(self._config):
if not attr.startswith("_"):
value = getattr(self._config, attr)
config[attr] = value
return config
def set_config(self, config: dict[str, Any]) -> dict[str, Any]:
for attr in dir(self._config):
if not attr.startswith("_"):
setattr(self._config, attr, config[attr])
return self.get_config()
config_manager = ConfigManager()