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

1080 statements  

« prev     ^ index     » next       coverage.py v7.15.1, created at 2026-07-14 20:06 +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 EmailBlocksBuilder, 

45 ParaBlock, 

46 QuoteBlock, 

47 UserInfo, 

48) 

49from couchers.email.locales import get_emails_i18next 

50from couchers.i18n import LocalizationContext 

51from couchers.i18n.localize import format_phone_number 

52from couchers.markup import html_link, html_mailto_link, markdown_to_plaintext 

53from couchers.notifications.quick_links import generate_quick_decline_link 

54from couchers.proto import conversations_pb2, events_pb2, notification_data_pb2 

55from couchers.utils import now, to_aware_datetime 

56 

57# Common string keys 

58_do_not_reply_request_string_key = "generic.do_not_reply_request" 

59 

60# Specific email definitions 

61 

62 

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

64class AccountDeletionStartedEmail(EmailBase): 

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

66 

67 deletion_link: str 

68 

69 @property 

70 def string_key_base(self) -> str: 

71 return "account_deletion.started" 

72 

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

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

75 builder.para(".purpose") 

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

77 return builder.build() 

78 

79 @classmethod 

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

81 return cls( 

82 user_name=user_name, 

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

84 ) 

85 

86 @classmethod 

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

88 return [ 

89 cls( 

90 user_name="Alice", 

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

92 ) 

93 ] 

94 

95 

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

97class AccountDeletionCompletedEmail(EmailBase): 

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

99 

100 undelete_link: str 

101 days: int 

102 

103 @property 

104 def string_key_base(self) -> str: 

105 return "account_deletion.completed" 

106 

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

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

109 builder.para(".purpose") 

110 builder.para(".farewell") 

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

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

113 return builder.build() 

114 

115 @classmethod 

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

117 return cls( 

118 user_name=user_name, 

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

120 days=data.undelete_days, 

121 ) 

122 

123 @classmethod 

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

125 return [ 

126 cls( 

127 user_name="Alice", 

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

129 days=30, 

130 ) 

131 ] 

132 

133 

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

135class AccountDeletionRecoveredEmail(EmailBase): 

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

137 

138 @property 

139 def string_key_base(self) -> str: 

140 return "account_deletion.recovered" 

141 

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

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

144 builder.para(".confirmation") 

145 builder.para(".login_instructions") 

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

147 builder.para(".redelete_instructions") 

148 return builder.build() 

149 

150 @classmethod 

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

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

153 

154 

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

156class ActivenessProbeEmail(EmailBase): 

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

158 

159 days_left: int 

160 

161 @property 

162 def string_key_base(self) -> str: 

163 return "activeness_probe" 

164 

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

166 builder = self._body_builder(loc_context) 

167 builder.para(".purpose") 

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

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

170 builder.para(".encouragement") 

171 

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

173 version = config.VERSION 

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

175 version = version_match[1] 

176 

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

178 return builder.build() 

179 

180 @classmethod 

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

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

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

184 

185 @classmethod 

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

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

188 

189 

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

191class APIKeyIssuedEmail(EmailBase): 

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

193 

194 api_key: str 

195 expiry: datetime 

196 

197 @property 

198 def string_key_base(self) -> str: 

199 return "api_key_issued" 

200 

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

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

203 builder.para(".header") 

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

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

206 builder.para(".usage_warning") 

207 builder.para(".policy_warning", {"terms_url": urls.terms_of_service_url()}) 

208 return builder.build() 

209 

210 @classmethod 

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

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

213 

214 @classmethod 

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

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

217 

218 

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

220class BadgeChangedEmail(EmailBase): 

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

222 

223 badge_name: str 

224 added: bool 

225 

226 @property 

227 def string_key_base(self) -> str: 

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

229 

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

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

232 

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

234 builder = self._body_builder(loc_context) 

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

236 return builder.build() 

237 

238 @classmethod 

239 def from_notification( 

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

241 ) -> Self: 

242 return cls( 

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

244 ) 

245 

246 @classmethod 

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

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

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

250 

251 

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

253class BirthdateChangedEmail(EmailBase): 

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

255 

256 new_birthdate: date 

257 

258 @property 

259 def string_key_base(self) -> str: 

260 return "birthdate_changed" 

261 

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

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

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

265 return builder.build() 

266 

267 @classmethod 

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

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

270 

271 @classmethod 

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

273 return [ 

274 cls( 

275 user_name="Alice", 

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

277 ) 

278 ] 

279 

280 

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

282class ChatMessageReceivedEmail(EmailBase): 

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

284 

285 group_chat_title: str | None # None if direct message 

286 author: UserInfo 

287 text: str 

288 view_url: str 

289 

290 @property 

291 def string_key_base(self) -> str: 

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

293 

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

295 return self._localize( 

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

297 ) 

298 

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

300 return self.text 

301 

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

303 builder = self._body_builder(loc_context) 

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

305 builder.user(self.author) 

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

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

308 return builder.build() 

309 

310 @classmethod 

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

312 return cls( 

313 user_name, 

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

315 text=data.text, 

316 group_chat_title=data.group_chat_title or None, 

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

318 ) 

319 

320 @classmethod 

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

322 prototype = cls( 

323 user_name="Alice", 

324 group_chat_title=None, 

325 author=UserInfo.dummy_bob(), 

326 text="Hi Alice!", 

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

328 ) 

329 return [ 

330 replace(prototype, group_chat_title=None), 

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

332 ] 

