Coverage for slidge/db/models.py: 95%

1import logging 

2import warnings 

3from collections.abc import Callable 

4from datetime import UTC, date, datetime 

5from enum import IntEnum 

6from functools import lru_cache 

7from pathlib import Path 

8from typing import Any, Literal 

9from xml.etree import ElementTree as ET 

10 

11import slixmpp.plugins.xep_0234.stanza 

12import slixmpp.plugins.xep_0264.stanza 

13import slixmpp.plugins.xep_0300.stanza 

14import slixmpp.plugins.xep_0372.stanza 

15import slixmpp.plugins.xep_0385.stanza 

16import slixmpp.plugins.xep_0447.stanza 

17import sqlalchemy as sa 

18import sqlalchemy.orm 

19from slixmpp import JID, Message 

20from slixmpp.plugins.xep_0292.stanza import VCard4 

21from slixmpp.plugins.xep_0297 import Forwarded 

22from slixmpp.plugins.xep_0444 import XEP_0444 

23from slixmpp.types import MucAffiliation, MucRole 

24from sqlalchemy import JSON, Column, ForeignKey, Index, Table, UniqueConstraint 

25from sqlalchemy.orm import Mapped, mapped_column, relationship, sessionmaker 

26 

27from ..core import config 

28from ..util.types import AvatarMetadata, ClientType, Hat, MucType 

29from .meta import Base, JSONSerializable, JSONSerializableTypes 

30 

31 

32class ArchivedMessageSource(IntEnum): 

33 """ 

34 Whether an archived message comes from ``LegacyMUC.backfill()`` or was received 

35 as a "live" message. 

36 """ 

37 

38 LIVE = 1 

39 BACKFILL = 2 

40 

41 

42class _JidMixin: 

43 jid_localpart: Mapped[str] 

44 

45 @property 

46 def jid(self) -> JID: 

47 return JID(f"{self.jid_localpart}@{config.JID}") 

48 

49 

50class _EnsureLoadedMixin: 

51 id: int | None 

52 

53 def ensure_loaded( 

54 self, 

55 group: str, 

56 session_maker: sessionmaker[Any] | Callable[[], sa.orm.Session], 

57 ) -> None: 

58 if self.id is None: 

59 return 

60 

61 if sa.orm.object_session(self) is not None: 

62 return 

63 

64 state = sa.inspect(self) 

65 if state is None: 

66 return 

67 

68 group_keys = _group_keys( 

69 self.__class__, # type:ignore[arg-type] 

70 group, 

71 ) 

72 

73 unloaded = group_keys & state.unloaded 

74 

75 if not unloaded: 

76 return 

77 

78 cls = self.__class__ 

79 stmt = sa.select(*(getattr(cls, key) for key in unloaded)).where( 

80 cls.id == self.id # type:ignore[arg-type] 

81 ) 

82 with session_maker() as session, session.no_autoflush: 

83 row = session.execute(stmt).one() 

84 

85 for key, value in zip(unloaded, row): 

86 sa.orm.attributes.set_committed_value(self, key, value) 

87 

88 

89@lru_cache 

90def _group_keys(cls: type[Base], group: str) -> frozenset[str]: 

91 keys = frozenset(a.key for a in sa.inspect(cls).column_attrs if a.group == group) 

92 if not keys: 

93 raise ValueError(f"{cls.__name__} has no deferred group {group!r}") 

94 return keys 

95 

96 

97class GatewayUser(Base): 

98 """ 

99 A user, registered to the gateway component. 

100 """ 

101 

102 __tablename__ = "user_account" 

103 id: Mapped[int] = mapped_column(primary_key=True) 

104 jid: Mapped[JID] = mapped_column(unique=True) 

105 registration_date: Mapped[datetime] = mapped_column( 

106 sa.DateTime, server_default=sa.func.now() 

107 ) 

108 

109 legacy_module_data: Mapped[JSONSerializable] = mapped_column(default={}) 

110 """ 

111 Arbitrary non-relational data that legacy modules can use 

112 """ 

113 preferences: Mapped[JSONSerializable] = mapped_column(default={}) 

114 avatar_hash: Mapped[str | None] = mapped_column(default=None) 

