Coverage for core/repositories/settings_repository.py: 32.65%

1""" 

2系统设置数据仓库 

3""" 

4 

5import json 

6from datetime import datetime 

7from typing import Optional 

8 

9from core.models.settings import (AssetModeType, DataSourceConfig, 

10 LongportConfig, UserSettings) 

11from infrastructure.database.redis_client import RedisClient 

12 

13 

14class SettingsRepository: 

15 def __init__(self, redis_client: Optional[RedisClient] = None): 

16 self.redis = redis_client or RedisClient() 

17 self.settings_key_prefix = "user_settings:" 

18 

19 def _get_user_settings_key(self, user_id: str) -> str: 

20 """获取用户设置键""" 

21 return f"{self.settings_key_prefix}{user_id}" 

22 

23 def get_user_settings(self, user_id: str) -> Optional[UserSettings]: 

24 """获取用户设置""" 

25 settings_key = self._get_user_settings_key(user_id) 

26 data = self.redis.get(settings_key) 

27 

28 if data and isinstance(data, dict): 

29 return UserSettings(**data) 

30 return None 

31 

32 def save_user_settings(self, user_settings: UserSettings) -> bool: 

33 """保存用户设置""" 

34 try: 

35 settings_key = self._get_user_settings_key(user_settings.user_id) 

36 settings_dict = user_settings.model_dump() 

37 

38 print(f"保存用户设置: {settings_key}, 数据: {settings_dict}") 

39 

40 # 设置过期时间为30天 

41 self.redis.setex(settings_key, 86400 * 30, settings_dict) 

42 

43 # 验证保存是否成功 

44 saved_data = self.redis.get(settings_key) 

45 print(f"验证保存结果: {saved_data}") 

46 

47 return True 

48 except Exception as e: 

49 print(f"保存用户设置失败: {e}") 

50 return False 

51 

52 def update_data_source_config( 

53 self, user_id: str, data_source_config: DataSourceConfig 

54 ) -> bool: 

55 """更新数据源配置""" 

56 try: 

57 user_settings = self.get_user_settings(user_id) 

58 if not user_settings: 

59 user_settings = UserSettings(user_id=user_id) 

60 

61 user_settings.data_source_config = data_source_config 

62 user_settings.updated_at = datetime.utcnow().isoformat() 

63 

64 return self.save_user_settings(user_settings) 

65 except Exception as e: 

66 print(f"更新数据源配置失败: {e}") 

67 return False 

68 

69 def update_longport_config( 

70 self, user_id: str, longport_config: LongportConfig 

71 ) -> bool: 

72 """更新长桥证券配置""" 

73 try: 

74 user_settings = self.get_user_settings(user_id) 

75 if not user_settings: 

76 user_settings = UserSettings(user_id=user_id) 

77 

78 user_settings.longport_config = longport_config 

79 user_settings.updated_at = datetime.utcnow().isoformat() 

80 

81 return self.save_user_settings(user_settings) 

82 except Exception as e: 

83 print(f"更新长桥证券配置失败: {e}") 

84 return False 

85 

86 def get_data_source_config(self, user_id: str) -> Optional[DataSourceConfig]: 

87 """获取数据源配置""" 

88 user_settings = self.get_user_settings(user_id) 

89 return user_settings.data_source_config if user_settings else None 

90 

91 def get_longport_config(self, user_id: str) -> Optional[LongportConfig]: 

92 """获取长桥证券配置""" 

93 user_settings = self.get_user_settings(user_id) 

94 return user_settings.longport_config if user_settings else None 

95 

96 def get_asset_mode(self, user_id: str) -> AssetModeType: 

97 """获取资产模式""" 

98 user_settings = self.get_user_settings(user_id) 

99 # print(f"获取资产模式: user_id={user_id}, settings={user_settings}") # 已禁用调试输出 

100 return user_settings.asset_mode if user_settings else AssetModeType.TRADING 

101 

102 def update_asset_mode(self, user_id: str, asset_mode: AssetModeType) -> bool: 

103 """更新资产模式""" 

104 try: 

105 user_settings = self.get_user_settings(user_id) 

106 if not user_settings: 

107 user_settings = UserSettings(user_id=user_id) 

108 

109 user_settings.asset_mode = asset_mode 

110 user_settings.updated_at = datetime.utcnow().isoformat() 

111 return self.save_user_settings(user_settings) 

112 except Exception as e: 

113 print(f"更新资产模式失败: {e}") 

114 return False 

115 

116 def update_data_source_type(self, user_id: str, data_source_type: str) -> bool: 

117 """更新数据源类型""" 

118 try: 

119 user_settings = self.get_user_settings(user_id) 

120 if not user_settings: 

121 user_settings = UserSettings(user_id=user_id) 

122 

123 if not user_settings.data_source_config: 

124 user_settings.data_source_config = DataSourceConfig() 

125 

126 # 设置数据源类型 

127 user_settings.data_source_config.type = data_source_type 

128 user_settings.updated_at = datetime.utcnow().isoformat() 

129 

130 return self.save_user_settings(user_settings) 

131 except Exception as e: 

132 print(f"更新数据源类型失败: {e}") 

133 return False 

134 

135 def get_data_source_type(self, user_id: str) -> Optional[str]: 

136 """获取数据源类型""" 

137 user_settings = self.get_user_settings(user_id) 

138 if user_settings and user_settings.data_source_config: 

139 return user_settings.data_source_config.type 

140 return None 

141 

142 def get_user_setting(self, user_id: str, setting_key: str) -> Optional[str]: 

143 """获取用户的特定设置""" 

144 user_settings = self.get_user_settings(user_id) 

145 if not user_settings: 

146 return None 

147 

148 # 根据设置键返回对应的值 

149 if setting_key == "data_source_type": 

150 return self.get_data_source_type(user_id) 

151 elif setting_key == "asset_mode": 

152 return user_settings.asset_mode.value if user_settings.asset_mode else None 

153 

154 return None