import os import re from pathlib import Path from typing import Final from urllib.parse import urlsplit, urlunsplit from dotenv import dotenv_values, load_dotenv BACKEND_ROOT: Final[Path] = Path(__file__).resolve().parents[1] PROJECT_ROOT: Final[Path] = BACKEND_ROOT.parent # Load env files used by this project. # Priority (high -> low): # 1) process environment # 2) project/.env.prod # 3) backend/.env # # We keep process-provided env untouched, while allowing .env.prod to override backend/.env. _process_env_keys = set(os.environ.keys()) _backend_env_values = dotenv_values(BACKEND_ROOT / ".env") _prod_env_values = dotenv_values(PROJECT_ROOT / ".env.prod") load_dotenv(BACKEND_ROOT / ".env", override=False) for _k, _v in _prod_env_values.items(): if _v is None: continue if _k in _process_env_keys: continue os.environ[_k] = str(_v) def _env_bool(name: str, default: bool) -> bool: raw = os.getenv(name) if raw is None: return default return str(raw).strip().lower() in {"1", "true", "yes", "on"} def _env_int(name: str, default: int, min_value: int, max_value: int) -> int: raw = os.getenv(name) if raw is None: return default try: value = int(str(raw).strip()) except Exception: value = default return max(min_value, min(max_value, value)) def _normalize_extension(raw: str) -> str: text = str(raw or "").strip().lower() if not text: return "" if text.startswith("*."): text = text[1:] if not text.startswith("."): text = f".{text}" if not re.fullmatch(r"\.[a-z0-9][a-z0-9._+-]{0,31}", text): return "" return text def _env_extensions(name: str, default: tuple[str, ...]) -> tuple[str, ...]: raw = os.getenv(name) if raw is None: source = list(default) else: source = re.split(r"[,;\s]+", str(raw)) rows: list[str] = [] for item in source: ext = _normalize_extension(item) if ext and ext not in rows: rows.append(ext) if raw is None: return tuple(rows or list(default)) return tuple(rows) def _normalize_origin(raw: str) -> str: text = str(raw or "").strip() if not text: return "" try: parsed = urlsplit(text) except Exception: return "" scheme = str(parsed.scheme or "").strip().lower() netloc = str(parsed.netloc or "").strip().lower() if scheme not in {"http", "https"} or not netloc: return "" return urlunsplit((scheme, netloc, "", "", "")) def _env_origins(name: str, default: tuple[str, ...]) -> tuple[str, ...]: raw = os.getenv(name) source = list(default) if raw is None else re.split(r"[,;\s]+", str(raw)) rows: list[str] = [] for item in source: origin = _normalize_origin(item) if origin and origin not in rows: rows.append(origin) return tuple(rows) def _normalize_dir_path(path_value: str) -> str: raw = str(path_value or "").strip() if not raw: return raw raw = os.path.expandvars(os.path.expanduser(raw)) p = Path(raw) if p.is_absolute(): return str(p) return str((BACKEND_ROOT / p).resolve()) DATA_ROOT: Final[str] = _normalize_dir_path(os.getenv("DATA_ROOT", str(PROJECT_ROOT / "data"))) BOTS_WORKSPACE_ROOT: Final[str] = _normalize_dir_path( os.getenv("BOTS_WORKSPACE_ROOT", str(PROJECT_ROOT / "workspace" / "bots")) ) RUNTIME_DATA_ROOT: Final[Path] = Path(DATA_ROOT).resolve() RUNTIME_TEMPLATES_ROOT: Final[Path] = (RUNTIME_DATA_ROOT / "templates").resolve() RUNTIME_SKILLS_ROOT: Final[Path] = (RUNTIME_DATA_ROOT / "skills").resolve() RUNTIME_MODEL_ROOT: Final[Path] = (RUNTIME_DATA_ROOT / "model").resolve() def _normalize_database_url(url: str) -> str: raw = str(url or "").strip() prefix = "sqlite:///" if not raw.startswith(prefix): return raw path_part = raw[len(prefix) :] if not path_part or path_part.startswith("/"): return raw abs_path = (BACKEND_ROOT / path_part).resolve() return f"{prefix}{abs_path.as_posix()}" def _database_engine(url: str) -> str: raw = str(url or "").strip().lower() if raw.startswith("sqlite"): return "sqlite" if raw.startswith("postgresql"): return "postgresql" if raw.startswith("mysql"): return "mysql" if "+" in raw: return raw.split("+", 1)[0] if "://" in raw: return raw.split("://", 1)[0] return "unknown" def _mask_database_url(url: str) -> str: raw = str(url or "").strip() if not raw or raw.startswith("sqlite"): return raw try: parsed = urlsplit(raw) if parsed.password is None: return raw host = parsed.hostname or "" if parsed.port: host = f"{host}:{parsed.port}" auth = parsed.username or "" if auth: auth = f"{auth}:***@{host}" else: auth = host netloc = auth return urlunsplit((parsed.scheme, netloc, parsed.path, parsed.query, parsed.fragment)) except Exception: return raw _db_env = str(os.getenv("DATABASE_URL") or "").strip() if not _db_env: raise RuntimeError("DATABASE_URL is not set in environment. PostgreSQL is required.") DATABASE_URL: Final[str] = _normalize_database_url(_db_env) DATABASE_ENGINE: Final[str] = _database_engine(DATABASE_URL) DATABASE_URL_DISPLAY: Final[str] = _mask_database_url(DATABASE_URL) DATABASE_ECHO: Final[bool] = _env_bool("DATABASE_ECHO", True) DATABASE_POOL_SIZE: Final[int] = _env_int("DATABASE_POOL_SIZE", 20, 1, 200) DATABASE_MAX_OVERFLOW: Final[int] = _env_int("DATABASE_MAX_OVERFLOW", 40, 0, 200) DATABASE_POOL_TIMEOUT: Final[int] = _env_int("DATABASE_POOL_TIMEOUT", 30, 1, 300) DATABASE_POOL_RECYCLE: Final[int] = _env_int("DATABASE_POOL_RECYCLE", 1800, 30, 86400) DEFAULT_UPLOAD_MAX_MB: Final[int] = 100 DEFAULT_PAGE_SIZE: Final[int] = 10 DEFAULT_CHAT_PULL_PAGE_SIZE: Final[int] = 60 DEFAULT_COMMAND_AUTO_UNLOCK_SECONDS: Final[int] = 10 DEFAULT_AUTH_TOKEN_TTL_HOURS: Final[int] = _env_int("AUTH_TOKEN_TTL_HOURS", 24, 1, 720) DEFAULT_AUTH_TOKEN_MAX_ACTIVE: Final[int] = _env_int("AUTH_TOKEN_MAX_ACTIVE", 2, 1, 20) DEFAULT_BOT_SYSTEM_TIMEZONE: Final[str] = str( os.getenv("DEFAULT_BOT_SYSTEM_TIMEZONE") or os.getenv("TZ") or "Asia/Shanghai" ).strip() or "Asia/Shanghai" DEFAULT_WORKSPACE_DOWNLOAD_EXTENSIONS: Final[tuple[str, ...]] = ( ".pdf", ".doc", ".docx", ".xls", ".xlsx", ".xlsm", ".ppt", ".pptx", ".odt", ".ods", ".odp", ".wps", ) STT_ENABLED_DEFAULT: Final[bool] = True STT_MODEL: Final[str] = str(os.getenv("STT_MODEL") or "ggml-small-q8_0.bin").strip() _DEFAULT_STT_MODEL_DIR: Final[Path] = RUNTIME_MODEL_ROOT _configured_stt_model_dir = _normalize_dir_path(os.getenv("STT_MODEL_DIR", str(_DEFAULT_STT_MODEL_DIR))) if _configured_stt_model_dir and not Path(_configured_stt_model_dir).exists() and _DEFAULT_STT_MODEL_DIR.exists(): STT_MODEL_DIR: Final[str] = str(_DEFAULT_STT_MODEL_DIR) else: STT_MODEL_DIR: Final[str] = _configured_stt_model_dir STT_DEVICE: Final[str] = str(os.getenv("STT_DEVICE") or "cpu").strip().lower() or "cpu" DEFAULT_STT_MAX_AUDIO_SECONDS: Final[int] = 20 DEFAULT_STT_DEFAULT_LANGUAGE: Final[str] = "zh" DEFAULT_STT_FORCE_SIMPLIFIED: Final[bool] = True DEFAULT_STT_AUDIO_PREPROCESS: Final[bool] = True DEFAULT_STT_AUDIO_FILTER: Final[str] = "highpass=f=120,lowpass=f=7600,afftdn=nf=-20" DEFAULT_STT_INITIAL_PROMPT: Final[str] = ( "以下内容可能包含简体中文和英文术语。请优先输出简体中文,英文单词、缩写、品牌名和数字保持原文,不要翻译。" ) REDIS_ENABLED: Final[bool] = _env_bool("REDIS_ENABLED", False) REDIS_URL: Final[str] = str(os.getenv("REDIS_URL") or "").strip() REDIS_PREFIX: Final[str] = str(os.getenv("REDIS_PREFIX") or "dashboard_nanobot").strip() or "dashboard_nanobot" REDIS_DEFAULT_TTL: Final[int] = _env_int("REDIS_DEFAULT_TTL", 60, 1, 86400) PANEL_ACCESS_PASSWORD: Final[str] = str(os.getenv("PANEL_ACCESS_PASSWORD") or "").strip() CORS_ALLOWED_ORIGINS: Final[tuple[str, ...]] = _env_origins( "CORS_ALLOWED_ORIGINS", ( "http://localhost:5173", "http://127.0.0.1:5173", "http://localhost:4173", "http://127.0.0.1:4173", ), ) APP_HOST: Final[str] = str(os.getenv("APP_HOST") or "0.0.0.0").strip() APP_PORT: Final[int] = _env_int("APP_PORT", 8000, 1, 65535) APP_RELOAD: Final[bool] = _env_bool("APP_RELOAD", False) AGENT_MD_TEMPLATES_FILE: Final[Path] = RUNTIME_TEMPLATES_ROOT / "agent_md_templates.json" TOPIC_PRESETS_TEMPLATES_FILE: Final[Path] = RUNTIME_TEMPLATES_ROOT / "topic_presets.json"