Coverage for core/services/settings_service.py: 42.00%

1""" 

2系统设置服务 

3""" 

4 

5from typing import Optional 

6 

7from core.data_source.adapters.trade_adapter import TradeDataSourceAdapter 

8from core.data_source.factories.client_factory import ClientFactory 

9from core.data_source.factories.config_factory import ConfigFactory 

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

11 LongportConfig) 

12from core.models.user import User 

13from core.repositories.settings_repository import SettingsRepository 

14 

15 

16class SettingsService: 

17 def __init__(self, settings_repository: Optional[SettingsRepository] = None): 

18 self.settings_repository = settings_repository or SettingsRepository() 

19 

20 def get_data_source_config(self, current_user: User) -> Optional[DataSourceConfig]: 

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

22 return self.settings_repository.get_data_source_config(current_user.id) 

23 

24 def update_data_source_config( 

25 self, current_user: User, data_source_config: DataSourceConfig 

26 ) -> bool: 

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

28 return self.settings_repository.update_data_source_config( 

29 current_user.id, data_source_config 

30 ) 

31 

32 def get_longport_config(self, current_user: User) -> Optional[LongportConfig]: 

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

34 return self.settings_repository.get_longport_config(current_user.id) 

35 

36 def update_longport_config( 

37 self, current_user: User, longport_config: LongportConfig 

38 ) -> bool: 

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

40 return self.settings_repository.update_longport_config( 

41 current_user.id, longport_config 

42 ) 

43 

44 def get_asset_mode(self, current_user: User) -> AssetModeType: 

45 """获取资产模式""" 

46 return self.settings_repository.get_asset_mode(current_user.id) 

47 

48 def update_asset_mode(self, current_user: User, asset_mode: AssetModeType) -> bool: 

49 """更新资产模式""" 

50 return self.settings_repository.update_asset_mode(current_user.id, asset_mode) 

51 

52 def refresh_longport_token(self, current_user: User) -> dict: 

53 """刷新长桥访问令牌""" 

54 try: 

55 # 使用TradeDataSourceAdapter刷新token 

56 trade_adapter = TradeDataSourceAdapter(current_user.id) 

57 result = trade_adapter.refresh_access_token() 

58 

59 if result and result.get("success"): 

60 new_token = result.get("new_token") 

61 if new_token: 

62 # 获取当前长桥配置 

63 current_config = self.get_longport_config(current_user) 

64 if current_config: 

65 # 更新access_token 

66 updated_config = LongportConfig( 

67 app_key=current_config.app_key, 

68 app_secret=current_config.app_secret, 

69 access_token=new_token, 

70 language=current_config.language, 

71 enable_overnight=current_config.enable_overnight, 

72 ) 

73 

74 # 保存到数据库 

75 success = self.update_longport_config( 

76 current_user, updated_config 

77 ) 

78 if success: 

79 # 清理用户相关缓存 

80 self._clear_user_caches(current_user.id) 

81 return { 

82 "success": True, 

83 "message": "Token刷新成功", 

84 "new_token": new_token, 

85 } 

86 else: 

87 return { 

88 "success": False, 

89 "message": "Token刷新成功但保存到数据库失败", 

90 } 

91 else: 

92 return {"success": False, "message": "未找到长桥配置"} 

93 else: 

94 return {"success": False, "message": "Token刷新成功但未返回新token"} 

95 else: 

96 return result or {"success": False, "message": "Token刷新失败"} 

97 except Exception as e: 

98 return {"success": False, "message": f"Token刷新失败: {str(e)}"} 

99 

100 def _clear_user_caches(self, user_id: str): 

101 """清理用户相关缓存""" 

102 try: 

103 # 清理配置缓存 

104 config_factory = ConfigFactory() 

105 config_factory.clear_cache(user_id) 

106 

107 # 清理客户端缓存 

108 client_factory = ClientFactory() 

109 client_factory.clear_cache(user_id) 

110 except Exception as e: 

111 print(f"清理用户缓存失败: {e}")