Coverage for documents/data_models.py: 32%

1import dataclasses 

2import datetime 

3from enum import IntEnum 

4from pathlib import Path 

5from typing import TypedDict 

6 

7import magic 

8from guardian.shortcuts import get_groups_with_perms 

9from guardian.shortcuts import get_users_with_perms 

10 

11 

12class StoredBarcode(TypedDict): 

13 """ 

14 A detected barcode as it is stored with a document 

15 """ 

16 

17 page: int # 1-indexed 

18 value: str 

19 format: str # a DocumentBarcode.Format value 

20 

21 

22@dataclasses.dataclass 

23class DocumentMetadataOverrides: 

24 """ 

25 Manages overrides for document fields which normally would 

26 be set from content or matching. All fields default to None, 

27 meaning no override is happening 

28 """ 

29 

30 filename: str | None = None 

31 title: str | None = None 

32 correspondent_id: int | None = None 

33 document_type_id: int | None = None 

34 tag_ids: list[int] | None = None 

35 storage_path_id: int | None = None 

36 created: datetime.date | None = None 

37 asn: int | None = None 

38 owner_id: int | None = None 

39 view_users: list[int] | None = None 

40 view_groups: list[int] | None = None 

41 change_users: list[int] | None = None 

42 change_groups: list[int] | None = None 

43 custom_fields: dict | None = None 

44 skip_asn_if_exists: bool = False 

45 version_label: str | None = None 

46 actor_id: int | None = None 

47 remote_ocr: bool = False 

48 barcodes: list[StoredBarcode] | None = None 

49 

50 def update(self, other: "DocumentMetadataOverrides") -> "DocumentMetadataOverrides": 

51 """ 

52 Merges two DocumentMetadataOverrides objects such that object B's overrides 

53 are applied to object A or merged if multiple are accepted. 

54 

55 The update is an in-place modification of self 

56 """ 

57 # only if empty 

58 if other.title is not None: 

59 self.title = other.title 

60 if other.correspondent_id is not None: 

61 self.correspondent_id = other.correspondent_id 

62 if other.document_type_id is not None: 

63 self.document_type_id = other.document_type_id 

64 if other.storage_path_id is not None: 

65 self.storage_path_id = other.storage_path_id 

66 if other.owner_id is not None: 

67 self.owner_id = other.owner_id 

68 if other.actor_id is not None: 

69 self.actor_id = other.actor_id 

70 if other.skip_asn_if_exists: 

71 self.skip_asn_if_exists = True 

72 if other.remote_ocr: 

73 self.remote_ocr = True 

74 if other.version_label is not None: 

75 self.version_label = other.version_label 

76 

77 # merge 

78 if self.tag_ids is None: 

79 self.tag_ids = other.tag_ids 

80 elif other.tag_ids is not None: 

81 self.tag_ids.extend(other.tag_ids) 

82 self.tag_ids = list(set(self.tag_ids)) 

83 

84 if self.view_users is None: 

85 self.view_users = other.view_users 

86 elif other.view_users is not None: 

87 self.view_users.extend(other.view_users) 

88 self.view_users = list(set(self.view_users)) 

89 

90 if self.view_groups is None: 

91 self.view_groups = other.view_groups 

92 elif other.view_groups is not None: 

93 self.view_groups.extend(other.view_groups) 

94 self.view_groups = list(set(self.view_groups)) 

95 

96 if self.change_users is None: 

97 self.change_users = other.change_users 

98 elif other.change_users is not None: 

99 self.change_users.extend(other.change_users) 

100 self.change_users = list(set(self.change_users)) 

101 

102 if self.change_groups is None: 

103 self.change_groups = other.change_groups 

104 elif other.change_groups is not None: 

105 self.change_groups.extend(other.change_groups) 

106 self.change_groups = list(set(self.change_groups)) 

107 

108 if self.custom_fields is None: 

109 self.custom_fields = other.custom_fields 