115 """ 

116 Hash of the user's avatar, to avoid re-publishing the same avatar on the 

117 legacy network 

118 """ 

119 

120 contacts: Mapped[list["Contact"]] = relationship( 

121 back_populates="user", cascade="all, delete-orphan" 

122 ) 

123 rooms: Mapped[list["Room"]] = relationship( 

124 back_populates="user", cascade="all, delete-orphan" 

125 ) 

126 attachments: Mapped[list["Attachment"]] = relationship(cascade="all, delete-orphan") 

127 spaces: Mapped[list["Space"]] = relationship( 

128 back_populates="user", cascade="all, delete-orphan" 

129 ) 

130 

131 def __repr__(self) -> str: 

132 return f"User(id={self.id!r}, jid={self.jid!r})" 

133 

134 def get(self, field: str, default: str = "") -> JSONSerializableTypes: 

135 # """ 

136 # Get fields from the registration form (required to comply with slixmpp backend protocol) 

137 # 

138 # :param field: Name of the field 

139 # :param default: Default value to return if the field is not present 

140 # 

141 # :return: Value of the field 

142 # """ 

143 return self.legacy_module_data.get(field, default) 

144 

145 @property 

146 def registration_form(self) -> dict[str, Any]: 

147 # Kept for retrocompat, should be 

148 # FIXME: delete me 

149 warnings.warn( 

150 "GatewayUser.registration_form is deprecated.", DeprecationWarning 

151 ) 

152 return self.legacy_module_data 

153 

154 

155class Avatar(Base): 

156 """ 

157 Avatars of contacts, rooms and participants. 

158 

159 To comply with XEPs, we convert them all to PNG before storing them. 

160 """ 

161 

162 __tablename__ = "avatar" 

163 

164 id: Mapped[int] = mapped_column(primary_key=True) 

165 

166 hash: Mapped[str | None] = mapped_column(unique=True, nullable=True) 

167 height: Mapped[int] = mapped_column() 

168 width: Mapped[int] = mapped_column() 

169 

170 legacy_id: Mapped[str | None] = mapped_column(unique=True, nullable=True) 

171 

172 http_id: Mapped[str | None] = mapped_column(nullable=True) 

173 http_type: Mapped[str | None] = mapped_column(nullable=True) 

174 http_bytes: Mapped[int | None] = mapped_column(nullable=True) 

175 http_url: Mapped[str | None] = mapped_column(nullable=True) 

176 http_height: Mapped[int | None] = mapped_column(nullable=True) 

177 http_width: Mapped[int | None] = mapped_column(nullable=True) 

178 

179 # this is only used when avatars are available as HTTP URLs and do not 

180 # have a legacy_id 

181 url: Mapped[str | None] = mapped_column(unique=True, default=None) 

182 etag: Mapped[str | None] = mapped_column(default=None) 

183 last_modified: Mapped[str | None] = mapped_column(default=None) 

184 

185 contacts: Mapped[list["Contact"]] = relationship(back_populates="avatar") 

186 rooms: Mapped[list["Room"]] = relationship(back_populates="avatar") 

187 

188 def set_http_metadata(self, meta: AvatarMetadata | None) -> None: 

189 if meta is None: 

190 return 

191 self.http_id = meta.id 

192 self.http_bytes = meta.bytes 

193 self.http_type = f"image/{meta.type}" 

194 self.http_url = meta.url 

195 self.http_height = meta.height 

196 self.http_width = meta.width 

197 

198 

199space_owner_association = Table( 

200 "space_owner_association", 

201 Base.metadata, 

202 Column("space_id", ForeignKey("space.id"), primary_key=True), 

203 Column("contact_id", ForeignKey("contact.id"), primary_key=True), 

204) 

205 

206 

207class Contact(Base, _EnsureLoadedMixin, _JidMixin): 

208 """ 

209 Legacy contacts 

210 """ 

211 

212 __tablename__ = "contact" 

213 __table_args__ = ( 

214 UniqueConstraint( 

215 "user_account_id", "legacy_id", name="uq_contact_user_account_id_legacy_id" 

216 ), 

217 UniqueConstraint( 

218 "user_account_id", 

219 "jid_localpart", 

220 name="uq_contact_user_account_id_jid_localpart", 

221 ), 

222 ) 

