Coverage for app/backend/src/couchers/email/emails.py: 95%

1080 statements  

« prev     ^ index     » next       coverage.py v7.14.3, created at 2026-06-30 23:29 +0000

1""" 

2Defines data models for each email we sent out to users. 

3 

4Email writing style guidelines 

5 

6Subject line: 

7 Single sentence of the form "who did what" (avoid passive voice when possible) 

8 No punctuation*. 

9 Do not quote people, community, group or event names. 

10 No need to refer to "Couchers.org", the sender name already does. 

11 

12Preview line: 

13 Quoted user content (e.g. comment/reference text), otherwise none. 

14 Don't repeat or paraphrase the subject. 

15 

16Purpose line of the body (first paragraph after the greeting): 

17 Usually a single sentence stating the purpose of the email. 

18 Similar or identical to the subject but may include additional info (e.g. dates). 

19 Should be punctuated with a period*, or end with a colon (':') if we then quote user content. 

20 

21Other instructions for body text: 

22 A single purpose line is enough for day-to-day notifications, no need for further prose. 

23 Highlight key pieces of info (names, locations, dates) using <b> tags. 

24 Highlight important passages using <strong> tags. 

25 Provide a link or instructions if the user has follow-up actions. 

26 

27* Some key emails like new accounts might use an exclamation mark (limit to 1) and more personal prose. 

28""" 

29 

30import re 

31from dataclasses import dataclass, replace 

32from datetime import UTC, date, datetime 

33from typing import Self, assert_never 

34 

35from markupsafe import Markup, escape 

36 

37from couchers import urls 

38from couchers.config import config 

39from couchers.constants import LATEST_RELEASE_BLOG_URL 

40from couchers.email.blocks import ( 

41 ActionBlock, 

42 EmailBase, 

43 EmailBlock, 

44 ParaBlock, 

45 QuoteBlock, 

46 UserInfo, 

47) 

48from couchers.email.locales import get_emails_i18next 

49from couchers.i18n import LocalizationContext 

50from couchers.i18n.localize import format_phone_number 

51from couchers.markup import html_link, html_mailto_link, markdown_to_plaintext 

52from couchers.notifications.quick_links import generate_quick_decline_link 

53from couchers.proto import conversations_pb2, events_pb2, notification_data_pb2 

54from couchers.utils import now, to_aware_datetime 

55 

56# Common string keys 

57_do_not_reply_request_string_key = "generic.do_not_reply_request" 

58 

59# Specific email definitions 

60 

61 

62@dataclass(kw_only=True, slots=True) 

63class AccountDeletionStartedEmail(EmailBase): 

64 """Sent to a user to confirm their account deletion request.""" 

65 

66 deletion_link: str 

67 

68 @property 

69 def string_key_base(self) -> str: 

70 return "account_deletion.started" 

71 

72 def get_body_blocks(self, loc_context: LocalizationContext) -> list[EmailBlock]: 

73 builder = self._body_builder(loc_context, security_warning=True) 

74 builder.para(".purpose") 

75 builder.action(self.deletion_link, ".confirm_action") 

76 return builder.build() 

77 

78 @classmethod 

79 def from_notification(cls, data: notification_data_pb2.AccountDeletionStart, *, user_name: str) -> Self: 

80 return cls( 

81 user_name=user_name, 

82 deletion_link=urls.delete_account_link(account_deletion_token=data.deletion_token), 

83 ) 

84 

85 @classmethod 

86 def test_instances(cls) -> list[Self]: 

87 return [ 

88 cls( 

89 user_name="Alice", 

90 deletion_link="https://couchers.org/delete-account?token=xxx", 

91 ) 

92 ] 

93 

94 

95@dataclass(kw_only=True, slots=True) 

96class AccountDeletionCompletedEmail(EmailBase): 

97 """Sent to a user after their account has been deleted.""" 

98 

99 undelete_link: str 

100 days: int 

101 

102 @property 

103 def string_key_base(self) -> str: 

104 return "account_deletion.completed" 

105 

106 def get_body_blocks(self, loc_context: LocalizationContext) -> list[EmailBlock]: 

107 builder = self._body_builder(loc_context, security_warning=True) 

108 builder.para(".purpose") 

109 builder.para(".farewell") 

110 builder.para(".recovery_instructions_days", {"count": self.days}) 

111 builder.action(self.undelete_link, ".recover_action") 

112 return builder.build() 

113 

114 @classmethod 

115 def from_notification(cls, data: notification_data_pb2.AccountDeletionComplete, *, user_name: str) -> Self: 

116 return cls( 

117 user_name=user_name, 

118 undelete_link=urls.recover_account_link(account_undelete_token=data.undelete_token), 

119 days=data.undelete_days, 

120 ) 

121 

122 @classmethod 

123 def test_instances(cls) -> list[Self]: 

124 return [ 

125 cls( 

126 user_name="Alice", 

127 undelete_link="https://couchers.org/recover-account?token=xxx", 

128 days=30, 

129 ) 

130 ] 

131 

132 

133@dataclass(kw_only=True, slots=True) 

134class AccountDeletionRecoveredEmail(EmailBase): 

135 """Sent to a user after their account deletion has been cancelled.""" 

136 

137 @property 

138 def string_key_base(self) -> str: 

139 return "account_deletion.recovered" 

140 

141 def get_body_blocks(self, loc_context: LocalizationContext) -> list[EmailBlock]: 

142 builder = self._body_builder(loc_context, security_warning=True) 

143 builder.para(".confirmation") 

144 builder.para(".login_instructions") 

145 builder.action(urls.app_link(), ".login_action") 

146 builder.para(".redelete_instructions") 

147 return builder.build() 

148 

149 @classmethod 

150 def test_instances(cls) -> list[Self]: 

151 return [cls(user_name="Alice")] 

152 

153 

154@dataclass(kw_only=True, slots=True) 

155class ActivenessProbeEmail(EmailBase): 

156 """Sent to a host to check if they are still open to hosting.""" 

157 

158 days_left: int 

159 

160 @property 

161 def string_key_base(self) -> str: 

162 return "activeness_probe" 

163 

164 def get_body_blocks(self, loc_context: LocalizationContext) -> list[EmailBlock]: 

165 builder = self._body_builder(loc_context) 

166 builder.para(".purpose") 

167 builder.para(".instructions_days", {"count": self.days_left}) 

168 builder.action(urls.app_link(), ".login_action") 

169 builder.para(".encouragement") 

170 

171 # Extract major.minor from the version string. "v1.3.18927" -> "1.3" 

172 version = config.VERSION 

173 if version_match := re.search(r"^v?(\d+\.\d+)\b", version): 173 ↛ 174line 173 didn't jump to line 174 because the condition on line 173 was never true

174 version = version_match[1] 

175 

176 builder.para(".latest_release", {"version": version, "blog_url": LATEST_RELEASE_BLOG_URL}) 

177 return builder.build() 

178 

179 @classmethod 

180 def from_notification(cls, data: notification_data_pb2.ActivenessProbe, *, user_name: str) -> Self: 

181 days_left = (to_aware_datetime(data.deadline) - now()).days 

182 return cls(user_name=user_name, days_left=days_left) 

183 

184 @classmethod 

185 def test_instances(cls) -> list[Self]: 

186 return [cls(user_name="Alice", days_left=7)] 

187 

188 

189@dataclass(kw_only=True, slots=True) 

190class APIKeyIssuedEmail(EmailBase): 

191 """Sent to a user to notify them that their API key was issued.""" 

192 

193 api_key: str 

194 expiry: datetime 

195 

196 @property 

197 def string_key_base(self) -> str: 

198 return "api_key_issued" 

199 

200 def get_body_blocks(self, loc_context: LocalizationContext) -> list[EmailBlock]: 

201 builder = self._body_builder(loc_context, security_warning=True) 

202 builder.para(".header") 

203 builder.quote(self.api_key, markdown=False) 

204 builder.para(".expiry", {"datetime": loc_context.localize_datetime(self.expiry)}) 

205 builder.para(".usage_warning") 

206 builder.para(".policy_warning") 

207 return builder.build() 

208 

209 @classmethod 

210 def from_notification(cls, data: notification_data_pb2.ApiKeyCreate, *, user_name: str) -> Self: 

211 return cls(user_name=user_name, api_key=data.api_key, expiry=data.expiry.ToDatetime(tzinfo=UTC)) 

212 

213 @classmethod 

214 def test_instances(cls) -> list[Self]: 

215 return [cls(user_name="Alice", api_key="my_api_key_123", expiry=datetime(2099, 12, 31, 23, 59, 59, tzinfo=UTC))] 

216 

217 

218@dataclass(kw_only=True, slots=True) 

219class BadgeChangedEmail(EmailBase): 

220 """Sent to a user to notify them that a badge was added or removed from their profile.""" 

221 

222 badge_name: str 

223 added: bool 

224 

225 @property 

226 def string_key_base(self) -> str: 

227 return "badges.added" if self.added else "badges.removed" 

228 

229 def get_subject_line(self, loc_context: LocalizationContext) -> str: 

230 return self._localize(loc_context, ".subject", {"name": self.badge_name}) 

231 

232 def get_body_blocks(self, loc_context: LocalizationContext) -> list[EmailBlock]: 

233 builder = self._body_builder(loc_context) 

234 builder.para(".purpose", {"name": self.badge_name}) 

235 return builder.build() 

236 

237 @classmethod 

238 def from_notification( 

239 cls, data: notification_data_pb2.BadgeAdd | notification_data_pb2.BadgeRemove, *, user_name: str 

240 ) -> Self: 

241 return cls( 

242 user_name=user_name, badge_name=data.badge_name, added=isinstance(data, notification_data_pb2.BadgeAdd) 

243 ) 

244 

245 @classmethod 

246 def test_instances(cls) -> list[Self]: 

247 prototype = cls(user_name="Alice", badge_name="Founder", added=True) 

248 return [replace(prototype, added=True), replace(prototype, added=False)] 

249 

250 

251@dataclass(kw_only=True, slots=True) 

252class BirthdateChangedEmail(EmailBase): 

253 """Sent to a user to notify them that their birthdate was changed.""" 

254 

255 new_birthdate: date 

256 

257 @property 

258 def string_key_base(self) -> str: 

259 return "birthdate_changed" 

260 

261 def get_body_blocks(self, loc_context: LocalizationContext) -> list[EmailBlock]: 

262 builder = self._body_builder(loc_context, security_warning=True) 

263 builder.para(".purpose", {"date": loc_context.localize_date(self.new_birthdate)}) 

264 return builder.build() 

265 

266 @classmethod 

267 def from_notification(cls, data: notification_data_pb2.BirthdateChange, *, user_name: str) -> Self: 

268 return cls(user_name=user_name, new_birthdate=date.fromisoformat(data.birthdate)) 

269 

270 @classmethod 

271 def test_instances(cls) -> list[Self]: 

272 return [ 

273 cls( 

274 user_name="Alice", 

275 new_birthdate=date(1990, 1, 1), 

276 ) 

277 ] 

278 

279 

280@dataclass(kw_only=True, slots=True) 

281class ChatMessageReceivedEmail(EmailBase): 

282 """Sent to a user when they receive a new chat message.""" 

283 

284 group_chat_title: str | None # None if direct message 

285 author: UserInfo 

286 text: str 

287 view_url: str 

288 

289 @property 