333 

334 

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

336class ChatMessagesMissedEmail(EmailBase): 

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

338 

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

340 class Entry: 

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

342 

343 group_chat_title: str | None # None if direct message 

344 missed_count: int 

345 latest_message_author: UserInfo 

346 latest_message_text: str 

347 view_url: str 

348 

349 entries: list[Entry] 

350 

351 @property 

352 def string_key_base(self) -> str: 

353 return "chat_messages.missed" 

354 

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

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

357 

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

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

360 return None 

361 return self.entries[0].latest_message_text 

362 

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

364 builder = self._body_builder(loc_context) 

365 builder.para(".purpose") 

366 for entry in self.entries: 

367 if entry.group_chat_title is None: 

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

369 else: 

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

371 builder.user(entry.latest_message_author) 

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

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

374 return builder.build() 

375 

376 @classmethod 

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

378 missed_entries = [ 

379 cls.Entry( 

380 group_chat_title=message.group_chat_title or None, 

381 missed_count=message.unseen_count, 

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

383 latest_message_text=message.text, 

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

385 ) 

386 for message in data.messages 

387 ] 

388 

389 return cls(user_name, entries=missed_entries) 

390 

391 @classmethod 

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

393 entry_prototype = ChatMessagesMissedEmail.Entry( 

394 group_chat_title=None, 

395 missed_count=1, 

396 latest_message_author=UserInfo.dummy_bob(), 

397 latest_message_text="Hello!", 

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

399 ) 

400 return [ 

401 cls( 

402 user_name="Alice", 

403 entries=[ 

404 replace(entry_prototype, group_chat_title=None), 

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

406 ], 

407 ) 

408 ] 

409 

410 

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

412class DiscussionCreatedEmail(EmailBase): 

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

414 

415 author: UserInfo 

416 title: str 

417 parent_context: str # Community or group name 

418 markdown_text: str 

419 view_link: str 

420 

421 @property 

422 def string_key_base(self) -> str: 

423 return "discussions.created" 

424 

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

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

427 

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

429 return markdown_to_plaintext(self.markdown_text) 

430 

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

432 builder = self._body_builder(loc_context) 

433 builder.para( 

434 ".purpose", 

435 { 

436 "author": self.author.name, 

437 "parent_context": self.parent_context, 

438 }, 

439 ) 

440 builder.user(self.author) 

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

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

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

444 return builder.build() 

445 

446 @classmethod 

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

448 discussion = data.discussion 

449 return cls( 

450 user_name=user_name, 

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

452 title=discussion.title, 

453 parent_context=discussion.owner_title, 

454 markdown_text=discussion.content, 

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

456 ) 

457 

458 @classmethod 

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

460 return [ 

461 cls( 

462 user_name="Alice", 

463 author=UserInfo.dummy_bob(), 

464 title="Best hiking trails near Berlin", 

465 parent_context="Berlin", 

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

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

468 ) 

469 ] 

470 

471 

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

473class DiscussionCommentEmail(EmailBase): 

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

475 

476 author: UserInfo 

477 discussion_title: str 

478 discussion_parent_context: str # Community or group name 

479 markdown_text: str 

480 view_link: str 

481 

482 @property 

483 def string_key_base(self) -> str: 

484 return "discussions.comment" 

485 

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

487 return self._localize( 

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

489 ) 

490 

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

492 return markdown_to_plaintext(self.markdown_text) 

493 

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

495 builder = self._body_builder(loc_context) 

496 builder.para( 

497 ".purpose", 

498 { 

499 "author": self.author.name, 

500 "discussion_title": self.discussion_title, 

501 "parent_context": self.discussion_parent_context, 

502 }, 

503 ) 

504 builder.user(self.author) 

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

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

507 return builder.build() 

508 

509 @classmethod 

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

511 discussion = data.discussion 

512 return cls( 

513 user_name=user_name, 

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

515 discussion_title=discussion.title, 

516 discussion_parent_context=discussion.owner_title, 

517 markdown_text=data.reply.content, 

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

519 ) 

520 

521 @classmethod 

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

523 return [ 

524 cls( 

525 user_name="Alice", 

526 author=UserInfo.dummy_bob(), 

527 discussion_title="Best hiking trails near Berlin", 

528 discussion_parent_context="Berlin", 

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

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

531 ) 

532 ] 

533 

534 

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

536class DonationReceivedEmail(EmailBase): 

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

538 

539 amount: int 

540 receipt_url: str 

541 

542 @property 

543 def string_key_base(self) -> str: 

544 return "donation_received" 

545 

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

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

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

549 builder.para(".contribution_impact") 

550 builder.para(".invoice_receipt_info") 

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

552 builder.para(".tax_acknowledgment") 

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

554 builder.para("generic.thanks") 

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

556 return builder.build() 

557 

558 @classmethod 

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

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

561 

562 @classmethod 

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

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

565 

566 

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

568class EmailChangedEmail(EmailBase): 

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

570 

571 new_email: str 

572 

573 @property 

574 def string_key_base(self) -> str: 

575 return "email_change.initiated" 

576 

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

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

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

580 return builder.build() 

581 

582 @classmethod 

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

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

585 

586 @classmethod 

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

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

589 

590 

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

592class EmailChangeConfirmationEmail(EmailBase): 

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

594 

595 old_email: str 

596 confirm_url: str 

597 

598 @property 

599 def string_key_base(self) -> str: 

600 return "email_change.confirmation" 

601 

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

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

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

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