223 

224 id: Mapped[int] = mapped_column(primary_key=True) 

225 user_account_id: Mapped[int] = mapped_column( 

226 ForeignKey("user_account.id"), deferred=True 

227 ) 

228 user: Mapped[GatewayUser] = relationship(lazy=True, back_populates="contacts") 

229 legacy_id: Mapped[str] = mapped_column(nullable=False) 

230 

231 jid_localpart: Mapped[str] = mapped_column(nullable=False) 

232 

233 avatar_id: Mapped[int | None] = mapped_column( 

234 ForeignKey("avatar.id"), nullable=True 

235 ) 

236 avatar: Mapped[Avatar | None] = relationship(back_populates="contacts") 

237 

238 nick: Mapped[str | None] = mapped_column(nullable=True) 

239 

240 cached_presence: Mapped[bool] = mapped_column( 

241 default=False, deferred=True, deferred_group="presence" 

242 ) 

243 last_seen: Mapped[datetime | None] = mapped_column( 

244 nullable=True, deferred=True, deferred_group="presence" 

245 ) 

246 ptype: Mapped[str | None] = mapped_column( 

247 nullable=True, deferred=True, deferred_group="presence" 

248 ) 

249 pstatus: Mapped[str | None] = mapped_column( 

250 nullable=True, deferred=True, deferred_group="presence" 

251 ) 

252 pshow: Mapped[str | None] = mapped_column( 

253 nullable=True, deferred=True, deferred_group="presence" 

254 ) 

255 

256 is_friend: Mapped[bool] = mapped_column(default=False) 

257 added_to_roster: Mapped[bool] = mapped_column(default=False) 

258 sent_order: Mapped[list["ContactSent"]] = relationship( 

259 back_populates="contact", cascade="all, delete-orphan" 

260 ) 

261 

262 extra_attributes: Mapped[JSONSerializable | None] = mapped_column( 

263 default=None, nullable=True 

264 ) 

265 updated: Mapped[bool] = mapped_column(default=False) 

266 

267 vcard_fetched: Mapped[bool] = mapped_column(default=False) 

268 # vcard fields 

269 full_name: Mapped[str | None] = mapped_column( 

270 nullable=True, deferred=True, deferred_group="vcard" 

271 ) 

272 birthday: Mapped[date | None] = mapped_column( 

273 nullable=True, deferred=True, deferred_group="vcard" 

274 ) 

275 phones: Mapped[list[str] | None] = mapped_column( 

276 JSON, nullable=True, deferred=True, deferred_group="vcard" 

277 ) 

278 note: Mapped[str | None] = mapped_column( 

279 nullable=True, deferred=True, deferred_group="vcard" 

280 ) 

281 url: Mapped[str | None] = mapped_column( 

282 nullable=True, deferred=True, deferred_group="vcard" 

283 ) 

284 email: Mapped[str | None] = mapped_column( 

285 nullable=True, deferred=True, deferred_group="vcard" 

286 ) 

287 country: Mapped[str | None] = mapped_column( 

288 nullable=True, deferred=True, deferred_group="vcard" 

289 ) 

290 locality: Mapped[str | None] = mapped_column( 

291 nullable=True, deferred=True, deferred_group="vcard" 

292 ) 

293 pronouns: Mapped[str | None] = mapped_column( 

294 nullable=True, deferred=True, deferred_group="vcard" 

295 ) 

296 

297 # No "merge" cascade: committing a Contact must never write its Participants. 

298 # Otherwise a stale, transient Participant left in this (loaded) collection 

299 # by the back_populates of Participant.contact gets INSERTed a second time on 

300 # the next Contact.commit(). 

301 participants: Mapped[list["Participant"]] = relationship( 

302 back_populates="contact", cascade="save-update" 

303 ) 

304 

305 client_type: Mapped[ClientType] = mapped_column(nullable=False, default="pc") 

306 

307 messages: Mapped[list["DirectMessages"]] = relationship( 

308 cascade="all, delete-orphan" 

309 ) 

310 threads: Mapped[list["DirectThreads"]] = relationship(cascade="all, delete-orphan") 

311 

312 spaces_created: Mapped[list["Space"]] = relationship( 

313 back_populates="creator", 

314 cascade="all, delete-orphan", 

315 ) 