290 def string_key_base(self) -> str: 

291 return f"chat_messages.received.{'direct' if self.group_chat_title is None else 'group'}" 

292 

293 def get_subject_line(self, loc_context: LocalizationContext) -> str: 

294 return self._localize( 

295 loc_context, ".subject", {"author": self.author.name, "group": self.group_chat_title or ""} 

296 ) 

297 

298 def get_preview_line(self, loc_context: LocalizationContext) -> str | None: 

299 return self.text 

300 

301 def get_body_blocks(self, loc_context: LocalizationContext) -> list[EmailBlock]: 

302 builder = self._body_builder(loc_context) 

303 builder.para(".purpose", {"author": self.author.name, "group": self.group_chat_title or ""}) 

304 builder.user(self.author) 

305 builder.quote(self.text, markdown=False) 

306 builder.action(self.view_url, ".view_action") 

307 return builder.build() 

308 

309 @classmethod 

310 def from_notification(cls, data: notification_data_pb2.ChatMessage, *, user_name: str) -> Self: 

311 return cls( 

312 user_name, 

313 author=UserInfo.from_protobuf(data.author), 

314 text=data.text, 

315 group_chat_title=data.group_chat_title or None, 

316 view_url=urls.chat_link(chat_id=data.group_chat_id), 

317 ) 

318 

319 @classmethod 

320 def test_instances(cls) -> list[Self]: 

321 prototype = cls( 

322 user_name="Alice", 

323 group_chat_title=None, 

324 author=UserInfo.dummy_bob(), 

325 text="Hi Alice!", 

326 view_url="https://couchers.org/messages/chats/123", 

327 ) 

328 return [ 

329 replace(prototype, group_chat_title=None), 

330 replace(prototype, group_chat_title="Best friends"), 

331 ] 

332 

333 

334@dataclass(kw_only=True, slots=True) 

335class ChatMessagesMissedEmail(EmailBase): 

336 """Sent to a user after they've missed new chat messages.""" 

337 

338 @dataclass(kw_only=True, slots=True) 

339 class Entry: 

340 """Entry for each chat with missed messages.""" 

341 

342 group_chat_title: str | None # None if direct message 

343 missed_count: int 

344 latest_message_author: UserInfo 

345 latest_message_text: str 

346 view_url: str 

347 

348 entries: list[Entry] 

349 

350 @property 

351 def string_key_base(self) -> str: 

352 return "chat_messages.missed" 

353 

354 def get_subject_line(self, loc_context: LocalizationContext) -> str: 

355 return self._localize(loc_context, ".subject") 

356 

357 def get_preview_line(self, loc_context: LocalizationContext) -> str | None: 

358 if len(self.entries) != 1: 

359 return None 

360 return self.entries[0].latest_message_text 

361 

362 def get_body_blocks(self, loc_context: LocalizationContext) -> list[EmailBlock]: 

363 builder = self._body_builder(loc_context) 

364 builder.para(".purpose") 

365 for entry in self.entries: 

366 if entry.group_chat_title is None: 

367 builder.para(".count_in_dm", {"count": entry.missed_count, "author": entry.latest_message_author.name}) 

368 else: 

369 builder.para(".count_in_group", {"count": entry.missed_count, "group": entry.group_chat_title}) 

370 builder.user(entry.latest_message_author) 

371 builder.quote(entry.latest_message_text, markdown=False) 

372 builder.action(entry.view_url, ".view_action") 

373 return builder.build() 

374 

375 @classmethod 

376 def from_notification(cls, data: notification_data_pb2.ChatMissedMessages, *, user_name: str) -> Self: 

377 missed_entries = [ 

378 cls.Entry( 

379 group_chat_title=message.group_chat_title or None, 

380 missed_count=message.unseen_count, 

381 latest_message_author=UserInfo.from_protobuf(message.author), 

382 latest_message_text=message.text, 

383 view_url=urls.chat_link(chat_id=message.group_chat_id), 

384 ) 

385 for message in data.messages 

386 ] 

387 

388 return cls(user_name, entries=missed_entries) 

389 

390 @classmethod 

391 def test_instances(cls) -> list[Self]: 

392 entry_prototype = ChatMessagesMissedEmail.Entry( 

393 group_chat_title=None, 

394 missed_count=1, 

395 latest_message_author=UserInfo.dummy_bob(), 

396 latest_message_text="Hello!", 

397 view_url="https://couchers.org/messages/chats/123", 

398 ) 

399 return [ 

400 cls( 

401 user_name="Alice", 

402 entries=[ 

403 replace(entry_prototype, group_chat_title=None), 

404 replace(entry_prototype, group_chat_title="Best friends"), 

405 ], 

406 ) 

407 ] 

408 

409 

410@dataclass(kw_only=True, slots=True) 

411class DiscussionCreatedEmail(EmailBase): 

412 """Sent to a user when a new discussion is created in a community they follow.""" 

413 

414 author: UserInfo 

415 title: str 

416 parent_context: str # Community or group name 

417 markdown_text: str 

418 view_link: str 

419 

420 @property 

421 def string_key_base(self) -> str: 

422 return "discussions.created" 

423 

424 def get_subject_line(self, loc_context: LocalizationContext) -> str: 

425 return self._localize(loc_context, ".subject", {"author": self.author.name, "title": self.title}) 

426 

427 def get_preview_line(self, loc_context: LocalizationContext) -> str | None: 

428 return markdown_to_plaintext(self.markdown_text) 

429 

430 def get_body_blocks(self, loc_context: LocalizationContext) -> list[EmailBlock]: 

431 builder = self._body_builder(loc_context) 

432 builder.para( 

433 ".purpose", 

434 { 

435 "author": self.author.name, 

436 "parent_context": self.parent_context, 

437 }, 

438 ) 

439 builder.user(self.author) 

440 builder.block(ParaBlock(text=Markup(f"<b>{escape(self.title)}</b>"))) 

441 builder.quote(self.markdown_text, markdown=True) 

442 builder.action(self.view_link, ".view_action") 

443 return builder.build() 

444 

445 @classmethod 

446 def from_notification(cls, data: notification_data_pb2.DiscussionCreate, *, user_name: str) -> Self: 

447 discussion = data.discussion 

448 return cls( 

449 user_name=user_name, 

450 author=UserInfo.from_protobuf(data.author), 

451 title=discussion.title, 

452 parent_context=discussion.owner_title, 

453 markdown_text=discussion.content, 

454 view_link=urls.discussion_link(discussion_id=discussion.discussion_id, slug=discussion.slug), 

455 ) 

456 

457 @classmethod 

458 def test_instances(cls) -> list[Self]: 

459 return [ 

460 cls( 

461 user_name="Alice", 

462 author=UserInfo.dummy_bob(), 

463 title="Best hiking trails near Berlin", 

464 parent_context="Berlin", 

465 markdown_text="I've been exploring the area and found some **great** spots...", 

466 view_link="https://couchers.org/discussions/123", 

467 ) 

468 ] 

469 

470 

471@dataclass(kw_only=True, slots=True) 

472class DiscussionCommentEmail(EmailBase): 

473 """Sent to a user when someone comments on a discussion they follow.""" 

474 

475 author: UserInfo 

476 discussion_title: str 

477 discussion_parent_context: str # Community or group name 

478 markdown_text: str 

479 view_link: str 

480 

481 @property 

482 def string_key_base(self) -> str: 

483 return "discussions.comment" 

484 

485 def get_subject_line(self, loc_context: LocalizationContext) -> str: 

486 return self._localize( 

487 loc_context, ".subject", {"author": self.author.name, "discussion_title": self.discussion_title} 

488 ) 

489 

490 def get_preview_line(self, loc_context: LocalizationContext) -> str | None: 

491 return markdown_to_plaintext(self.markdown_text) 

492 

493 def get_body_blocks(self, loc_context: LocalizationContext) -> list[EmailBlock]: 

494 builder = self._body_builder(loc_context) 

495 builder.para( 

496 ".purpose", 

497 { 

498 "author": self.author.name, 

499 "discussion_title": self.discussion_title, 

500 "parent_context": self.discussion_parent_context, 

501 }, 

502 ) 

503 builder.user(self.author) 

504 builder.quote(self.markdown_text, markdown=True) 

505 builder.action(self.view_link, ".view_action") 

506 return builder.build() 

507 

508 @classmethod 

509 def from_notification(cls, data: notification_data_pb2.DiscussionComment, *, user_name: str) -> Self: 

510 discussion = data.discussion 

511 return cls( 

512 user_name=user_name, 

513 author=UserInfo.from_protobuf(data.author), 

514 discussion_title=discussion.title, 

515 discussion_parent_context=discussion.owner_title, 

516 markdown_text=data.reply.content, 

517 view_link=urls.discussion_link(discussion_id=discussion.discussion_id, slug=discussion.slug), 

518 ) 

519 

520 @classmethod 

521 def test_instances(cls) -> list[Self]: 

522 return [ 

523 cls( 

524 user_name="Alice", 

525 author=UserInfo.dummy_bob(), 

526 discussion_title="Best hiking trails near Berlin", 

527 discussion_parent_context="Berlin", 

528 markdown_text="Great recommendations, I also **love** the Grünewald forest!", 

529 view_link="https://couchers.org/discussions/123", 

530 ) 

531 ] 

532 

533 

534@dataclass(kw_only=True, slots=True) 

535class DonationReceivedEmail(EmailBase): 

536 """Sent to a user to thank them for a donation.""" 

537 

538 amount: int 

539 receipt_url: str 

540 

541 @property 

542 def string_key_base(self) -> str: 

543 return "donation_received" 

544 

545 def get_body_blocks(self, loc_context: LocalizationContext) -> list[EmailBlock]: 

546 builder = self._body_builder(loc_context, default_closing=False) 

547 builder.para(".purpose", {"amount_with_currency": f"${self.amount}"}) 

548 builder.para(".contribution_impact") 

549 builder.para(".invoice_receipt_info") 

550 builder.action(self.receipt_url, ".download_invoice") 

551 builder.para(".tax_acknowledgment") 

552 builder.para(".questions_contact", {"email_link": html_mailto_link("donations@couchers.org")}) 

553 builder.para("generic.thanks") 

554 builder.para("generic.closing_lines.founders") 

555 return builder.build() 

556 

557 @classmethod 

558 def from_notification(cls, data: notification_data_pb2.DonationReceived, *, user_name: str) -> Self: 

559 return cls(user_name=user_name, amount=data.amount, receipt_url=data.receipt_url) 

560 

561 @classmethod 

562 def test_instances(cls) -> list[Self]: 

563 return [cls(user_name="Alice", amount=25, receipt_url="https://couchers.org/receipts/123")] 

564 

565 

566@dataclass(kw_only=True, slots=True) 

567class EmailChangedEmail(EmailBase): 

568 """Sent to a user to notify them that their email address was changed.""" 

569 

570 new_email: str 

571 

572 @property 

573 def string_key_base(self) -> str: 

574 return "email_change.initiated" 

575 

576 def get_body_blocks(self, loc_context: LocalizationContext) -> list[EmailBlock]: 

577 builder = self._body_builder(loc_context, security_warning=True) 

578 builder.para(".purpose", {"email_address": self.new_email}) 

579 return builder.build() 

580 

581 @classmethod 

582 def from_notification(cls, data: notification_data_pb2.EmailAddressChange, *, user_name: str) -> Self: 

