buff2.0/backend-admin/app/models/system_config.py

"""
系统配置模型（纵表设计）
"""
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()