DataOps-platform/app/config/config.py

import os
import platform
from collections.abc import Mapping


def get_bool_env(name: str, default: bool = False) -> bool:
    """Return a boolean value from an environment variable."""
    value = os.environ.get(name)
    if value is None:
        return default
    return value.strip().lower() in {"1", "true", "yes", "on"}


def _bounded_int(name: str, default: int, minimum: int, maximum: int) -> int:
    try:
        value = int(os.environ.get(name, str(default)))
    except (TypeError, ValueError) as exc:
        raise ValueError(f"{name} must be an integer") from exc
    if value < minimum or value > maximum:
        raise ValueError(f"{name} must be between {minimum} and {maximum}")
    return value


def _apply_datasource_pool_overrides(
    settings: dict, overrides: Mapping[str, int]
) -> dict:
    bounds = {
        "pool_size": (1, 10),
        "max_overflow": (0, 10),
    }
    unsupported = sorted(set(overrides) - set(bounds))
    if unsupported:
        raise ValueError(
            f"Unsupported data source pool overrides: {', '.join(unsupported)}"
        )

    resolved = dict(settings)
    for name, value in overrides.items():
        minimum, maximum = bounds[name]
        if type(value) is not int or value < minimum or value > maximum:
            raise ValueError(f"{name} must be between {minimum} and {maximum}")
        resolved[name] = value
    return resolved


def datasource_pool_settings(
    overrides: Mapping[str, int] = None,
) -> dict:
    """Return bounded settings for external business data-source pools."""
    settings = {
        "pool_size": _bounded_int("DATASOURCE_POOL_SIZE", 2, 1, 10),
        "max_overflow": _bounded_int("DATASOURCE_MAX_OVERFLOW", 3, 0, 10),
        "pool_timeout": _bounded_int("DATASOURCE_POOL_TIMEOUT", 10, 1, 60),
        "pool_recycle": _bounded_int(
            "DATASOURCE_POOL_RECYCLE", 1800, 60, 86400
        ),
        "idle_ttl": _bounded_int(
            "DATASOURCE_POOL_IDLE_TTL", 900, 60, 86400
        ),
        "max_idle_pools": _bounded_int(
            "DATASOURCE_MAX_IDLE_POOLS", 20, 1, 100
        ),
        "query_timeout": _bounded_int(
            "DATASOURCE_QUERY_TIMEOUT", 30, 1, 300
        ),
    }
    return _apply_datasource_pool_overrides(settings, overrides or {})


def get_environment():
    """
    获取当前运行环境
    优先级：
    1. 环境变量 FLASK_ENV
    2. 根据操作系统自动判断（Windows -> development, Linux -> production）
    """
    # 首先检查环境变量
    env = os.environ.get("FLASK_ENV")
    if env:
        return env.lower()

    # 根据操作系统判断
    system = platform.system().lower()
    if system == "windows":
        return "development"
    elif system == "linux":
        return "production"
    else:
        return "development"  # 其他系统默认使用开发环境


def resolve_log_file(default_filename: str = "flask_production.log") -> str:
    """Resolve application log path to an absolute file under LOG_DIR."""
    raw = os.environ.get("LOG_FILE", default_filename)
    if os.path.isabs(raw):
        return raw

    app_dir = os.environ.get("APP_DIR", os.getcwd())
    log_dir = os.environ.get("LOG_DIR", os.path.join(app_dir, "logs"))
    return os.path.join(log_dir, os.path.basename(raw))


def _clean_env(name: str, default: str = "") -> str:
    value = os.environ.get(name)
    if value is None:
        return default
    return value.strip().strip("\r\n\t")


def load_production_env_file() -> None:
    """Load production env file into os.environ without overriding existing keys."""
    env_file = os.environ.get(
        "APP_ENV_FILE",
        os.environ.get("ENV_FILE", "/etc/dataops-platform/dataops.env"),
    )
    if not env_file or not os.path.isfile(env_file):
        return
    if not os.access(env_file, os.R_OK):
        return
    try:
        from dotenv import load_dotenv

        load_dotenv(env_file, override=False)
    except Exception:
        return