316 spaces_owned: Mapped[list["Space"]] = relationship( 

317 back_populates="owners", 

318 secondary=space_owner_association, 

319 ) 

320 

321 last_sent_msg_legacy_id: Mapped[str | None] = mapped_column( 

322 deferred=True, deferred_group="last_msg" 

323 ) 

324 last_sent_msg_date: Mapped[datetime | None] = mapped_column( 

325 deferred=True, deferred_group="last_msg" 

326 ) 

327 

328 def vcard(self, domain: str) -> VCard4: 

329 vcard = VCard4() 

330 vcard.add_impp(f"xmpp:{self.jid_localpart}@{domain}") 

331 

332 if n := self.nick: 

333 vcard.add_nickname(n) 

334 if self.full_name: 

335 vcard["full_name"] = self.full_name 

336 elif n: 

337 vcard["full_name"] = n 

338 

339 if self.birthday: 

340 vcard["birthday"] = self.birthday 

341 

342 if self.note: 

343 vcard.add_note(self.note) 

344 if self.url: 

345 vcard.add_url(self.url) 

346 if self.email: 

347 vcard.add_email(self.email) 

348 

349 if self.phones: 

350 for p in self.phones: 

351 vcard.add_tel(p) 

352 if self.country or self.locality: 

353 vcard.add_address(self.country, self.locality) 

354 if self.pronouns: 

355 vcard["pronouns"]["text"] = self.pronouns 

356 

357 return vcard 

358 

359 

360class ContactSent(Base): 

361 """ 

362 Keep track of XMPP msg ids sent by a specific contact for networks in which 

363 all messages need to be marked as read. 

364 

365 (XMPP displayed markers convey a "read up to here" semantic.) 

366 """ 

367 

368 __tablename__ = "contact_sent" 

369 __table_args__ = ( 

370 UniqueConstraint( 

371 "contact_id", "msg_id", name="uq_contact_sent_contact_id_msg_id" 

372 ), 

373 ) 

374 

375 id: Mapped[int] = mapped_column(primary_key=True) 

376 contact_id: Mapped[int] = mapped_column(ForeignKey("contact.id")) 

377 contact: Mapped[Contact] = relationship(back_populates="sent_order") 

378 msg_id: Mapped[str] = mapped_column() 

379 

380 

381class Room(Base, _EnsureLoadedMixin, _JidMixin): 

382 """ 

383 Legacy room 

384 """ 

385 

386 __table_args__ = ( 

387 UniqueConstraint( 

388 "user_account_id", "legacy_id", name="uq_room_user_account_id_legacy_id" 

389 ), 

390 UniqueConstraint( 

391 "user_account_id", 

392 "jid_localpart", 

393 name="uq_room_user_account_id_jid_localpart", 

394 ), 

395 ) 

396 

397 __tablename__ = "room" 

398 id: Mapped[int] = mapped_column(primary_key=True) 

399 user_account_id: Mapped[int] = mapped_column( 

400 ForeignKey("user_account.id"), deferred=True 

401 ) 

402 user: Mapped[GatewayUser] = relationship(lazy=True, back_populates="rooms") 

403 legacy_id: Mapped[str] = mapped_column(nullable=False) 

404 

405 jid_localpart: Mapped[str] = mapped_column(nullable=False) 

406 

407 avatar_id: Mapped[int | None] = mapped_column( 

408 ForeignKey("avatar.id"), nullable=True 

409 ) 

410 avatar: Mapped[Avatar | None] = relationship(back_populates="rooms") 

411 

412 name: Mapped[str | None] = mapped_column(nullable=True) 

413 

414 description: Mapped[str | None] = mapped_column( 

415 nullable=True, deferred=True, deferred_group="disco" 

416 ) 

417 n_participants: Mapped[int | None] = mapped_column( 

418 default=None, deferred=True, deferred_group="disco" 

419 ) 

420 

421 subject: Mapped[str | None] = mapped_column( 

422 nullable=True, deferred=True, deferred_group="subject" 

423 ) 

424 subject_date: Mapped[datetime | None] = mapped_column( 

425 nullable=True, deferred=True, deferred_group="subject" 

426 ) 

