""" Qihu-AIFlow 智能自动化PaaS平台 - 后端配置 """ import os import logging from pathlib import Path from logging.handlers import TimedRotatingFileHandler from app.core.observability import TraceIdFilter, JsonFormatter from dotenv import load_dotenv # 加载 .env 文件(Docker 环境跳过,直接使用容器环境变量) if not (Path("/.dockerenv").exists() or Path("/run/.containerenv").exists()): env_path = Path("/data/share/project/.env") if env_path.exists(): load_dotenv(env_path, override=False) # 数据库配置 DATABASE_URL = f"mysql+pymysql://{os.getenv('MYSQL_USER', 'root')}:{os.getenv('MYSQL_PASSWORD', 'pwd2026')}@{os.getenv('MYSQL_HOST', 'localhost')}:{os.getenv('MYSQL_PORT', '3306')}/{os.getenv('MYSQL_DATABASE', 'aiflow')}?charset=utf8mb4" # Redis配置 REDIS_HOST = os.getenv("REDIS_HOST", "localhost") REDIS_PORT = int(os.getenv("REDIS_PORT", "6379")) REDIS_PASSWORD = os.getenv("REDIS_PASSWORD", "pwd2026") REDIS_URL = f"redis://:{REDIS_PASSWORD}@{REDIS_HOST}:{REDIS_PORT}/0" # JWT配置 SECRET_KEY = os.getenv("SECRET_KEY", "aiflow-secret-key-change-in-production-2026") ALGORITHM = "HS256" ACCESS_TOKEN_EXPIRE_MINUTES = 60 * 24 # 24小时 # 应用配置 APP_NAME = "Qihu-AIFlow" APP_VERSION = "1.0.0" DEBUG = os.getenv("DEBUG", "false").lower() == "true" # 服务地址 # 三级优先级: # 1. docker-compose.yml 的环境变量(生产部署) # 2. .env 文件中的配置(本地开发) # 3. 硬编码正式域名(最终兜底) # 注意:如果环境变量是临时开发地址(如team-coding子域名),强制使用.env中的正式域名 APP_ACCESS_URL = os.getenv("APP_ACCESS_URL", "") # 检测是否为临时/开发环境域名,这些域名在生产不可用 _TEMP_DOMAIN_PATTERNS = ["team-coding", "localhost", "127.0.0.1", "0.0.0.0"] _is_temp_url = any(p in (APP_ACCESS_URL or "") for p in _TEMP_DOMAIN_PATTERNS) if not APP_ACCESS_URL or _is_temp_url: # 尝试从.env文件读取正式域名(Docker环境跳过.env,直接用docker-compose变量) try: from dotenv import load_dotenv as _load_dotenv _env_path = Path("/data/share/project/.env") if _env_path.exists(): _load_dotenv(_env_path, override=True) env_url = os.getenv("APP_ACCESS_URL", "") if env_url: APP_ACCESS_URL = env_url _is_temp_url = any(p in (APP_ACCESS_URL or "") for p in _TEMP_DOMAIN_PATTERNS) except Exception: pass # 最终兜底:如果仍然是临时URL或为空,使用硬编码正式域名 if not APP_ACCESS_URL or _is_temp_url: APP_ACCESS_URL = "https://bot.qihuyun.com" # 上传文件配置(兼容 Docker /app 和本地 /data/share/project 路径) _DATA_BASE = Path("/app/data") if Path("/.dockerenv").exists() else Path("/data/share/project/data") UPLOAD_DIR = _DATA_BASE / "uploads" UPLOAD_DIR.mkdir(parents=True, exist_ok=True) MAX_UPLOAD_SIZE = 50 * 1024 * 1024 # 50MB # ============================================================ # 日志配置 - 按天轮转输出到文件 # ============================================================ LOG_DIR = _DATA_BASE / "logs" LOG_DIR.mkdir(parents=True, exist_ok=True) # 慢请求阈值(秒) SLOW_REQUEST_THRESHOLD = float(os.getenv("SLOW_REQUEST_THRESHOLD", "3.0")) def setup_file_logging(): """配置应用日志按天轮转输出到文件""" use_json = os.getenv("AIFLOW_JSON_LOGS", "0") == "1" if use_json: log_format = JsonFormatter() else: log_format = logging.Formatter( "%(asctime)s [%(levelname)s] %(name)s [tid:%(trace_id)s] - %(message)s", datefmt="%Y-%m-%d %H:%M:%S", ) # 应用日志 - 按天轮转,保留30天 app_handler = TimedRotatingFileHandler( filename=str(LOG_DIR / "app.log"), when="midnight", interval=1, backupCount=30, encoding="utf-8", ) app_handler.setFormatter(log_format) app_handler.setLevel(logging.INFO) # 错误日志 - 按天轮转,保留30天 error_handler = TimedRotatingFileHandler( filename=str(LOG_DIR / "error.log"), when="midnight", interval=1, backupCount=30, encoding="utf-8", ) error_handler.setFormatter(log_format) error_handler.setLevel(logging.ERROR) # 访问日志 - 按天轮转,保留14天 access_handler = TimedRotatingFileHandler( filename=str(LOG_DIR / "access.log"), when="midnight", interval=1, backupCount=14, encoding="utf-8", ) access_handler.setFormatter(log_format) access_handler.setLevel(logging.INFO) # 配置根logger root_logger = logging.getLogger() root_logger.setLevel(logging.INFO) root_logger.addHandler(app_handler) root_logger.addHandler(error_handler) # TraceIdFilter 必须挂在 handler 上,而非 root logger。 # logger 级 filter 只对「直接经该 logger 发射」的记录生效;子 logger 的记录向上冒泡 # 到 root 的 handler 时不会经过 root 的 logger 级 filter,导致 %(trace_id)s 缺失并抛 # KeyError('trace_id')。handler 级 filter 对经它格式化的每条记录都生效。 _tid_filter = TraceIdFilter() app_handler.addFilter(_tid_filter) error_handler.addFilter(_tid_filter) access_handler.addFilter(_tid_filter) # 配置访问日志专用logger access_logger = logging.getLogger("aiflow.access") access_logger.addHandler(access_handler) access_logger.propagate = False access_logger.addFilter(TraceIdFilter()) return access_logger