def is_placeholder_env_value(value: str) -> bool:
    """Detect template placeholders that should not override production defaults."""
    if not value:
        return True
    lower = value.lower()
    if lower.startswith("replace-"):
        return True
    placeholders = (
        "replace-password",
        "replace-neo4j-password",
        "replace-minio",
        "replace-with-a-long-random-secret",
        "replace-n8n-api-key",
        "your-api-key",
    )
    if any(item in lower for item in placeholders):
        return True
    if "dataops_user@" in lower and "127.0.0.1" in lower:
        return True
    return lower in {"127.0.0.1:9000", "localhost:9000"}


def _is_local_minio_host(host: str) -> bool:
    if not host:
        return False
    return host.split(":")[0].lower() in {"127.0.0.1", "localhost"}


def _apply_minio_config(app, defaults: dict) -> None:
    """Apply MinIO settings as a whole to avoid host/user/password mismatch."""
    host = _clean_env("MINIO_HOST")
    user = _clean_env("MINIO_USER")
    password = _clean_env("MINIO_PASSWORD")
    bucket = _clean_env("MINIO_BUCKET")
    prefix = _clean_env("MINIO_PREFIX")
    is_production = os.environ.get("FLASK_ENV", "").lower() == "production"

    use_defaults = (
        is_placeholder_env_value(user)
        or is_placeholder_env_value(password)
        or is_placeholder_env_value(host)
        or (is_production and _is_local_minio_host(host))
    )

    if use_defaults:
        app.config["MINIO_HOST"] = defaults["MINIO_HOST"]
        app.config["MINIO_USER"] = defaults["MINIO_USER"]
        app.config["MINIO_PASSWORD"] = defaults["MINIO_PASSWORD"]
        app.config["MINIO_BUCKET"] = defaults["MINIO_BUCKET"]
        app.config["PREFIX"] = defaults["MINIO_PREFIX"]
        app.config["MINIO_SECURE"] = defaults["MINIO_SECURE"]
        return

    app.config["MINIO_HOST"] = host or defaults["MINIO_HOST"]
    app.config["MINIO_USER"] = user or defaults["MINIO_USER"]
    app.config["MINIO_PASSWORD"] = password or defaults["MINIO_PASSWORD"]
    app.config["MINIO_BUCKET"] = bucket or defaults["MINIO_BUCKET"]
    app.config["PREFIX"] = prefix if prefix else defaults["MINIO_PREFIX"]
    app.config["MINIO_SECURE"] = get_bool_env("MINIO_SECURE", defaults["MINIO_SECURE"])


PRODUCTION_SERVICE_DEFAULTS = {
    "SQLALCHEMY_DATABASE_URI": "postgresql://postgres:postgres@127.0.0.1:5432/dataops",
    "NEO4J_URI": "bolt://127.0.0.1:7687",
    "NEO4J_HTTP_URI": "http://127.0.0.1:7474",
    "NEO4J_USER": "neo4j",
    "NEO4J_PASSWORD": "Passw0rd",
    "NEO4J_ENCRYPTED": False,
    "MINIO_HOST": "127.0.0.1:9000",
    "MINIO_USER": "citu-test",
    "MINIO_PASSWORD": "citu-test",
    "MINIO_SECURE": False,
    "MINIO_BUCKET": "dataops-bucket",
    "MINIO_PREFIX": "",
}


def _apply_config_from_env(app, config_key: str, env_name: str, default):
    value = _clean_env(env_name)
    if value and not is_placeholder_env_value(value):
        app.config[config_key] = value
        return value
    app.config[config_key] = default
    return default