427 subject_setter: Mapped[str | None] = mapped_column( 

428 nullable=True, deferred=True, deferred_group="subject" 

429 ) 

430 

431 muc_type: Mapped[MucType] = mapped_column(default=MucType.CHANNEL) 

432 

433 user_nick: Mapped[str | None] = mapped_column() 

434 user_resources: Mapped[str | None] = mapped_column(nullable=True) 

435 

436 participants_filled: Mapped[bool] = mapped_column(default=False) 

437 history_filled: Mapped[bool] = mapped_column(default=False) 

438 

439 extra_attributes: Mapped[JSONSerializable | None] = mapped_column(default=None) 

440 updated: Mapped[bool] = mapped_column(default=False) 

441 

442 participants: Mapped[list["Participant"]] = relationship( 

443 back_populates="room", 

444 primaryjoin="Participant.room_id == Room.id", 

445 cascade="all, delete-orphan", 

446 ) 

447 

448 archive: Mapped[list["ArchivedMessage"]] = relationship( 

449 cascade="all, delete-orphan" 

450 ) 

451 

452 space_id: Mapped[int | None] = mapped_column(ForeignKey("space.id"), nullable=True) 

453 space: Mapped["Space | None"] = relationship(back_populates="rooms") 

454 

455 def load_participants( 

456 self, orm: sessionmaker[Any] | Callable[[], sa.orm.Session], with_presence: bool 

457 ) -> None: 

458 if self.id is None: 

459 return 

460 

461 stmt = ( 

462 _STMT_ROOM_LOAD_PARTICIPANTS_WITH_PRESENCE 

463 if with_presence 

464 else _STMT_ROOM_LOAD_PARTICIPANTS 

465 ).where(Room.id == self.id) 

466 

467 if (existing_session := sa.orm.object_session(self)) is None: 

468 with orm() as session, session.no_autoflush: 

469 session.add(self) 

470 session.execute(stmt).all() 

471 else: 

472 existing_session.execute(stmt).all() 

473 

474 

475class ArchivedMessage(Base): 

476 """ 

477 Messages of rooms, that we store to act as a MAM server 

478 """ 

479 

480 __tablename__ = "mam" 

481 __table_args__ = ( 

482 UniqueConstraint("room_id", "stanza_id", name="uq_mam_room_id_stanza_id"), 

483 Index("ix_mam_room_id_legacy_id", "room_id", "legacy_id"), 

484 Index("ix_mam_room_id_timestamp", "room_id", "timestamp"), 

485 ) 

486 

487 id: Mapped[int] = mapped_column(primary_key=True) 

488 room_id: Mapped[int] = mapped_column(ForeignKey("room.id"), nullable=False) 

489 room: Mapped[Room] = relationship(lazy=True, back_populates="archive") 

490 

491 stanza_id: Mapped[str] = mapped_column(nullable=False) 

492 origin_id: Mapped[str | None] = mapped_column(nullable=True) 

493 timestamp: Mapped[datetime] = mapped_column(nullable=False) 

494 

495 author_resource: Mapped[str | None] = mapped_column(nullable=True) 

496 author_affiliation: Mapped[str | None] = mapped_column() 

497 author_role: Mapped[str | None] = mapped_column() 

498 author_jid_localpart: Mapped[str | None] = mapped_column() 

499 

500 from_user: Mapped[bool] = mapped_column() 

501 

502 occupant_id: Mapped[str | None] = mapped_column() 

503 

504 source: Mapped[ArchivedMessageSource] = mapped_column(nullable=False) 

505 legacy_id: Mapped[str | None] = mapped_column(nullable=True) 

506 thread_legacy_id: Mapped[str | None] = mapped_column(nullable=True) 

507 

508 # payload is for XML we don't have dedicated columns for 

509 payload: Mapped[str | None] = mapped_column(nullable=True) 

510 

511 displayed_id: Mapped[str | None] = mapped_column(nullable=True) 

512 reaction_id: Mapped[str | None] = mapped_column(nullable=True) 

513 reaction_emojis: Mapped[list[str] | None] = mapped_column(JSON, nullable=True) 

514 body: Mapped[str | None] = mapped_column(nullable=True) 

