(params, record, attachments)
| 173 | |
| 174 | |
| 175 | def upload_attachments(params, record, attachments): |
| 176 | # type: (KeeperParams, Union[PasswordRecord, TypedRecord], List[UploadTask]) -> None |
| 177 | cryptor = crypto.StreamCrypter() |
| 178 | if isinstance(record, PasswordRecord): |
| 179 | cryptor.is_gcm = False |
| 180 | if not isinstance(record.attachments, list): |
| 181 | record.attachments = [] |
| 182 | thumbs = [x for x in attachments if x.thumbnail is not None] |
| 183 | rq = { |
| 184 | 'command': 'request_upload', |
| 185 | 'file_count': len(attachments), |
| 186 | 'thumbnail_count': len(thumbs), |
| 187 | } |
| 188 | rs = api.communicate(params, rq) |
| 189 | file_uploads = rs['file_uploads'] |
| 190 | thumb_uploads = rs['thumbnail_uploads'] |
| 191 | thumb_pos = 0 |
| 192 | for i, task in enumerate(attachments): |
| 193 | uo = file_uploads[i] |
| 194 | attachment_id = uo['file_id'] |
| 195 | attachment_key = utils.generate_aes_key() |
| 196 | cryptor.key = attachment_key |
| 197 | atta = AttachmentFile() |
| 198 | task.prepare() |
| 199 | with task.open() as task_stream, cryptor.set_stream(task_stream, True) as crypto_stream: |
| 200 | files = { |
| 201 | uo['file_parameter']: (attachment_id, crypto_stream, 'application/octet-stream') |
| 202 | } |
| 203 | response = requests.post(uo['url'], files=files, data=uo['parameters']) |
| 204 | if response.status_code == uo['success_status_code']: |
| 205 | atta.id = attachment_id |
| 206 | atta.name = task.name or '' |
| 207 | atta.title = task.title or '' |
| 208 | atta.mime_type = task.mime_type or '' |
| 209 | atta.last_modified = utils.current_milli_time() |
| 210 | atta.key = utils.base64_url_encode(attachment_key) |
| 211 | atta.size = task.size |
| 212 | record.attachments.append(atta) |
| 213 | else: |
| 214 | raise Exception(f'Uploading file {task.name}: HTTP status code {response.status_code}') |
| 215 | if isinstance(task.thumbnail, bytes) and thumb_pos < len(thumbs) and thumb_pos < len(thumb_uploads): |
| 216 | thumb_task = thumbs[thumb_pos] |
| 217 | tuo = thumb_uploads[thumb_pos] |
| 218 | thumb_pos += 1 |
| 219 | atta.thumbnails = [] |
| 220 | with io.BytesIO(task.thumbnail) as thumb_stream, \ |
| 221 | cryptor.set_stream(thumb_stream, True) as crypto_stream: |
| 222 | files = { |
| 223 | tuo['file_parameter']: (tuo['file_id'], crypto_stream, 'application/octet-stream') |
| 224 | } |
| 225 | response = requests.post(tuo['url'], files=files, data=tuo['parameters']) |
| 226 | if response.status_code == uo['success_status_code']: |
| 227 | thumb = AttachmentFileThumb() |
| 228 | thumb.id = tuo['file_id'] |
| 229 | thumb.type = thumb_task.mime_type |
| 230 | thumb.size = len(task.thumbnail) |
| 231 | atta.thumbnails.append(thumb) |
| 232 | else: |
nothing calls this directly
no test coverage detected