Coverage for slidge/core/mixins/attachment.py: 88%

1from __future__ import annotations 

2 

3import base64 

4import contextlib 

5import functools 

6import io 

7import logging 

8import os 

9import shutil 

10import stat 

11import tempfile 

12import warnings 

13from collections.abc import AsyncIterator, Collection, Sequence 

14from datetime import datetime 

15from itertools import chain 

16from mimetypes import guess_extension, guess_type 

17from pathlib import Path 

18from typing import IO, Any, cast 

19from urllib.parse import quote as urlquote 

20from uuid import uuid4 

21from xml.etree import ElementTree as ET 

22 

23import aiohttp 

24import thumbhash 

25from PIL import Image, ImageOps 

26from slixmpp import JID, Iq, Message 

27from slixmpp.plugins.xep_0264.stanza import Thumbnail 

28from slixmpp.plugins.xep_0447.stanza import StatelessFileSharing 

29 

30from ...db.avatar import avatar_cache 

31from ...db.models import Attachment 

32from ...util.types import LegacyAttachment, MessageReference 

33from ...util.util import fix_namespaces, fix_suffix 

34from .. import config 

35from .message_text import TextMessageMixin 

36 

37 

38class AttachmentMixin(TextMessageMixin): 

39 @property 

40 def __is_component(self) -> bool: 

41 return self.session is NotImplemented 

42 

43 async def __upload(self, att: _AttachmentWithName) -> str: 

44 assert config.UPLOAD_SERVICE 

45 att = await _ensure_metadata(att) 

46 iq_slot = await self.__request_upload_slot( 

47 config.UPLOAD_SERVICE, 

48 att.name, 

49 att.size, 

50 att.content_type, 

51 ) 

52 if iq_slot["type"] == "error": 

53 # COMPAT: In theory, __request_upload_slot() raises IqError, but 

54 # there is a bug in prosody's mod_privilege where the outer 

55 # IQ type is (illegally) not set to error when the inner IQ 

56 # has type='error'. 

57 raise RuntimeError(f"Error while requesting upload slot: {iq_slot}") 

58 slot = iq_slot.get_plugin("http_upload_slot", check=True) 

59 if slot is None: 

60 raise RuntimeError(f"No upload slot in this IQ: {iq_slot}") 

61 put = slot["put"]["url"] 

62 assert isinstance(put, str) 

63 if not put: 

64 raise RuntimeError(f"Cannot find a PUT URL in: {slot}") 

65 get = slot["get"]["url"] 

66 assert isinstance(get, str) 

67 if not get: 

68 raise RuntimeError(f"Cannot find a GET URL in: {slot}") 

69 headers = { 

70 "Content-Length": str(att.size), 

71 "Content-Type": att.content_type, 

72 **{header["name"]: header["value"] for header in slot["put"]["headers"]}, 

73 } 

74 

75 async with ( 

76 aiohttp.ClientSession() as http, 

77 _get_data(att, http) as data, 

78 http.put(slot["put"]["url"], data=data, headers=headers) as resp, 

79 ): 

80 resp.raise_for_status() 

81 

82 return get 

83 

84 async def __request_upload_slot( 

85 self, 

86 upload_service: JID | str, 

87 filename: str, 

88 size: int, 

89 content_type: str, 

90 ) -> Iq: 

91 iq_request = self.xmpp.make_iq_get(ito=upload_service) 

92 request = iq_request["http_upload_request"] 

93 request["filename"] = filename 

94 request["size"] = str(size) 

95 request["content-type"] = content_type 

96 if not self.__is_component: 

97 iq_request.set_from(self.session.user_jid) 

98 try: 

99 return await self.xmpp.plugin["xep_0356"].send_privileged_iq(iq_request) 

100 except Exception as e: # noqa: BLE001 

101 warnings.warn( 

102 "Could not request upload slot on behalf of " 

103 f"{self.session.user_jid}: {e}." 

104 "Falling back to not using privileges." 

105 ) 

106 fix_namespaces(iq_request.xml, "jabber:client", "jabber:component:accept") 

107 iq_request.set_from(config.UPLOAD_REQUESTER or self.xmpp.boundjid) 

