""" 系统配置模型(纵表设计) """ from sqlalchemy import Column, Integer, String, DateTime, SmallInteger, Text, Enum from sqlalchemy.orm import Session from app.database import Base, get_db from app.utils.timezone import beijing_now from typing import Dict, Any, Optional import json class SystemConfig(Base): """系统配置表(纵表设计)""" __tablename__ = "system_config" id = Column(Integer, primary_key=True, index=True, comment="主键ID") config_key = Column(String(100), nullable=False, index=True, comment="配置键名") config_value = Column(String(500), nullable=False, comment="配置值") config_type = Column(Enum('int', 'string', 'bool', 'json'), default='string', comment="配置值类型") app_name = Column(String(100), default='backend-admin', comment="所属应用:backend-admin、backend-client,多个用逗号拼接") description = Column(String(255), comment="配置描述") is_system = Column(SmallInteger, default=1, comment="是否为系统配置:1-是,0-否") is_editable = Column(SmallInteger, default=1, comment="是否可编辑:1-是,0-否") is_deleted = Column(SmallInteger, default=0, comment="是否删除:0-否,1-是") created_at = Column(DateTime, default=beijing_now, comment="创建时间") updated_at = Column(DateTime, default=beijing_now, onupdate=beijing_now, comment="更新时间") @staticmethod def get_config_value(config_key: str, default_value: Any = None) -> Any: """获取配置值""" try: db = next(get_db()) config = db.query(SystemConfig).filter( SystemConfig.config_key == config_key, SystemConfig.is_deleted == 0 ).first() if not config: return default_value # 根据类型转换值 if config.config_type == 'int': return int(config.config_value) elif config.config_type == 'bool': return config.config_value.lower() in ('1', 'true', 'yes') elif config.config_type == 'json': return json.loads(config.config_value) else: return config.config_value except Exception as e: print(f"获取配置失败 {config_key}: {e}") return default_value finally: if 'db' in locals(): db.close() @staticmethod def get_all_configs() -> Dict[str, Any]: """获取所有配置""" try: db = next(get_db()) configs = db.query(SystemConfig).filter( SystemConfig.is_deleted == 0 ).all() result = {} for config in configs: value = SystemConfig.get_config_value(config.config_key) result[config.config_key] = value return result except Exception as e: print(f"获取所有配置失败: {e}") return {} finally: if 'db' in locals(): db.close() @staticmethod def set_config_value(config_key: str, value: Any, config_type: str = 'string', description: str = '') -> bool: """设置配置值""" try: db = next(get_db()) # 查找现有配置 config = db.query(SystemConfig).filter( SystemConfig.config_key == config_key, SystemConfig.is_deleted == 0 ).first() # 转换值 if config_type == 'bool': str_value = '1' if value else '0' elif config_type == 'json': str_value = json.dumps(value) else: str_value = str(value) if config: # 更新现有配置 config.config_value = str_value config.config_type = config_type if description: config.description = description config.updated_at = beijing_now() else: # 创建新配置 new_config = SystemConfig( config_key=config_key, config_value=str_value, config_type=config_type, description=description, is_system=0, # 用户设置的配置 is_editable=1 ) db.add(new_config) db.commit() return True except Exception as e: print(f"设置配置失败 {config_key}: {e}") if 'db' in locals(): db.rollback() return False finally: if 'db' in locals(): db.close()