583 return cls(user_name=user_name, new_email=data.new_email) 

584 

585 @classmethod 

586 def test_instances(cls) -> list[Self]: 

587 return [cls(user_name="Alice", new_email="alice@example.com")] 

588 

589 

590@dataclass(kw_only=True, slots=True) 

591class EmailChangeConfirmationEmail(EmailBase): 

592 """Sent to a user to confirm their new email address.""" 

593 

594 old_email: str 

595 confirm_url: str 

596 

597 @property 

598 def string_key_base(self) -> str: 

599 return "email_change.confirmation" 

600 

601 def get_body_blocks(self, loc_context: LocalizationContext) -> list[EmailBlock]: 

602 builder = self._body_builder(loc_context, security_warning=True) 

603 builder.para(".purpose", {"old_email": self.old_email}) 

604 builder.action(self.confirm_url, ".confirm_action") 

605 return builder.build() 

606 

607 @classmethod 

608 def test_instances(cls) -> list[Self]: 

609 return [cls(user_name="Alice", old_email="alice@example.com", confirm_url="https://example.com")] 

610 

611 

612@dataclass(kw_only=True, slots=True) 

613class EmailVerifiedEmail(EmailBase): 

614 """Sent to a user to notify them that their new email address has been verified.""" 

615 

616 @property 

617 def string_key_base(self) -> str: 

618 return "email_change.verified" 

619 

620 def get_body_blocks(self, loc_context: LocalizationContext) -> list[EmailBlock]: 

621 builder = self._body_builder(loc_context, security_warning=True) 

622 builder.para(".purpose") 

623 return builder.build() 

624 

625 @classmethod 

626 def test_instances(cls) -> list[Self]: 

627 return [cls(user_name="Alice")] 

628 

629 

630@dataclass(kw_only=True, slots=True) 

631class EventInfo: 

632 """Common display fields for an event, extracted from its proto representation.""" 

633 

634 title: str 

635 start_time: datetime 

636 end_time: datetime 

637 address: str | None # The None case handles legacy online events 

638 view_url: str 

639 description_markdown: str 

640 

641 def get_details_block(self, loc_context: LocalizationContext) -> EmailBlock: 

642 # TODO(#8695): Support localized time ranges 

643 start_time_display = loc_context.localize_datetime(self.start_time, with_year=False, with_day_of_week=True) 

644 end_time_display = loc_context.localize_datetime(self.end_time, with_year=False, with_day_of_week=True) 

645 time_range_display = f"{start_time_display} - {end_time_display}" 

646 

647 html = f"<b>{escape(self.title)}</b>" 

648 html += "<br>" 

649 html += time_range_display 

650 if self.address: 

651 html += "<br>" 

652 html += f"<i>{escape(self.address)}</i>" 

653 

654 return ParaBlock(text=Markup(html)) 

655 

656 def get_description_block(self) -> EmailBlock: 

657 return QuoteBlock(text=Markup(self.description_markdown), markdown=True) 

658 

659 def get_view_action_block(self, loc_context: LocalizationContext) -> EmailBlock: 

660 view_action_text = loc_context.localize_string("events.generic.view_action", i18next=get_emails_i18next()) 

661 return ActionBlock(text=view_action_text, target_url=self.view_url) 

662 

663 @classmethod 

664 def from_proto(cls, event: events_pb2.Event) -> EventInfo: 

665 return cls( 

666 title=event.title, 

667 start_time=event.start_time.ToDatetime(tzinfo=UTC), 

668 end_time=event.end_time.ToDatetime(tzinfo=UTC), 

669 # Backcompat (2026-06): We might still have queued notifications referencing events with online_information. 

670 address=(event.location.address or None) if event.HasField("location") else None, 

671 view_url=urls.event_link(occurrence_id=event.event_id, slug=event.slug), 

672 description_markdown=event.content or "", 

673 ) 

674 

675 @staticmethod 

676 def dummy() -> EventInfo: 

677 return EventInfo( 

678 title="Berlin Meetup", 

679 start_time=datetime(2025, 7, 15, 18, 0, 0, tzinfo=UTC), 

680 end_time=datetime(2025, 7, 15, 21, 0, 0, tzinfo=UTC), 

681 address="Alexanderplatz, Berlin", 

682 view_url="https://couchers.org/events/123/berlin-community-meetup", 

683 description_markdown="Come join us for our monthly meetup!", 

684 ) 

685 

686 

687@dataclass(kw_only=True, slots=True) 

688class EventCreatedEmail(EmailBase): 

689 """Sent when a user is invited to an event (create_approved) or a new event is created (create_any).""" 

690 

691 inviting_user: UserInfo 

692 event_info: EventInfo 

693 community_name: str | None 

694 community_url: str | None 

695 is_invite: bool # True = create_approved (invitation), False = create_any 

696 

697 @property 

698 def string_key_base(self) -> str: 

699 return f"events.created.{'invitation' if self.is_invite else 'notification'}" 

700 

701 def get_subject_line(self, loc_context: LocalizationContext) -> str: 

702 return self._localize( 

703 loc_context, ".subject", {"user": self.inviting_user.name, "title": self.event_info.title} 

704 ) 

705 

706 def get_preview_line(self, loc_context: LocalizationContext) -> str | None: 

707 return markdown_to_plaintext(self.event_info.description_markdown) 

708 

709 def get_body_blocks(self, loc_context: LocalizationContext) -> list[EmailBlock]: 

710 builder = self._body_builder(loc_context) 

711 if self.community_name: 

712 builder.para(".purpose_with_community", {"user": self.inviting_user.name, "community": self.community_name}) 

713 else: 

714 builder.para(".purpose_no_community", {"user": self.inviting_user.name}) 

715 builder.block(self.event_info.get_details_block(loc_context)) 

716 builder.user(self.inviting_user) 

717 builder.block(self.event_info.get_description_block()) 

718 builder.block(self.event_info.get_view_action_block(loc_context)) 

719 return builder.build() 

720 

721 @classmethod 

722 def from_notification(cls, data: notification_data_pb2.EventCreate, *, user_name: str, is_invite: bool) -> Self: 

723 has_community = bool(data.in_community.community_id) 

724 community_url = ( 

725 urls.community_link(node_id=data.in_community.community_id, slug=data.in_community.slug) 

726 if has_community 

727 else None 

728 ) 

729 return cls( 

730 user_name=user_name, 

731 inviting_user=UserInfo.from_protobuf(data.inviting_user), 

732 event_info=EventInfo.from_proto(data.event), 

733 community_name=data.in_community.name if has_community else None, 

734 community_url=community_url, 

735 is_invite=is_invite, 

736 ) 

737 

738 @classmethod 

739 def test_instances(cls) -> list[Self]: 

740 prototype = cls( 

741 user_name="Alice", 

742 inviting_user=UserInfo.dummy_bob(), 

743 event_info=EventInfo.dummy(), 

744 community_name="Berlin", 

745 community_url="https://couchers.org/community/1/berlin-community", 

746 is_invite=True, 

747 ) 

748 return [ 

749 replace(prototype, is_invite=True), 

750 replace(prototype, is_invite=True, community_name=None, community_url=None), 

751 replace(prototype, is_invite=False), 

752 replace(prototype, is_invite=False, community_name=None, community_url=None), 

753 ] 

754 

755 

756@dataclass(kw_only=True, slots=True) 

757class EventUpdatedEmail(EmailBase): 

758 """Sent to subscribers when an event is updated.""" 

759 

760 updating_user: UserInfo 

761 event_info: EventInfo 

762 updated_items: list[notification_data_pb2.EventUpdateItem.ValueType] 

763 

764 @property 

765 def string_key_base(self) -> str: 

766 return "events.updated" 

767 

768 def get_subject_line(self, loc_context: LocalizationContext) -> str: 

769 return self._localize( 

770 loc_context, ".subject", {"user": self.updating_user.name, "title": self.event_info.title} 

771 ) 

772 

773 def get_body_blocks(self, loc_context: LocalizationContext) -> list[EmailBlock]: 

774 builder = self._body_builder(loc_context) 

775 

776 updated_items_string_keys = list( 

777 filter(None, (type(self)._updated_item_to_string_key(i) for i in self.updated_items)) 

778 ) 

779 if updated_items_string_keys: 

780 updated_items_text = loc_context.localize_list( 

781 [self._localize(loc_context, key) for key in updated_items_string_keys] 

782 ) 

783 builder.para(".purpose_with_items", {"user": self.updating_user.name, "items_list": updated_items_text}) 

784 else: 

785 builder.para(".purpose_generic", {"user": self.updating_user.name}) 

786 

787 builder.block(self.event_info.get_details_block(loc_context)) 

788 builder.user(self.updating_user) 

789 builder.block(self.event_info.get_description_block()) 

790 builder.block(self.event_info.get_view_action_block(loc_context)) 

791 return builder.build() 

792 

793 @classmethod 

794 def from_notification(cls, data: notification_data_pb2.EventUpdate, *, user_name: str) -> Self: 

795 updated_items: list[notification_data_pb2.EventUpdateItem.ValueType] = [] 

796 if data.updated_enum_items: 796 ↛ 798line 796 didn't jump to line 798 because the condition on line 796 was always true

797 updated_items.extend(data.updated_enum_items) 

798 elif data.updated_str_items: 

799 for updated_str_item in data.updated_str_items: 

800 if updated_enum_item := cls._updated_item_str_to_enum(updated_str_item): 

801 updated_items.append(updated_enum_item) 

802 

803 return cls( 

804 user_name=user_name, 

805 updating_user=UserInfo.from_protobuf(data.updating_user), 

806 event_info=EventInfo.from_proto(data.event), 

807 updated_items=updated_items, 

808 ) 

809 

810 # TODO(#9117): Backcompat. Remove update_str_items fallback once known unused. 

811 @staticmethod 

812 def _updated_item_str_to_enum(value: str) -> notification_data_pb2.EventUpdateItem.ValueType | None: 

813 match value: 

814 case "title": 

815 return notification_data_pb2.EventUpdateItem.EVENT_UPDATE_ITEM_TITLE 

816 case "content": 

817 return notification_data_pb2.EventUpdateItem.EVENT_UPDATE_ITEM_CONTENT 

818 case "location": 

819 return notification_data_pb2.EventUpdateItem.EVENT_UPDATE_ITEM_LOCATION 

820 case "start time": 

821 return notification_data_pb2.EventUpdateItem.EVENT_UPDATE_ITEM_START_TIME 

822 case "end time": 

823 return notification_data_pb2.EventUpdateItem.EVENT_UPDATE_ITEM_END_TIME 

824 case _: 

825 return None 

826 

827 @staticmethod 

828 def _updated_item_to_string_key(value: notification_data_pb2.EventUpdateItem.ValueType) -> str | None: 

829 match value: 

830 case notification_data_pb2.EventUpdateItem.EVENT_UPDATE_ITEM_TITLE: 

831 return ".item_names.title" 

832 case notification_data_pb2.EventUpdateItem.EVENT_UPDATE_ITEM_CONTENT: 

833 return ".item_names.content" 

834 case notification_data_pb2.EventUpdateItem.EVENT_UPDATE_ITEM_LOCATION: 

835 return ".item_names.location" 

836 case notification_data_pb2.EventUpdateItem.EVENT_UPDATE_ITEM_START_TIME: 