515 thread: Mapped[str | None] = mapped_column(nullable=True) 

516 

517 displayed_by_user: Mapped[bool] = mapped_column(default=False, nullable=True) 

518 

519 def stanza( 

520 self, ns: str = "jabber:client", user_jid: JID | str | None = None 

521 ) -> Message: 

522 msg = Message( 

523 stype="groupchat", 

524 sfrom=f"{self.room.jid}/{self.author_resource}" 

525 if self.author_resource 

526 else self.room.jid, 

527 ) 

528 if ns != "jabber:client": 

529 msg.xml.tag = f"{{{ns}}}message" 

530 msg.namespace = ns 

531 

532 msg["stanza_id"]["id"] = self.stanza_id 

533 msg["stanza_id"]["by"] = self.room.jid 

534 

535 if self.origin_id: 

536 msg["id"] = msg["origin_id"]["id"] = self.origin_id 

537 else: 

538 msg["id"] = self.stanza_id 

539 

540 if self.occupant_id: 

541 msg["occupant-id"]["id"] = self.occupant_id 

542 elif self.author_jid_localpart: 

543 msg["occupant-id"]["id"] = f"{self.author_jid_localpart}@{config.JID}" 

544 else: 

545 log.warning("Cannot set an occupant ID for this archived message") 

546 

547 if self.author_affiliation: 

548 msg["muc"]["affiliation"] = self.author_affiliation 

549 if self.author_role: 

550 msg["muc"]["role"] = self.author_role 

551 if self.from_user: 

552 msg["muc"]["jid"] = str(user_jid) 

553 elif self.author_jid_localpart: 

554 msg["muc"]["jid"] = JID(f"{self.author_jid_localpart}@{config.JID}") 

555 

556 if self.payload: 

557 for child in ET.fromstring(f'<x xmlns="{ns}">{self.payload}</x>'): 

558 msg.append(child) 

559 if self.displayed_id: 

560 msg["displayed"]["id"] = self.displayed_id 

561 if self.reaction_id: 

562 XEP_0444.set_reactions(msg, self.reaction_id, self.reaction_emojis or []) 

563 if self.body: 

564 msg["body"] = self.body 

565 if self.thread: 

566 msg["thread"] = self.thread 

567 

568 return msg 

569 

570 def forwarded(self, user_jid: JID | str) -> Forwarded: 

571 forwarded = Forwarded() 

572 forwarded["delay"]["stamp"] = self.timestamp.replace(tzinfo=UTC) 

573 forwarded.append(self.stanza(user_jid=user_jid)) 

574 return forwarded 

575 

576 

577class _LegacyToXmppIdsBase: 

578 """ 

579 XMPP-client generated IDs, and mapping to the corresponding legacy IDs. 

580 

581 A single legacy ID can map to several XMPP ids. 

582 """ 

583 

584 id: Mapped[int] = mapped_column(primary_key=True) 

585 legacy_id: Mapped[str] = mapped_column(nullable=False) 

586 xmpp_id: Mapped[str] = mapped_column(nullable=False) 

587 

588 

589class DirectMessages(_LegacyToXmppIdsBase, Base): 

590 __tablename__ = "direct_msg" 

591 __table_args__ = (Index("ix_direct_msg_legacy_id", "legacy_id", "foreign_key"),) 

592 foreign_key: Mapped[int] = mapped_column(ForeignKey("contact.id"), nullable=False) 

593 

594 

595class DirectThreads(_LegacyToXmppIdsBase, Base): 

596 __tablename__ = "direct_thread" 

597 __table_args__ = (Index("ix_direct_direct_thread_id", "legacy_id", "foreign_key"),) 

598 foreign_key: Mapped[int] = mapped_column(ForeignKey("contact.id"), nullable=False) 

599 

600 

601class Attachment(Base): 

602 """ 

603 Legacy attachments 

604 """ 

605 

606 __tablename__ = "attachment" 

607 __table_args__ = ( 

608 UniqueConstraint( 

609 "user_account_id", 

610 "legacy_file_id", 

611 name="uq_attachment_user_account_id_legacy_file_id", 

612 ), 

613 ) 

614 

615 id: Mapped[int] = mapped_column(primary_key=True) 