def apply_runtime_env_config(app) -> None:
    """Re-read env-backed settings when the worker starts (after dataops.env is loaded)."""
    load_production_env_file()
    defaults = PRODUCTION_SERVICE_DEFAULTS

    _apply_config_from_env(
        app, "SQLALCHEMY_DATABASE_URI", "DATABASE_URL", defaults["SQLALCHEMY_DATABASE_URI"]
    )
    _apply_config_from_env(app, "NEO4J_URI", "NEO4J_URI", defaults["NEO4J_URI"])
    _apply_config_from_env(
        app, "NEO4J_HTTP_URI", "NEO4J_HTTP_URI", defaults["NEO4J_HTTP_URI"]
    )
    _apply_config_from_env(app, "NEO4J_USER", "NEO4J_USER", defaults["NEO4J_USER"])
    _apply_config_from_env(
        app, "NEO4J_PASSWORD", "NEO4J_PASSWORD", defaults["NEO4J_PASSWORD"]
    )
    app.config["NEO4J_ENCRYPTED"] = get_bool_env(
        "NEO4J_ENCRYPTED", defaults["NEO4J_ENCRYPTED"]
    )
    _apply_minio_config(app, defaults)

    secret_key = _clean_env("SECRET_KEY")
    if secret_key and not is_placeholder_env_value(secret_key):
        app.config["SECRET_KEY"] = secret_key

    deepseek_key = _clean_env("DEEPSEEK_API_KEY")
    llm_key = deepseek_key or _clean_env("LLM_API_KEY")
    if deepseek_key and not is_placeholder_env_value(deepseek_key):
        app.config["DEEPSEEK_API_KEY"] = deepseek_key
        app.config["LLM_API_KEY"] = deepseek_key
    elif llm_key and not is_placeholder_env_value(llm_key):
        app.config["LLM_API_KEY"] = llm_key

    _apply_config_from_env(
        app,
        "LLM_BASE_URL",
        "LLM_BASE_URL",
        "https://api.deepseek.com",
    )

    llm_model = _clean_env("LLM_MODEL_NAME")
    if llm_model:
        app.config["LLM_MODEL_NAME"] = llm_model

    llm_reasoning = _clean_env("LLM_REASONING_EFFORT")
    if llm_reasoning:
        app.config["LLM_REASONING_EFFORT"] = llm_reasoning

    pool_settings = datasource_pool_settings()
    app.config.update(
        {
            "DATASOURCE_POOL_SIZE": pool_settings["pool_size"],
            "DATASOURCE_MAX_OVERFLOW": pool_settings["max_overflow"],
            "DATASOURCE_POOL_TIMEOUT": pool_settings["pool_timeout"],
            "DATASOURCE_POOL_RECYCLE": pool_settings["pool_recycle"],
            "DATASOURCE_POOL_IDLE_TTL": pool_settings["idle_ttl"],
            "DATASOURCE_MAX_IDLE_POOLS": pool_settings["max_idle_pools"],
            "DATASOURCE_QUERY_TIMEOUT": pool_settings["query_timeout"],
            "DATASOURCE_CREDENTIAL_MASTER_KEY": _clean_env(
                "DATASOURCE_CREDENTIAL_MASTER_KEY"
            ),
            "DATASOURCE_CREDENTIAL_KEY_VERSION": _clean_env(
                "DATASOURCE_CREDENTIAL_KEY_VERSION", "v1"
            ),
            "DATASOURCE_CERT_DIR": _clean_env(
                "DATASOURCE_CERT_DIR",
                "/etc/dataops-platform/datasource-certs",
            ),
        }
    )


def log_llm_env_status(app) -> None:
    """Log LLM env load result after logging is configured."""
    deepseek_key = _clean_env("DEEPSEEK_API_KEY") or _clean_env("LLM_API_KEY")
    if not deepseek_key:
        deepseek_key = str(
            app.config.get("DEEPSEEK_API_KEY") or app.config.get("LLM_API_KEY") or ""
        ).strip().strip("\r\n\t")
    env_file = os.environ.get("APP_ENV_FILE", "/etc/dataops-platform/dataops.env")

    if deepseek_key:
        app.logger.info(
            "DeepSeek API Key 已加载 (长度=%s, base_url=%s, model=%s)",
            len(deepseek_key),
            app.config.get("LLM_BASE_URL", "https://api.deepseek.com"),
            app.config.get("LLM_MODEL_NAME", "deepseek-chat"),
        )
        return

    if os.path.isfile(env_file) and not os.access(env_file, os.R_OK):
        app.logger.error(
            f"无法读取 {env_file}，Supervisor 用户需有读权限。"
            f"请执行: sudo chown root:{os.environ.get('APP_USER', 'ubuntu')} "
            f"{env_file} && sudo chmod 640 {env_file}"
        )
        return

    app.logger.warning(
        f"DeepSeek API Key 未配置，请在 {env_file} 中设置 DEEPSEEK_API_KEY"
    )


