80b96d236f
把原本散落在 /root/zikai 的 FastAPI 服务整理到 server/ 子目录， 开启独立 git 与 venv。运行时数据（config.yaml / keys / uploads） 按 .gitignore 留在工作目录但不入仓。
104 lines
2.8 KiB
Python
104 lines
2.8 KiB
Python
"""运行时配置：所有参数与凭据均从项目根目录的 config.yaml 读取。"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import os
|
||
from functools import lru_cache
|
||
from pathlib import Path
|
||
|
||
import yaml
|
||
from pydantic import BaseModel, Field, field_validator
|
||
|
||
PROJECT_ROOT = Path(__file__).resolve().parent.parent
|
||
DEFAULT_CONFIG_PATH = PROJECT_ROOT / "config.yaml"
|
||
|
||
|
||
class ServerConfig(BaseModel):
|
||
host: str = "127.0.0.1"
|
||
port: int = 6867
|
||
workers: int = 1
|
||
|
||
|
||
class DatabaseConfig(BaseModel):
|
||
host: str = "127.0.0.1"
|
||
port: int = 3306
|
||
user: str = "zikai_filesvc"
|
||
password: str = ""
|
||
database: str = "zikai_filesvc"
|
||
pool_size: int = 5
|
||
pool_recycle: int = 1800
|
||
|
||
|
||
class StorageConfig(BaseModel):
|
||
upload_dir: str = "./uploads"
|
||
chunk_bytes: int = 1024 * 1024
|
||
sha256_on_upload: bool = True
|
||
|
||
|
||
class SftpUser(BaseModel):
|
||
username: str
|
||
password_hash: str = "" # bcrypt
|
||
public_key: str | None = None # 可选的内联 OpenSSH 公钥
|
||
|
||
|
||
class SftpConfig(BaseModel):
|
||
enabled: bool = True
|
||
host: str = "0.0.0.0"
|
||
port: int = 2022
|
||
host_key_path: str = "./keys/ssh_host_ed25519_key"
|
||
authorized_keys_path: str = "./keys/authorized_keys"
|
||
users: list[SftpUser] = Field(default_factory=list)
|
||
|
||
|
||
class DocsConfig(BaseModel):
|
||
"""/docs、/redoc、/openapi.json 的 Basic Auth 凭据（明文）。"""
|
||
|
||
enabled: bool = True
|
||
username: str = "admin"
|
||
password: str = "" # 留空 → /docs 返回 503，避免误暴露
|
||
realm: str = "zikai docs"
|
||
|
||
@field_validator("password", mode="before")
|
||
@classmethod
|
||
def _coerce(cls, v):
|
||
# 兼容 YAML 把纯数字密码解析成 int 的情况。
|
||
return "" if v is None else str(v)
|
||
|
||
|
||
class Settings(BaseModel):
|
||
server: ServerConfig = ServerConfig()
|
||
database: DatabaseConfig = DatabaseConfig()
|
||
storage: StorageConfig = StorageConfig()
|
||
sftp: SftpConfig = SftpConfig()
|
||
docs: DocsConfig = DocsConfig()
|
||
|
||
def db_url(self) -> str:
|
||
c = self.database
|
||
return (
|
||
f"mysql+pymysql://{c.user}:{c.password}@{c.host}:{c.port}/{c.database}"
|
||
"?charset=utf8mb4"
|
||
)
|
||
|
||
def resolved_upload_dir(self) -> Path:
|
||
return (PROJECT_ROOT / self.storage.upload_dir).resolve()
|
||
|
||
|
||
def _load_yaml(path: Path) -> dict:
|
||
if not path.exists():
|
||
raise FileNotFoundError(
|
||
f"未找到配置文件 {path}；请先 cp config.example.yaml config.yaml 并填值。"
|
||
)
|
||
return yaml.safe_load(path.read_text(encoding="utf-8")) or {}
|
||
|
||
|
||
@lru_cache(maxsize=1)
|
||
def get_settings() -> Settings:
|
||
path = Path(os.getenv("CONFIG_PATH", str(DEFAULT_CONFIG_PATH)))
|
||
return Settings.model_validate(_load_yaml(path))
|
||
|
||
|
||
def reload_settings() -> Settings:
|
||
"""清缓存并重新读取，供脚本与测试使用。"""
|
||
get_settings.cache_clear()
|
||
return get_settings()
|