Coverage for api/services/notifications/builder.py: 19%

1import copy 

2import re 

3from typing import Any, TypeAlias 

4 

5from errors import log_error 

6from logging_config import logger 

7from sotrans_models.models.services.notification_presets import ( 

8 NOTIFICATION_PRESETS_MAPPING, 

9) 

10from sotrans_models.models.services.notifications import ( 

11 CastType, 

12 NotificationCreateBatchModel, 

13 NotificationCreateModel, 

14 NotificationCreateMulticastModel, 

15 NotificationPresetModel, 

16 NotificationUserModel, 

17) 

18from sotrans_models.models.users import SotransUserDBModel 

19 

20 

21def _get_object_field(object_: dict | list, field_path: list[str]) -> str: 

22 try: 

23 field_value: Any = "(не найдено)" 

24 if isinstance(object_, dict): 

25 field_value = object_.get(field_path[0]) 

26 elif isinstance(object_, list): 

27 index = int(field_path[0]) 

28 field_value = object_[index] 

29 if isinstance(field_value, dict) or isinstance(field_value, list): 

30 return _get_object_field(field_value, field_path[1:]) 

31 logger.info(object_) 

32 logger.info(field_value) 

33 if not field_value: 

34 return "(не найдено)" 

35 return str(field_value) 

36 except (IndexError, ValueError): 

37 logger.error( 

38 "Invalid field_path {0} for object {1}".format(field_path, object_) 

39 ) 

40 return "(не найдено)" 

41 

42 

43UserType: TypeAlias = NotificationUserModel | dict | SotransUserDBModel 

44UsersType: TypeAlias = list[UserType] 

45 

46 

47def _fill_text_from_object(object_: dict, text: str) -> str: 

48 matches = re.finditer(r"\*([-\w]+,?)+\*", text, re.MULTILINE) 

49 for match in matches: 

50 match_text = str(match.group()) 

51 field_path = match_text[1:-1].split(",") 

52 object_field_value = _get_object_field(object_, field_path) 

53 logger.info(object_field_value) 

54 text = text.replace( 

55 match_text, _get_object_field(object_, field_path), 1 

56 ) 

57 return text 

58 

59 

60def _get_user_dict(user: UserType): 

61 if isinstance(user, NotificationUserModel) or isinstance( 

62 user, SotransUserDBModel 

63 ): 

64 return user.model_dump(format_ids=False) 

65 elif isinstance(user, dict): 

66 return user 

67 raise ValueError( 

68 "User must be of type NotificationUserModel, SotransUserDBModel or dict" 

69 ) 

70 

71 

72def _get_users_dict(users: UsersType): 

73 return [_get_user_dict(user) for user in users] 

74 

75 

76DOC_TRANSLATION_TABLE = { 

77 "DriverDocumentType.drivers_license": "водительское удостоверение", 

78 "DriverDocumentType.passport": "паспорт", 

79 "TruckDocumentType.sts": "СТС тягача", 

80 "TruckDocumentType.lease_contract": "договор аренды тягача", 

81 "TrailerDocumentType.sts": "СТС прицепа", 

82 "TrailerDocumentType.lease_contract": "договор аренды прицепа", 

83 "OrganizationDocumentType.tax_authorities_certificate": "свидетельство о постановке на учет в налоговом органе", 

84 "OrganizationDocumentType.registration_certificate": "свидетельство о государственной регистрации юридического лица", 

85 "OrganizationDocumentType.director_or_proprietor_passport": "Копия паспорта генерального директора или индивидуального предпринимателя", 

86 "OrganizationDocumentType.director_appointment": "приказ о назначении директора", 

87 "OrganizationDocumentType.decision_protocol": "протокол решения общего собрания учредителей о создании юр. лица", 

88 "OrganizationDocumentType.partner_card": "карта партнёра", 

89 "RequestDocumentType.order_request": "запрос на заказ", 

90 "RequestDocumentType.draft": "драфт", 

91 "RequestDocumentType.order_request_draft": "драфт запроса на заказ", 

92 "UserDocumentType.photo": "фото пользователя", 

93} 

94 

95 

96def _doc_verification_translation(payload_dict: dict): 

97 if payload_dict.get("type") not in ( 

98 "success_document_verification", 

99 "failed_document_verification", 

100 ): 

101 return 

102 for k in DOC_TRANSLATION_TABLE: 

103 if k in payload_dict["title"]: 

104 old_title: str = payload_dict["title"] 

