"""应用配置。 从 `backend/.env` 读取配置项,字段名与 `.env` 中的变量名一一对应 (pydantic-settings 默认大小写不敏感,`app_secret_key` ←→ `APP_SECRET_KEY`)。 设计取舍: - `APP_SECRET_KEY` 无默认值 —— 缺失时启动即失败。空密钥签发 JWT 属安全问题 而非可用性问题,必须 fail-fast,不能容忍"后端起得来但令牌可被伪造"。 - 数据库/Redis 连接不在此处建立 —— 依赖不可用时仍应能启动应用、查看接口文档。 """ from functools import lru_cache from pathlib import Path from urllib.parse import quote from pydantic import Field, field_validator from pydantic_settings import BaseSettings, SettingsConfigDict from sqlalchemy import URL # backend/ 目录:本文件位于 backend/app/core/config.py,向上三级 BACKEND_DIR = Path(__file__).resolve().parents[2] ENV_FILE = BACKEND_DIR / ".env" class Settings(BaseSettings): """应用配置项。""" model_config = SettingsConfigDict( env_file=str(ENV_FILE), # Windows 默认编码为 cp936,不显式指定会导致中文值解析出错 env_file_encoding="utf-8", case_sensitive=False, # .env 中可能含本服务用不到的变量(如 docker 编排专用项),忽略而非报错 extra="ignore", ) # ===== 应用 ===== app_env: str = "dev" upload_root: Path | None = Field(default=None, validation_alias="UPLOAD_DIR") @field_validator("upload_root", mode="before") @classmethod def _empty_upload_root(cls, value): return None if value in (None, "") else value app_secret_key: str = Field(description="JWT 签名密钥,至少 32 字符", repr=False) payroll_maker_user_id: int | None = Field( default=None, ge=1, description="指定工资制作人的现有账号ID;未设置则业务只读" ) payroll_allow_legacy_calculation: bool = False payroll_approval_enabled: bool = False # ===== 单进程运行监控;历史只清理监控新表 ===== monitor_enabled: bool = True monitor_cpu_threshold: float = Field(default=85, ge=1, le=100) monitor_memory_threshold: float = Field(default=85, ge=1, le=100) monitor_disk_threshold: float = Field(default=90, ge=1, le=100) monitor_resource_duration_seconds: int = Field(default=300, ge=15) monitor_slow_request_ms: int = Field(default=2000, ge=1) monitor_error_rate_threshold: float = Field(default=5, ge=0.1, le=100) monitor_error_min_requests: int = Field(default=20, ge=1) # ===== 鉴权 ===== # 规范第 13 节:Token 有效期 ≤ 2 小时 jwt_expire_minutes: int = Field(default=120, ge=1, le=120) # 规范第 6 节:验证码 300s captcha_expire_seconds: int = Field(default=300, ge=60, le=300) login_fail_limit: int = Field(default=5, ge=1) login_lock_minutes: int = Field(default=15, ge=1) # ===== MySQL ===== mysql_host: str = "127.0.0.1" mysql_port: int = 3306 mysql_database: str = "payroll" mysql_user: str = "payroll" mysql_password: str = Field(default="", repr=False) # ===== Redis ===== redis_host: str = "127.0.0.1" redis_port: int = 6379 redis_password: str = Field(default="", repr=False) redis_db: int = 0 # ===== T+ 只读 SQL Server ===== # T+ 数据库默认关闭,避免开发者未配置驱动时影响主业务启动。 t_plus_enabled: bool = False t_plus_host: str = "127.0.0.1" t_plus_port: int = 1433 t_plus_database: str = "" t_plus_user: str = "" t_plus_password: str = Field(default="", repr=False) t_plus_allow_dev_sa: bool = False t_plus_connection_file: str = "" # 部署前必须指定已在 SQL Server 2008 R2 实测的驱动,不自动选择最新版。 t_plus_odbc_driver: str = "" t_plus_encrypt: bool = True # 默认校验服务器证书;自签回退只允许隔离开发显式开启,避免中间人替换证书。 t_plus_trust_server_certificate: bool = False t_plus_connect_timeout: int = 10 t_plus_query_timeout: int = Field(default=120, ge=1, le=600) t_plus_query_dir: str = "" t_plus_automation_enabled: bool = False t_plus_schedule_weekday: int = Field(default=0, ge=0, le=6) t_plus_schedule_time: str = Field(default="02:00", pattern=r"^([01][0-9]|2[0-3]):[0-5][0-9]$") t_plus_stop_file: str = "" @field_validator("app_secret_key") @classmethod def _check_secret_key(cls, v: str) -> str: """校验密钥强度:过短的密钥可被暴力破解,直接拒绝启动。""" if len(v) < 32: raise ValueError("APP_SECRET_KEY 长度不足 32 字符,请在 backend/.env 中重新生成") return v @property def database_url(self) -> URL: """数据库连接 URL。运行期与 Alembic 迁移共用同一份。 用 SQLAlchemy 的 URL.create 而非手工拼接字符串 —— 密码含 @ : / 等 特殊字符时手工拼接会产生语法错误。 驱动为同步的 pymysql:本项目并发极低,异步无收益;而绩效核算与 pandas 报表都是同步阻塞代码,同步方案不会与之冲突。 """ return URL.create( drivername="mysql+pymysql", username=self.mysql_user, password=self.mysql_password, host=self.mysql_host, port=self.mysql_port, database=self.mysql_database, query={"charset": "utf8mb4"}, ) @property def redis_url(self) -> str: """Redis 连接 URL。密码为空时不拼接认证段。""" auth = f":{quote(self.redis_password, safe='')}@" if self.redis_password else "" return f"redis://{auth}{self.redis_host}:{self.redis_port}/{self.redis_db}" @property def t_plus_database_url(self) -> URL: """构造 T+ 只读 SQL Server 连接 URL。 连接 URL 只在启用 T+ 适配器时使用;密码由 SQLAlchemy URL 对象保存, 日志中调用 ``render_as_string()`` 时仍会默认隐藏密码。 ODBC 的分号和花括号由方言统一转义,调用方不得再手工拼接连接串。 """ query = { "driver": self.t_plus_odbc_driver, "Encrypt": "yes" if self.t_plus_encrypt else "no", "TrustServerCertificate": "yes" if self.t_plus_trust_server_certificate else "no", "Connection Timeout": str(self.t_plus_connect_timeout), } return URL.create( drivername="mssql+pyodbc", username=self.t_plus_user, password=self.t_plus_password, host=self.t_plus_host, port=self.t_plus_port, database=self.t_plus_database, query=query, ) @property def upload_dir(self) -> Path: """上传文件根目录(公司 logo 等):backend/uploads/。 不叫 static/ 是刻意的:静态目录将来可能整体交给前端构建产物, 运行期上传物必须与之隔离;该目录不入版本库(.gitignore)。 """ return self.upload_root.resolve() if self.upload_root else BACKEND_DIR / "uploads" @lru_cache def get_settings() -> Settings: """获取配置单例。 进程内只解析一次 .env;测试中如需覆盖,调用方应使用 `get_settings.cache_clear()` 配合环境变量注入。 """ return Settings() # type: ignore[call-arg]