VWED_server/config/settings.py
2025-07-14 10:29:37 +08:00

446 lines
18 KiB
Python
Raw Permalink Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
系统配置模块
提供集中的配置管理,支持开发环境、测试环境和生产环境
"""
import os
from typing import Dict, Any, Optional, List
from urllib.parse import quote_plus
from pydantic import Field
from pydantic_settings import BaseSettings
# 定义数据目录路径
DATA_DIR = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "data")
# 确保数据目录存在
os.makedirs(DATA_DIR, exist_ok=True)
# 服务器配置
default_server_config = {
'host': '0.0.0.0',
'port': 8000,
'workers': 1,
'reload': True,
'log_level': 'info'
}
# 生产环境服务器配置
production_server_config = {
'host': '0.0.0.0',
'port': 8000,
'workers': 1, # 推荐根据CPU核心数调整: (CPU核心数 × 2) + 1
'reload': False, # 生产环境必须关闭热重载
'log_level': 'warning' # 生产环境减少日志输出
}
# 测试环境服务器配置
test_server_config = {
'host': '127.0.0.1',
'port': 8001,
'workers': 2, # 测试环境适中配置
'reload': False,
'log_level': 'info'
}
# 数据库配置
# default_db_config = {
# 'dialect': 'mysql',
# 'driver': 'pymysql',
# 'username': 'remote_admin',
# 'password': 'aiit@0619',
# 'host': '192.168.189.82',
# 'port': 3306,
# 'database': 'vwed_task',
# 'charset': 'utf8mb4'
# }
default_db_config = {
'dialect': 'mysql',
'driver': 'pymysql',
'username': 'root',
'password': 'root',
'host': 'localhost',
'port': 3306,
'database': 'vwed_task',
'charset': 'utf8mb4'
}
# Redis配置
default_redis_config = {
'host': 'localhost',
'port': 6379,
'db': 0,
'password': None,
'prefix': 'vwed:',
'socket_timeout': 5,
'socket_connect_timeout': 5,
'decode_responses': True
}
# test_redis_config = {
# 'host': 'localhost',
# 'port': 6379,
# 'db': 1,
# 'password': None,
# 'prefix': 'vwed_test:',
# 'decode_responses': True
# }
# # 库位服务API端点映射
# storage_api_endpoints = {
# "batch_setting_site": "/site/batch-setting",
# "get_idle_crowded_site": "/site/idle-crowded",
# "get_idle_site": "/site/idle",
# "get_locked_sites_by_task_record_id": "/site/locked-by-task",
# "get_site_attr": "/site/attr",
# "query_idle_site": "/site/query",
# "set_site_attr": "/site/attr",
# "set_site_content": "/site/content",
# "set_site_empty": "/site/empty",
# "set_site_filled": "/site/filled",
# "set_site_locked": "/site/lock",
# "set_site_tags": "/site/tags",
# "set_site_unlocked": "/site/unlock"
# }
# # 库位服务API HTTP方法映射
# storage_api_methods = {
# "batch_setting_site": "POST",
# "get_idle_crowded_site": "GET",
# "get_idle_site": "GET",
# "get_locked_sites_by_task_record_id": "GET",
# "get_site_attr": "GET",
# "query_idle_site": "GET",
# "set_site_attr": "PUT",
# "set_site_content": "PUT",
# "set_site_empty": "PUT",
# "set_site_filled": "PUT",
# "set_site_locked": "PUT",
# "set_site_tags": "PUT",
# "set_site_unlocked": "PUT"
# }
# 机器人调度服务API端点映射
robot_api_endpoints = {
"vehicle_station": "/robot/station",
"get_battery_level": "/robot/battery",
"get_pgv_code": "/robot/pgv-code"
}
# 机器人调度服务API HTTP方法映射
robot_api_methods = {
"vehicle_station": "GET",
"get_battery_level": "GET",
"get_pgv_code": "GET"
}
# 呼叫器设备服务API端点映射
calldevice_api_endpoints = {
"get_device_state": "/jeecg-boot/device/getDeviceState",
"init_device": "/jeecg-boot/device/initDevice",
"set_device_state": "/jeecg-boot/device/setDeviceState"
}
# 呼叫器设备服务API HTTP方法映射
calldevice_api_methods = {
"get_device_state": "POST",
"init_device": "POST",
"set_device_state": "POST"
}
# 任务执行API端点映射
task_execution_api_endpoints = {
"run_task": "/api/vwed-task-edit/run",
"stop_task": "/api/vwed-task-record/stop"
}
# 任务执行API HTTP方法映射
task_execution_api_methods = {
"run_task": "POST",
"stop_task": "POST"
}
# 外部服务API配置
external_api_config = {
# "storage": {
# "base_url": "http://localhost:8080/api/storage",
# "endpoints": storage_api_endpoints,
# "methods": storage_api_methods
# },
"robot": {
"base_url": "http://localhost:8080/api/robot",
"endpoints": robot_api_endpoints,
"methods": robot_api_methods
},
"calldevice": {
"base_url": "http://82.156.39.91:18080",
"init_base_url": "http://82.156.39.91:18080",
"endpoints": calldevice_api_endpoints,
"methods": calldevice_api_methods
},
"task_execution": {
"base_url": "http://127.0.0.1:8000",
"endpoints": task_execution_api_endpoints,
"methods": task_execution_api_methods
}
}
def get_config_for_env(env: str, config_type: str) -> Dict[str, Any]:
"""根据环境获取配置"""
if config_type == 'server':
if env == 'production':
return production_server_config
elif env == 'test':
return test_server_config
return default_server_config
elif config_type == 'db':
# 目前只有默认数据库配置
return default_db_config
elif config_type == 'redis':
# if env == 'test':
# return test_redis_config
return default_redis_config
return {}
class BaseConfig(BaseSettings):
"""基础配置类"""
# 应用信息
APP_NAME: str = "VWED任务系统"
APP_VERSION: str = "1.0.0"
APP_DESCRIPTION: str = "自动化移动机器人(AMR)任务管理系统"
# 系统设置
DEBUG: bool = False
API_PREFIX: str = "/api"
# 获取当前环境
_env: str = os.getenv("APP_ENV", "production").lower()
# 服务配置
_server_config = get_config_for_env(_env, 'server')
SERVER_HOST: str = Field(default=_server_config['host'], env="HOST")
SERVER_PORT: int = Field(default=_server_config['port'], env="PORT")
SERVER_WORKERS: int = Field(default=_server_config['workers'], env="WORKERS")
SERVER_RELOAD: bool = Field(default=_server_config['reload'], env="RELOAD")
SERVER_LOG_LEVEL: str = Field(default=_server_config['log_level'], env="LOG_LEVEL")
# Modbus设备配置
MODBUS_MOCK_MODE: bool = Field(default=True, env="MODBUS_MOCK_MODE") # 测试模式True表示启用模拟数据
MODBUS_TIMEOUT: float = Field(default=2.0, env="MODBUS_TIMEOUT") # Modbus通信超时时间
MODBUS_RETRY_INTERVAL: float = Field(default=1.0, env="MODBUS_RETRY_INTERVAL") # 重试间隔时间(秒)
MODBUS_POLL_INTERVAL: float = Field(default=0.5, env="MODBUS_POLL_INTERVAL") # 轮询间隔时间(秒)
# 数据库连接配置
_db_config = get_config_for_env(_env, 'db')
DB_DIALECT: str = Field(default=_db_config['dialect'], env="DB_DIALECT")
DB_DRIVER: str = Field(default=_db_config['driver'], env="DB_DRIVER")
DB_USER: str = Field(default=_db_config['username'], env="DB_USER")
DB_PASSWORD: str = Field(default=_db_config['password'], env="DB_PASSWORD")
DB_HOST: str = Field(default=_db_config['host'], env="DB_HOST")
DB_PORT: str = Field(default=str(_db_config['port']), env="DB_PORT")
DB_NAME: str = Field(default=_db_config['database'], env="DB_NAME")
DB_CHARSET: str = Field(default=_db_config['charset'], env="DB_CHARSET")
DB_ECHO: bool = False # 是否输出SQL语句
DB_POOL_SIZE: int = 10
DB_MAX_OVERFLOW: int = 20
DB_POOL_RECYCLE: int = 3600 # 连接池回收时间,防止连接过期
# Redis配置
_redis_config = get_config_for_env(_env, 'redis')
REDIS_HOST: str = Field(default=_redis_config['host'], env="REDIS_HOST")
REDIS_PORT: int = Field(default=_redis_config['port'], env="REDIS_PORT")
REDIS_DB: int = Field(default=_redis_config['db'], env="REDIS_DB")
REDIS_PASSWORD: Optional[str] = Field(default=_redis_config['password'], env="REDIS_PASSWORD")
REDIS_PREFIX: str = Field(default=_redis_config['prefix'], env="REDIS_PREFIX")
REDIS_SOCKET_TIMEOUT: int = Field(default=_redis_config['socket_timeout'], env="REDIS_SOCKET_TIMEOUT")
REDIS_SOCKET_CONNECT_TIMEOUT: int = Field(default=_redis_config['socket_connect_timeout'], env="REDIS_SOCKET_CONNECT_TIMEOUT")
REDIS_DECODE_RESPONSES: bool = Field(default=_redis_config['decode_responses'], env="REDIS_DECODE_RESPONSES")
# 安全设置
SECRET_KEY: str = Field(default="YOUR_SECRET_KEY_HERE", env="SECRET_KEY")
ACCESS_TOKEN_EXPIRE_MINUTES: int = 60 * 24 # 1天
# 外部服务API通用配置
API_TIMEOUT: int = Field(default=30, env="API_TIMEOUT") # 请求超时时间(秒)
API_TOKEN: Optional[str] = Field(default=None, env="API_TOKEN") # API通用认证令牌
API_MOCK_MODE: bool = Field(default=False, env="API_MOCK_MODE") # 是否启用API模拟模式
# 库位服务API配置
# STORAGE_API_BASE_URL: str = Field(default=external_api_config["storage"]["base_url"], env="STORAGE_API_BASE_URL")
# STORAGE_API_ENDPOINTS: Dict[str, str] = external_api_config["storage"]["endpoints"]
# STORAGE_API_METHODS: Dict[str, str] = external_api_config["storage"]["methods"]
# STORAGE_API_TIMEOUT: int = Field(default=30, env="STORAGE_API_TIMEOUT")
# STORAGE_API_TOKEN: Optional[str] = Field(default=None, env="STORAGE_API_TOKEN")
# STORAGE_API_MOCK_MODE: bool = Field(default=False, env="STORAGE_API_MOCK_MODE")
# 机器人调度服务API配置
ROBOT_API_BASE_URL: str = Field(default=external_api_config["robot"]["base_url"], env="ROBOT_API_BASE_URL")
ROBOT_API_ENDPOINTS: Dict[str, str] = external_api_config["robot"]["endpoints"]
ROBOT_API_METHODS: Dict[str, str] = external_api_config["robot"]["methods"]
ROBOT_API_TIMEOUT: int = Field(default=30, env="ROBOT_API_TIMEOUT")
ROBOT_API_TOKEN: Optional[str] = Field(default=None, env="ROBOT_API_TOKEN")
ROBOT_API_MOCK_MODE: bool = Field(default=False, env="ROBOT_API_MOCK_MODE")
# 呼叫器设备服务API配置
CALLDEVICE_API_BASE_URL: str = Field(default=external_api_config["calldevice"]["base_url"], env="CALLDEVICE_API_BASE_URL")
CALLDEVICE_API_INIT_BASE_URL: str = Field(default=external_api_config["calldevice"]["init_base_url"], env="CALLDEVICE_API_INIT_BASE_URL")
CALLDEVICE_API_ENDPOINTS: Dict[str, str] = external_api_config["calldevice"]["endpoints"]
CALLDEVICE_API_METHODS: Dict[str, str] = external_api_config["calldevice"]["methods"]
CALLDEVICE_API_TIMEOUT: int = Field(default=10, env="CALLDEVICE_API_TIMEOUT") # 获取设备状态的超时时间
CALLDEVICE_API_INIT_TIMEOUT: int = Field(default=30, env="CALLDEVICE_API_INIT_TIMEOUT") # 初始化设备的超时时间
CALLDEVICE_API_RESET_TIMEOUT: int = Field(default=30, env="CALLDEVICE_API_RESET_TIMEOUT") # 复位按钮的超时时间
CALLDEVICE_API_TOKEN: Optional[str] = Field(default=None, env="CALLDEVICE_API_TOKEN")
CALLDEVICE_API_MOCK_MODE: bool = Field(default=False, env="CALLDEVICE_API_MOCK_MODE")
# 任务执行API配置
TASK_EXECUTION_API_BASE_URL: str = Field(default=external_api_config["task_execution"]["base_url"], env="TASK_EXECUTION_API_BASE_URL")
TASK_EXECUTION_API_ENDPOINTS: Dict[str, str] = external_api_config["task_execution"]["endpoints"]
TASK_EXECUTION_API_METHODS: Dict[str, str] = external_api_config["task_execution"]["methods"]
TASK_EXECUTION_API_TIMEOUT: int = Field(default=30, env="TASK_EXECUTION_API_TIMEOUT")
# CORS设置
CORS_ORIGINS: List[str] = ["*"]
CORS_ALLOW_CREDENTIALS: bool = True
CORS_ALLOW_METHODS: List[str] = ["*"]
CORS_ALLOW_HEADERS: List[str] = ["*"]
# 日志设置
LOG_LEVEL: str = "INFO"
LOG_FILE: str = "logs/app.log"
LOG_FORMAT: str = "%(asctime)s - %(name)s - %(levelname)s - %(message)s"
# 任务调度设置
TASK_SCHEDULER_WORKER_COUNT: int = 5 # 任务调度器工作线程数量
# 脚本执行配置
SCRIPT_TIMEOUT: int = os.getenv("SCRIPT_TIMEOUT", 60) # 脚本执行超时时间(秒)
SCRIPT_MAX_WORKERS: int = os.getenv("SCRIPT_MAX_WORKERS", 5) # 脚本执行最大线程数
# 任务导出加密配置
TASK_EXPORT_ENCRYPTION_KEY: str = Field(default="vwed_task_export_secret_key", env="TASK_EXPORT_ENCRYPTION_KEY") # 导出加密密钥
TASK_EXPORT_ENCRYPTION_ALGORITHM: str = Field(default="AES", env="TASK_EXPORT_ENCRYPTION_ALGORITHM") # 加密算法
TASK_EXPORT_IV: str = Field(default="vwed1234task5678", env="TASK_EXPORT_IV") # 初始化向量
# 增强版任务调度器配置
TASK_SCHEDULER_MIN_WORKER_COUNT: int = 15 # 最小工作线程数
TASK_SCHEDULER_MAX_WORKER_COUNT: int = 30 # 最大工作线程数
TASK_SCHEDULER_QUEUE_COUNT: int = 3 # 队列数量
TASK_SCHEDULER_QUEUE_THRESHOLD_PERCENTILES: List[float] = [0.1, 0.3, 1.0] # 队列阈值百分比配置
TASK_SCHEDULER_WORKER_RATIOS: List[float] = [0.6, 0.3, 0.1] # 工作线程分配比例
TASK_SCHEDULER_TASK_TIMEOUT: int = 3600 # 任务超时时间(秒)
TASK_SCHEDULER_MAX_RETRY_COUNT: int = 3 # 最大重试次数
TASK_SCHEDULER_RETRY_DELAY: int = 60 # 重试基础延迟(秒)
TASK_SCHEDULER_BACKUP_INTERVAL: int = 300 # 备份间隔(秒)
TASK_SCHEDULER_BACKUP_DIR: str = os.path.join(DATA_DIR, "task_backups") # 备份目录
TASK_SCHEDULER_MAX_BACKUPS: int = 5 # 最大备份数
TASK_SCHEDULER_ZOMBIE_TASK_CHECK_INTERVAL: int = 60 # 僵尸任务检查间隔(秒)
TASK_SCHEDULER_CPU_THRESHOLD: float = 80.0 # CPU使用率阈值百分比
TASK_SCHEDULER_MEMORY_THRESHOLD: float = 80.0 # 内存使用率阈值(百分比)
TASK_SCHEDULER_AUTO_SCALE_INTERVAL: int = 120 # 自动扩缩容间隔(秒)
TASK_SCHEDULER_WORKER_HEARTBEAT_INTERVAL: int = 120 # 心跳间隔(秒)
# 告警同步配置
ALERT_SYNC_ENABLED: bool = Field(default=True, env="ALERT_SYNC_ENABLED") # 是否启用告警同步
ALERT_SYNC_HOST: str = Field(default="192.168.189.80", env="ALERT_SYNC_HOST") # 主系统IP
ALERT_SYNC_PORT: int = Field(default=8080, env="ALERT_SYNC_PORT") # 主系统端口
ALERT_SYNC_API_PATH: str = Field(default="/jeecg-boot/warning", env="ALERT_SYNC_API_PATH") # 告警API路径
ALERT_SYNC_TIMEOUT: int = Field(default=10, env="ALERT_SYNC_TIMEOUT") # 请求超时时间(秒)
ALERT_SYNC_RETRY_COUNT: int = Field(default=3, env="ALERT_SYNC_RETRY_COUNT") # 重试次数
ALERT_SYNC_RETRY_DELAY: int = Field(default=1, env="ALERT_SYNC_RETRY_DELAY") # 重试延迟(秒)
ALERT_SYNC_BATCH_SIZE: int = Field(default=10, env="ALERT_SYNC_BATCH_SIZE") # 批量发送大小
ALERT_SYNC_QUEUE_SIZE: int = Field(default=1000, env="ALERT_SYNC_QUEUE_SIZE") # 队列大小
ALERT_SYNC_MIN_LEVEL: str = Field(default="WARNING", env="ALERT_SYNC_MIN_LEVEL") # 最小告警级别
# 地图数据库区容量配置
# 密集库区容量配置
MAP_DENSE_STORAGE_BASE_CAPACITY: int = Field(default=50, env="MAP_DENSE_STORAGE_BASE_CAPACITY") # 密集库区基础容量
MAP_DENSE_STORAGE_CAPACITY_PER_POINT: int = Field(default=10, env="MAP_DENSE_STORAGE_CAPACITY_PER_POINT") # 密集库区每个动作点增加的容量
MAP_DENSE_STORAGE_LAYER_MULTIPLIER: float = Field(default=1.5, env="MAP_DENSE_STORAGE_LAYER_MULTIPLIER") # 密集库区分层倍数
# 一般库区容量配置
MAP_GENERAL_STORAGE_BASE_CAPACITY: int = Field(default=30, env="MAP_GENERAL_STORAGE_BASE_CAPACITY") # 一般库区基础容量
MAP_GENERAL_STORAGE_CAPACITY_PER_POINT: int = Field(default=15, env="MAP_GENERAL_STORAGE_CAPACITY_PER_POINT") # 一般库区每个动作点增加的容量
MAP_GENERAL_STORAGE_LAYER_MULTIPLIER: float = Field(default=1.2, env="MAP_GENERAL_STORAGE_LAYER_MULTIPLIER") # 一般库区分层倍数
@property
def DATABASE_URL(self) -> str:
"""构建数据库连接URL"""
if self.DB_DIALECT == 'sqlite':
return f"sqlite:///{self.DB_NAME}"
# 对用户名和密码进行URL编码避免特殊字符如@符号)造成解析错误
encoded_user = quote_plus(self.DB_USER)
encoded_password = quote_plus(self.DB_PASSWORD)
return f"{self.DB_DIALECT}+{self.DB_DRIVER}://{encoded_user}:{encoded_password}@{self.DB_HOST}:{self.DB_PORT}/{self.DB_NAME}?charset={self.DB_CHARSET}"
@property
def DATABASE_ARGS(self) -> Dict[str, Any]:
"""构建数据库连接参数"""
args = {
"pool_size": self.DB_POOL_SIZE,
"max_overflow": self.DB_MAX_OVERFLOW,
"pool_recycle": self.DB_POOL_RECYCLE,
"echo": self.DB_ECHO
}
return args
@property
def REDIS_URL(self) -> str:
"""构建Redis连接URL"""
if self.REDIS_PASSWORD:
# 对Redis密码进行URL编码避免特殊字符如@符号)造成解析错误
encoded_password = quote_plus(self.REDIS_PASSWORD)
return f"redis://:{encoded_password}@{self.REDIS_HOST}:{self.REDIS_PORT}/{self.REDIS_DB}"
return f"redis://{self.REDIS_HOST}:{self.REDIS_PORT}/{self.REDIS_DB}"
@property
def ALERT_SYNC_URL(self) -> str:
"""构建告警同步URL"""
return f"http://{self.ALERT_SYNC_HOST}:{self.ALERT_SYNC_PORT}{self.ALERT_SYNC_API_PATH}"
# 更新为Pydantic v2的配置方式
model_config = {
"env_file": ".env",
"env_file_encoding": "utf-8",
"case_sensitive": True
}
class DevelopmentConfig(BaseConfig):
"""开发环境配置"""
DEBUG: bool = True
DB_ECHO: bool = True # 开发环境输出SQL语句
LOG_LEVEL: str = "DEBUG"
SERVER_RELOAD: bool = BaseConfig().SERVER_RELOAD
STORAGE_API_MOCK_MODE: bool = True # 开发环境默认使用API模拟模式
ROBOT_API_MOCK_MODE: bool = True # 开发环境默认使用机器人API模拟模式
# 开发环境可以使用较小的容量配置便于测试
MAP_DENSE_STORAGE_BASE_CAPACITY: int = 20
MAP_DENSE_STORAGE_CAPACITY_PER_POINT: int = 5
MAP_GENERAL_STORAGE_BASE_CAPACITY: int = 15
MAP_GENERAL_STORAGE_CAPACITY_PER_POINT: int = 8
# 根据环境变量选择配置
def get_config():
"""根据环境变量获取配置"""
return DevelopmentConfig()
# 导出配置
settings = get_config()
# 添加LogConfig类以供logger.py使用
class LogConfig:
"""日志配置类"""
@staticmethod
def as_dict():
"""返回日志配置字典"""
return {
"level": settings.LOG_LEVEL,
"file": settings.LOG_FILE,
"format": settings.LOG_FORMAT
}