110 elif other.custom_fields is not None: 

111 self.custom_fields.update(other.custom_fields) 

112 

113 return self 

114 

115 @staticmethod 

116 def from_document(doc) -> "DocumentMetadataOverrides": 

117 """ 

118 Fills in the overrides from a document object 

119 """ 

120 overrides = DocumentMetadataOverrides() 

121 overrides.title = doc.title 

122 overrides.correspondent_id = doc.correspondent.id if doc.correspondent else None 

123 overrides.document_type_id = doc.document_type.id if doc.document_type else None 

124 overrides.storage_path_id = doc.storage_path.id if doc.storage_path else None 

125 overrides.owner_id = doc.owner.id if doc.owner else None 

126 overrides.tag_ids = list(doc.tags.values_list("id", flat=True)) 

127 overrides.created = doc.created 

128 

129 overrides.view_users = list( 

130 get_users_with_perms( 

131 doc, 

132 only_with_perms_in=["view_document"], 

133 ).values_list("id", flat=True), 

134 ) 

135 overrides.change_users = list( 

136 get_users_with_perms( 

137 doc, 

138 only_with_perms_in=["change_document"], 

139 ).values_list("id", flat=True), 

140 ) 

141 overrides.custom_fields = { 

142 custom_field.field.id: custom_field.value 

143 for custom_field in doc.custom_fields.all() 

144 } 

145 

146 groups_with_perms = get_groups_with_perms( 

147 doc, 

148 attach_perms=True, 

149 ) 

150 overrides.view_groups = [ 

151 group.id 

152 for group in groups_with_perms 

153 if "view_document" in groups_with_perms[group] 

154 ] 

155 overrides.change_groups = [ 

156 group.id 

157 for group in groups_with_perms 

158 if "change_document" in groups_with_perms[group] 

159 ] 

160 

161 return overrides 

162 

163 

164class DocumentSource(IntEnum): 

165 """ 

166 The source of an incoming document. May have other uses in the future 

167 """ 

168 

169 ConsumeFolder = 1 

170 ApiUpload = 2 

171 MailFetch = 3 

172 WebUI = 4 

173 

174 

175@dataclasses.dataclass 

176class ConsumableDocument: 

177 """ 

178 Encapsulates an incoming document, either from consume folder, API upload 

179 or mail fetching and certain useful operations on it. 

180 """ 

181 

182 source: DocumentSource 

183 original_file: Path 

184 root_document_id: int | None = None 

185 original_path: Path | None = None 

186 mailrule_id: int | None = None 

187 mime_type: str = dataclasses.field(init=False, default=None) 

188 

189 def __post_init__(self) -> None: 

190 """ 

191 After a dataclass is initialized, this is called to finalize some data 

192 1. Make sure the original path is an absolute, fully qualified path 

193 2. Get the mime type of the file 

194 """ 

195 # Always fully qualify the path first thing 

196 # Just in case, convert to a path if it's a str 

197 self.original_file = Path(self.original_file).resolve() 

198 

199 # Get the file type once at init 

200 # Note this function isn't called when the object is unpickled 

201 self.mime_type = magic.from_file(self.original_file, mime=True) 

202 

203 

204class ConsumeFileDuplicateResult(TypedDict): 

205 """Returned by consume_file when the file is rejected as a duplicate.""" 

206 

207 duplicate_of: int 

208 duplicate_in_trash: bool 

209 

210 

211class ConsumeFileSuccessResult(TypedDict): 

212 """Returned by consume_file when the document is created successfully.""" 

213 

214 document_id: int 

215 

216 

217class ConsumeFileStoppedResult(TypedDict): 

218 """Returned by consume_file when a plugin raises StopConsumeTaskError. 

219 

220 Examples: barcode split dispatched child tasks, double-sided scan waiting 

221 for the second half, workflow deleted the document during consumption. 

222 """ 

223 

224 reason: str