DataOps-platform/app/__init__.py

import logging
import os
import uuid

from flask import Flask, jsonify
from flask_cors import CORS
from flask_sqlalchemy import SQLAlchemy

from app.config.config import (
    apply_runtime_env_config,
    config,
    current_env,
    log_llm_env_status,
    log_service_env_status,
    validate_production_database_identity,
)
from app.config.cors import CORS_OPTIONS

db = SQLAlchemy()


def create_app():
    """Create and configure the Flask application"""
    app = Flask(__name__)

    # 加载配置
    app.config.from_object(config[current_env])
    apply_runtime_env_config(app)

    # 初始化扩展
    # 配置CORS以解决跨域问题
    CORS(app, **CORS_OPTIONS)
    db.init_app(app)
    with app.app_context():
        validate_production_database_identity(app, db.engine)

    # 注册蓝图
    from app.api.business_domain import bp as business_domain_bp
    from app.api.data_development import bp as data_development_bp
    from app.api.data_factory import bp as data_factory_bp
    from app.api.data_flow import bp as data_flow_bp
    from app.api.data_interface import bp as data_interface_bp
    from app.api.data_rules import bp as data_rules_bp
    from app.api.data_service import bp as data_service_bp
    from app.api.data_source import bp as data_source_bp
    from app.api.graph import bp as graph_bp
    from app.api.knowledge_base import bp as knowledge_base_bp
    from app.api.meta_data import bp as meta_bp
    from app.api.system import bp as system_bp

    app.register_blueprint(meta_bp, url_prefix="/api/meta")
    app.register_blueprint(data_interface_bp, url_prefix="/api/interface")
    app.register_blueprint(data_rules_bp, url_prefix="/api/rules")
    app.register_blueprint(graph_bp, url_prefix="/api/graph")
    app.register_blueprint(system_bp, url_prefix="/api/system")
    app.register_blueprint(data_source_bp, url_prefix="/api/datasource")
    app.register_blueprint(data_development_bp, url_prefix="/api/development/v1")
    app.register_blueprint(data_flow_bp, url_prefix="/api/dataflow")
    app.register_blueprint(business_domain_bp, url_prefix="/api/bd")
    app.register_blueprint(data_factory_bp, url_prefix="/api/datafactory")
    app.register_blueprint(data_service_bp, url_prefix="/api/dataservice")
    app.register_blueprint(knowledge_base_bp, url_prefix="/api/knowledge")

    from app.core.system.permissions import configure_api_authorization

    configure_api_authorization(app)

    # Configure global response headers
    configure_response_headers(app)

    # Configure logging
    configure_logging(app)
    log_llm_env_status(app)
    log_service_env_status(app)

    # 添加全局异常处理器
    configure_error_handlers(app)

    # 输出启动信息（生产环境由 Gunicorn 按 LISTEN_PORT 监听，此处 PORT 与配置一致）
    port = app.config["PORT"]
    app.logger.info(
        f"Starting server in {current_env} mode on port {port} "
        f"(LISTEN_PORT={os.environ.get('LISTEN_PORT', port)})"
    )

    return app


def configure_response_headers(app):
    """Configure global response headers for JSON content"""

    @app.after_request
    def after_request(response):
        from flask import request

        # 检查是否是API路径
        if request.path.startswith("/api/"):
            # 排除文件下载和特殊响应类型
            excluded_types = [
                "application/octet-stream",
                "application/pdf",
                "image/",
                "text/csv",
                "application/vnd.ms-excel",
                "application/vnd.openxmlformats-officedocument",
            ]
            if response.content_type and any(
                ct in response.content_type for ct in excluded_types
            ):
                # 保持原有的文件类型不变
                pass
            elif response.content_type and "application/json" in response.content_type:
                # 确保JSON响应设置正确的Content-Type和charset
                ct = "application/json; charset=utf-8"
                response.headers["Content-Type"] = ct
            elif (
                not response.content_type
                or response.content_type == "text/html; charset=utf-8"
                or response.content_type == "text/plain"
            ):
                # 对于API路由，默认设置为JSON
                ct = "application/json; charset=utf-8"
                response.headers["Content-Type"] = ct

            # 确保CORS头部不被覆盖
            if "Access-Control-Allow-Origin" not in response.headers:
                # 动态设置Origin，支持任意前端地址
                origin = request.headers.get("Origin")
                if origin:
                    # 允许任意Origin（最灵活的配置）
                    response.headers["Access-Control-Allow-Origin"] = origin
                else:
                    # 如果没有Origin头部，设置为通配符
                    response.headers["Access-Control-Allow-Origin"] = "*"

            # 专门处理预检请求（OPTIONS方法）
            if request.method == "OPTIONS":
                origin = request.headers.get("Origin", "*")
                response.headers["Access-Control-Allow-Origin"] = origin
                methods = "GET, POST, PUT, DELETE, OPTIONS"
                response.headers["Access-Control-Allow-Methods"] = methods
                headers = (
                    "Content-Type, Authorization, X-Requested-With, "
                    "Accept, Origin, Cache-Control, X-File-Name, If-Match, "
                    "X-Agent-Credential, X-Edge-Enrollment, X-Edge-Credential, "
                    "X-Edge-Certificate-SHA256"
                )
                response.headers["Access-Control-Allow-Headers"] = headers
                response.headers["Access-Control-Max-Age"] = "86400"
                return response

            # 根据配置设置凭据支持
            from app.config.cors import ALLOW_ALL_ORIGINS

            if "Access-Control-Allow-Credentials" not in response.headers:
                if ALLOW_ALL_ORIGINS:
                    # 通配符时不支持凭据
                    response.headers["Access-Control-Allow-Credentials"] = "false"
                else:
                    response.headers["Access-Control-Allow-Credentials"] = "true"

            if "Access-Control-Allow-Methods" not in response.headers:
                methods = "GET, POST, PUT, DELETE, OPTIONS"
                response.headers["Access-Control-Allow-Methods"] = methods
            if "Access-Control-Allow-Headers" not in response.headers:
                headers = (
                    "Content-Type, Authorization, X-Requested-With, Accept, Origin, "
                    "If-Match, X-Agent-Credential, X-Edge-Enrollment, "
                    "X-Edge-Credential, X-Edge-Certificate-SHA256"
                )
                response.headers["Access-Control-Allow-Headers"] = headers

            # 添加安全头部
            if "X-Content-Type-Options" not in response.headers:
                response.headers["X-Content-Type-Options"] = "nosniff"
            if "X-Frame-Options" not in response.headers:
                response.headers["X-Frame-Options"] = "DENY"
            if "X-XSS-Protection" not in response.headers:
                response.headers["X-XSS-Protection"] = "1; mode=block"
            if "Referrer-Policy" not in response.headers:
                response.headers["Referrer-Policy"] = "no-referrer"
            if "Permissions-Policy" not in response.headers:
                response.headers["Permissions-Policy"] = (
                    "camera=(), microphone=(), geolocation=()"
                )
            if (
                request.path.startswith("/api/system/auth")
                or request.path.startswith(
                    "/api/system/governance-audit"
                )
            ):
                response.headers["Cache-Control"] = "no-store"

        if request.path.startswith("/api/") and request.path != "/api/system/health":
            app.logger.info(
                "%s %s -> %s",
                request.method,
                request.path,
                response.status_code,
            )

        return response


