"""分层配置:dev / prod 全部走环境变量,启动即校验。""" from __future__ import annotations import os from functools import lru_cache from typing import Annotated, Literal from pydantic import Field, field_validator from pydantic_settings import BaseSettings, NoDecode, SettingsConfigDict DEV_JWT_SECRET = "dev-only-change-me" # noqa: S105 - 仅开发默认值,生产启动时会被校验拦下 class Settings(BaseSettings): model_config = SettingsConfigDict( env_file=".env", env_file_encoding="utf-8", extra="ignore", case_sensitive=False ) # ── 基础 ────────────────────────────────────────────── app_name: str = "达道Task" env: Literal["dev", "test", "prod"] = "dev" api_prefix: str = "/api/v1" cors_origins: list[str] = Field( default_factory=lambda: ["http://localhost:5180", "http://127.0.0.1:5180"] ) # ── 数据库 ──────────────────────────────────────────── # 应用角色:受 RLS 约束,非 owner、非 superuser database_url: str = "postgresql+asyncpg://dadao_app:dadao_dev_pw@127.0.0.1:5433/dadaotask" # 迁移角色:owner,可绕过 RLS,仅 alembic / 运维脚本使用 migration_database_url: str = ( "postgresql+asyncpg://dadao_owner:dadao_dev_pw@127.0.0.1:5433/dadaotask" ) # 测试专用库:测试会 TRUNCATE 所有表,绝不能落到开发库上。 # 留空则退回 database_url(CI 里直接给 DATABASE_URL 即可)。 test_database_url: str | None = None test_migration_database_url: str | None = None db_echo: bool = False db_pool_size: int = 10 # ── Redis ───────────────────────────────────────────── redis_url: str = "redis://127.0.0.1:6380/0" # ── 业务 ────────────────────────────────────────────── # "今天/本周"这类视图按用户时区切日界,不能按 UTC —— 中国用户 23:30 建的任务 # 在 UTC 下已经是"明天"了。M7 多地域时改成按工作区配置。 default_timezone: str = "Asia/Shanghai" # ── 认证 ────────────────────────────────────────────── jwt_secret: str = DEV_JWT_SECRET jwt_algorithm: str = "HS256" access_token_minutes: int = 30 refresh_token_days: int = 14 sms_code_ttl_seconds: int = 300 # 同一手机号两次发码的最小间隔;测试里设为 0 可关闭节流 sms_throttle_seconds: int = 60 sms_dev_code: str = "123456" sms_provider: Literal["dev", "aliyun", "tencent"] = "dev" # 内测开关:生产默认禁止固定验证码。确实要在内网试用时显式打开, # 启动会打 WARNING(见 main.py),其余生产校验照旧——绝不能配合公网暴露。 internal_test_login: bool = False # 登录方式开关:手机验证码这条**默认关掉**。内测期的固定验证码等于"知道手机号就能进任意账号", # 关掉后 /auth/sms/send 与 /auth/login 统一回 2017,登录走「手机号 / 邮箱 + 密码」。 # 将来接上真短信通道把它打开即可,登录页会按 /auth/methods 自动列出可用的方式。 sms_login_enabled: bool = False # 平台管理员:能改系统设置(SMTP 等)。按手机号白名单,不挂工作区角色—— # 系统设置是**跨租户**的,某个工作区的 owner 也不该看到平台的发信配置。 # `NoDecode` + 下面的 before-validator:这个值环境变量里是逗号分隔的一串, # 不让 pydantic-settings 先按 JSON 解析(那样 `1360...,1390...` 会直接启动失败)。 platform_admin_phones: Annotated[list[str], NoDecode] = Field( default_factory=lambda: ["13605513683"] ) # ── 邮件(SMTP)──────────────────────────────────────── # 以「系统设置」页里存的值为准(DB),这里只是首次启动前的兜底。 # 改 SMTP 不该要人登服务器改 env 再重启——所以真配置在 DB 里(见 services/mail.py)。 smtp_host: str | None = None smtp_port: int = 465 smtp_username: str | None = None smtp_password: str | None = None smtp_from_name: str = "达道Task" # 465 走 SSL,587 走 STARTTLS;126/163 这类用 465 + SSL smtp_use_ssl: bool = True smtp_timeout_seconds: int = 10 # ── 邮箱注册 ────────────────────────────────────────── # 手机验证码关掉以后,新用户靠「邮箱 + 验证码」自己注册进来。 # 注册必须真的发出验证码(发信配置在「系统设置」里),所以这个开关跟 SMTP 是否配好绑定: # 没配 SMTP 时发码接口会回 2020,而不是假装发出去。 email_register_enabled: bool = True email_code_ttl_seconds: int = 600 # 同一邮箱两次发码的最小间隔;测试里设为 0 可关闭节流 email_code_throttle_seconds: int = 60 # 一个邮箱最多试错几次验证码,超了就作废(防止把 6 位数当彩票刮) email_code_max_attempts: int = 5 # 开发/联调用:把验证码直接回给调用方,**不发真信**。 # 生产打开等于任何人都能注册任意邮箱,所以 validate_runtime() 会直接拒绝启动。 email_code_dev_echo: bool = False # ── 幂等 ────────────────────────────────────────────── idempotency_enabled: bool = True idempotency_ttl_seconds: int = 86_400 # ── 附件 ────────────────────────────────────────────── # 本地磁盘目录(相对 backend/ 或绝对路径)。换对象存储时只改 services/storage.py。 upload_dir: str = "var/uploads" # 单文件上限。20MB 够放文档与截图;再大就该走对象存储 + 分片了。 max_upload_mb: int = 20 # 图片直链的签名有效期。短一点更安全,太长等于把凭据长期贴在页面上; # 30 分钟够一次阅读/编辑,过期后刷新页面会重新签发。 attachment_url_ttl_seconds: int = 1800 # ── 知识检索(M3)────────────────────────────────────── # 切片大小与重叠:中文一段 600 字大约 400~600 token,够表达一个完整片段; # 重叠 80 字是为了不把跨段的话切断(检索命中时前后文还读得懂)。 chunk_max_chars: int = 600 chunk_overlap_chars: int = 80 # 嵌入向量提供方: # hash —— 内置的字符 n-gram 哈希向量,离线可用、零依赖、可复现; # 质量不如真模型,但能把"检索链路"跑通,且换模型时接口不变。 # openai —— 任何 OpenAI 兼容的 /embeddings 端点(BYOK)。 embedding_provider: Literal["hash", "openai"] = "hash" embedding_base_url: str = "https://api.deepseek.com/v1" embedding_api_key: str | None = None embedding_model: str = "text-embedding-3-small" # 混合检索里向量与词法的权重。纯词法搜不到同义改写,纯向量搜不准专有名词, # 所以默认各占一半,可用环境变量调。 search_vector_weight: float = 0.6 search_lexical_weight: float = 0.4 # 低于这个分数就不返回:宁可"没找到",也不要塞一条不相干的让人误信。 search_min_score: float = 0.06 # ── 知识库向量库(独立实例,M3 起用)─────────────────── # 刻意与业务库分开放:fnsvr-01 上的 PG 没有 pgvector,也不该为了装扩展去动它; # 向量检索吃内存,和任务库挤在同一实例会互相拖累。 # 未配置时所有向量相关能力(M3 才启用)直接不可用,不会静默降级成假结果。 vector_database_url: str | None = None embedding_dim: int = 1024 embedding_model: str = "BAAI/bge-m3" # ── 前端产物 ────────────────────────────────────────── # 生产没有 nginx 时由后端直接托管前端构建产物(同源,/api 无跨域)。 # 指向 web/dist 的绝对路径;留空则只提供 API。 web_dist_dir: str | None = None # ── 安卓安装包(OTA)────────────────────────────────── # 放 latest.json + 对应 apk 的目录;留空则用 backend/static/apk。 # 发布目录是构建产物,不入库,部署脚本也不清理(git reset 不动未跟踪文件)。 apk_dir: str | None = None # ── 构建标记(部署校验用)────────────────────────────── build_commit: str = "unknown" build_time: str = "unknown" @field_validator("cors_origins", "platform_admin_phones", mode="before") @classmethod def _split_origins(cls, value: object) -> object: """支持 CORS_ORIGINS="a,b" 这种写法(平台管理员白名单同理)。""" if isinstance(value, str) and not value.startswith("["): return [item.strip() for item in value.split(",") if item.strip()] return value def validate_runtime(self) -> None: """生产环境的致命配置错误,必须在启动时炸掉,而不是等待请求。""" if self.env != "prod": return problems: list[str] = [] if self.jwt_secret == DEV_JWT_SECRET or len(self.jwt_secret) < 32: problems.append("JWT_SECRET 仍是开发默认值或长度不足 32") if "dadao_dev_pw" in self.database_url: problems.append("DATABASE_URL 仍在用开发默认口令") if self.sms_login_enabled and self.sms_provider == "dev": problems.append( "短信验证码登录开着(SMS_LOGIN_ENABLED=true),但 SMS_PROVIDER=dev 只会发固定验证码" "(等于知道手机号就能进);接真短信通道之前请把 SMS_LOGIN_ENABLED 关掉" ) if self.email_code_dev_echo: problems.append( "EMAIL_CODE_DEV_ECHO=true 会把注册验证码直接回给调用方," "等于任何人都能注册任意邮箱;生产必须关掉(只在本地联调时打开)" ) if problems: raise RuntimeError("配置校验失败:" + ";".join(problems)) def resolved_database_url(self) -> str: """应用实际连的库:测试环境优先用独立测试库(测试会 TRUNCATE)。""" if self.env == "test" and self.test_database_url: return self.test_database_url return self.database_url def resolved_migration_database_url(self) -> str: """迁移/清表实际连的库:必须和应用同一个库,否则清表清到别处去了。""" if self.env == "test" and self.test_migration_database_url: return self.test_migration_database_url return self.migration_database_url def resolved_vector_database_url(self) -> str | None: """向量库连接串;没配就是 None,调用方必须显式处理"没有向量能力"。""" return self.vector_database_url @lru_cache def get_settings() -> Settings: return Settings() def build_commit() -> str: """优先读环境变量(CI/镜像注入),退回 git。""" env_value = os.getenv("BUILD_COMMIT") if env_value: return env_value return get_settings().build_commit