837 return ".item_names.start_time" 

838 case notification_data_pb2.EventUpdateItem.EVENT_UPDATE_ITEM_END_TIME: 

839 return ".item_names.end_time" 

840 case _: 

841 return None 

842 

843 @classmethod 

844 def test_instances(cls) -> list[Self]: 

845 prototype = cls( 

846 user_name="Alice", 

847 updating_user=UserInfo.dummy_bob(), 

848 event_info=EventInfo.dummy(), 

849 updated_items=[], 

850 ) 

851 return [ 

852 replace(prototype, updated_items=[]), 

853 replace(prototype, updated_items=notification_data_pb2.EventUpdateItem.values()), 

854 ] 

855 

856 

857@dataclass(kw_only=True, slots=True) 

858class EventOrganizerInvitedEmail(EmailBase): 

859 """Sent when a user is invited to co-organize an event.""" 

860 

861 inviting_user: UserInfo 

862 event_info: EventInfo 

863 

864 @property 

865 def string_key_base(self) -> str: 

866 return "events.organizer_invited" 

867 

868 def get_subject_line(self, loc_context: LocalizationContext) -> str: 

869 return self._localize( 

870 loc_context, 

871 ".subject", 

872 {"user": self.inviting_user.name, "title": self.event_info.title}, 

873 ) 

874 

875 def get_body_blocks(self, loc_context: LocalizationContext) -> list[EmailBlock]: 

876 builder = self._body_builder(loc_context) 

877 builder.para(".purpose", {"user": self.inviting_user.name, "title": self.event_info.title}) 

878 builder.block(self.event_info.get_details_block(loc_context)) 

879 builder.user(self.inviting_user) 

880 builder.block(self.event_info.get_description_block()) 

881 builder.block(self.event_info.get_view_action_block(loc_context)) 

882 builder.para(_do_not_reply_request_string_key, epilogue=True) 

883 return builder.build() 

884 

885 @classmethod 

886 def from_notification(cls, data: notification_data_pb2.EventInviteOrganizer, *, user_name: str) -> Self: 

887 return cls( 

888 user_name=user_name, 

889 inviting_user=UserInfo.from_protobuf(data.inviting_user), 

890 event_info=EventInfo.from_proto(data.event), 

891 ) 

892 

893 @classmethod 

894 def test_instances(cls) -> list[Self]: 

895 return [cls(user_name="Alice", inviting_user=UserInfo.dummy_bob(), event_info=EventInfo.dummy())] 

896 

897 

898@dataclass(kw_only=True, slots=True) 

899class EventCommentEmail(EmailBase): 

900 """Sent to subscribers when someone comments on an event.""" 

901 

902 author: UserInfo 

903 event_info: EventInfo 

904 comment_markdown: str 

905 

906 @property 

907 def string_key_base(self) -> str: 

908 return "events.comment" 

909 

910 def get_subject_line(self, loc_context: LocalizationContext) -> str: 

911 return self._localize(loc_context, ".subject", {"author": self.author.name, "title": self.event_info.title}) 

912 

913 def get_preview_line(self, loc_context: LocalizationContext) -> str | None: 

914 return markdown_to_plaintext(self.comment_markdown) 

915 

916 def get_body_blocks(self, loc_context: LocalizationContext) -> list[EmailBlock]: 

917 builder = self._body_builder(loc_context) 

918 builder.para(".purpose", {"author": self.author.name}) 

919 builder.block(self.event_info.get_details_block(loc_context)) 

920 builder.user(self.author) 

921 builder.quote(self.comment_markdown, markdown=True) 

922 builder.block(self.event_info.get_view_action_block(loc_context)) 

923 return builder.build() 

924 

925 @classmethod 

926 def from_notification(cls, data: notification_data_pb2.EventComment, *, user_name: str) -> Self: 

927 return cls( 

928 user_name=user_name, 

929 author=UserInfo.from_protobuf(data.author), 

930 event_info=EventInfo.from_proto(data.event), 

931 comment_markdown=data.reply.content, 

932 ) 

933 

934 @classmethod 

935 def test_instances(cls) -> list[Self]: 

936 return [ 

937 cls( 

938 user_name="Alice", 

939 author=UserInfo.dummy_bob(), 

940 event_info=EventInfo.dummy(), 

941 comment_markdown="Looking forward to it, see you all there!", 

942 ) 

943 ] 

944 

945 

946@dataclass(kw_only=True, slots=True) 

947class EventReminderEmail(EmailBase): 

948 """Sent to subscribers as a reminder that an event starts soon.""" 

949 

950 event_info: EventInfo 

951 

952 @property 

953 def string_key_base(self) -> str: 

954 return "events.reminder" 

955 

956 def get_subject_line(self, loc_context: LocalizationContext) -> str: 

957 return self._localize(loc_context, ".subject", {"title": self.event_info.title}) 

958 

959 def get_body_blocks(self, loc_context: LocalizationContext) -> list[EmailBlock]: 

960 builder = self._body_builder(loc_context) 

961 builder.para(".purpose") 

962 builder.block(self.event_info.get_details_block(loc_context)) 

963 builder.block(self.event_info.get_description_block()) 

964 builder.block(self.event_info.get_view_action_block(loc_context)) 

965 return builder.build() 

966 

967 @classmethod 

968 def from_notification(cls, data: notification_data_pb2.EventReminder, *, user_name: str) -> Self: 

969 return cls( 

970 user_name=user_name, 

971 event_info=EventInfo.from_proto(data.event), 

972 ) 

973 

974 @classmethod 

975 def test_instances(cls) -> list[Self]: 

976 return [cls(user_name="Alice", event_info=EventInfo.dummy())] 

977 

978 

979@dataclass(kw_only=True, slots=True) 

980class EventCancelledEmail(EmailBase): 

981 """Sent to subscribers when an event is cancelled.""" 

982 

983 cancelling_user: UserInfo 

984 event_info: EventInfo 

985 

986 @property 

987 def string_key_base(self) -> str: 

988 return "events.cancel" 

989 

990 def get_subject_line(self, loc_context: LocalizationContext) -> str: 

991 return self._localize( 

992 loc_context, 

993 ".subject", 

994 {"user": self.cancelling_user.name, "title": self.event_info.title}, 

995 ) 

996 

997 def get_body_blocks(self, loc_context: LocalizationContext) -> list[EmailBlock]: 

998 builder = self._body_builder(loc_context) 

999 builder.para(".purpose", {"user": self.cancelling_user.name}) 

1000 builder.block(self.event_info.get_details_block(loc_context)) 

1001 builder.user(self.cancelling_user) 

1002 builder.quote(self.event_info.description_markdown, markdown=True) 

1003 builder.block(self.event_info.get_view_action_block(loc_context)) 

1004 return builder.build() 

1005 

1006 @classmethod 

1007 def from_notification(cls, data: notification_data_pb2.EventCancel, *, user_name: str) -> Self: 

1008 return cls( 

1009 user_name=user_name, 

1010 cancelling_user=UserInfo.from_protobuf(data.cancelling_user), 

1011 event_info=EventInfo.from_proto(data.event), 

1012 ) 

1013 

1014 @classmethod 

1015 def test_instances(cls) -> list[Self]: 

1016 return [cls(user_name="Alice", cancelling_user=UserInfo.dummy_bob(), event_info=EventInfo.dummy())] 

1017 

1018 

1019@dataclass(kw_only=True, slots=True) 

1020class EventDeletedEmail(EmailBase): 

1021 """Sent to subscribers when a moderator deletes an event.""" 

1022 

1023 event_info: EventInfo 

1024 

1025 @property 

1026 def string_key_base(self) -> str: 

1027 return "events.deleted" 

1028 

1029 def get_subject_line(self, loc_context: LocalizationContext) -> str: 

1030 return self._localize(loc_context, ".subject", {"title": self.event_info.title}) 

1031 

1032 def get_body_blocks(self, loc_context: LocalizationContext) -> list[EmailBlock]: 

1033 builder = self._body_builder(loc_context) 

1034 builder.para(".purpose") 

1035 builder.block(self.event_info.get_details_block(loc_context)) 

1036 return builder.build() 

1037 

1038 @classmethod 

1039 def from_notification(cls, data: notification_data_pb2.EventDelete, *, user_name: str) -> Self: 

1040 return cls( 

1041 user_name=user_name, 

1042 event_info=EventInfo.from_proto(data.event), 

1043 ) 

1044 

1045 @classmethod 

1046 def test_instances(cls) -> list[Self]: 

1047 return [ 

1048 cls( 

1049 user_name="Alice", 

1050 event_info=EventInfo.dummy(), 

1051 ) 

1052 ] 

1053 

1054 

1055@dataclass(kw_only=True, slots=True) 

1056class FriendReferenceReceivedEmail(EmailBase): 

1057 """Sent to a user when they receive a friend reference.""" 

1058 

1059 from_user: UserInfo 

1060 text: str 

1061 

1062 @property 

1063 def string_key_base(self) -> str: 

1064 return "references.received.friend" 

1065 

1066 def get_subject_line(self, loc_context: LocalizationContext) -> str: 

1067 return self._localize(loc_context, ".subject", {"name": self.from_user.name}) 

1068 

1069 def get_preview_line(self, loc_context: LocalizationContext) -> str | None: 

1070 return self.text 

1071 

1072 def get_body_blocks(self, loc_context: LocalizationContext) -> list[EmailBlock]: 

1073 builder = self._body_builder(loc_context) 

1074 builder.para(".purpose", {"name": self.from_user.name}) 

1075 builder.user(self.from_user) 

1076 builder.quote(self.text, markdown=False) 

1077 builder.action(urls.profile_references_link(), "references.received.view_action") 

1078 return builder.build() 

1079 

1080 @classmethod 

1081 def from_notification(cls, data: notification_data_pb2.ReferenceReceiveFriend, *, user_name: str) -> Self: 

1082 return cls(user_name=user_name, from_user=UserInfo.from_protobuf(data.from_user), text=data.text) 

1083 

1084 @classmethod 

1085 def test_instances(cls) -> list[Self]: 

1086 return [ 

1087 cls( 

1088 user_name="Alice", 

1089 from_user=UserInfo.dummy_bob(), 

1090 text="Alice is a wonderful person and a great travel companion!", 

1091 ) 

1092 ] 

1093 

1094 

1095@dataclass(kw_only=True, slots=True) 

1096class FriendRequestReceivedEmail(EmailBase): 

1097 """Sent to a user when they receive a friend request.""" 

1098 

1099 befriender: UserInfo 

1100 

1101 @property 

1102 def string_key_base(self) -> str: 

1103 return "friend_requests.received" 

1104 

1105 def get_subject_line(self, loc_context: LocalizationContext) -> str: 

1106 return self._localize(loc_context, ".subject", {"name": self.befriender.name}) 

1107 

1108 def get_body_blocks(self, loc_context: LocalizationContext) -> list[EmailBlock]: 

1109 builder = self._body_builder(loc_context) 

1110 builder.para(".purpose", {"name": self.befriender.name}) 

1111 builder.user(self.befriender) 

1112 builder.action(urls.friend_requests_link(), ".view_action") 

1113 builder.para(".closing") 

1114 builder.para(_do_not_reply_request_string_key, epilogue=True) 

1115 return builder.build() 

1116 

1117 @classmethod 