108 return await iq_request.send() # type:ignore[no-any-return] 

109 

110 @staticmethod 

111 async def __no_upload( 

112 att: _AttachmentWithName, legacy_file_id: str | None 

113 ) -> tuple[Path, str]: 

114 file_id = uuid4().hex if legacy_file_id is None else legacy_file_id 

115 assert config.NO_UPLOAD_PATH is not None 

116 assert config.NO_UPLOAD_URL_PREFIX is not None 

117 destination_dir = Path(config.NO_UPLOAD_PATH) / file_id 

118 

119 if destination_dir.exists(): 

120 log.debug("Dest dir exists: %s", destination_dir) 

121 files = [f for f in destination_dir.glob("**/*") if f.is_file()] 

122 if len(files) == 1: 

123 log.debug( 

124 "Found the legacy attachment '%s' at '%s'", 

125 legacy_file_id, 

126 files[0], 

127 ) 

128 name = files[0].name 

129 uu = files[0].parent.name # anti-obvious url trick, see below 

130 return files[0], f"{file_id}/{uu}/{name}" 

131 else: 

132 log.warning( 

133 ( 

134 "There are several or zero files in %s, " 

135 "slidge doesn't know which one to pick among %s. " 

136 "Removing the dir." 

137 ), 

138 destination_dir, 

139 files, 

140 ) 

141 shutil.rmtree(destination_dir) 

142 

143 log.debug("Did not find a file in: %s", destination_dir) 

144 # let's use a UUID to avoid URLs being too obvious 

145 uu = str(uuid4()) 

146 destination_dir = destination_dir / uu 

147 destination_dir.mkdir(parents=True) 

148 

149 assert att.name 

150 destination = destination_dir / att.name 

151 if att.path: 

152 assert isinstance(att.path, Path) 

153 try: 

154 destination.hardlink_to(att.path) 

155 except OSError as e: 

156 if is_temp_path(att.path): 

157 shutil.copy2(att.path, destination) 

158 else: 

159 log.debug("Could not hardlink: %s, attempting symlink", e) 

160 try: 

161 destination.symlink_to(att.path) 

162 except OSError as e: 

163 log.debug("Could not symlink: %s, copying data", e) 

164 shutil.copy2(att.path, destination) 

165 elif att.data: 

166 destination.write_bytes(att.data) 

167 else: 

168 with destination.open("wb") as f: 

169 if att.aio_stream: 

170 async for chunk in att.aio_stream: 

171 f.write(chunk) 

172 elif att.stream: 

173 shutil.copyfileobj(att.stream, f) 

174 elif att.url: 

175 async with ( 

176 aiohttp.ClientSession() as http, 

177 http.get(att.url) as resp, 

178 ): 

179 resp.raise_for_status() 

180 async for chunk in resp.content.iter_chunked(64 * 1024): 

181 f.write(chunk) 

182 else: 

183 raise RuntimeError 

184 

185 if config.NO_UPLOAD_FILE_READ_OTHERS: 

186 log.debug("Changing perms of %s", destination) 

187 destination.chmod(destination.stat().st_mode | stat.S_IROTH) 

188 

189 url = f"{file_id}/{uu}/{att.name}" 

190 return destination, url 

191 

192 async def __valid_url(self, url: str) -> bool: 

193 async with self.session.http.head(url) as r: 

194 return r.status < 400 

195 

196 async def __get_stored(self, attachment: LegacyAttachment) -> Attachment: 

197 if attachment.legacy_file_id is not None and not self.__is_component: 

198 with self.xmpp.store.session() as orm: 

199 stored = ( 

200 orm.query(Attachment) 

201 .filter_by( 

202 legacy_file_id=str(attachment.legacy_file_id), 

203 user_account_id=self.session.user_pk, 

204 ) 

205 .one_or_none() 

206 ) 

207 if stored is not None: 

208 if not await self.__valid_url(stored.url): 

209 stored.url = None # type:ignore 

210 return stored 