606 return builder.build() 

607 

608 @classmethod 

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

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

611 

612 

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

614class EmailVerifiedEmail(EmailBase): 

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

616 

617 @property 

618 def string_key_base(self) -> str: 

619 return "email_change.verified" 

620 

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

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

623 builder.para(".purpose") 

624 return builder.build() 

625 

626 @classmethod 

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

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

629 

630 

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

632class EventInfo: 

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

634 

635 title: str 

636 start_time: datetime 

637 end_time: datetime 

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

639 view_url: str 

640 description_markdown: str 

641 

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

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

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

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

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

647 

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

649 html += "<br>" 

650 html += time_range_display 

651 if self.address: 

652 html += "<br>" 

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

654 

655 return ParaBlock(text=Markup(html)) 

656 

657 def get_description_block(self) -> EmailBlock: 

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

659 

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

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

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

663 

664 @classmethod 

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

666 return cls( 

667 title=event.title, 

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

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

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

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

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

673 description_markdown=event.content or "", 

674 ) 

675 

676 @staticmethod 

677 def dummy() -> EventInfo: 

678 return EventInfo( 

679 title="Berlin Meetup", 

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

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

682 address="Alexanderplatz, Berlin", 

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

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

685 ) 

686 

687 

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

689class EventCreatedEmail(EmailBase): 

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

691 

692 inviting_user: UserInfo 

693 event_info: EventInfo 

694 community_name: str | None 

695 community_url: str | None 

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

697 

698 @property 

699 def string_key_base(self) -> str: 

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

701 

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

703 return self._localize( 

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

705 ) 

706 

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

708 return markdown_to_plaintext(self.event_info.description_markdown) 

709 

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

711 builder = self._body_builder(loc_context) 

712 if self.community_name: 

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

714 else: 

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

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

717 builder.user(self.inviting_user) 

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

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

720 return builder.build() 

721 

722 @classmethod 

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

724 has_community = bool(data.in_community.community_id) 

725 community_url = ( 

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

727 if has_community 

728 else None 

729 ) 

730 return cls( 

731 user_name=user_name, 

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

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

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

735 community_url=community_url, 

736 is_invite=is_invite, 

737 ) 

738 

739 @classmethod 

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

741 prototype = cls( 

742 user_name="Alice", 

743 inviting_user=UserInfo.dummy_bob(), 

744 event_info=EventInfo.dummy(), 

745 community_name="Berlin", 

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

747 is_invite=True, 

748 ) 

749 return [ 

750 replace(prototype, is_invite=True), 

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

752 replace(prototype, is_invite=False), 

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

754 ] 

755 

756 

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

758class EventUpdatedEmail(EmailBase): 

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

760 

761 updating_user: UserInfo 

762 event_info: EventInfo 

763 updated_items: list[notification_data_pb2.EventUpdateItem.ValueType] 

764 

765 @property 

766 def string_key_base(self) -> str: 

767 return "events.updated" 

768 

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

770 return self._localize( 

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

772 ) 

773 

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

775 builder = self._body_builder(loc_context) 

776 

777 updated_items_string_keys = list( 

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

779 ) 

780 if updated_items_string_keys: 

781 updated_items_text = loc_context.localize_list( 

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

783 ) 

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

785 else: 

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

787 

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

789 builder.user(self.updating_user) 

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

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

792 return builder.build() 

793 

794 @classmethod 

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

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

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

798 updated_items.extend(data.updated_enum_items) 

799 elif data.updated_str_items: 

800 for updated_str_item in data.updated_str_items: 

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

802 updated_items.append(updated_enum_item) 

803 

804 return cls( 

805 user_name=user_name, 

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

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

808 updated_items=updated_items, 

809 ) 

810 

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

812 @staticmethod 

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

814 match value: 

815 case "title": 

816 return notification_data_pb2.EventUpdateItem.EVENT_UPDATE_ITEM_TITLE 

817 case "content": 

818 return notification_data_pb2.EventUpdateItem.EVENT_UPDATE_ITEM_CONTENT 

819 case "location": 

820 return notification_data_pb2.EventUpdateItem.EVENT_UPDATE_ITEM_LOCATION 

821 case "start time": 

822 return notification_data_pb2.EventUpdateItem.EVENT_UPDATE_ITEM_START_TIME 

823 case "end time": 

824 return notification_data_pb2.EventUpdateItem.EVENT_UPDATE_ITEM_END_TIME 

825 case _: 

826 return None 

827 

828 @staticmethod 

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

830 match value: 

831 case notification_data_pb2.EventUpdateItem.EVENT_UPDATE_ITEM_TITLE: 

832 return ".item_names.title" 

833 case notification_data_pb2.EventUpdateItem.EVENT_UPDATE_ITEM_CONTENT: 

834 return ".item_names.content" 

835 case notification_data_pb2.EventUpdateItem.EVENT_UPDATE_ITEM_LOCATION: 

836 return ".item_names.location" 

837 case notification_data_pb2.EventUpdateItem.EVENT_UPDATE_ITEM_START_TIME: 

838 return ".item_names.start_time" 

839 case notification_data_pb2.EventUpdateItem.EVENT_UPDATE_ITEM_END_TIME: 

840 return ".item_names.end_time" 

841 case _: 

842 return None 

843 

844 @classmethod 

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

846 prototype = cls( 

847 user_name="Alice", 

848 updating_user=UserInfo.dummy_bob(), 

849 event_info=EventInfo.dummy(), 

850 updated_items=[], 

851 ) 

