Coverage for documents/workflows/mutations.py: 5%

1import logging 

2 

3from django.utils import timezone 

4from guardian.shortcuts import remove_perm 

5 

6from documents.data_models import DocumentMetadataOverrides 

7from documents.models import CustomFieldInstance 

8from documents.models import Document 

9from documents.models import WorkflowAction 

10from documents.permissions import set_permissions_for_object 

11from documents.templating.workflows import parse_w_workflow_placeholders 

12 

13logger = logging.getLogger("paperless.workflows.mutations") 

14 

15 

16def apply_assignment_to_document( 

17 action: WorkflowAction, 

18 document: Document, 

19 logging_group, 

20): 

21 """ 

22 Apply assignment actions to a Document instance. 

23 

24 action: WorkflowAction, annotated with 'has_assign_*' boolean fields 

25 """ 

26 if action.has_assign_tags: 

27 # Apply to a freshly-fetched instance rather than the shared `document`. 

28 # Document.tags.add() fires an m2m_changed signal that ultimately calls 

29 # instance.refresh_from_db(), which would discard any other unsaved 

30 # assignment fields (e.g. storage_path) already staged on `document`. 

31 Document.objects.get(pk=document.pk).add_nested_tags(action.assign_tags.all()) 

32 

33 if action.assign_correspondent: 

34 document.correspondent = action.assign_correspondent 

35 

36 if action.assign_document_type: 

37 document.document_type = action.assign_document_type 

38 

39 if action.assign_storage_path: 

40 document.storage_path = action.assign_storage_path 

41 

42 if action.assign_owner: 

43 document.owner = action.assign_owner 

44 

45 if action.assign_title: 

46 try: 

47 title = parse_w_workflow_placeholders( 

48 action.assign_title, 

49 document.correspondent.name if document.correspondent else "", 

50 document.document_type.name if document.document_type else "", 

51 document.owner.username if document.owner else "", 

52 timezone.localtime(document.added), 

53 document.original_filename or "", 

54 document.filename or "", 

55 document.created, 

56 "", # dont pass the title to avoid recursion 

57 "", # no urls in titles 

58 document.pk, 

59 ) 

60 if title: 

61 document.title = title 

62 except Exception: # pragma: no cover 

63 logger.exception( 

64 f"Error occurred parsing title assignment '{action.assign_title}', falling back to original", 

65 extra={"group": logging_group}, 

66 ) 

67 

68 if any( 

69 [ 

70 action.has_assign_view_users, 

71 action.has_assign_view_groups, 

72 action.has_assign_change_users, 

73 action.has_assign_change_groups, 

74 ], 

75 ): 

76 permissions = { 

77 "view": { 

78 "users": action.assign_view_users.values_list("id", flat=True), 

79 "groups": action.assign_view_groups.values_list("id", flat=True), 

80 }, 

81 "change": { 

82 "users": action.assign_change_users.values_list("id", flat=True), 

83 "groups": action.assign_change_groups.values_list("id", flat=True), 

84 }, 

85 } 

86 set_permissions_for_object( 

87 permissions=permissions, 

88 object=document, 

89 merge=True, 

90 ) 

91 

92 if action.has_assign_custom_fields: 

93 for field in action.assign_custom_fields.all(): 

94 value_field_name = CustomFieldInstance.get_value_field_name( 

95 data_type=field.data_type, 

96 ) 

97 args = { 

98 value_field_name: action.assign_custom_fields_values.get( 

99 str(field.pk), 

100 None, 

101 ), 

102 } 

103 # for some reason update_or_create doesn't work here 

104 instance = CustomFieldInstance.objects.filter( 

105 field=field, 

106 document=document, 

107 ).first() 

108 # empty string is indistinguishable from no value in the UI 

109 if instance and args[value_field_name] not in (None, ""): 

110 setattr(instance, value_field_name, args[value_field_name]) 

111 instance.save() 

112 elif not instance: 

113 CustomFieldInstance.objects.create( 

114 **args, 

115 field=field, 

116 document=document, 

117 ) 

118 

119 

120def apply_assignment_to_overrides( 

121 action: WorkflowAction, 

122 overrides: DocumentMetadataOverrides, 

123): 