211 return Attachment( 

212 user_account_id=None if self.__is_component else self.session.user_pk, 

213 legacy_file_id=None 

214 if attachment.legacy_file_id is None 

215 else str(attachment.legacy_file_id), 

216 url=attachment.url if config.USE_ATTACHMENT_ORIGINAL_URLS else None, 

217 ) 

218 

219 async def __get_url(self, att: LegacyAttachment, stored: Attachment) -> str: 

220 att = _ensure_name(att) 

221 

222 if len(att.name) > config.ATTACHMENT_MAXIMUM_FILE_NAME_LENGTH: 

223 log.debug("Trimming long filename: %s", att.name) 

224 base, ext = os.path.splitext(att.name) 

225 att.name = ( 

226 base[: config.ATTACHMENT_MAXIMUM_FILE_NAME_LENGTH - len(ext)] + ext 

227 ) 

228 

229 if config.FIX_FILENAME_SUFFIX_MIME_TYPE and isinstance(att.path, Path): 

230 att.name, att.content_type = fix_suffix( 

231 att.path, att.content_type, att.name 

232 ) 

233 

234 att.legacy_file_id = stored.legacy_file_id 

235 

236 if config.NO_UPLOAD_PATH: 

237 att.path, new_url = await self.__no_upload(att, stored.legacy_file_id) 

238 new_url = (config.NO_UPLOAD_URL_PREFIX or "") + "/" + urlquote(new_url) 

239 else: 

240 new_url = await self.__upload(att) 

241 

242 if stored.legacy_file_id: 

243 stored.url = new_url 

244 

245 return new_url 

246 

247 def __set_sims( 

248 self, 

249 msg: Message, 

250 uploaded_url: str, 

251 attachment: LegacyAttachment, 

252 stored: Attachment, 

253 thumbnail: Thumbnail | None, 

254 ) -> None: 

255 if stored.sims is not None: 

256 ref = self.xmpp.plugin["xep_0372"].stanza.Reference( 

257 xml=ET.fromstring(stored.sims) 

258 ) 

259 msg.append(ref) 

260 return 

261 

262 if attachment.data or attachment.stream or attachment.path: 

263 ref = self.xmpp.plugin["xep_0385"].get_sims( 

264 attachment.path, # type:ignore[arg-type] 

265 [uploaded_url], 

266 attachment.content_type, 

267 attachment.caption, 

268 file=attachment.stream, 

269 data=attachment.data, 

270 ) 

271 else: 

272 sims = self.xmpp.plugin["xep_0385"].stanza.Sims() 

273 ref = self.xmpp.plugin["xep_0372"].stanza.Reference() 

274 ref["uri"] = uploaded_url 

275 ref["type"] = "data" 

276 sims["sources"].append(ref) 

277 ref = self.xmpp.plugin["xep_0372"].stanza.Reference() 

278 ref.append(sims) 

279 ref["type"] = "data" 

280 sims.enable("file") 

281 if attachment.content_type: 

282 sims["file"]["media-type"] = attachment.content_type 

283 if attachment.caption: 

284 sims["file"]["desc"] = attachment.caption 

285 if attachment.size: 

286 sims["file"]["size"] = attachment.size 

287 if attachment.name: 

288 ref["sims"]["file"]["name"] = attachment.name 

289 if attachment.aio_stream is not None: 

290 # revove the date in case we have downloaded the file to disk 

291 # because no size was provided by the legacy module 

292 ref["sims"]["file"]["date"] = None 

293 

294 if thumbnail is not None: 

295 ref["sims"]["file"].append(thumbnail) 

296 

297 stored.sims = str(ref) 

298 msg.append(ref) 

299 

300 def __set_sfs( 

301 self, 

302 msg: Message, 

303 uploaded_url: str, 

304 attachment: LegacyAttachment, 

305 stored: Attachment, 

306 thumbnail: Thumbnail | None = None, 

307 ) -> None: 

308 if stored.sfs is not None: 

309 msg.append(StatelessFileSharing(xml=ET.fromstring(stored.sfs))) 

310 return 

311 

312 if attachment.path or attachment.data or attachment.stream: 