def log_service_env_status(app) -> None:
    """Log database/Neo4j endpoints after logging is configured."""
    db_uri = str(app.config.get("SQLALCHEMY_DATABASE_URI", ""))
    db_host = db_uri.split("@")[-1] if "@" in db_uri else db_uri
    app.logger.info(
        "服务连接配置: PostgreSQL=%s, Neo4j=%s, MinIO=%s (user=%s)",
        db_host,
        app.config.get("NEO4J_URI"),
        app.config.get("MINIO_HOST"),
        _mask_secret(str(app.config.get("MINIO_USER", ""))),
    )
    if is_placeholder_env_value(_clean_env("DATABASE_URL")):
        app.logger.warning(
            "DATABASE_URL 仍为模板占位符，已回退到默认生产库 "
            f"{PRODUCTION_SERVICE_DEFAULTS['SQLALCHEMY_DATABASE_URI'].split('@')[-1]}"
        )
    if is_placeholder_env_value(_clean_env("NEO4J_PASSWORD")):
        app.logger.warning(
            "NEO4J_PASSWORD 仍为模板占位符，已回退到 config.py 中的默认生产配置"
        )
    minio_user_env = _clean_env("MINIO_USER")
    minio_host_env = _clean_env("MINIO_HOST")
    if (
        is_placeholder_env_value(minio_user_env)
        or is_placeholder_env_value(minio_host_env)
        or (
            os.environ.get("FLASK_ENV", "").lower() == "production"
            and _is_local_minio_host(minio_host_env)
        )
    ):
        app.logger.warning(
            "MinIO 配置仍为模板或 localhost，已回退到默认生产 MinIO "
            f"{PRODUCTION_SERVICE_DEFAULTS['MINIO_HOST']}"
        )


def _mask_secret(value: str) -> str:
    if not value:
        return "(empty)"
    if len(value) <= 4:
        return "****"
    return f"{value[:4]}****"