124 """ 

125 Apply assignment actions to DocumentMetadataOverrides. 

126 

127 action: WorkflowAction, annotated with 'has_assign_*' boolean fields 

128 """ 

129 if action.has_assign_tags: 

130 if overrides.tag_ids is None: 

131 overrides.tag_ids = [] 

132 tag_ids_to_add: set[int] = set() 

133 for tag in action.assign_tags.all(): 

134 tag_ids_to_add.add(tag.pk) 

135 tag_ids_to_add.update(int(pk) for pk in tag.get_ancestors_pks()) 

136 

137 overrides.tag_ids = list(set(overrides.tag_ids) | tag_ids_to_add) 

138 

139 if action.assign_correspondent: 

140 overrides.correspondent_id = action.assign_correspondent.pk 

141 

142 if action.assign_document_type: 

143 overrides.document_type_id = action.assign_document_type.pk 

144 

145 if action.assign_storage_path: 

146 overrides.storage_path_id = action.assign_storage_path.pk 

147 

148 if action.assign_owner: 

149 overrides.owner_id = action.assign_owner.pk 

150 

151 if action.assign_title: 

152 overrides.title = action.assign_title 

153 

154 if any( 

155 [ 

156 action.has_assign_view_users, 

157 action.has_assign_view_groups, 

158 action.has_assign_change_users, 

159 action.has_assign_change_groups, 

160 ], 

161 ): 

162 overrides.view_users = list( 

163 set( 

164 (overrides.view_users or []) 

165 + list(action.assign_view_users.values_list("id", flat=True)), 

166 ), 

167 ) 

168 overrides.view_groups = list( 

169 set( 

170 (overrides.view_groups or []) 

171 + list(action.assign_view_groups.values_list("id", flat=True)), 

172 ), 

173 ) 

174 overrides.change_users = list( 

175 set( 

176 (overrides.change_users or []) 

177 + list(action.assign_change_users.values_list("id", flat=True)), 

178 ), 

179 ) 

180 overrides.change_groups = list( 

181 set( 

182 (overrides.change_groups or []) 

183 + list(action.assign_change_groups.values_list("id", flat=True)), 

184 ), 

185 ) 

186 

187 if action.has_assign_custom_fields: 

188 if overrides.custom_fields is None: 

189 overrides.custom_fields = {} 

190 overrides.custom_fields.update( 

191 { 

192 field.pk: action.assign_custom_fields_values.get( 

193 str(field.pk), 

194 None, 

195 ) 

196 for field in action.assign_custom_fields.all() 

197 }, 

198 ) 

199 

200 

201def apply_removal_to_document( 

202 action: WorkflowAction, 

203 document: Document, 

204): 

205 """ 

206 Apply removal actions to a Document instance. 

207 

208 action: WorkflowAction, annotated with 'has_remove_*' boolean fields 

209 """ 

210 

211 if action.remove_all_tags: 

212 Document.objects.get(pk=document.pk).tags.clear() 

213 else: 

214 tag_ids_to_remove: set[int] = set() 

215 for tag in action.remove_tags.all(): 

216 tag_ids_to_remove.add(tag.pk) 

217 tag_ids_to_remove.update(int(pk) for pk in tag.get_descendants_pks()) 

218 

219 if tag_ids_to_remove: 

220 Document.objects.get(pk=document.pk).tags.remove(*tag_ids_to_remove) 

221 

222 if action.remove_all_correspondents or ( 

223 document.correspondent 

224 and action.remove_correspondents.filter(pk=document.correspondent.pk).exists() 

225 ): 

226 document.correspondent = None 

227 

228 if action.remove_all_document_types or ( 

229 document.document_type 

230 and action.remove_document_types.filter(pk=document.document_type.pk).exists() 

231 ): 

232 document.document_type = None 

233 

234 if action.remove_all_storage_paths or ( 

235 document.storage_path 

236 and action.remove_storage_paths.filter(pk=document.storage_path.pk).exists() 

237 ): 

238 document.storage_path = None 

239 

240 if action.remove_all_owners or ( 

241 document.owner and action.remove_owners.filter(pk=document.owner.pk).exists() 

242 ): 

243 document.owner = None 

244 

245 if action.remove_all_permissions: 

246 permissions = { 

247 "view": {"users": [], "groups": []}, 

248 "change": {"users": [], "groups": []}, 

249 } 