313 sfs = self.xmpp.plugin["xep_0447"].get_sfs( # type:ignore 

314 attachment.path, 

315 [uploaded_url], 

316 attachment.content_type, 

317 attachment.caption, 

318 data=attachment.data, 

319 file=attachment.stream, 

320 ) 

321 else: 

322 sfs = self.xmpp.plugin["xep_0447"].stanza.StatelessFileSharing() 

323 ref = self.xmpp.plugin["xep_0447"].stanza.UrlData() 

324 ref["target"] = uploaded_url 

325 sfs["sources"].append(ref) 

326 sfs.enable("file") 

327 if attachment.content_type: 

328 sfs["file"]["media-type"] = attachment.content_type 

329 if attachment.name: 

330 sfs["file"]["name"] = attachment.name 

331 if attachment.disposition: 

332 sfs["disposition"] = attachment.disposition 

333 else: 

334 del sfs["disposition"] 

335 if thumbnail is not None: 

336 sfs["file"].append(thumbnail) 

337 if attachment.aio_stream is not None: 

338 # revove the date in case we have downloaded the file to disk 

339 # because no size was provided by the legacy module 

340 sfs["file"]["date"] = None 

341 if attachment.size: 

342 sfs["file"]["size"] = attachment.size 

343 stored.sfs = str(sfs) 

344 msg.append(sfs) 

345 

346 def __send_url( 

347 self, 

348 msg: Message, 

349 legacy_msg_id: str | None, 

350 uploaded_url: str, 

351 caption: str | None = None, 

352 carbon: bool = False, 

353 when: datetime | None = None, 

354 correction: bool = False, 

355 **kwargs: Any, # noqa:ANN401 

356 ) -> list[Message]: 

357 msg["oob"]["url"] = uploaded_url 

358 msg["body"] = uploaded_url 

359 if "sfs" in msg: 

360 msg["fallback"].enable("body") 

361 msg["fallback"]["for"] = self.xmpp.plugin["xep_0447"].stanza.NAMESPACE 

362 if caption: 

363 if correction: 

364 if not legacy_msg_id: 

365 raise TypeError 

366 msg["replace"]["id"] = self._replace_id(legacy_msg_id) 

367 elif legacy_msg_id: 

368 self._set_msg_id(msg, legacy_msg_id) 

369 m1 = self._send(msg, carbon=carbon, correction=correction, **kwargs) 

370 m2 = self.send_text( 

371 caption, legacy_msg_id=None, when=when, carbon=carbon, **kwargs 

372 ) 

373 return [m1, m2] if m2 else [m1] 

374 else: 

375 if correction: 

376 if not legacy_msg_id: 

377 raise TypeError 

378 msg["replace"]["id"] = self._replace_id(legacy_msg_id) 

379 elif legacy_msg_id: 

380 self._set_msg_id(msg, legacy_msg_id) 

381 return [self._send(msg, carbon=carbon, **kwargs)] 

382 

383 def __get_base_message( 

384 self, 

385 legacy_msg_id: str | None = None, 

386 reply_to: MessageReference | None = None, 

387 when: datetime | None = None, 

388 thread: str | None = None, 

389 carbon: bool = False, 

390 correction: bool = False, 

391 mto: JID | None = None, 

392 ) -> Message: 

393 if correction: 

394 if not legacy_msg_id: 

395 raise TypeError 

396 xmpp_ids = self._legacy_to_xmpp(legacy_msg_id) 

397 if xmpp_ids: 

398 original_xmpp_id = xmpp_ids[0] 

399 for xmpp_id in xmpp_ids: 

400 if xmpp_id == original_xmpp_id: 

401 continue 

402 self.retract(xmpp_id, thread) 

403 

404 if reply_to is not None and reply_to.body: 

405 # We cannot have a "quote fallback" for attachments since most (all?) 

406 # XMPP clients will only treat a message as an attachment if the 

407 # body is the URL and nothing else. 

408 reply_to_for_attachment: MessageReference | None = MessageReference( 

409 reply_to.legacy_id, reply_to.author 

410 ) 

411 else: 

412 reply_to_for_attachment = reply_to 

413 

414 return self._make_message( 

415 when=when, 

416 reply_to=reply_to_for_attachment, 

417 carbon=carbon, 

418 mto=mto, 

419 thread=thread, 

420 ) 