class BaseConfig:
    """基础配置类，包含所有环境共享的配置"""

    SECRET_KEY = os.environ.get("SECRET_KEY") or "you-will-never-guess"
    RULE_GENERATION_RECEIPT_SECRET = os.environ.get(
        "RULE_GENERATION_RECEIPT_SECRET"
    )
    JSON_AS_ASCII = False
    JSONIFY_PRETTYPRINT_REGULAR = True
    JSON_SORT_KEYS = False

    # 平台特定配置
    PLATFORM = platform.system().lower()

    # 文件上传配置
    ALLOWED_EXTENSIONS = {
        "txt",
        "pdf",
        "png",
        "jpg",
        "jpeg",
        "gif",
        "xlsx",
        "xls",
        "csv",
        "sql",
        "dll",
        "docx",
        "doc",
    }

    # 数据抽取配置
    DATA_EXTRACT_BATCH_SIZE = 1000  # 每批处理的记录数

    # PostgreSQL 基础配置
    SQLALCHEMY_ENGINE_OPTIONS = {
        "pool_pre_ping": True,
        "pool_recycle": 300,
        "pool_size": 10,
        "max_overflow": 20,
    }

    _DATASOURCE_POOL_SETTINGS = datasource_pool_settings()
    DATASOURCE_POOL_SIZE = _DATASOURCE_POOL_SETTINGS["pool_size"]
    DATASOURCE_MAX_OVERFLOW = _DATASOURCE_POOL_SETTINGS["max_overflow"]
    DATASOURCE_POOL_TIMEOUT = _DATASOURCE_POOL_SETTINGS["pool_timeout"]
    DATASOURCE_POOL_RECYCLE = _DATASOURCE_POOL_SETTINGS["pool_recycle"]
    DATASOURCE_POOL_IDLE_TTL = _DATASOURCE_POOL_SETTINGS["idle_ttl"]
    DATASOURCE_MAX_IDLE_POOLS = _DATASOURCE_POOL_SETTINGS["max_idle_pools"]
    DATASOURCE_QUERY_TIMEOUT = _DATASOURCE_POOL_SETTINGS["query_timeout"]
    DATASOURCE_CREDENTIAL_MASTER_KEY = os.environ.get(
        "DATASOURCE_CREDENTIAL_MASTER_KEY", ""
    )
    DATASOURCE_CREDENTIAL_KEY_VERSION = os.environ.get(
        "DATASOURCE_CREDENTIAL_KEY_VERSION", "v1"
    )
    DATASOURCE_CERT_DIR = os.environ.get(
        "DATASOURCE_CERT_DIR",
        "/etc/dataops-platform/datasource-certs",
    )

    # DeepSeek LLM（OpenAI 兼容接口）
    DEEPSEEK_API_KEY = os.environ.get("DEEPSEEK_API_KEY", "")
    LLM_BASE_URL = os.environ.get("LLM_BASE_URL", "https://api.deepseek.com")
    LLM_MODEL_NAME = os.environ.get("LLM_MODEL_NAME", "deepseek-chat")
    LLM_REASONING_EFFORT = os.environ.get("LLM_REASONING_EFFORT", "high")
    # 兼容旧环境变量名 LLM_API_KEY
    LLM_API_KEY = DEEPSEEK_API_KEY or os.environ.get("LLM_API_KEY", "")

    # 数据治理知识库
    KNOWLEDGE_ENABLED = os.environ.get("KNOWLEDGE_ENABLED", "true").lower() == "true"
    KNOWLEDGE_VECTOR_TOP_K = int(os.environ.get("KNOWLEDGE_VECTOR_TOP_K", "40"))
    KNOWLEDGE_RRF_K = int(os.environ.get("KNOWLEDGE_RRF_K", "60"))
    KNOWLEDGE_RERANK_TOP_K = int(os.environ.get("KNOWLEDGE_RERANK_TOP_K", "8"))
    KNOWLEDGE_EVIDENCE_TOKEN_BUDGET = int(
        os.environ.get("KNOWLEDGE_EVIDENCE_TOKEN_BUDGET", "8000")
    )
    KNOWLEDGE_IMPACT_MAX_HOPS = int(os.environ.get("KNOWLEDGE_IMPACT_MAX_HOPS", "3"))
    KNOWLEDGE_IMPACT_MAX_POINTS = int(
        os.environ.get("KNOWLEDGE_IMPACT_MAX_POINTS", "1000")
    )
    KNOWLEDGE_LIGHTRAG_ENABLED = (
        os.environ.get("KNOWLEDGE_LIGHTRAG_ENABLED", "false").lower() == "true"
    )
    KNOWLEDGE_LIGHTRAG_SHADOW_ONLY = (
        os.environ.get("KNOWLEDGE_LIGHTRAG_SHADOW_ONLY", "true").lower() == "true"
    )
    KNOWLEDGE_LIGHTRAG_BASE_URL = os.environ.get(
        "KNOWLEDGE_LIGHTRAG_BASE_URL", "http://lightrag:9621"
    )
    KNOWLEDGE_LIGHTRAG_API_KEY = os.environ.get("KNOWLEDGE_LIGHTRAG_API_KEY", "")
    QWEN_EMBEDDING_BASE_URL = os.environ.get("QWEN_EMBEDDING_BASE_URL", "")
    QWEN_EMBEDDING_API_KEY = os.environ.get("QWEN_EMBEDDING_API_KEY", "")
    QWEN_EMBEDDING_MODEL = os.environ.get("QWEN_EMBEDDING_MODEL", "text-embedding-v3")
    QWEN_EMBEDDING_DIMENSION = int(os.environ.get("QWEN_EMBEDDING_DIMENSION", "1024"))

    # 日志基础配置
    LOG_FORMAT = "%(asctime)s - %(levelname)s - %(filename)s - %(funcName)s - %(lineno)s - %(message)s"
    LOG_ENCODING = "UTF-8"
    LOG_ENABLED = True

    # DataFlow 配置
    DATAFLOW_SCHEMA = os.environ.get("DATAFLOW_SCHEMA", "dags")

    # n8n 工作流引擎配置
    N8N_API_URL = os.environ.get("N8N_API_URL", "")
    N8N_API_KEY = os.environ.get("N8N_API_KEY", "")
    N8N_API_TIMEOUT = int(os.environ.get("N8N_API_TIMEOUT", "30"))

    # DataOps 平台 API 基础 URL（用于 n8n 工作流回调等）
    API_BASE_URL = os.environ.get("API_BASE_URL", "http://127.0.0.1:5500/api")


