74 lines
2.4 KiB
Python
74 lines
2.4 KiB
Python
"""
|
||
配置相关数据访问层
|
||
"""
|
||
|
||
from typing import List, Dict, Optional
|
||
from sqlalchemy.orm import Session
|
||
from models.sur_config import SurConfig, SurConfigBase
|
||
from config import settings
|
||
|
||
|
||
class SurConfigRepository:
|
||
"""配置数据访问类"""
|
||
|
||
def __init__(self, db: Session):
|
||
self.db = db
|
||
|
||
def get_face_config_params(self) -> Dict[str, str]:
|
||
"""
|
||
获取人脸识别配置参数
|
||
|
||
返回:
|
||
配置参数字典 {config_key: config_value}
|
||
"""
|
||
try:
|
||
# 查询全局人脸配置
|
||
config_records = self.db.query(SurConfig).filter(
|
||
SurConfig.scope == settings.SUR_CONFIG_SCOPE_GLOBAL,
|
||
SurConfig.config_type == settings.SUR_CONFIG_TYPE_FACE
|
||
).all()
|
||
|
||
# 查询配置组对应的基础配置
|
||
config_group_ids = [record.config_group_id for record in config_records if record.config_group_id]
|
||
|
||
if config_group_ids:
|
||
base_configs = self.db.query(SurConfigBase).filter(
|
||
SurConfigBase.group_id.in_(config_group_ids)
|
||
).all()
|
||
|
||
# 合并配置
|
||
config_dict = {}
|
||
for record in config_records:
|
||
if record.config_key and record.config_value:
|
||
config_dict[record.config_key] = record.config_value
|
||
|
||
return config_dict
|
||
|
||
return {}
|
||
|
||
except Exception as e:
|
||
print(f"获取人脸配置参数失败: {e}")
|
||
return {}
|
||
|
||
def get_face_config_value(self, config_key: str) -> Optional[str]:
|
||
"""
|
||
获取指定配置键的值
|
||
|
||
参数:
|
||
config_key: 配置键
|
||
|
||
返回:
|
||
配置值，如果不存在返回None
|
||
"""
|
||
try:
|
||
config_record = self.db.query(SurConfig).filter(
|
||
SurConfig.scope == settings.SUR_CONFIG_SCOPE_GLOBAL,
|
||
SurConfig.config_type == settings.SUR_CONFIG_TYPE_FACE,
|
||
SurConfig.config_key == config_key
|
||
).first()
|
||
|
||
return config_record.config_value if config_record else None
|
||
|
||
except Exception as e:
|
||
print(f"获取配置值失败 {config_key}: {e}")
|
||
return None |