1118 def from_notification(cls, data: notification_data_pb2.FriendRequestCreate, *, user_name: str) -> Self: 

1119 return cls(user_name=user_name, befriender=UserInfo.from_protobuf(data.other_user)) 

1120 

1121 @classmethod 

1122 def test_instances(cls) -> list[Self]: 

1123 return [ 

1124 cls( 

1125 user_name="Alice", 

1126 befriender=UserInfo.dummy_bob(), 

1127 ) 

1128 ] 

1129 

1130 

1131@dataclass(kw_only=True, slots=True) 

1132class FriendRequestAcceptedEmail(EmailBase): 

1133 """Sent to a user when their friend request is accepted.""" 

1134 

1135 new_friend: UserInfo 

1136 

1137 @property 

1138 def string_key_base(self) -> str: 

1139 return "friend_requests.accepted" 

1140 

1141 def get_subject_line(self, loc_context: LocalizationContext) -> str: 

1142 return self._localize(loc_context, ".subject", {"name": self.new_friend.name}) 

1143 

1144 def get_body_blocks(self, loc_context: LocalizationContext) -> list[EmailBlock]: 

1145 builder = self._body_builder(loc_context) 

1146 builder.para(".purpose", {"name": self.new_friend.name}) 

1147 builder.user(self.new_friend) 

1148 builder.action(self.new_friend.profile_url, ".view_action") 

1149 builder.para(".closing") 

1150 return builder.build() 

1151 

1152 @classmethod 

1153 def from_notification(cls, data: notification_data_pb2.FriendRequestAccept, *, user_name: str) -> Self: 

1154 return cls(user_name=user_name, new_friend=UserInfo.from_protobuf(data.other_user)) 

1155 

1156 @classmethod 

1157 def test_instances(cls) -> list[Self]: 

1158 return [ 

1159 cls( 

1160 user_name="Alice", 

1161 new_friend=UserInfo.dummy_bob(), 

1162 ) 

1163 ] 

1164 

1165 

1166@dataclass(kw_only=True, slots=True) 

1167class GenderChangedEmail(EmailBase): 

1168 """Sent to a user to notify them that their gender was changed.""" 

1169 

1170 new_gender: str 

1171 

1172 @property 

1173 def string_key_base(self) -> str: 

1174 return "gender_changed" 

1175 

1176 def get_body_blocks(self, loc_context: LocalizationContext) -> list[EmailBlock]: 

1177 builder = self._body_builder(loc_context, security_warning=True) 

1178 builder.para(".purpose", {"gender": self.new_gender}) 

1179 return builder.build() 

1180 

1181 @classmethod 

1182 def from_notification(cls, data: notification_data_pb2.GenderChange, *, user_name: str) -> Self: 

1183 return cls(user_name=user_name, new_gender=data.gender) 

1184 

1185 @classmethod 

1186 def test_instances(cls) -> list[Self]: 

1187 return [ 

1188 cls( 

1189 user_name="Alice", 

1190 new_gender="Male", 

1191 ) 

1192 ] 

1193 

1194 

1195@dataclass(kw_only=True, slots=True) 

1196class HostRequestCreatedEmail(EmailBase): 

1197 """Sent to a host when a surfer sends them a new host request.""" 

1198 

1199 surfer: UserInfo 

1200 from_date: date 

1201 to_date: date 

1202 text: str 

1203 view_link: str 

1204 quick_decline_link: str 

1205 

1206 @property 

1207 def string_key_base(self) -> str: 

1208 return "host_requests.created" 

1209 

1210 def get_subject_line(self, loc_context: LocalizationContext) -> str: 

1211 return self._localize(loc_context, ".subject", {"surfer_name": self.surfer.name}) 

1212 

1213 def get_preview_line(self, loc_context: LocalizationContext) -> str | None: 

1214 return self.text 

1215 

1216 def get_body_blocks(self, loc_context: LocalizationContext) -> list[EmailBlock]: 

1217 builder = self._body_builder(loc_context) 

1218 builder.para(".purpose", {"surfer_name": self.surfer.name}) 

1219 builder.user( 

1220 self.surfer, 

1221 "host_requests.generic.date_range", 

1222 { 

1223 "from_date": _localize_host_request_date(self.from_date, loc_context), 

1224 "to_date": _localize_host_request_date(self.to_date, loc_context), 

1225 }, 

1226 ) 

1227 builder.quote(self.text, markdown=False) 

1228 builder.action(self.view_link, "host_requests.generic.view_action") 

1229 builder.action(self.quick_decline_link, "host_requests.generic.quick_decline_action") 

1230 builder.para(".respond_encouragement") 

1231 builder.para(_do_not_reply_request_string_key, epilogue=True) 

1232 return builder.build() 

1233 

1234 @classmethod 

1235 def from_notification(cls, data: notification_data_pb2.HostRequestCreate, *, user_name: str) -> Self: 

1236 return cls( 

1237 user_name, 

1238 surfer=UserInfo.from_protobuf(data.surfer), 

1239 from_date=date.fromisoformat(data.host_request.from_date), 

1240 to_date=date.fromisoformat(data.host_request.to_date), 

1241 text=data.text, 

1242 view_link=urls.host_request(host_request_id=data.host_request.host_request_id), 

1243 quick_decline_link=generate_quick_decline_link(data.host_request), 

1244 ) 

1245 

1246 @classmethod 

1247 def test_instances(cls) -> list[Self]: 

1248 return [ 

1249 cls( 

1250 user_name="Alice", 

1251 surfer=UserInfo.dummy_bob(), 

1252 from_date=date(2025, 6, 1), 

1253 to_date=date(2025, 6, 7), 

1254 text="Hey, I'd love to stay for a few nights!", 

1255 view_link="https://couchers.org/requests/123", 

1256 quick_decline_link="https://couchers.org/requests/123/decline?token=xxx", 

1257 ) 

1258 ] 

1259 

1260 

1261@dataclass(kw_only=True, slots=True) 

1262class HostRequestReminderEmail(EmailBase): 

1263 """Sent to a host as a reminder to respond to a pending host request.""" 

1264 

1265 surfer: UserInfo 

1266 from_date: date 

1267 to_date: date 

1268 view_link: str 

1269 quick_decline_link: str 

1270 

1271 @property 

1272 def string_key_base(self) -> str: 

1273 return "host_requests.reminder" 

1274 

1275 def get_subject_line(self, loc_context: LocalizationContext) -> str: 

1276 return self._localize(loc_context, ".subject", {"surfer_name": self.surfer.name}) 

1277 

1278 def get_body_blocks(self, loc_context: LocalizationContext) -> list[EmailBlock]: 

1279 builder = self._body_builder(loc_context) 

1280 builder.para(".purpose", {"surfer_name": self.surfer.name}) 

1281 builder.user( 

1282 self.surfer, 

1283 "host_requests.generic.date_range", 

1284 { 

1285 "from_date": _localize_host_request_date(self.from_date, loc_context), 

1286 "to_date": _localize_host_request_date(self.to_date, loc_context), 

1287 }, 

1288 ) 

1289 builder.action(self.view_link, "host_requests.generic.view_action") 

1290 builder.action(self.quick_decline_link, "host_requests.generic.quick_decline_action") 

1291 builder.para(_do_not_reply_request_string_key, epilogue=True) 

1292 return builder.build() 

1293 

1294 @classmethod 

1295 def from_notification(cls, data: notification_data_pb2.HostRequestReminder, *, user_name: str) -> Self: 

1296 return cls( 

1297 user_name, 

1298 surfer=UserInfo.from_protobuf(data.surfer), 

1299 from_date=date.fromisoformat(data.host_request.from_date), 

1300 to_date=date.fromisoformat(data.host_request.to_date), 

1301 view_link=urls.host_request(host_request_id=data.host_request.host_request_id), 

1302 quick_decline_link=generate_quick_decline_link(data.host_request), 

1303 ) 

1304 

1305 @classmethod 

1306 def test_instances(cls) -> list[Self]: 

1307 return [ 

1308 cls( 

1309 user_name="Alice", 

1310 surfer=UserInfo.dummy_bob(), 

1311 from_date=date(2025, 6, 1), 

1312 to_date=date(2025, 6, 7), 

1313 view_link="https://couchers.org/requests/123", 

1314 quick_decline_link="https://couchers.org/requests/123/decline?token=xxx", 

1315 ) 

1316 ] 

1317 

1318 

1319@dataclass(kw_only=True, slots=True) 

1320class HostRequestMessageEmail(EmailBase): 

1321 """Sent when a user sends a message in an existing host request.""" 

1322 

1323 other_user: UserInfo 

1324 from_date: date 

1325 to_date: date 

1326 text: str 

1327 from_host: bool 

1328 view_link: str 

1329 

1330 @property 

1331 def string_key_base(self) -> str: 

1332 variant = "from_host" if self.from_host else "from_surfer" 

1333 return f"host_requests.message.{variant}" 

1334 

1335 def get_subject_line(self, loc_context: LocalizationContext) -> str: 

1336 return self._localize(loc_context, ".subject", {"other_name": self.other_user.name}) 

1337 

1338 def get_preview_line(self, loc_context: LocalizationContext) -> str | None: 

1339 return self.text 

1340 

1341 def get_body_blocks(self, loc_context: LocalizationContext) -> list[EmailBlock]: 

1342 builder = self._body_builder(loc_context) 

1343 builder.para(".purpose", {"other_name": self.other_user.name}) 

1344 builder.user( 

1345 self.other_user, 

1346 "host_requests.generic.date_range", 

1347 { 

1348 "from_date": _localize_host_request_date(self.from_date, loc_context), 

1349 "to_date": _localize_host_request_date(self.to_date, loc_context), 

1350 }, 

1351 ) 

1352 builder.quote(self.text, markdown=False) 

1353 builder.action(self.view_link, "host_requests.generic.view_action") 

1354 builder.para(_do_not_reply_request_string_key, epilogue=True) 

1355 return builder.build() 

1356 

1357 @classmethod 

1358 def from_notification(cls, data: notification_data_pb2.HostRequestMessage, *, user_name: str) -> Self: 

1359 return cls( 

1360 user_name, 

1361 other_user=UserInfo.from_protobuf(data.user), 

1362 from_date=date.fromisoformat(data.host_request.from_date), 

1363 to_date=date.fromisoformat(data.host_request.to_date), 

1364 text=data.text, 

1365 from_host=not data.am_host, 

1366 view_link=urls.host_request(host_request_id=data.host_request.host_request_id), 

1367 ) 

1368 

1369 @classmethod 

1370 def test_instances(cls) -> list[Self]: 

1371 prototype = cls( 

1372 user_name="Alice", 

1373 other_user=UserInfo.dummy_bob(), 

1374 from_date=date(2025, 6, 1), 

1375 to_date=date(2025, 6, 7), 

1376 text="Looking forward to it, see you soon!", 

1377 from_host=True, 

1378 view_link="https://couchers.org/requests/123", 

1379 ) 

1380 return [replace(prototype, from_host=True), replace(prototype, from_host=False)] 

1381 

1382 

1383@dataclass(kw_only=True, slots=True) 

1384class HostRequestMissedMessagesEmail(EmailBase): 

1385 """Sent as a digest when a user has missed messages in a host request.""" 

1386 

1387 other_user: UserInfo 

1388 from_date: date 

1389 to_date: date 

1390 from_host: bool 