852 return [ 

853 replace(prototype, updated_items=[]), 

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

855 ] 

856 

857 

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

859class EventOrganizerInvitedEmail(EmailBase): 

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

861 

862 inviting_user: UserInfo 

863 event_info: EventInfo 

864 

865 @property 

866 def string_key_base(self) -> str: 

867 return "events.organizer_invited" 

868 

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

870 return self._localize( 

871 loc_context, 

872 ".subject", 

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

874 ) 

875 

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

877 builder = self._body_builder(loc_context) 

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

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

880 builder.user(self.inviting_user) 

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

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

883 builder.para(_do_not_reply_request_string_key, epilogue=True) 

884 return builder.build() 

885 

886 @classmethod 

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

888 return cls( 

889 user_name=user_name, 

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

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

892 ) 

893 

894 @classmethod 

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

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

897 

898 

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

900class EventCommentEmail(EmailBase): 

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

902 

903 author: UserInfo 

904 event_info: EventInfo 

905 comment_markdown: str 

906 

907 @property 

908 def string_key_base(self) -> str: 

909 return "events.comment" 

910 

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

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

913 

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

915 return markdown_to_plaintext(self.comment_markdown) 

916 

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

918 builder = self._body_builder(loc_context) 

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

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

921 builder.user(self.author) 

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

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

924 return builder.build() 

925 

926 @classmethod 

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

928 return cls( 

929 user_name=user_name, 

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

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

932 comment_markdown=data.reply.content, 

933 ) 

934 

935 @classmethod 

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

937 return [ 

938 cls( 

939 user_name="Alice", 

940 author=UserInfo.dummy_bob(), 

941 event_info=EventInfo.dummy(), 

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

943 ) 

944 ] 

945 

946 

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

948class EventReminderEmail(EmailBase): 

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

950 

951 event_info: EventInfo 

952 

953 @property 

954 def string_key_base(self) -> str: 

955 return "events.reminder" 

956 

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

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

959 

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

961 builder = self._body_builder(loc_context) 

962 builder.para(".purpose") 

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

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

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

966 return builder.build() 

967 

968 @classmethod 

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

970 return cls( 

971 user_name=user_name, 

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

973 ) 

974 

975 @classmethod 

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

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

978 

979 

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

981class EventCancelledEmail(EmailBase): 

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

983 

984 cancelling_user: UserInfo 

985 event_info: EventInfo 

986 

987 @property 

988 def string_key_base(self) -> str: 

989 return "events.cancel" 

990 

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

992 return self._localize( 

993 loc_context, 

994 ".subject", 

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

996 ) 

997 

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

999 builder = self._body_builder(loc_context) 

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

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

1002 builder.user(self.cancelling_user) 

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

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

1005 return builder.build() 

1006 

1007 @classmethod 

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

1009 return cls( 

1010 user_name=user_name, 

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

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

1013 ) 

1014 

1015 @classmethod 

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

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

1018 

1019 

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

1021class EventDeletedEmail(EmailBase): 

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

1023 

1024 event_info: EventInfo 

1025 

1026 @property 

1027 def string_key_base(self) -> str: 

1028 return "events.deleted" 

1029 

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

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

1032 

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

1034 builder = self._body_builder(loc_context) 

1035 builder.para(".purpose") 

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

1037 return builder.build() 

1038 

1039 @classmethod 

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

1041 return cls( 

1042 user_name=user_name, 

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

1044 ) 

1045 

1046 @classmethod 

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

1048 return [ 

1049 cls( 

1050 user_name="Alice", 

1051 event_info=EventInfo.dummy(), 

1052 ) 

1053 ] 

1054 

1055 

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

1057class FriendReferenceReceivedEmail(EmailBase): 

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

1059 

1060 from_user: UserInfo 

1061 text: str 

1062 

1063 @property 

1064 def string_key_base(self) -> str: 

1065 return "references.received.friend" 

1066 

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

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

1069 

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

1071 return self.text 

1072 

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

1074 builder = self._body_builder(loc_context) 

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

1076 builder.user(self.from_user) 

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

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

1079 return builder.build() 

1080 

1081 @classmethod 

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

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

1084 

1085 @classmethod 

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

1087 return [ 

1088 cls( 

1089 user_name="Alice", 

1090 from_user=UserInfo.dummy_bob(), 

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

1092 ) 

1093 ] 

1094 

1095 

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

1097class FriendRequestReceivedEmail(EmailBase): 

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

1099 

1100 befriender: UserInfo 

1101 

1102 @property 

1103 def string_key_base(self) -> str: 

1104 return "friend_requests.received" 

1105 

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

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

1108 

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

1110 builder = self._body_builder(loc_context) 

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

1112 builder.user(self.befriender) 

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

1114 builder.para(".closing") 

1115 builder.para(_do_not_reply_request_string_key, epilogue=True) 

1116 return builder.build() 

1117 

1118 @classmethod 

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

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

1121 

1122 @classmethod 

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

1124 return [ 

1125 cls( 

1126 user_name="Alice", 

1127 befriender=UserInfo.dummy_bob(), 

1128 ) 

1129 ] 

1130 

1131 

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

1133class FriendRequestAcceptedEmail(EmailBase): 

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

1135 

1136 new_friend: UserInfo 

1137 

1138 @property 

1139 def string_key_base(self) -> str: 

1140 return "friend_requests.accepted" 

1141 

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

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

1144 

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

