zTools2/app/config.py

"""运行时配置：所有参数与凭据均从项目根目录的 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
    # 分片上传会话的暂存目录（相对 upload_dir），完整文件拼接在此完成
    chunk_session_dir: str = "./.work"
    # 分片上传会话的存活秒数：超过该时长无活动的 pending 会话由后台 reaper 清理
    chunk_session_ttl_seconds: int = 300


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 TunnelUser(BaseModel):
    """一个反向隧道用户：凭据 + 固定的隧道端口与本地端口。"""

    username: str
    password_hash: str = ""        # bcrypt，与 SFTP 用户同款
    # 该 user 在 server 侧绑定的隧道端口（SSH remote forward 的 listen port）
    tunnel_port: int = 0
    # 该 user 要暴露的本地服务端口（仅用于记录，实际转发由 user 端完成）
    local_port: int = 0


class TunnelConfig(BaseModel):
    """反向隧道总开关与用户列表。"""

    enabled: bool = False
    users: list[TunnelUser] = Field(default_factory=list)

    def find_user(self, username: str) -> TunnelUser | None:
        return next((u for u in self.users if u.username == username), None)


class WhiteboardConfig(BaseModel):
    """共享白板配置。

    白板本身无鉴权（任何人凭 /whiteboard/{id} 即可访问并实时协作）；
    管理页（/whiteboard-admin、/api/admin/whiteboards）走 docs 同款 Basic Auth。
    心跳按 heartbeat_interval_seconds 发送，连续丢失 heartbeat_miss_threshold 次即判失活。
    """

    heartbeat_interval_seconds: int = 3
    heartbeat_miss_threshold: int = 5
    # board_id 合法字符集与长度上限，防路径/注入
    max_board_id_length: int = 64
    # 列表/管理页分页默认值
    list_limit: int = 100


class Settings(BaseModel):
    server: ServerConfig = ServerConfig()
    database: DatabaseConfig = DatabaseConfig()
    storage: StorageConfig = StorageConfig()
    sftp: SftpConfig = SftpConfig()
    docs: DocsConfig = DocsConfig()
    tunnel: TunnelConfig = TunnelConfig()
    whiteboard: WhiteboardConfig = WhiteboardConfig()

    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()