1391 view_link: str 

1392 

1393 @property 

1394 def string_key_base(self) -> str: 

1395 variant = "from_host" if self.from_host else "from_surfer" 

1396 return f"host_requests.missed_messages.{variant}" 

1397 

1398 def get_subject_line(self, loc_context: LocalizationContext) -> str: 

1399 return self._localize(loc_context, ".subject", {"other_name": self.other_user.name}) 

1400 

1401 def get_body_blocks(self, loc_context: LocalizationContext) -> list[EmailBlock]: 

1402 builder = self._body_builder(loc_context) 

1403 builder.para(".purpose", {"other_name": self.other_user.name}) 

1404 builder.user( 

1405 self.other_user, 

1406 "host_requests.generic.date_range", 

1407 { 

1408 "from_date": _localize_host_request_date(self.from_date, loc_context), 

1409 "to_date": _localize_host_request_date(self.to_date, loc_context), 

1410 }, 

1411 ) 

1412 builder.action(self.view_link, "host_requests.generic.view_action") 

1413 builder.para(_do_not_reply_request_string_key, epilogue=True) 

1414 return builder.build() 

1415 

1416 @classmethod 

1417 def from_notification(cls, data: notification_data_pb2.HostRequestMissedMessages, *, user_name: str) -> Self: 

1418 return cls( 

1419 user_name, 

1420 other_user=UserInfo.from_protobuf(data.user), 

1421 from_date=date.fromisoformat(data.host_request.from_date), 

1422 to_date=date.fromisoformat(data.host_request.to_date), 

1423 from_host=not data.am_host, 

1424 view_link=urls.host_request(host_request_id=data.host_request.host_request_id), 

1425 ) 

1426 

1427 @classmethod 

1428 def test_instances(cls) -> list[Self]: 

1429 prototype = cls( 

1430 user_name="Alice", 

1431 other_user=UserInfo.dummy_bob(), 

1432 from_date=date(2025, 6, 1), 

1433 to_date=date(2025, 6, 7), 

1434 from_host=True, 

1435 view_link="https://couchers.org/requests/123", 

1436 ) 

1437 return [replace(prototype, from_host=True), replace(prototype, from_host=False)] 

1438 

1439 

1440@dataclass(kw_only=True, slots=True) 

1441class HostRequestStatusChangedEmail(EmailBase): 

1442 """Sent when a host request is accepted, declined, confirmed, or cancelled.""" 

1443 

1444 other_user: UserInfo 

1445 from_date: date 

1446 to_date: date 

1447 new_status: conversations_pb2.HostRequestStatus.ValueType 

1448 view_link: str 

1449 

1450 @property 

1451 def string_key_base(self) -> str: 

1452 base_key = "host_requests.status_changed" 

1453 match self.new_status: 

1454 case conversations_pb2.HOST_REQUEST_STATUS_ACCEPTED: 

1455 return f"{base_key}.accepted_by_host" 

1456 case conversations_pb2.HOST_REQUEST_STATUS_REJECTED: 

1457 return f"{base_key}.declined_by_host" 

1458 case conversations_pb2.HOST_REQUEST_STATUS_CONFIRMED: 

1459 return f"{base_key}.confirmed_by_surfer" 

1460 case conversations_pb2.HOST_REQUEST_STATUS_CANCELLED: 1460 ↛ 1462line 1460 didn't jump to line 1462 because the pattern on line 1460 always matched

1461 return f"{base_key}.cancelled_by_surfer" 

1462 case _: 

1463 raise ValueError(f"Unexpected host request status: {self.new_status}") 

1464 

1465 def get_subject_line(self, loc_context: LocalizationContext) -> str: 

1466 return self._localize(loc_context, ".subject", {"other_name": self.other_user.name}) 

1467 

1468 def get_body_blocks(self, loc_context: LocalizationContext) -> list[EmailBlock]: 

1469 builder = self._body_builder(loc_context) 

1470 builder.para(".purpose", {"other_name": self.other_user.name}) 

1471 builder.user( 

1472 self.other_user, 

1473 "host_requests.generic.date_range", 

1474 { 

1475 "from_date": _localize_host_request_date(self.from_date, loc_context), 

1476 "to_date": _localize_host_request_date(self.to_date, loc_context), 

1477 }, 

1478 ) 

1479 builder.action(self.view_link, "host_requests.generic.view_action") 

1480 builder.para(_do_not_reply_request_string_key, epilogue=True) 

1481 return builder.build() 

1482 

1483 @classmethod 

1484 def from_notification( 

1485 cls, 

1486 data: notification_data_pb2.HostRequestAccept 

1487 | notification_data_pb2.HostRequestReject 

1488 | notification_data_pb2.HostRequestConfirm 

1489 | notification_data_pb2.HostRequestCancel, 

1490 *, 

1491 user_name: str, 

1492 ) -> Self: 

1493 other_user: UserInfo 

1494 new_status: conversations_pb2.HostRequestStatus.ValueType 

1495 match data: 

1496 case notification_data_pb2.HostRequestAccept(): 

1497 other_user = UserInfo.from_protobuf(data.host) 

1498 new_status = conversations_pb2.HostRequestStatus.HOST_REQUEST_STATUS_ACCEPTED 

1499 case notification_data_pb2.HostRequestReject(): 1499 ↛ 1500line 1499 didn't jump to line 1500 because the pattern on line 1499 never matched

1500 other_user = UserInfo.from_protobuf(data.host) 

1501 new_status = conversations_pb2.HostRequestStatus.HOST_REQUEST_STATUS_REJECTED 

1502 case notification_data_pb2.HostRequestConfirm(): 

1503 other_user = UserInfo.from_protobuf(data.surfer) 

1504 new_status = conversations_pb2.HostRequestStatus.HOST_REQUEST_STATUS_CONFIRMED 

1505 case notification_data_pb2.HostRequestCancel(): 1505 ↛ 1508line 1505 didn't jump to line 1508 because the pattern on line 1505 always matched

1506 other_user = UserInfo.from_protobuf(data.surfer) 

1507 new_status = conversations_pb2.HostRequestStatus.HOST_REQUEST_STATUS_CANCELLED 

1508 case _: 

1509 # Enable mypy's exhaustiveness checking 

1510 assert_never("Unexpected host request status changed notification data type.") 

1511 

1512 return cls( 

1513 user_name, 

1514 other_user=other_user, 

1515 from_date=date.fromisoformat(data.host_request.from_date), 

1516 to_date=date.fromisoformat(data.host_request.to_date), 

1517 new_status=new_status, 

1518 view_link=urls.host_request(host_request_id=data.host_request.host_request_id), 

1519 ) 

1520 

1521 @classmethod 

1522 def test_instances(cls) -> list[Self]: 

1523 prototype = cls( 

1524 user_name="Alice", 

1525 other_user=UserInfo.dummy_bob(), 

1526 from_date=date(2025, 6, 1), 

1527 to_date=date(2025, 6, 7), 

1528 new_status=conversations_pb2.HOST_REQUEST_STATUS_ACCEPTED, 

1529 view_link="https://couchers.org/requests/123", 

1530 ) 

1531 return [ 

1532 replace(prototype, new_status=conversations_pb2.HOST_REQUEST_STATUS_ACCEPTED), 

1533 replace(prototype, new_status=conversations_pb2.HOST_REQUEST_STATUS_REJECTED), 

1534 replace(prototype, new_status=conversations_pb2.HOST_REQUEST_STATUS_CONFIRMED), 

1535 replace(prototype, new_status=conversations_pb2.HOST_REQUEST_STATUS_CANCELLED), 

1536 ] 

1537 

1538 

1539@dataclass(kw_only=True, slots=True) 

1540class HostReferenceReceivedEmail(EmailBase): 

1541 """Sent to a user when they receive a reference from a past host or surfer.""" 

1542 

1543 from_user: UserInfo 

1544 text: str | None # None if hidden because receiver hasn't written their reference yet. 

1545 surfed: bool # True if I was the surfer, False if I was the host 

1546 leave_reference_url: str 

1547 

1548 @property 

1549 def string_key_base(self) -> str: 

1550 return "references.received" 

1551 

1552 @property 

1553 def string_role_subkey(self) -> str: 

1554 return "surfed" if self.surfed else "hosted" 

1555 

1556 def get_subject_line(self, loc_context: LocalizationContext) -> str: 

1557 return self._localize(loc_context, ".subject", {"name": self.from_user.name}) 

1558 

1559 def get_preview_line(self, loc_context: LocalizationContext) -> str | None: 

1560 return self.text 

1561 

1562 def get_body_blocks(self, loc_context: LocalizationContext) -> list[EmailBlock]: 

1563 builder = self._body_builder(loc_context) 

1564 builder.para(f".{self.string_role_subkey}.purpose", {"name": self.from_user.name}) 

1565 builder.user(self.from_user) 

1566 if self.text: 

1567 builder.quote(self.text, markdown=False) 

1568 builder.action(urls.profile_references_link(), ".view_action") 

1569 else: 

1570 builder.para(f".{self.string_role_subkey}.reciprocate_encouragement", {"name": self.from_user.name}) 

1571 builder.action(self.leave_reference_url, "references.write_action", {"name": self.from_user.name}) 

1572 return builder.build() 

1573 

1574 @classmethod 

1575 def from_notification( 

1576 cls, data: notification_data_pb2.ReferenceReceiveHostRequest, *, user_name: str, surfed: bool 

1577 ) -> Self: 

1578 return cls( 

1579 user_name=user_name, 

1580 from_user=UserInfo.from_protobuf(data.from_user), 

1581 text=data.text or None, 

1582 surfed=surfed, 

1583 leave_reference_url=urls.leave_reference_link( 

1584 reference_type="surfed" if surfed else "hosted", 

1585 to_user_id=str(data.from_user.user_id), 

1586 host_request_id=str(data.host_request_id), 

1587 ), 

1588 ) 

1589 

1590 @classmethod 

1591 def test_instances(cls) -> list[Self]: 

1592 prototype = cls( 

1593 user_name="Alice", 

1594 from_user=UserInfo.dummy_bob(), 

1595 text="Alice was a fantastic guest!", 

1596 surfed=True, 

1597 leave_reference_url="https://couchers.org/leave-reference/123", 

1598 ) 

1599 return [ 

1600 replace(prototype, surfed=True, text="Alice was a fantastic guest!"), 

1601 replace(prototype, surfed=True, text=None), 

1602 replace(prototype, surfed=False, text="Bob was a wonderful host!"), 

1603 replace(prototype, surfed=False, text=None), 

1604 ] 

1605 

1606 

1607@dataclass(kw_only=True, slots=True) 

1608class HostReferenceReminderEmail(EmailBase): 

1609 """Sent as a reminder to write a reference after a stay.""" 

1610 

1611 other_user: UserInfo 

1612 days_left: int 

1613 surfed: bool # True if I was the surfer, False if I was the host 

1614 leave_reference_url: str 

1615 

1616 @property 

1617 def string_key_base(self) -> str: 

1618 return "references.reminder" 

1619 

1620 @property 

1621 def string_role_subkey(self) -> str: 

1622 return "surfed" if self.surfed else "hosted" 

1623 

1624 def get_subject_line(self, loc_context: LocalizationContext) -> str: 

1625 return self._localize( 

1626 loc_context, 

1627 ".subject_days", 

1628 {"name": self.other_user.name, "count": self.days_left}, 

1629 ) 