1146 builder = self._body_builder(loc_context) 

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

1148 builder.user(self.new_friend) 

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

1150 builder.para(".closing") 

1151 return builder.build() 

1152 

1153 @classmethod 

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

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

1156 

1157 @classmethod 

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

1159 return [ 

1160 cls( 

1161 user_name="Alice", 

1162 new_friend=UserInfo.dummy_bob(), 

1163 ) 

1164 ] 

1165 

1166 

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

1168class GenderChangedEmail(EmailBase): 

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

1170 

1171 new_gender: str 

1172 

1173 @property 

1174 def string_key_base(self) -> str: 

1175 return "gender_changed" 

1176 

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

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

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

1180 return builder.build() 

1181 

1182 @classmethod 

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

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

1185 

1186 @classmethod 

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

1188 return [ 

1189 cls( 

1190 user_name="Alice", 

1191 new_gender="Male", 

1192 ) 

1193 ] 

1194 

1195 

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

1197class HostRequestCreatedEmail(EmailBase): 

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

1199 

1200 surfer: UserInfo 

1201 from_date: date 

1202 to_date: date 

1203 text: str 

1204 view_link: str 

1205 quick_decline_link: str 

1206 

1207 @property 

1208 def string_key_base(self) -> str: 

1209 return "host_requests.created" 

1210 

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

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

1213 

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

1215 return self.text 

1216 

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

1218 builder = self._body_builder(loc_context) 

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

1220 builder.user( 

1221 self.surfer, 

1222 "host_requests.generic.date_range", 

1223 { 

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

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

1226 }, 

1227 ) 

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

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

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

1231 builder.para(".respond_encouragement") 

1232 builder.para(_do_not_reply_request_string_key, epilogue=True) 

1233 return builder.build() 

1234 

1235 @classmethod 

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

1237 return cls( 

1238 user_name, 

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

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

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

1242 text=data.text, 

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

1244 quick_decline_link=generate_quick_decline_link(data.host_request), 

1245 ) 

1246 

1247 @classmethod 

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

1249 return [ 

1250 cls( 

1251 user_name="Alice", 

1252 surfer=UserInfo.dummy_bob(), 

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

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

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

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

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

1258 ) 

1259 ] 

1260 

1261 

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

1263class HostRequestReminderEmail(EmailBase): 

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

1265 

1266 surfer: UserInfo 

1267 from_date: date 

1268 to_date: date 

1269 view_link: str 

1270 quick_decline_link: str 

1271 

1272 @property 

1273 def string_key_base(self) -> str: 

1274 return "host_requests.reminder" 

1275 

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

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

1278 

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

1280 builder = self._body_builder(loc_context) 

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

1282 builder.user( 

1283 self.surfer, 

1284 "host_requests.generic.date_range", 

1285 { 

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

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

1288 }, 

1289 ) 

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

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

1292 builder.para(_do_not_reply_request_string_key, epilogue=True) 

1293 return builder.build() 

1294 

1295 @classmethod 

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

1297 return cls( 

1298 user_name, 

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

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

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

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

1303 quick_decline_link=generate_quick_decline_link(data.host_request), 

1304 ) 

1305 

1306 @classmethod 

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

1308 return [ 

1309 cls( 

1310 user_name="Alice", 

1311 surfer=UserInfo.dummy_bob(), 

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

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

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

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

1316 ) 

1317 ] 

1318 

1319 

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

1321class HostRequestMessageEmail(EmailBase): 

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

1323 

1324 other_user: UserInfo 

1325 from_date: date 

1326 to_date: date 

1327 text: str 

1328 from_host: bool 

1329 view_link: str 

1330 

1331 @property 

1332 def string_key_base(self) -> str: 

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

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

1335 

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

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

1338 

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

1340 return self.text 

1341 

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

1343 builder = self._body_builder(loc_context) 

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

1345 builder.user( 

1346 self.other_user, 

1347 "host_requests.generic.date_range", 

1348 { 

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

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

1351 }, 

1352 ) 

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

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

1355 builder.para(_do_not_reply_request_string_key, epilogue=True) 

1356 return builder.build() 

1357 

1358 @classmethod 

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

1360 return cls( 

1361 user_name, 

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

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

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

1365 text=data.text, 

1366 from_host=not data.am_host, 

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

1368 ) 

1369 

1370 @classmethod 

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

1372 prototype = cls( 

1373 user_name="Alice", 

1374 other_user=UserInfo.dummy_bob(), 

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

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

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

1378 from_host=True, 

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

1380 ) 

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

1382 

1383 

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

1385class HostRequestMissedMessagesEmail(EmailBase): 

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

1387 

1388 other_user: UserInfo 

1389 from_date: date 

1390 to_date: date 

1391 from_host: bool 

1392 view_link: str 

1393 

1394 @property 

1395 def string_key_base(self) -> str: 

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

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

1398 

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

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

1401 

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

1403 builder = self._body_builder(loc_context) 

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

1405 builder.user( 

1406 self.other_user, 

1407 "host_requests.generic.date_range", 

1408 { 

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

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

1411 }, 

1412 ) 

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

1414 builder.para(_do_not_reply_request_string_key, epilogue=True) 

1415 return builder.build() 

1416 

1417 @classmethod 

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

1419 return cls( 

1420 user_name, 

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

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

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

1424 from_host=not data.am_host, 

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

1426 ) 

1427 

1428 @classmethod 

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