105 new_title = old_title.replace(k, DOC_TRANSLATION_TABLE[k]) 

106 payload_dict["title"] = new_title 

107 return 

108 

109 

110# Inefficient due to object copies, but this is an experiment 

111class NotificationBuilder: 

112 # not scary until class copies itself 

113 # payload_dict: dict = {} 

114 preset: NotificationPresetModel 

115 

116 def __init__(self): 

117 self.payload_dict = {} 

118 

119 def _copy(self): 

120 return copy.deepcopy(self) 

121 

122 def _fill_payload_fields(self, object_: dict): 

123 if self.payload_dict.get("title"): 

124 self.payload_dict["title"] = _fill_text_from_object( 

125 object_, self.payload_dict["title"] 

126 ) 

127 if self.payload_dict.get("description"): 

128 self.payload_dict["description"] = _fill_text_from_object( 

129 object_, self.payload_dict["description"] 

130 ) 

131 

132 def with_preset(self, preset: NotificationPresetModel): 

133 copy_self = self._copy() 

134 copy_self.preset = preset 

135 copy_self.payload_dict = { 

136 **copy_self.payload_dict, 

137 **copy_self.preset.default_info_model.model_dump(), 

138 } 

139 return copy_self 

140 

141 def with_preset_type(self, preset_type: str): 

142 if preset_type not in NOTIFICATION_PRESETS_MAPPING: 

143 raise ValueError( 

144 f"preset_type {preset_type} is not in NOTIFICATION_PRESET_MAPPING" 

145 ) 

146 return self.with_preset(NOTIFICATION_PRESETS_MAPPING[preset_type]) 

147 

148 def with_object(self, object_: dict, fill_payload_fields: bool = True): 

149 copy_self = self._copy() 

150 copy_self.payload_dict["object"] = object_ 

151 if fill_payload_fields: 

152 copy_self._fill_payload_fields(object_) 

153 return copy_self 

154 

155 def with_user(self, user: UserType): 

156 copy_self = self._copy() 

157 copy_self.payload_dict["user"] = _get_user_dict(user) 

158 print(copy_self.payload_dict) 

159 return copy_self 

160 

161 def with_users( 

162 self, 

163 users: UsersType, 

164 ): 

165 copy_self = self._copy() 

166 copy_self.payload_dict["users"] = [ 

167 _get_user_dict(user) for user in users 

168 ] 

169 return copy_self 

170 

171 def with_cast_one(self, cast_type: CastType, recipient: UserType | None): 

172 if not recipient: 

173 log_error( 

174 f"No recipient provided for builder method with_cast_one, ${self.payload_dict}" 

175 ) 

176 return self 

177 copy_self = self._copy() 

178 current_casts = copy_self.payload_dict.get("casts", []) 

179 current_casts.append( 

180 {"cast_type": cast_type, "recipient": _get_user_dict(recipient)} 

181 ) 

182 copy_self.payload_dict["casts"] = current_casts 

183 return copy_self 

184 

185 def with_cast_many( 

186 self, cast_type: CastType, recipients: UsersType | None 

187 ): 

188 if not recipients: 

189 # log_warning( 

190 # f"No recipient provided for builder method with_cast_many, ${self.payload_dict}" 

191 # ) 

192 return self 

193 copy_self = self._copy() 

194 current_casts = copy_self.payload_dict.get("casts", []) 

195 current_casts.append( 

196 {"cast_type": cast_type, "recipients": _get_users_dict(recipients)} 

197 ) 

198 copy_self.payload_dict["casts"] = current_casts 

199 return copy_self 

200 

201 def build( 

202 self, 

203 ) -> ( 

204 NotificationCreateBatchModel 

205 | NotificationCreateModel 

206 | NotificationCreateMulticastModel 

207 | None 

208 ): 

209 _doc_verification_translation(self.payload_dict) 

210 if "user" in self.payload_dict: 

211 return NotificationCreateModel(**self.payload_dict) 

212 elif "users" in self.payload_dict: 

213 return NotificationCreateBatchModel(**self.payload_dict) 

214 elif "casts" in self.payload_dict: 

215 return NotificationCreateMulticastModel(**self.payload_dict) 

216 return None 

217 # raise ValueError("No user or users") 

218 

219 

220# print(NotificationBuilder().with_user(SotransUserDBModel(id=UUID(int=0),)).with_preset_type("success_inn_verification").build()) 

221# print(NotificationBuilder().with_users([SotransUserDBModel(id=UUID(int=0),)]).with_preset_type("success_inn_verification").build())