document-management/backend/dms/config.py

"""DMS环境配置。"""

from __future__ import annotations

import json
import os
from pathlib import Path
from typing import Any
from zoneinfo import ZoneInfo, ZoneInfoNotFoundError

import shutil

from dms.storage.envelope import load_kek


def _json_serializer(value: Any) -> str:
    """SQLAlchemy JSON 列序列化：原文中文（ensure_ascii=False），对齐达梦存量与 Java 版。"""
    return json.dumps(value, ensure_ascii=False)



BACKEND_ROOT = Path(__file__).resolve().parents[1]
DEFAULT_STORAGE_ROOT = BACKEND_ROOT / "dms-storage"
DEFAULT_UI_DICTIONARY_CONFIG_PATH = (
    BACKEND_ROOT / "dms" / "resources" / "ui-dictionaries.zh-CN.json"
)
DEFAULT_DOCUMENT_RELATION_PRESET_CONFIG_PATH = (
    BACKEND_ROOT / "dms" / "resources" / "document-relation-presets.zh-CN.json"
)
# 当前数据库：达梦 DM（dm+dmPython 方言，默认端口 5236，schema 由用户 DMS 决定）。
# 原 MySQL 默认链接（保留备查，回切时恢复）：
# "mysql+pymysql://dms_app:change-me@127.0.0.1:3306/dms?charset=utf8mb4"
DEFAULT_DATABASE_URL = "dm+dmPython://DMS:change-me@127.0.0.1:5236"


def _positive_int(name: str, default: int) -> int:
    raw_value = os.environ.get(name)
    if raw_value is None or not raw_value.strip():
        return default
    try:
        value = int(raw_value)
    except ValueError as exc:
        raise ValueError(f"{name}必须是正整数") from exc
    if value <= 0:
        raise ValueError(f"{name}必须是正整数")
    return value


def _non_negative_int(name: str, default: int) -> int:
    """非负整数（0 有语义：表示关闭该能力，如空闲登出）。"""
    raw_value = os.environ.get(name)
    if raw_value is None or not raw_value.strip():
        return default
    try:
        value = int(raw_value)
    except ValueError as exc:
        raise ValueError(f"{name}必须是非负整数") from exc
    if value < 0:
        raise ValueError(f"{name}必须是非负整数")
    return value


def resolve_backup_config_path() -> Path:
    """备份配置文件路径：环境变量覆盖，缺省为工作目录（与 Java 版口径一致）。"""
    value = os.environ.get("DMS_BACKUP_CONFIG_PATH", "").strip()
    if value:
        return Path(value).expanduser().resolve()
    return Path("dms-backup-config.json").resolve()


def resolve_env_path() -> Path:
    """.env 路径（备份 best-effort 副本用）：环境变量覆盖，缺省为工作目录。"""
    value = os.environ.get("DMS_ENV_FILE", "").strip()
    if value:
        return Path(value).expanduser().resolve()
    return Path(".env").resolve()