1430 prototype = cls( 

1431 user_name="Alice", 

1432 other_user=UserInfo.dummy_bob(), 

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

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

1435 from_host=True, 

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

1437 ) 

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

1439 

1440 

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

1442class HostRequestStatusChangedEmail(EmailBase): 

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

1444 

1445 other_user: UserInfo 

1446 from_date: date 

1447 to_date: date 

1448 new_status: conversations_pb2.HostRequestStatus.ValueType 

1449 view_link: str 

1450 

1451 @property 

1452 def string_key_base(self) -> str: 

1453 base_key = "host_requests.status_changed" 

1454 match self.new_status: 

1455 case conversations_pb2.HOST_REQUEST_STATUS_ACCEPTED: 

1456 return f"{base_key}.accepted_by_host" 

1457 case conversations_pb2.HOST_REQUEST_STATUS_REJECTED: 

1458 return f"{base_key}.declined_by_host" 

1459 case conversations_pb2.HOST_REQUEST_STATUS_CONFIRMED: 

1460 return f"{base_key}.confirmed_by_surfer" 

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

1462 return f"{base_key}.cancelled_by_surfer" 

1463 case _: 

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

1465 

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

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

1468 

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

1470 builder = self._body_builder(loc_context) 

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

1472 builder.user( 

1473 self.other_user, 

1474 "host_requests.generic.date_range", 

1475 { 

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

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

1478 }, 

1479 ) 

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

1481 builder.para(_do_not_reply_request_string_key, epilogue=True) 

1482 return builder.build() 

1483 

1484 @classmethod 

1485 def from_notification( 

1486 cls, 

1487 data: notification_data_pb2.HostRequestAccept 

1488 | notification_data_pb2.HostRequestReject 

1489 | notification_data_pb2.HostRequestConfirm 

1490 | notification_data_pb2.HostRequestCancel, 

1491 *, 

1492 user_name: str, 

1493 ) -> Self: 

1494 other_user: UserInfo 

1495 new_status: conversations_pb2.HostRequestStatus.ValueType 

1496 match data: 

1497 case notification_data_pb2.HostRequestAccept(): 

1498 other_user = UserInfo.from_protobuf(data.host) 

1499 new_status = conversations_pb2.HostRequestStatus.HOST_REQUEST_STATUS_ACCEPTED 

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

1501 other_user = UserInfo.from_protobuf(data.host) 

1502 new_status = conversations_pb2.HostRequestStatus.HOST_REQUEST_STATUS_REJECTED 

1503 case notification_data_pb2.HostRequestConfirm(): 

1504 other_user = UserInfo.from_protobuf(data.surfer) 

1505 new_status = conversations_pb2.HostRequestStatus.HOST_REQUEST_STATUS_CONFIRMED 

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

1507 other_user = UserInfo.from_protobuf(data.surfer) 

1508 new_status = conversations_pb2.HostRequestStatus.HOST_REQUEST_STATUS_CANCELLED 

1509 case _: 

1510 # Enable mypy's exhaustiveness checking 

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

1512 

1513 return cls( 

1514 user_name, 

1515 other_user=other_user, 

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

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

1518 new_status=new_status, 

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

1520 ) 

1521 

1522 @classmethod 

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

1524 prototype = cls( 

1525 user_name="Alice", 

1526 other_user=UserInfo.dummy_bob(), 

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

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

1529 new_status=conversations_pb2.HOST_REQUEST_STATUS_ACCEPTED, 

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

1531 ) 

1532 return [ 

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

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

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

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

1537 ] 

1538 

1539 

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

1541class HostReferenceReceivedEmail(EmailBase): 

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

1543 

1544 from_user: UserInfo 

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

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

1547 leave_reference_url: str 

1548 

1549 @property 

1550 def string_key_base(self) -> str: 

1551 return "references.received" 

1552 

1553 @property 

1554 def string_role_subkey(self) -> str: 

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

1556 

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

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

1559 

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

1561 return self.text 

1562 

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

1564 builder = self._body_builder(loc_context) 

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

1566 builder.user(self.from_user) 

1567 if self.text: 

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

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

1570 else: 

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

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

1573 return builder.build() 

1574 

1575 @classmethod 

1576 def from_notification( 

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

1578 ) -> Self: 

1579 return cls( 

1580 user_name=user_name, 

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

1582 text=data.text or None, 

1583 surfed=surfed, 

1584 leave_reference_url=urls.leave_reference_link( 

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

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

1587 host_request_id=str(data.host_request_id), 

1588 ), 

1589 ) 

1590 

1591 @classmethod 

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

1593 prototype = cls( 

1594 user_name="Alice", 

1595 from_user=UserInfo.dummy_bob(), 

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

1597 surfed=True, 

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

1599 ) 

1600 return [ 

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

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

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

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

1605 ] 

1606 

1607 

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

1609class HostReferenceReminderEmail(EmailBase): 

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

1611 

1612 other_user: UserInfo 

1613 days_left: int 

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

1615 leave_reference_url: str 

1616 

1617 @property 

1618 def string_key_base(self) -> str: 

1619 return "references.reminder" 

1620 

1621 @property 

1622 def string_role_subkey(self) -> str: 

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

1624 

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

1626 return self._localize( 

1627 loc_context, 

1628 ".subject_days", 

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

1630 ) 

1631 

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

1633 builder = self._body_builder(loc_context) 

1634 builder.para( 

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

1636 ) 

1637 builder.user(self.other_user) 

1638 builder.action( 

1639 self.leave_reference_url, 

1640 "references.write_action", 

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

1642 ) 

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