class DevelopmentConfig(BaseConfig):
    """Windows 开发环境配置"""

    FLASK_ENV = "development"
    DEBUG = True
    PORT = 5500

    # 开发环境 MinIO 配置
    MINIO_HOST = "localhost:9000"
    MINIO_USER = "citu-test"
    MINIO_PASSWORD = "citu-test"
    MINIO_SECURE = False
    MINIO_BUCKET = "dataops-bucket"
    PREFIX = ""

    # 开发环境 PostgreSQL 配置
    SQLALCHEMY_DATABASE_URI = "postgresql://postgres:postgres@localhost:5432/dataops"

    # 开发环境 Neo4j 配置
    NEO4J_URI = "bolt://localhost:7687"
    NEO4J_HTTP_URI = "http://localhost:7474"
    NEO4J_USER = "neo4j"
    NEO4J_PASSWORD = "Passw0rd"
    NEO4J_ENCRYPTED = False

    # 开发环境文件路径配置
    UPLOAD_BASE_PATH = "C:\\tmp\\upload"
    ARCHIVE_BASE_PATH = "C:\\tmp\\archive"

    # 开发环境日志配置
    LOG_LEVEL = "DEBUG"
    LOG_FILE = resolve_log_file("flask_development.log")
    LOG_TO_CONSOLE = True

class ProductionConfig(BaseConfig):
    """Linux 生产环境配置"""

    FLASK_ENV = "production"
    DEBUG = False
    # 与 run_dataops.sh / dataops.env 中 LISTEN_PORT 保持一致（默认 5500，供 Nginx 反代）
    PORT = int(os.environ.get("LISTEN_PORT", os.environ.get("PORT", "5500")))

    # 生产环境 MinIO 配置
    MINIO_HOST = os.environ.get("MINIO_HOST", "127.0.0.1:9000")
    MINIO_USER = os.environ.get("MINIO_USER", "citu-test")
    MINIO_PASSWORD = os.environ.get("MINIO_PASSWORD", "citu-test")
    MINIO_SECURE = get_bool_env("MINIO_SECURE", False)
    MINIO_BUCKET = os.environ.get("MINIO_BUCKET", "dataops-bucket")
    PREFIX = os.environ.get("MINIO_PREFIX", "")

    # 生产环境 PostgreSQL 配置
    SQLALCHEMY_DATABASE_URI = os.environ.get(
        "DATABASE_URL", "postgresql://postgres:postgres@127.0.0.1:5432/dataops"
    )

    # 生产环境 Neo4j 配置
    NEO4J_URI = os.environ.get("NEO4J_URI", "bolt://127.0.0.1:7687")
    NEO4J_HTTP_URI = os.environ.get("NEO4J_HTTP_URI", "http://127.0.0.1:7474")
    NEO4J_USER = os.environ.get("NEO4J_USER", "neo4j")
    NEO4J_PASSWORD = os.environ.get("NEO4J_PASSWORD", "Passw0rd")
    NEO4J_ENCRYPTED = get_bool_env("NEO4J_ENCRYPTED", False)

    # 生产环境文件路径配置
    UPLOAD_BASE_PATH = os.environ.get("UPLOAD_BASE_PATH", "/data/upload")
    ARCHIVE_BASE_PATH = os.environ.get("ARCHIVE_BASE_PATH", "/data/archive")

    # 生产环境日志配置
    LOG_LEVEL = os.environ.get("LOG_LEVEL", "INFO")
    LOG_FILE = resolve_log_file("flask_production.log")
    LOG_TO_CONSOLE = get_bool_env("LOG_TO_CONSOLE", False)

# 配置字典
config = {
    "development": DevelopmentConfig,
    "production": ProductionConfig,
    "default": DevelopmentConfig,
}

# 获取当前环境
current_env = get_environment()