def load_dms_config() -> dict[str, Any]:
    """从环境变量加载DMS配置，不读取或覆盖现有AI配置。"""
    storage_value = os.environ.get("DMS_STORAGE_ROOT", "").strip()
    storage_root = Path(storage_value or DEFAULT_STORAGE_ROOT).expanduser()
    max_upload_size = _positive_int(
        "DMS_MAX_UPLOAD_SIZE", 100 * 1024 * 1024
    )
    max_file_size = _positive_int(
        "DMS_MAX_FILE_SIZE", min(50 * 1024 * 1024, max_upload_size)
    )
    if max_file_size > max_upload_size:
        raise ValueError("DMS_MAX_FILE_SIZE不得超过DMS_MAX_UPLOAD_SIZE")
    database_url = os.environ.get("DMS_DATABASE_URL", "").strip()
    ui_dictionary_config_value = os.environ.get(
        "DMS_UI_DICTIONARY_CONFIG_PATH", ""
    ).strip()
    ui_dictionary_config_path = Path(
        ui_dictionary_config_value or DEFAULT_UI_DICTIONARY_CONFIG_PATH
    ).expanduser()

    relation_preset_config_value = os.environ.get(
        "DMS_DOCUMENT_RELATION_PRESET_CONFIG_PATH", ""
    ).strip()
    relation_preset_config_path = Path(
        relation_preset_config_value
        or DEFAULT_DOCUMENT_RELATION_PRESET_CONFIG_PATH
    ).expanduser()

    business_timezone = (
        os.environ.get("DMS_BUSINESS_TIMEZONE", "").strip()
        or "Asia/Shanghai"
    )
    try:
        ZoneInfo(business_timezone)
    except (ZoneInfoNotFoundError, ValueError) as exc:
        raise ValueError("DMS_BUSINESS_TIMEZONE必须是有效的IANA时区") from exc

    office_preview_enabled = (
        os.environ.get("DMS_OFFICE_PREVIEW_ENABLED", "true").strip().lower()
        in {"1", "true", "yes", "on"}
    )
    libreoffice_executable = (
        os.environ.get("DMS_LIBREOFFICE_EXECUTABLE", "").strip()
        or shutil.which("soffice")
        or shutil.which("libreoffice")
        or ""
    )
    libreoffice_timeout_seconds = _positive_int(
        "DMS_LIBREOFFICE_TIMEOUT_SECONDS", 60
    )
    libreoffice_max_concurrency = _positive_int(
        "DMS_LIBREOFFICE_MAX_CONCURRENCY", 2
    )

    resolved_database_url = database_url or DEFAULT_DATABASE_URL
    # init_command（SET time_zone）为 pymysql 专属参数，达梦 dmPython 不接受；
    # 仅在 MySQL 链接时下发。
    connect_args = (
        {"init_command": "SET time_zone = '+00:00'"}
        if resolved_database_url.startswith(("mysql", "jdbc:mysql"))
        else {}
    )

    engine_options: dict[str, Any] = {
        "isolation_level": "READ COMMITTED",
        "pool_pre_ping": True,
        "pool_recycle": 1800,
        "connect_args": connect_args,
    }
    # json_serializer 同为 MySQL/通用方言参数，达梦方言（DMDialect_dmPython）
    # 不接受该 kwarg（create_engine 直接报错）；仅非达梦链接下发。
    if not resolved_database_url.startswith(("dm", "jdbc:dm")):
        # JSON 列统一原文中文序列化（ensure_ascii=False），与存量数据及
        # Java 版 Jackson 写入格式一致；否则 \uXXXX 转义会导致 LIKE 检索漏配。
        engine_options["json_serializer"] = _json_serializer

    return {
        "SQLALCHEMY_DATABASE_URI": resolved_database_url,
        "SQLALCHEMY_TRACK_MODIFICATIONS": False,
        "SQLALCHEMY_ENGINE_OPTIONS": engine_options,
        "DMS_STORAGE_ROOT": str(storage_root.resolve()),
        "DMS_MAX_UPLOAD_SIZE": max_upload_size,
        "DMS_MAX_FILE_SIZE": max_file_size,
        "MAX_CONTENT_LENGTH": max_upload_size,
        "DMS_BATCH_MAX_FILES": _positive_int("DMS_BATCH_MAX_FILES", 50),
        "DMS_BUSINESS_TIMEZONE": business_timezone,
        "DMS_JWT_SECRET": os.environ.get("DMS_JWT_SECRET", ""),
        "DMS_JWT_ISSUER": "dms",
        "DMS_JWT_EXPIRES_SECONDS": _positive_int(
            "DMS_JWT_EXPIRES_SECONDS", 3600
        ),
        "DMS_JWT_KEEP_SIGNED_IN_EXPIRES_SECONDS": 604800,
        # 页面空闲自动登出（契约1.29）：分钟数，0 表示关闭
        "DMS_IDLE_LOGOUT_MINUTES": _non_negative_int(
            "DMS_IDLE_LOGOUT_MINUTES", 5
        ),
        # 契约1.21：应用层信封加密 KEK（base64 编码 32 字节）；未配置则明文落盘
        "DMS_STORAGE_KEK_BYTES": load_kek(
            os.environ.get("DMS_STORAGE_KEK", "")
        ),
        "DMS_UI_DICTIONARY_CONFIG_PATH": str(
            ui_dictionary_config_path.resolve()
        ),
        "DMS_DOCUMENT_RELATION_PRESET_CONFIG_PATH": str(
            relation_preset_config_path.resolve()
        ),
        "DMS_OFFICE_PREVIEW_ENABLED": office_preview_enabled,
        "DMS_LIBREOFFICE_EXECUTABLE": libreoffice_executable,
        "DMS_LIBREOFFICE_TIMEOUT_SECONDS": libreoffice_timeout_seconds,
        "DMS_LIBREOFFICE_MAX_CONCURRENCY": libreoffice_max_concurrency,
        # 运行日志文件（契约1.23）：默认 logs/dms-server.log，OFF/NONE 关闭
        "DMS_LOG_FILE": (
            os.environ.get("DMS_LOG_FILE", "").strip() or "logs/dms-server.log"
        ),
        # 资源采样（契约1.23）：采样间隔秒数与保留天数
        "DMS_METRIC_INTERVAL_SECONDS": _positive_int(
            "DMS_METRIC_INTERVAL_SECONDS", 30
        ),
        "DMS_METRIC_RETENTION_DAYS": _positive_int(
            "DMS_METRIC_RETENTION_DAYS", 7
        ),
        # 系统管理页监控信息展示（契约1.26）：false 时仅显示维护操作与备份管理
        "DMS_SYSTEM_MONITOR_ENABLED": (
            os.environ.get("DMS_SYSTEM_MONITOR_ENABLED", "false").strip().lower()
            in {"1", "true", "yes", "on"}
        ),
        # 登录图片验证码（安全需求4）：缺省启用；关闭时登录回归三键请求体
        "DMS_CAPTCHA_ENABLED": (
            os.environ.get("DMS_CAPTCHA_ENABLED", "true").strip().lower()
            in {"1", "true", "yes", "on"}
        ),
        # 超级管理员登录开关（安全需求7）：false 时内置admin禁止登录（报错与
        # 密码错误一致）、用户列表隐藏admin、创建同名用户报"不能创建该名称的用户"
        "DMS_ALLOW_ADMIN_LOGIN": (
            os.environ.get("DMS_ALLOW_ADMIN_LOGIN", "true").strip().lower()
            in {"1", "true", "yes", "on"}
        ),
    }