1644 builder.para(".visibility_note") 

1645 return builder.build() 

1646 

1647 @classmethod 

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

1649 return cls( 

1650 user_name=user_name, 

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

1652 days_left=data.days_left, 

1653 surfed=surfed, 

1654 leave_reference_url=urls.leave_reference_link( 

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

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

1657 host_request_id=str(data.host_request_id), 

1658 ), 

1659 ) 

1660 

1661 @classmethod 

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

1663 prototype = cls( 

1664 user_name="Alice", 

1665 other_user=UserInfo.dummy_bob(), 

1666 days_left=7, 

1667 surfed=True, 

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

1669 ) 

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

1671 

1672 

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

1674class ModeratorNoteEmail(EmailBase): 

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

1676 

1677 @property 

1678 def string_key_base(self) -> str: 

1679 return "moderator_note" 

1680 

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

1682 builder = self._body_builder(loc_context) 

1683 builder.para(".purpose") 

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

1685 # letting them use the platform. 

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

1687 return builder.build() 

1688 

1689 @classmethod 

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

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

1692 

1693 

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

1695class NewBlogPostEmail(EmailBase): 

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

1697 

1698 title: str 

1699 blurb: str 

1700 url: str 

1701 

1702 @property 

1703 def string_key_base(self) -> str: 

1704 return "new_blog_post" 

1705 

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

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

1708 

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

1710 return self.blurb 

1711 

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

1713 builder = self._body_builder(loc_context) 

1714 builder.para(".purpose") 

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

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

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

1718 return builder.build() 

1719 

1720 @classmethod 

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

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

1723 

1724 @classmethod 

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

1726 return [ 

1727 cls( 

1728 user_name="Alice", 

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

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

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

1732 ) 

1733 ] 

1734 

1735 

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

1737class OnboardingReminderEmail(EmailBase): 

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

1739 

1740 initial: bool 

1741 

1742 @property 

1743 def string_key_base(self) -> str: 

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

1745 

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

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

1748 if self.initial: 

1749 self._build_body_initial(builder) 

1750 else: 

1751 self._build_body_followup(builder) 

1752 return builder.build() 

1753 

1754 def _build_body_initial(self, builder: EmailBlocksBuilder) -> None: 

1755 builder.para(".welcome") 

1756 builder.para(".early_user_role") 

1757 builder.para(".complete_profile_request") 

1758 builder.para(".profile_importance") 

1759 builder.action(urls.edit_profile_link(), "onboarding_reminder.complete_profile_action") 

1760 builder.para(".share_request") 

1761 builder.para( 

1762 "generic.closing_lines.aapeli", 

1763 { 

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

1765 }, 

1766 ) 

1767 

1768 def _build_body_followup(self, builder: EmailBlocksBuilder) -> None: 

1769 builder.para(".intro") 

1770 builder.para(".request") 

1771 builder.action(urls.edit_profile_link(), "onboarding_reminder.complete_profile_action") 

1772 builder.para("generic.thanks") 

1773 builder.para( 

1774 "generic.closing_lines.emily", 

1775 { 

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

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

1778 }, 

1779 ) 

1780 

1781 @classmethod 

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

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

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

1785 

1786 

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

1788class PasswordChangedEmail(EmailBase): 

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

1790 

1791 @property 

1792 def string_key_base(self) -> str: 

1793 return "password_changed" 

1794 

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

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

1797 builder.para(".purpose") 

1798 return builder.build() 

1799 

1800 @classmethod 

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

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

1803 

1804 

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

1806class PasswordResetCompletedEmail(EmailBase): 

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

1808 

1809 @property 

1810 def string_key_base(self) -> str: 

1811 return "password_reset.completed" 

1812 

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

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

1815 builder.para(".purpose") 

1816 return builder.build() 

1817 

1818 @classmethod 

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

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

1821 

1822 

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

1824class PasswordResetStartedEmail(EmailBase): 

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

1826 

1827 password_reset_link: str 

1828 

1829 @property 

1830 def string_key_base(self) -> str: 

1831 return "password_reset.started" 

1832 

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

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

1835 builder.para(".purpose") 

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

1837 return builder.build() 

1838 

1839 @classmethod 

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

1841 return cls( 

1842 user_name=user_name, 

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

1844 ) 

1845 

1846 @classmethod 

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

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

1849 

1850 

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

1852class PhoneNumberChangeEmail(EmailBase): 

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

1854 

1855 new_phone_number: str 

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

1857 

1858 @property 

1859 def string_key_base(self) -> str: 

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

1861 

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

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

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

1865 return builder.build() 

1866 

1867 @classmethod 

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

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

1870 

1871 @classmethod 

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

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

1874 

1875 @classmethod 

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

1877 prototype = cls( 

1878 user_name="Alice", 

1879 new_phone_number="+12223334444", 

1880 completed=False, 

1881 ) 

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

1883 

1884 

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

1886class PostalVerificationFailedEmail(EmailBase): 

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

1888 

1889 reason: notification_data_pb2.PostalVerificationFailReason.ValueType 

1890 

1891 @property 

1892 def string_key_base(self) -> str: 

1893 return "postal_verification.failed" 

1894 

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

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

1897 match self.reason: 

1898 case notification_data_pb2.POSTAL_VERIFICATION_FAIL_REASON_CODE_EXPIRED: 

1899 purpose_string_key = ".purpose.code_expired" 