421 

422 async def send_file( 

423 self, 

424 attachment: LegacyAttachment | Path | str, 

425 legacy_msg_id: str | None = None, 

426 *, 

427 reply_to: MessageReference | None = None, 

428 when: datetime | None = None, 

429 thread: str | None = None, 

430 **kwargs: Any, # noqa:ANN401 

431 ) -> tuple[str | None, list[Message]]: 

432 """ 

433 Send a single file from this :term:`XMPP Entity`. 

434 

435 :param attachment: The file to send. 

436 Ideally, a :class:`.LegacyAttachment` with a unique ``legacy_file_id`` 

437 attribute set, to optimise potential future reuses. 

438 It can also be: 

439 - a :class:`pathlib.Path` instance to point to a local file, or 

440 - a ``str``, representing a fetchable HTTP URL. 

441 :param legacy_msg_id: If you want to be able to transport read markers from the gateway 

442 user to the legacy network, specify this 

443 :param reply_to: Quote another message (:xep:`0461`) 

444 :param when: when the file was sent, for a "delay" tag (:xep:`0203`) 

445 :param thread: 

446 """ 

447 coro = self.__send_file( 

448 attachment, 

449 legacy_msg_id, 

450 reply_to=reply_to, 

451 when=when, 

452 thread=thread, 

453 **kwargs, 

454 ) 

455 if ( 

456 self.__is_component 

457 or not isinstance(attachment, LegacyAttachment) 

458 or attachment.legacy_file_id is None 

459 ): 

460 return await coro 

461 else: 

462 # prevents race conditions where we download the same thing several time 

463 # and end up attempting to insert it twice in the DB, raising an 

464 # IntegrityError. 

465 async with self.session.lock(("attachment", attachment.legacy_file_id)): 

466 return await coro 

467 

468 async def __send_file( 

469 self, 

470 attachment: LegacyAttachment | Path | str, 

471 legacy_msg_id: str | None = None, 

472 *, 

473 reply_to: MessageReference | None = None, 

474 when: datetime | None = None, 

475 thread: str | None = None, 

476 store_multi: bool = True, 

477 carbon: bool = False, 

478 mto: JID | None = None, 

479 correction: bool = False, 

480 **send_kwargs: Any, # noqa:ANN401 

481 ) -> tuple[str | None, list[Message]]: 

482 if isinstance(attachment, str): 

483 attachment = LegacyAttachment(url=attachment) 

484 elif isinstance(attachment, Path): 

485 attachment = LegacyAttachment(path=attachment) 

486 

487 msg = self.__get_base_message( 

488 legacy_msg_id=legacy_msg_id, 

489 reply_to=reply_to, 

490 when=when, 

491 thread=thread, 

492 carbon=carbon, 

493 correction=correction, 

494 mto=mto, 

495 ) 

496 if attachment.is_sticker: 

497 msg.enable("sticker") 

498 

499 stored = await self.__get_stored(attachment) 

500 

501 if attachment.content_type is None and ( 

502 name := (attachment.name or attachment.url or attachment.path) 

503 ): 

504 attachment.content_type, _ = guess_type(name) 

505 

506 if not attachment.name: 

507 if attachment.url: 

508 attachment.url.split("/")[-1] 

509 elif isinstance(attachment.path, Path): 

510 attachment.name = attachment.path.name 

511 else: 

512 attachment.name = uuid4().hex 

513 if attachment.content_type: 

514 ext = guess_extension(attachment.content_type) 

515 if ext: 

516 attachment.name += ext 

517 

518 if stored.url: 

519 new_url = stored.url 

520 else: 

521 try: 

522 new_url = await self.__get_url(attachment, stored) 

523 except Exception as e: 

524 log.error("Error with attachment: %s: %s", attachment, e) 

525 log.debug("", exc_info=e) 

526 msg["body"] = ( 

527 f"/me tried to send a file ({attachment.format_for_user()}), " 

528 f"but something went wrong: {e}. " 

529 ) 

530 self._set_msg_id(msg, legacy_msg_id) 