616 user_account_id: Mapped[int] = mapped_column( 

617 ForeignKey("user_account.id"), deferred=True 

618 ) 

619 user: Mapped[GatewayUser] = relationship(back_populates="attachments") 

620 

621 legacy_file_id: Mapped[str | None] = mapped_column(index=True, nullable=True) 

622 url: Mapped[str] = mapped_column(index=True, nullable=False) 

623 

624 size: Mapped[int | None] = mapped_column(nullable=True) 

625 sha256: Mapped[str | None] = mapped_column(nullable=True) 

626 

627 thumbhash: Mapped[str | None] = mapped_column(nullable=True) 

628 width: Mapped[int | None] = mapped_column(nullable=True) 

629 height: Mapped[int | None] = mapped_column(nullable=True) 

630 content_type: Mapped[str | None] = mapped_column(nullable=True) 

631 desc: Mapped[str | None] = mapped_column(nullable=True) 

632 name: Mapped[str | None] = mapped_column(nullable=True) 

633 date: Mapped[datetime | None] = mapped_column(nullable=True) 

634 disposition: Mapped[Literal["inline", "attachment"] | None] = mapped_column( 

635 nullable=True 

636 ) 

637 

638 def _hash(self) -> slixmpp.plugins.xep_0300.stanza.Hash | None: 

639 if self.sha256 is None: 

640 return None 

641 hash_elem = slixmpp.plugins.xep_0300.stanza.Hash() 

642 hash_elem["algo"] = "sha-256" 

643 hash_elem["value"] = self.sha256 

644 return hash_elem 

645 

646 def _thumbnail(self) -> slixmpp.plugins.xep_0264.stanza.Thumbnail | None: 

647 if self.thumbhash is None: 

648 return None 

649 thumbnail = slixmpp.plugins.xep_0264.stanza.Thumbnail() 

650 thumbnail["width"] = self.width 

651 thumbnail["height"] = self.height 

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

653 thumbnail["uri"] = "data:image/thumbhash;base64," + self.thumbhash 

654 return thumbnail 

655 

656 def sims(self) -> slixmpp.plugins.xep_0372.stanza.Reference: 

657 ref = slixmpp.plugins.xep_0372.stanza.Reference() 

658 ref["type"] = "data" 

659 ref["uri"] = self.url 

660 

661 sims = slixmpp.plugins.xep_0385.stanza.Sims() 

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

663 sims.enable("file") 

664 if self.content_type is not None: 

665 sims["file"]["media-type"] = self.content_type 

666 if self.desc is not None: 

667 sims["file"]["desc"] = self.desc 

668 if self.size is not None: 

669 sims["file"]["size"] = self.size 

670 if self.name is not None: 

671 sims["file"]["name"] = self.name 

672 if self.date is not None: 

673 sims["file"]["date"] = self.date 

674 if self.thumbhash is not None: 

675 sims["file"].append(self._thumbnail()) 

676 if self.sha256 is not None: 

677 sims["file"].append(self._hash()) 

678 

679 wrapper = slixmpp.plugins.xep_0372.stanza.Reference() 

680 wrapper["type"] = "data" 

681 wrapper.append(sims) 

682 

683 return wrapper 

684 

685 def sfs(self) -> slixmpp.plugins.xep_0447.stanza.StatelessFileSharing: 

686 ref = slixmpp.plugins.xep_0447.stanza.UrlData() 

687 ref["target"] = self.url 

688 

689 sfs = slixmpp.plugins.xep_0447.stanza.StatelessFileSharing() 

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

691 sfs.enable("file") 

692 if self.content_type is not None: 

693 sfs["file"]["media-type"] = self.content_type 

694 if self.name is not None: 

695 sfs["file"]["name"] = self.name 

696 if self.desc is not None: 

697 sfs["file"]["desc"] = self.desc 

698 if self.disposition is not None: 

699 sfs["disposition"] = self.disposition 

700 else: 

701 del sfs["disposition"] 

702 if self.date is not None: 

703 sfs["file"]["date"] = self.date 

704 if self.size is not None: 

705 sfs["file"]["size"] = self.size 

706 if self.width is not None: 

707 sfs["file"]["width"] = self.width 

708 if self.height is not None: 

709 sfs["file"]["height"] = self.height