1900 case notification_data_pb2.POSTAL_VERIFICATION_FAIL_REASON_TOO_MANY_ATTEMPTS: 

1901 purpose_string_key = ".purpose.too_many_attempts" 

1902 case _: 

1903 purpose_string_key = ".purpose.default" 

1904 builder.para(purpose_string_key) 

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

1906 return builder.build() 

1907 

1908 @classmethod 

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

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

1911 

1912 @classmethod 

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

1914 prototype = cls( 

1915 user_name="Alice", 

1916 reason=notification_data_pb2.POSTAL_VERIFICATION_FAIL_REASON_CODE_EXPIRED, 

1917 ) 

1918 return [ 

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

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

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

1922 ] 

1923 

1924 

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

1926class PostalVerificationPostcardSentEmail(EmailBase): 

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

1928 

1929 city: str 

1930 country: str 

1931 

1932 @property 

1933 def string_key_base(self) -> str: 

1934 return "postal_verification.postcard_sent" 

1935 

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

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

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

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

1940 return builder.build() 

1941 

1942 @classmethod 

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

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

1945 

1946 @classmethod 

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

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

1949 

1950 

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

1952class PostalVerificationSucceededEmail(EmailBase): 

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

1954 

1955 @property 

1956 def string_key_base(self) -> str: 

1957 return "postal_verification.succeeded" 

1958 

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

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

1961 builder.para(".purpose") 

1962 return builder.build() 

1963 

1964 @classmethod 

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

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

1967 

1968 

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

1970class SignupVerifyEmail(EmailBase): 

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

1972 

1973 verify_url: str 

1974 

1975 @property 

1976 def string_key_base(self) -> str: 

1977 return "signup.verify" 

1978 

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

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

1981 

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

1983 builder = self._body_builder(loc_context) 

1984 builder.para(".thanks") 

1985 builder.para(".instructions") 

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

1987 builder.para("signup.closing") 

1988 return builder.build() 

1989 

1990 @classmethod 

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

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

1993 

1994 

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

1996class SignupContinueEmail(EmailBase): 

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

1998 

1999 continue_url: str 

2000 

2001 @property 

2002 def string_key_base(self) -> str: 

2003 return "signup.continue" 

2004 

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

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

2007 

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

2009 builder = self._body_builder(loc_context) 

2010 builder.para(".purpose") 

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

2012 builder.para("signup.closing") 

2013 builder.para(".ignore_if_unexpected") 

2014 return builder.build() 

2015 

2016 @classmethod 

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

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

2019 

2020 

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

2022class StrongVerificationFailedEmail(EmailBase): 

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

2024 

2025 reason: notification_data_pb2.SVFailReason.ValueType 

2026 

2027 @property 

2028 def string_key_base(self) -> str: 

2029 return "strong_verification.failed" 

2030 

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

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

2033 match self.reason: 

2034 case notification_data_pb2.SV_FAIL_REASON_WRONG_BIRTHDATE_OR_GENDER: 

2035 purpose_string_key = ".purpose.wrong_birthdate_or_gender" 

2036 case notification_data_pb2.SV_FAIL_REASON_NOT_A_PASSPORT: 

2037 purpose_string_key = ".purpose.not_a_passport" 

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

2039 purpose_string_key = ".purpose.duplicate" 

2040 case _: 

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

2042 builder.para(purpose_string_key) 

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

2044 return builder.build() 

2045 

2046 @classmethod 

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

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

2049 

2050 @classmethod 

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

2052 prototype = cls( 

2053 user_name="Alice", 

2054 reason=notification_data_pb2.SV_FAIL_REASON_NOT_A_PASSPORT, 

2055 ) 

2056 return [ 

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

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

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

2060 ] 

2061 

2062 

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

2064class StrongVerificationSucceededEmail(EmailBase): 

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

2066 

2067 @property 

2068 def string_key_base(self) -> str: 

2069 return "strong_verification.succeeded" 

2070 

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

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

2073 builder.para(".purpose") 

2074 builder.para(".thanks_message") 

2075 builder.para(".cost_explanation") 

2076 builder.para(".donation_request") 

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

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

2079 return builder.build() 

2080 

2081 @classmethod 

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

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

2084 

2085 

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

2087class ThreadReplyEmail(EmailBase): 

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

2089 

2090 author: UserInfo 

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

2092 markdown_text: str 

2093 view_link: str 

2094 

2095 @property 

2096 def string_key_base(self) -> str: 

2097 return "thread_reply" 

2098 

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

2100 return self._localize( 

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

2102 ) 

2103 

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

2105 return markdown_to_plaintext(self.markdown_text) 

2106 

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

2108 builder = self._body_builder(loc_context) 

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

2110 builder.user(self.author) 

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

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

2113 return builder.build() 

2114 

2115 @classmethod 

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

2117 parent = data.WhichOneof("reply_parent") 

2118 if parent == "event": 

2119 parent_context = data.event.title 

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

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

2122 parent_context = data.discussion.title 

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

2124 else: 

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

2126 return cls( 

2127 user_name=user_name, 

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

2129 parent_context=parent_context, 

2130 markdown_text=data.reply.content, 

2131 view_link=view_link, 

2132 ) 

2133 

2134 @classmethod 

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

2136 return [ 

2137 cls( 

2138 user_name="Alice", 

2139 author=UserInfo.dummy_bob(), 

2140 parent_context="Best hiking trails near Berlin", 

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

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

2143 ) 

2144 ] 

2145 

2146 

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

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