1630 

1631 def get_body_blocks(self, loc_context: LocalizationContext) -> list[EmailBlock]: 

1632 builder = self._body_builder(loc_context) 

1633 builder.para( 

1634 f".{self.string_role_subkey}.purpose_days", {"name": self.other_user.name, "count": self.days_left} 

1635 ) 

1636 builder.user(self.other_user) 

1637 builder.action( 

1638 self.leave_reference_url, 

1639 "references.write_action", 

1640 {"name": self.other_user.name}, 

1641 ) 

1642 builder.para(".no_meeting_note", {"name": self.other_user.name}) 

1643 builder.para(".visibility_note") 

1644 return builder.build() 

1645 

1646 @classmethod 

1647 def from_notification(cls, data: notification_data_pb2.ReferenceReminder, *, user_name: str, surfed: bool) -> Self: 

1648 return cls( 

1649 user_name=user_name, 

1650 other_user=UserInfo.from_protobuf(data.other_user), 

1651 days_left=data.days_left, 

1652 surfed=surfed, 

1653 leave_reference_url=urls.leave_reference_link( 

1654 reference_type="surfed" if surfed else "hosted", 

1655 to_user_id=str(data.other_user.user_id), 

1656 host_request_id=str(data.host_request_id), 

1657 ), 

1658 ) 

1659 

1660 @classmethod 

1661 def test_instances(cls) -> list[Self]: 

1662 prototype = cls( 

1663 user_name="Alice", 

1664 other_user=UserInfo.dummy_bob(), 

1665 days_left=7, 

1666 surfed=True, 

1667 leave_reference_url="https://couchers.org/leave-reference/123", 

1668 ) 

1669 return [replace(prototype, surfed=True), replace(prototype, surfed=False)] 

1670 

1671 

1672@dataclass(kw_only=True, slots=True) 

1673class ModeratorNoteEmail(EmailBase): 

1674 """Sent to a user to notify them they have received a moderator note.""" 

1675 

1676 @property 

1677 def string_key_base(self) -> str: 

1678 return "moderator_note" 

1679 

1680 def get_body_blocks(self, loc_context: LocalizationContext) -> list[EmailBlock]: 

1681 builder = self._body_builder(loc_context) 

1682 builder.para(".purpose") 

1683 # Users with moderator notes are "jailed": any URL will show the note before 

1684 # letting them use the platform. 

1685 builder.action(urls.dashboard_link(), ".view_action") 

1686 return builder.build() 

1687 

1688 @classmethod 

1689 def test_instances(cls) -> list[Self]: 

1690 return [cls(user_name="Alice")] 

1691 

1692 

1693@dataclass(kw_only=True, slots=True) 

1694class NewBlogPostEmail(EmailBase): 

1695 """Sent to notify users of a new blog post.""" 

1696 

1697 title: str 

1698 blurb: str 

1699 url: str 

1700 

1701 @property 

1702 def string_key_base(self) -> str: 

1703 return "new_blog_post" 

1704 

1705 def get_subject_line(self, loc_context: LocalizationContext) -> str: 

1706 return self._localize(loc_context, ".subject", {"title": self.title}) 

1707 

1708 def get_preview_line(self, loc_context: LocalizationContext) -> str | None: 

1709 return self.blurb 

1710 

1711 def get_body_blocks(self, loc_context: LocalizationContext) -> list[EmailBlock]: 

1712 builder = self._body_builder(loc_context) 

1713 builder.para(".purpose") 

1714 builder.block(ParaBlock(text=Markup(f"<b>{escape(self.title)}</b>"))) 

1715 builder.quote(self.blurb, markdown=False) 

1716 builder.action(self.url, ".read_action") 

1717 return builder.build() 

1718 

1719 @classmethod 

1720 def from_notification(cls, data: notification_data_pb2.GeneralNewBlogPost, *, user_name: str) -> Self: 

1721 return cls(user_name=user_name, title=data.title, blurb=data.blurb, url=data.url) 

1722 

1723 @classmethod 

1724 def test_instances(cls) -> list[Self]: 

1725 return [ 

1726 cls( 

1727 user_name="Alice", 

1728 title="Exciting new features on Couchers.org", 

1729 blurb="We've launched some great new features including improved messaging and event discovery.", 

1730 url="https://couchers.org/blog/2025/01/01/new-features", 

1731 ) 

1732 ] 

1733 

1734 

1735@dataclass(kw_only=True, slots=True) 

1736class OnboardingReminderEmail(EmailBase): 

1737 """Onboarding email sent to new users; initial=True for the first email, False for the second.""" 

1738 

1739 initial: bool 

1740 

1741 @property 

1742 def string_key_base(self) -> str: 

1743 return f"onboarding_reminder.{'initial' if self.initial else 'follow_up'}" 

1744 

1745 def get_body_blocks(self, loc_context: LocalizationContext) -> list[EmailBlock]: 

1746 builder = self._body_builder(loc_context, default_closing=False) 

1747 edit_profile_url = urls.edit_profile_link() 

1748 if self.initial: 

1749 builder.para(".welcome") 

1750 builder.para(".early_user_role") 

1751 builder.para(".fill_in_profile") 

1752 builder.para(".edit_profile_prompt") 

1753 builder.action(edit_profile_url, "onboarding_reminder.edit_profile_action") 

1754 builder.para(".share_with_friends") 

1755 builder.para(".link", {"link": html_link(urls.app_link())}) 

1756 builder.para(".platform_under_development") 

1757 builder.para(".thanks_for_joining") 

1758 builder.para( 

1759 "generic.closing_lines.aapeli", 

1760 { 

1761 "profile_link": html_link("https://couchers.org/user/aapeli"), 

1762 }, 

1763 ) 

1764 else: 

1765 builder.para(".intro") 

1766 builder.para(".request") 

1767 builder.action(edit_profile_url, "onboarding_reminder.edit_profile_action") 

1768 builder.para("generic.thanks") 

1769 builder.para( 

1770 "generic.closing_lines.emily", 

1771 { 

1772 "email_link": html_mailto_link("community@couchers.org"), 

1773 "profile_link": html_link("https://couchers.org/user/emily"), 

1774 }, 

1775 ) 

1776 return builder.build() 

1777 

1778 @classmethod 

1779 def test_instances(cls) -> list[Self]: 

1780 prototype = cls(user_name="Alice", initial=True) 

1781 return [replace(prototype, initial=True), replace(prototype, initial=False)] 

1782 

1783 

1784@dataclass(kw_only=True, slots=True) 

1785class PasswordChangedEmail(EmailBase): 

1786 """Sent to a user to notify them that their login password was changed.""" 

1787 

1788 @property 

1789 def string_key_base(self) -> str: 

1790 return "password_changed" 

1791 

1792 def get_body_blocks(self, loc_context: LocalizationContext) -> list[EmailBlock]: 

1793 builder = self._body_builder(loc_context, security_warning=True) 

1794 builder.para(".purpose") 

1795 return builder.build() 

1796 

1797 @classmethod 

1798 def test_instances(cls) -> list[Self]: 

1799 return [cls(user_name="Alice")] 

1800 

1801 

1802@dataclass(kw_only=True, slots=True) 

1803class PasswordResetCompletedEmail(EmailBase): 

1804 """Sent to a user to confirm their password was successfully reset.""" 

1805 

1806 @property 

1807 def string_key_base(self) -> str: 

1808 return "password_reset.completed" 

1809 

1810 def get_body_blocks(self, loc_context: LocalizationContext) -> list[EmailBlock]: 

1811 builder = self._body_builder(loc_context, security_warning=True) 

1812 builder.para(".purpose") 

1813 return builder.build() 

1814 

1815 @classmethod 

1816 def test_instances(cls) -> list[Self]: 

1817 return [cls(user_name="Alice")] 

1818 

1819 

1820@dataclass(kw_only=True, slots=True) 

1821class PasswordResetStartedEmail(EmailBase): 

1822 """Sent to a user with a link to complete their password reset.""" 

1823 

1824 password_reset_link: str 

1825 

1826 @property 

1827 def string_key_base(self) -> str: 

1828 return "password_reset.started" 

1829 

1830 def get_body_blocks(self, loc_context: LocalizationContext) -> list[EmailBlock]: 

1831 builder = self._body_builder(loc_context, security_warning=True) 

1832 builder.para(".purpose") 

1833 builder.action(self.password_reset_link, ".reset_action") 

1834 return builder.build() 

1835 

1836 @classmethod 

1837 def from_notification(cls, data: notification_data_pb2.PasswordResetStart, *, user_name: str) -> Self: 

1838 return cls( 

1839 user_name=user_name, 

1840 password_reset_link=urls.password_reset_link(password_reset_token=data.password_reset_token), 

1841 ) 

1842 

1843 @classmethod 

1844 def test_instances(cls) -> list[Self]: 

1845 return [cls(user_name="Alice", password_reset_link="https://couchers.org/reset-password")] 

1846 

1847 

1848@dataclass(kw_only=True, slots=True) 

1849class PhoneNumberChangeEmail(EmailBase): 

1850 """Sent to a user to notify them that their phone number verification status was changed.""" 

1851 

1852 new_phone_number: str 

1853 completed: bool # False = started, True = completed 

1854 

1855 @property 

1856 def string_key_base(self) -> str: 

1857 return "phone_number_verification.verified" if self.completed else "phone_number_verification.started" 

1858 

1859 def get_body_blocks(self, loc_context: LocalizationContext) -> list[EmailBlock]: 

1860 builder = self._body_builder(loc_context, security_warning=True) 

1861 builder.para(".purpose", {"phone_number": format_phone_number(self.new_phone_number)}) 

1862 return builder.build() 

1863 

1864 @classmethod 

1865 def from_change_notification(cls, data: notification_data_pb2.PhoneNumberChange, *, user_name: str) -> Self: 

1866 return cls(user_name=user_name, new_phone_number=data.phone, completed=False) 

1867 

1868 @classmethod 

1869 def from_verify_notification(cls, data: notification_data_pb2.PhoneNumberVerify, *, user_name: str) -> Self: 

1870 return cls(user_name=user_name, new_phone_number=data.phone, completed=True) 

1871 

1872 @classmethod 

1873 def test_instances(cls) -> list[Self]: 

1874 prototype = cls( 

1875 user_name="Alice", 

1876 new_phone_number="+12223334444", 

1877 completed=False, 

1878 ) 

1879 return [replace(prototype, completed=False), replace(prototype, completed=True)] 

1880 

1881 

1882@dataclass(kw_only=True, slots=True) 

1883class PostalVerificationFailedEmail(EmailBase): 

1884 """Sent to a user when their postal verification attempt has failed.""" 

1885 

1886 reason: notification_data_pb2.PostalVerificationFailReason.ValueType 

1887 

1888 @property 

1889 def string_key_base(self) -> str: 

1890 return "postal_verification.failed" 

1891 

1892 def get_body_blocks(self, loc_context: LocalizationContext) -> list[EmailBlock]: 

1893 builder = self._body_builder(loc_context, security_warning=True) 

1894 match self.reason: 

1895 case notification_data_pb2.POSTAL_VERIFICATION_FAIL_REASON_CODE_EXPIRED: 

1896 purpose_string_key = ".purpose.code_expired" 