250 set_permissions_for_object( 

251 permissions=permissions, 

252 object=document, 

253 merge=False, 

254 ) 

255 

256 if any( 

257 [ 

258 action.has_remove_view_users, 

259 action.has_remove_view_groups, 

260 action.has_remove_change_users, 

261 action.has_remove_change_groups, 

262 ], 

263 ): 

264 for user in action.remove_view_users.all(): 

265 remove_perm("view_document", user, document) 

266 for user in action.remove_change_users.all(): 

267 remove_perm("change_document", user, document) 

268 for group in action.remove_view_groups.all(): 

269 remove_perm("view_document", group, document) 

270 for group in action.remove_change_groups.all(): 

271 remove_perm("change_document", group, document) 

272 

273 if action.remove_all_custom_fields: 

274 CustomFieldInstance.objects.filter(document=document).hard_delete() 

275 elif action.has_remove_custom_fields: 

276 CustomFieldInstance.objects.filter( 

277 field__in=action.remove_custom_fields.all(), 

278 document=document, 

279 ).hard_delete() 

280 

281 

282def apply_removal_to_overrides( 

283 action: WorkflowAction, 

284 overrides: DocumentMetadataOverrides, 

285): 

286 """ 

287 Apply removal actions to DocumentMetadataOverrides. 

288 

289 action: WorkflowAction, annotated with 'has_remove_*' boolean fields 

290 """ 

291 if action.remove_all_tags: 

292 overrides.tag_ids = None 

293 elif overrides.tag_ids: 

294 tag_ids_to_remove: set[int] = set() 

295 for tag in action.remove_tags.all(): 

296 tag_ids_to_remove.add(tag.pk) 

297 tag_ids_to_remove.update(int(pk) for pk in tag.get_descendants_pks()) 

298 

299 overrides.tag_ids = [t for t in overrides.tag_ids if t not in tag_ids_to_remove] 

300 

301 if action.remove_all_correspondents or ( 

302 overrides.correspondent_id 

303 and action.remove_correspondents.filter(pk=overrides.correspondent_id).exists() 

304 ): 

305 overrides.correspondent_id = None 

306 

307 if action.remove_all_document_types or ( 

308 overrides.document_type_id 

309 and action.remove_document_types.filter(pk=overrides.document_type_id).exists() 

310 ): 

311 overrides.document_type_id = None 

312 

313 if action.remove_all_storage_paths or ( 

314 overrides.storage_path_id 

315 and action.remove_storage_paths.filter(pk=overrides.storage_path_id).exists() 

316 ): 

317 overrides.storage_path_id = None 

318 

319 if action.remove_all_owners or ( 

320 overrides.owner_id 

321 and action.remove_owners.filter(pk=overrides.owner_id).exists() 

322 ): 

323 overrides.owner_id = None 

324 

325 if action.remove_all_permissions: 

326 overrides.view_users = None 

327 overrides.view_groups = None 

328 overrides.change_users = None 

329 overrides.change_groups = None 

330 elif any( 

331 [ 

332 action.has_remove_view_users, 

333 action.has_remove_view_groups, 

334 action.has_remove_change_users, 

335 action.has_remove_change_groups, 

336 ], 

337 ): 

338 if overrides.view_users: 

339 for user in action.remove_view_users.filter(pk__in=overrides.view_users): 

340 overrides.view_users.remove(user.pk) 

341 if overrides.change_users: 

342 for user in action.remove_change_users.filter( 

343 pk__in=overrides.change_users, 

344 ): 

345 overrides.change_users.remove(user.pk) 

346 if overrides.view_groups: 

347 for group in action.remove_view_groups.filter(pk__in=overrides.view_groups): 

348 overrides.view_groups.remove(group.pk) 

349 if overrides.change_groups: 

350 for group in action.remove_change_groups.filter( 

351 pk__in=overrides.change_groups, 

352 ): 

353 overrides.change_groups.remove(group.pk) 

354 

355 if action.remove_all_custom_fields: 

356 overrides.custom_fields = None 

357 elif action.has_remove_custom_fields and overrides.custom_fields: 

358 for field in action.remove_custom_fields.filter( 

359 pk__in=overrides.custom_fields.keys(), 

360 ): 

361 overrides.custom_fields.pop(field.pk, None)