def configure_logging(app):
    """Configure logging for the application"""
    if not app.config.get("LOG_ENABLED", True):
        return None

    log_file = os.path.abspath(
        app.config.get("LOG_FILE", f"flask_{app.config['FLASK_ENV']}.log")
    )
    log_dir = os.path.dirname(log_file)
    if log_dir:
        os.makedirs(log_dir, exist_ok=True)

    log_level_name = app.config.get("LOG_LEVEL", "INFO")
    log_level = getattr(logging, log_level_name)
    log_format = app.config.get(
        "LOG_FORMAT",
        "%(asctime)s - %(levelname)s - %(filename)s - "
        "%(funcName)s - %(lineno)s - %(message)s",
    )
    log_encoding = app.config.get("LOG_ENCODING", "UTF-8")
    log_to_console = app.config.get("LOG_TO_CONSOLE", True)

    logging_format = logging.Formatter(log_format)

    root_logger = logging.getLogger()
    root_logger.setLevel(log_level)
    root_logger.handlers.clear()

    file_handler = logging.FileHandler(log_file, encoding=log_encoding)
    file_handler.setLevel(log_level)
    file_handler.setFormatter(logging_format)
    root_logger.addHandler(file_handler)

    if log_to_console:
        console = logging.StreamHandler()
        console.setLevel(log_level)
        console.setFormatter(logging_format)
        root_logger.addHandler(console)

    # Flask 默认 logger 关闭 propagate，清空 handler 后需要显式开启
    app.logger.handlers.clear()
    app.logger.propagate = True
    app.logger.setLevel(log_level)

    for logger_name in ("app", "flask.app"):
        named_logger = logging.getLogger(logger_name)
        named_logger.handlers.clear()
        named_logger.propagate = True
        named_logger.setLevel(log_level)

    app.logger.info(f"日志配置完成: 级别={log_level_name}, 文件={log_file}")
    return logging.getLogger("app")


def configure_error_handlers(app):
    """Configure global error handlers for the application"""

    @app.errorhandler(Exception)
    def handle_exception(e):
        """全局异常处理器，捕获所有未处理的异常"""
        from app.core.data_source.redaction import sanitize_exception

        correlation_id = str(uuid.uuid4())
        app.logger.error(
            "未处理的异常 correlation_id=%s type=%s detail=%s",
            correlation_id,
            type(e).__name__,
            sanitize_exception(e, limit=300),
        )
        error_response = {
            "success": False,
            "message": "服务器内部错误",
            "data": None,
            "correlation_id": correlation_id,
        }

        return jsonify(error_response), 500

    @app.errorhandler(404)
    def handle_not_found(e):
        """处理404错误"""
        app.logger.warning(f"404错误: {str(e)}")
        return jsonify(
            {"success": False, "message": "请求的资源不存在", "data": None}
        ), 404

    @app.errorhandler(500)
    def handle_internal_error(e):
        """处理500错误"""
        correlation_id = str(uuid.uuid4())
        app.logger.error(
            "500错误 correlation_id=%s type=%s",
            correlation_id,
            type(e).__name__,
        )
        return jsonify(
            {
                "success": False,
                "message": "服务器内部错误",
                "data": None,
                "correlation_id": correlation_id,
            }
        ), 500