531 return None, [self._send(msg, **send_kwargs)] 

532 stored.url = new_url 

533 attachment.url = new_url 

534 thumbnail = await self.__get_thumbnail(attachment) 

535 self.__set_sims(msg, new_url, attachment, stored, thumbnail) 

536 self.__set_sfs(msg, new_url, attachment, stored, thumbnail) 

537 

538 if isinstance(attachment.path, Path) and is_temp_path( 

539 attachment.path, async_iterator_download_only=True 

540 ): 

541 attachment.path.unlink() 

542 attachment.path = None 

543 

544 if not self.__is_component: 

545 with self.xmpp.store.session(expire_on_commit=False) as orm: 

546 orm.add(stored) 

547 orm.commit() 

548 

549 msgs = self.__send_url( 

550 msg, 

551 legacy_msg_id, 

552 uploaded_url=new_url, 

553 caption=attachment.caption, 

554 carbon=carbon, 

555 when=when, 

556 correction=correction, 

557 **send_kwargs, 

558 ) 

559 if not self.__is_component and store_multi and legacy_msg_id: 

560 self.__store_multi(legacy_msg_id, msgs) 

561 

562 return new_url, msgs 

563 

564 def __send_body( 

565 self, 

566 body: str | None = None, 

567 legacy_msg_id: str | None = None, 

568 reply_to: MessageReference | None = None, 

569 when: datetime | None = None, 

570 thread: str | None = None, 

571 **kwargs: Any, # noqa:ANN401 

572 ) -> Message | None: 

573 if body: 

574 return self.send_text( 

575 body, 

576 legacy_msg_id, 

577 reply_to=reply_to, 

578 when=when, 

579 thread=thread, 

580 **kwargs, 

581 ) 

582 else: 

583 return None 

584 

585 async def send_files( 

586 self, 

587 attachments: Collection[LegacyAttachment], 

588 legacy_msg_id: str | None = None, 

589 body: str | None = None, 

590 *, 

591 reply_to: MessageReference | None = None, 

592 when: datetime | None = None, 

593 thread: str | None = None, 

594 body_first: bool = False, 

595 correction: bool = False, 

596 correction_event_id: str | None = None, 

597 **kwargs: Any, # noqa:ANN401 

598 ) -> None: 

599 # TODO: once the epic XEP-0385 vs XEP-0447 battle is over, pick 

600 # one and stop sending several attachments this way 

601 # we attach the legacy_message ID to the last message we send, because 

602 # we don't want several messages with the same ID (especially for MUC MAM) 

603 if not attachments and not body: 

604 # ignoring empty message 

605 return 

606 body_msg_id = ( 

607 legacy_msg_id if body_needs_msg_id(attachments, body, body_first) else None 

608 ) 

609 send_body = functools.partial( 

610 self.__send_body, 

611 body=body, 

612 reply_to=reply_to, 

613 when=when, 

614 thread=thread, 

615 correction=correction, 

616 legacy_msg_id=body_msg_id, 

617 correction_event_id=correction_event_id, 

618 **kwargs, 

619 ) 

620 all_msgs = [] 

621 if body_first: 

622 all_msgs.append(send_body()) 

623 for i, attachment in enumerate(attachments): 

624 legacy = legacy_msg_id if i == 0 and body_msg_id is None else None 

625 _url, msgs = await self.send_file( 

626 attachment, 

627 legacy, 

628 reply_to=reply_to, 

629 when=when, 

630 thread=thread, 

631 store_multi=False, 

632 **kwargs, 

633 ) 

634 all_msgs.extend(msgs) 

635 if not body_first: 

636 all_msgs.append(send_body()) 

637 self.__store_multi(legacy_msg_id, all_msgs) 

638 

639 def __store_multi( 

640 self, 

641 legacy_msg_id: str | None, 

642 all_msgs: Sequence[Message | None], 

643 ) -> None: 

644 if legacy_msg_id is None: 

645 return 

646 ids = [] 

647 for msg in all_msgs: 

648 if not msg: 

649 continue 

650 if stanza_id := msg.get_plugin("stanza_id", check=True): 