1897 case notification_data_pb2.POSTAL_VERIFICATION_FAIL_REASON_TOO_MANY_ATTEMPTS: 

1898 purpose_string_key = ".purpose.too_many_attempts" 

1899 case _: 

1900 purpose_string_key = ".purpose.default" 

1901 builder.para(purpose_string_key) 

1902 builder.action(urls.account_settings_link(), ".restart_action") 

1903 return builder.build() 

1904 

1905 @classmethod 

1906 def from_notification(cls, data: notification_data_pb2.PostalVerificationFailed, *, user_name: str) -> Self: 

1907 return cls(user_name=user_name, reason=data.reason) 

1908 

1909 @classmethod 

1910 def test_instances(cls) -> list[Self]: 

1911 prototype = cls( 

1912 user_name="Alice", 

1913 reason=notification_data_pb2.POSTAL_VERIFICATION_FAIL_REASON_CODE_EXPIRED, 

1914 ) 

1915 return [ 

1916 replace(prototype, reason=notification_data_pb2.POSTAL_VERIFICATION_FAIL_REASON_CODE_EXPIRED), 

1917 replace(prototype, reason=notification_data_pb2.POSTAL_VERIFICATION_FAIL_REASON_TOO_MANY_ATTEMPTS), 

1918 replace(prototype, reason=notification_data_pb2.POSTAL_VERIFICATION_FAIL_REASON_UNKNOWN), 

1919 ] 

1920 

1921 

1922@dataclass(kw_only=True, slots=True) 

1923class PostalVerificationPostcardSentEmail(EmailBase): 

1924 """Sent to a user to notify them that their verification postcard has been sent.""" 

1925 

1926 city: str 

1927 country: str 

1928 

1929 @property 

1930 def string_key_base(self) -> str: 

1931 return "postal_verification.postcard_sent" 

1932 

1933 def get_body_blocks(self, loc_context: LocalizationContext) -> list[EmailBlock]: 

1934 builder = self._body_builder(loc_context, security_warning=True) 

1935 builder.para(".purpose", {"city": self.city, "country": self.country}) 

1936 builder.action(urls.dashboard_link(), ".enter_code_action") 

1937 return builder.build() 

1938 

1939 @classmethod 

1940 def from_notification(cls, data: notification_data_pb2.PostalVerificationPostcardSent, *, user_name: str) -> Self: 

1941 return cls(user_name=user_name, city=data.city, country=data.country) 

1942 

1943 @classmethod 

1944 def test_instances(cls) -> list[Self]: 

1945 return [cls(user_name="Alice", city="New York", country="United States")] 

1946 

1947 

1948@dataclass(kw_only=True, slots=True) 

1949class PostalVerificationSucceededEmail(EmailBase): 

1950 """Sent to a user when their postal verification has succeeded.""" 

1951 

1952 @property 

1953 def string_key_base(self) -> str: 

1954 return "postal_verification.succeeded" 

1955 

1956 def get_body_blocks(self, loc_context: LocalizationContext) -> list[EmailBlock]: 

1957 builder = self._body_builder(loc_context, security_warning=True) 

1958 builder.para(".purpose") 

1959 return builder.build() 

1960 

1961 @classmethod 

1962 def test_instances(cls) -> list[Self]: 

1963 return [cls(user_name="Alice")] 

1964 

1965 

1966@dataclass(kw_only=True, slots=True) 

1967class SignupVerifyEmail(EmailBase): 

1968 """Sent to a user to verify their email address.""" 

1969 

1970 verify_url: str 

1971 

1972 @property 

1973 def string_key_base(self) -> str: 

1974 return "signup.verify" 

1975 

1976 def get_subject_line(self, loc_context: LocalizationContext) -> str: 

1977 return self._localize(loc_context, "signup.subject") 

1978 

1979 def get_body_blocks(self, loc_context: LocalizationContext) -> list[EmailBlock]: 

1980 builder = self._body_builder(loc_context) 

1981 builder.para(".thanks") 

1982 builder.para(".instructions") 

1983 builder.action(self.verify_url, ".confirm_action") 

1984 builder.para("signup.closing") 

1985 return builder.build() 

1986 

1987 @classmethod 

1988 def test_instances(cls) -> list[Self]: 

1989 return [cls(user_name="Alice", verify_url="https://example.com")] 

1990 

1991 

1992@dataclass(kw_only=True, slots=True) 

1993class SignupContinueEmail(EmailBase): 

1994 """Sent to a user to ask them to continue the signup process.""" 

1995 

1996 continue_url: str 

1997 

1998 @property 

1999 def string_key_base(self) -> str: 

2000 return "signup.continue" 

2001 

2002 def get_subject_line(self, loc_context: LocalizationContext) -> str: 

2003 return self._localize(loc_context, "signup.subject") 

2004 

2005 def get_body_blocks(self, loc_context: LocalizationContext) -> list[EmailBlock]: 

2006 builder = self._body_builder(loc_context) 

2007 builder.para(".purpose") 

2008 builder.action(self.continue_url, ".continue_action") 

2009 builder.para("signup.closing") 

2010 builder.para(".ignore_if_unexpected") 

2011 return builder.build() 

2012 

2013 @classmethod 

2014 def test_instances(cls) -> list[Self]: 

2015 return [cls(user_name="Alice", continue_url="https://example.com")] 

2016 

2017 

2018@dataclass(kw_only=True, slots=True) 

2019class StrongVerificationFailedEmail(EmailBase): 

2020 """Sent to a user when their strong verification attempt has failed.""" 

2021 

2022 reason: notification_data_pb2.SVFailReason.ValueType 

2023 

2024 @property 

2025 def string_key_base(self) -> str: 

2026 return "strong_verification.failed" 

2027 

2028 def get_body_blocks(self, loc_context: LocalizationContext) -> list[EmailBlock]: 

2029 builder = self._body_builder(loc_context, security_warning=True) 

2030 match self.reason: 

2031 case notification_data_pb2.SV_FAIL_REASON_WRONG_BIRTHDATE_OR_GENDER: 

2032 purpose_string_key = ".purpose.wrong_birthdate_or_gender" 

2033 case notification_data_pb2.SV_FAIL_REASON_NOT_A_PASSPORT: 

2034 purpose_string_key = ".purpose.not_a_passport" 

2035 case notification_data_pb2.SV_FAIL_REASON_DUPLICATE: 2035 ↛ 2037line 2035 didn't jump to line 2037 because the pattern on line 2035 always matched

2036 purpose_string_key = ".purpose.duplicate" 

2037 case _: 

2038 raise Exception("Shouldn't get here") 

2039 builder.para(purpose_string_key) 

2040 builder.action(urls.strong_verification_url(), ".restart_action") 

2041 return builder.build() 

2042 

2043 @classmethod 

2044 def from_notification(cls, data: notification_data_pb2.VerificationSVFail, *, user_name: str) -> Self: 

2045 return cls(user_name=user_name, reason=data.reason) 

2046 

2047 @classmethod 

2048 def test_instances(cls) -> list[Self]: 

2049 prototype = cls( 

2050 user_name="Alice", 

2051 reason=notification_data_pb2.SV_FAIL_REASON_NOT_A_PASSPORT, 

2052 ) 

2053 return [ 

2054 replace(prototype, reason=notification_data_pb2.SV_FAIL_REASON_WRONG_BIRTHDATE_OR_GENDER), 

2055 replace(prototype, reason=notification_data_pb2.SV_FAIL_REASON_NOT_A_PASSPORT), 

2056 replace(prototype, reason=notification_data_pb2.SV_FAIL_REASON_DUPLICATE), 

2057 ] 

2058 

2059 

2060@dataclass(kw_only=True, slots=True) 

2061class StrongVerificationSucceededEmail(EmailBase): 

2062 """Sent to a user when their strong verification has succeeded.""" 

2063 

2064 @property 

2065 def string_key_base(self) -> str: 

2066 return "strong_verification.succeeded" 

2067 

2068 def get_body_blocks(self, loc_context: LocalizationContext) -> list[EmailBlock]: 

2069 builder = self._body_builder(loc_context, security_warning=True) 

2070 builder.para(".purpose") 

2071 builder.para(".thanks_message") 

2072 builder.para(".cost_explanation") 

2073 builder.para(".donation_request") 

2074 donate_link = urls.donation_url() + "?utm_source=strong-verification-email" 

2075 builder.action(donate_link, ".donate_action") 

2076 return builder.build() 

2077 

2078 @classmethod 

2079 def test_instances(cls) -> list[Self]: 

2080 return [cls(user_name="Alice")] 

2081 

2082 

2083@dataclass(kw_only=True, slots=True) 

2084class ThreadReplyEmail(EmailBase): 

2085 """Sent to a user when someone replies in a comment thread they participated in.""" 

2086 

2087 author: UserInfo 

2088 parent_context: str # Title of the event or discussion being replied in 

2089 markdown_text: str 

2090 view_link: str 

2091 

2092 @property 

2093 def string_key_base(self) -> str: 

2094 return "thread_reply" 

2095 

2096 def get_subject_line(self, loc_context: LocalizationContext) -> str: 

2097 return self._localize( 

2098 loc_context, ".subject", {"author": self.author.name, "parent_context": self.parent_context} 

2099 ) 

2100 

2101 def get_preview_line(self, loc_context: LocalizationContext) -> str | None: 

2102 return markdown_to_plaintext(self.markdown_text) 

2103 

2104 def get_body_blocks(self, loc_context: LocalizationContext) -> list[EmailBlock]: 

2105 builder = self._body_builder(loc_context) 

2106 builder.para(".purpose", {"author": self.author.name, "parent_context": self.parent_context}) 

2107 builder.user(self.author) 

2108 builder.quote(self.markdown_text, markdown=True) 

2109 builder.action(self.view_link, ".view_action") 

2110 return builder.build() 

2111 

2112 @classmethod 

2113 def from_notification(cls, data: notification_data_pb2.ThreadReply, *, user_name: str) -> Self: 

2114 parent = data.WhichOneof("reply_parent") 

2115 if parent == "event": 

2116 parent_context = data.event.title 

2117 view_link = urls.event_link(occurrence_id=data.event.event_id, slug=data.event.slug) 

2118 elif parent == "discussion": 2118 ↛ 2122line 2118 didn't jump to line 2122 because the condition on line 2118 was always true

2119 parent_context = data.discussion.title 

2120 view_link = urls.discussion_link(discussion_id=data.discussion.discussion_id, slug=data.discussion.slug) 

2121 else: 

2122 raise Exception("Can only do replies to events and discussions") 

2123 return cls( 

2124 user_name=user_name, 

2125 author=UserInfo.from_protobuf(data.author), 

2126 parent_context=parent_context, 

2127 markdown_text=data.reply.content, 

2128 view_link=view_link, 

2129 ) 

2130 

2131 @classmethod 

2132 def test_instances(cls) -> list[Self]: 

2133 return [ 

2134 cls( 

2135 user_name="Alice", 

2136 author=UserInfo.dummy_bob(), 

2137 parent_context="Best hiking trails near Berlin", 

2138 markdown_text="I agree, the Grünewald is **amazing**!", 

2139 view_link="https://couchers.org/discussions/123", 

2140 ) 

2141 ] 

2142 

2143 

2144def _localize_host_request_date(value: date, loc_context: LocalizationContext) -> str: 

2145 return loc_context.localize_date(value, with_year=False, with_day_of_week=True)