651 ids.append(stanza_id["id"]) 

652 else: 

653 ids.append(msg.get_id()) 

654 with self.xmpp.store.session() as orm: 

655 self.xmpp.store.id_map.set_msg( 

656 orm, self._recipient_pk(), str(legacy_msg_id), ids, self.is_participant 

657 ) 

658 orm.commit() 

659 

660 async def __get_thumbnail(self, attachment: LegacyAttachment) -> Thumbnail | None: 

661 if attachment.content_type is None: 

662 return None 

663 if not (attachment.data or attachment.stream or attachment.path): 

664 return None 

665 if not attachment.content_type.startswith("image"): 

666 return None 

667 

668 try: 

669 h, x, y = await self.xmpp.loop.run_in_executor( 

670 avatar_cache._thread_pool, 

671 get_thumbhash, 

672 attachment, 

673 ) 

674 except Exception as e: 

675 log.debug("Could not generate a thumbhash", exc_info=e) 

676 return None 

677 

678 thumbnail = Thumbnail() 

679 thumbnail["width"] = x 

680 thumbnail["height"] = y 

681 thumbnail["media-type"] = "image/thumbhash" 

682 thumbnail["uri"] = "data:image/thumbhash;base64," + urlquote(h) 

683 return thumbnail 

684 

685 

686class _AttachmentWithName(LegacyAttachment): 

687 name: str 

688 path: Path | None 

689 

690 

691class _AttachmentWithMetadata(_AttachmentWithName): 

692 size: int 

693 content_type: str 

694 

695 

696def _ensure_name(att: LegacyAttachment) -> _AttachmentWithName: 

697 if not att.name: 

698 if att.path: 

699 att.name = Path(att.path).name 

700 elif att.url: 

701 att.name = att.url.split("/")[-1] 

702 else: 

703 att.name = "unnamed-file" 

704 return cast(_AttachmentWithName, att) 

705 

706 

707async def _ensure_metadata(att: _AttachmentWithName) -> _AttachmentWithMetadata: 

708 if att.size is None: 

709 if att.data: 

710 att.size = len(att.data) 

711 elif att.stream: 

712 att.stream.seek(0, io.SEEK_END) 

713 att.size = att.stream.tell() 

714 att.stream.seek(0) 

715 elif att.path: 

716 assert isinstance(att.path, Path) 

717 att.size = att.path.stat().st_size 

718 elif att.url: 

719 async with ( 

720 aiohttp.ClientSession() as http, 

721 http.head(att.url) as resp, 

722 ): 

723 att.size = resp.content_length 

724 elif att.aio_stream: 

725 warnings.warn("A size should be passed with async iterators") 

726 tmp_dir = Path(tempfile.mkdtemp(prefix=_TEMP_PREFIX)) 

727 with (tmp_dir / att.name).open("wb") as fp: 

728 async for chunk in att.aio_stream: 

729 fp.write(chunk) 

730 att.path = Path(fp.name) 

731 att.size = att.path.stat().st_size 

732 

733 if not att.content_type: 

734 att.content_type, _encoding = guess_type(att.name) 

735 if not att.content_type: 

736 att.content_type = "application/octet-stream" 

737 

738 return cast(_AttachmentWithMetadata, att) 

739 

740 

741def body_needs_msg_id( 

742 attachments: Collection[LegacyAttachment], body: str | None, body_first: bool 

743) -> bool: 

744 if attachments: 

745 return bool(body and body_first) 

746 else: 

747 return True 

748 

749 

750def get_thumbhash(att: LegacyAttachment) -> tuple[str, int, int]: 

751 img, width, height = get_image(att) 

752 rgba_2d = list(img.get_flattened_data()) 

753 rgba = list(chain(*rgba_2d)) # type:ignore[arg-type,var-annotated] 

754 ints = thumbhash.rgba_to_thumb_hash(img.width, img.height, rgba) 

755 return base64.b64encode(bytes(ints)).decode(), width, height 

756 

757 

758def get_image(att: LegacyAttachment) -> tuple[Image.Image, int, int]: 

759 if att.data: 

760 img = Image.open(io.BytesIO(att.data))