Coverage for app/venv/lib/python3.14/site-packages/weblate/accounts/notifications.py: 65%

1# Copyright © Michal Čihař <michal@weblate.org> 

2# 

3# SPDX-License-Identifier: GPL-3.0-or-later 

4 

5from __future__ import annotations 

6 

7from collections import defaultdict 

8from copy import copy 

9from email.utils import formataddr 

10from typing import TYPE_CHECKING, Any, ClassVar, cast 

11from uuid import uuid4 

12 

13from dateutil.relativedelta import relativedelta 

14from django.conf import settings 

15from django.core.exceptions import ObjectDoesNotExist 

16from django.core.signing import TimestampSigner 

17from django.db.models import IntegerChoices, Q 

18from django.template.loader import render_to_string 

19from django.urls import reverse 

20from django.utils import timezone 

21from django.utils.translation import ( 

22 get_language, 

23 get_language_bidi, 

24 gettext_lazy, 

25 override, 

26 pgettext_lazy, 

27) 

28 

29from weblate.accounts.tasks import EMAIL_BATCH_SIZE, queue_mails 

30from weblate.auth.models import User 

31from weblate.lang.models import Language 

32from weblate.logger import LOGGER 

33from weblate.trans.actions import ActionEvents 

34from weblate.trans.models import ( 

35 Alert, 

36 Change, 

37 Translation, 

38) 

39from weblate.utils.errors import report_error 

40from weblate.utils.markdown import get_mention_users 

41from weblate.utils.ratelimit import rate_limit_notify 

42from weblate.utils.site import get_site_domain, get_site_url 

43from weblate.utils.stats import prefetch_stats 

44from weblate.utils.version import USER_AGENT 

45 

46if TYPE_CHECKING: 46 ↛ 47line 46 didn't jump to line 47 because the condition on line 46 was never true

47 from collections.abc import Iterable 

48 

49 from django.db.models import QuerySet 

50 from django_stubs_ext import StrOrPromise 

51 

52 from weblate.accounts.models import Subscription 

53 from weblate.accounts.tasks import OutgoingEmail 

54 from weblate.trans.models import ( 

55 Announcement, 

56 Comment, 

57 Component, 

58 Project, 

59 Unit, 

60 ) 

61 

62 

63class NotificationFrequency(IntegerChoices): 

64 FREQ_NONE = 0, gettext_lazy("No notification") 

65 FREQ_INSTANT = 1, gettext_lazy("Instant notification") 

66 FREQ_DAILY = 2, gettext_lazy("Daily digest") 

67 FREQ_WEEKLY = 3, gettext_lazy("Weekly digest") 

68 FREQ_MONTHLY = 4, gettext_lazy("Monthly digest") 

69 

70 

71class NotificationScope(IntegerChoices): 

72 SCOPE_ALL = 0, "All" 

73 SCOPE_WATCHED = 10, "Watched" 

74 SCOPE_ADMIN = 20, "Administered" 

75 SCOPE_PROJECT = 30, "Project" 

76 SCOPE_COMPONENT = 40, "Component" 

77 

78 

79NOTIFICATIONS: list[type[Notification]] = [] 

80NOTIFICATIONS_ACTIONS: dict[int, list[type[Notification]]] = {} 

81 

82 

83def get_email_headers(notification: str) -> dict[str, str]: 

84 return { 

85 "X-Mailer": "Weblate" if settings.HIDE_VERSION else USER_AGENT, 

86 "X-Weblate-Notification": notification, 

87 "Message-ID": f"{uuid4()}@{get_site_domain()}", 

88 } 

89 

90 

91def register_notification(handler: type[Notification]) -> type[Notification]: 

92 """Register notification handler.""" 

93 NOTIFICATIONS.append(handler) 

94 for action in handler.actions: 

95 if action not in NOTIFICATIONS_ACTIONS: 

96 NOTIFICATIONS_ACTIONS[action] = [] 

97 NOTIFICATIONS_ACTIONS[action].append(handler) 

98 return handler 

99 

100 

101def is_notificable_action(action: int) -> bool: 

102 return action in NOTIFICATIONS_ACTIONS 

103 

104 

105def dispatch_changes_notifications(changes: Iterable[Change]) -> None: 

106 from weblate.accounts.tasks import notify_changes 

107 

108 notifiable: list[int] = [ 

109 change.pk for change in changes if is_notificable_action(change.action) 

110 ] 

111 if notifiable: 

112 notify_changes.delay_on_commit(notifiable) 

113 

114 

115class Notification: 

116 actions: Iterable[int] = () 

117 verbose: StrOrPromise = "" 

118 template_name: str = "" 

119 digest_template: str = "digest" 

120 filter_languages: bool = False 

121 ignore_watched: bool = False 

122 any_watched: bool = False 

123 required_attr: str | None = None 

124 skip_when_notify: ClassVar[set[type[Notification]]] = set() 

125 

126 def __init__( 

127 self, 

128 outgoing: list[OutgoingEmail], 

129 ) -> None: 

130 self.outgoing: list[OutgoingEmail] = outgoing 

131 self.subscription_cache: dict[int | None, list[Subscription]] = {} 

132 self.child_notify: list[Notification] | None = None 

133 

134 def get_language_filter( 

135 self, change: Change | None, translation: Translation | None 

136 ) -> Language | None: 

137 if self.filter_languages and translation is not None: 

138 return translation.language 

139 return None 

140 

141 @classmethod 

142 def get_freq_choices(cls) -> list[tuple[int, StrOrPromise]]: 

143 return NotificationFrequency.choices 

144 

145 @classmethod 

146 def get_choice(cls) -> tuple[str, StrOrPromise]: 

147 return (cls.get_name(), cls.verbose) 

148 

149 @classmethod 

150 def get_name(cls) -> str: 

151 return cls.__name__ 

152 

153 def filter_subscriptions(self, project: Project | None) -> list[Subscription]: 

154 from weblate.accounts.models import Subscription 

155 

156 result = Subscription.objects.filter(notification=self.get_name()) 

157 scopes: set[NotificationScope] = {NotificationScope.SCOPE_ALL} 

158 # special case for site-wide announcements 

159 if self.any_watched and not project: 159 ↛ 160line 159 didn't jump to line 160 because the condition on line 159 was never true

160 scopes.add(NotificationScope.SCOPE_WATCHED) 

161 

162 query = Q(scope__in=scopes) 

163 

164 if project: 164 ↛ 175line 164 didn't jump to line 175 because the condition on line 164 was always true

165 if not self.ignore_watched: 

166 query |= Q(scope=NotificationScope.SCOPE_WATCHED) & Q( 

167 user__profile__watched=project 

168 ) 

169 # Direct subscriptions 

170 query |= Q(project=project) | Q(component__project=project) 

171 # Admins for current project 

172 query |= Q(scope=NotificationScope.SCOPE_ADMIN) & Q( 

173 user__in=User.objects.all_admins(project) 

174 ) 

175 return list( 

176 result.filter(query) 

177 # Inactive users and bots 

178 .filter(Q(user__is_bot=False) & Q(user__is_active=True)) 

179 .order_by("user", "-scope") 

180 .prefetch_related("user", "user__profile", "user__profile__languages") 

181 ) 

182 

183 def get_subscriptions( 

184 self, 

185 change: Change | None, 

186 project: Project | None, 

187 component: Component | None, 

188 translation: Translation | None, 

189 users: list[int] | None, 

190 ) -> Iterable[Subscription]: 

191 lang_filter: Language | None = self.get_language_filter(change, translation) 

192 cache_key: int | None = project.pk if project else None 

193 if cache_key not in self.subscription_cache: 

194 self.subscription_cache[cache_key] = self.filter_subscriptions(project) 

195 for subscription in self.subscription_cache[cache_key]: 

196 # Users filter 

197 if users is not None and subscription.user_id not in users: 

198 continue 

199 

200 # Languages filter 

201 if ( 

202 lang_filter 

203 and lang_filter not in subscription.user.profile.languages.all() 

204 ): 

205 continue 

206 

207 # Component filter 

208 if subscription.component_id is not None and ( 208 ↛ 211line 208 didn't jump to line 211 because the condition on line 208 was never true

209 component is None or subscription.component_id != component.id 

210 ): 

211 continue 

212 

213 yield subscription 

214 

215 def missing_required_attrs(self, change: Change | None) -> bool: 

216 if not self.required_attr: 

217 return False 

218 if change is None: 218 ↛ 219line 218 didn't jump to line 219 because the condition on line 218 was never true

219 return True 

220 try: 

221 return getattr(change, self.required_attr) is None 

222 except ObjectDoesNotExist: 

223 return False 

224 

225 def get_users( 

226 self, 

227 frequency: NotificationFrequency, 

228 change: Change | None = None, 

229 project: Project | None = None, 

230 component: Component | None = None, 

231 translation: Translation | None = None, 

232 users: list[int] | None = None, 

233 ) -> Iterable[User]: 

234 if self.missing_required_attrs(change): 234 ↛ 235line 234 didn't jump to line 235 because the condition on line 234 was never true

235 return 

236 if change is not None: 236 ↛ 240line 236 didn't jump to line 240 because the condition on line 236 was always true

237 project = change.project 

238 component = change.component 

239 translation = change.translation 

240 last_user = None 

241 subscriptions = self.get_subscriptions( 

242 change, project, component, translation, users 

243 ) 

244 for subscription in subscriptions: 

245 user = subscription.user 

246 # Skip notification in some cases 

247 if ( 

248 # Lower priority subscription for user 

249 (user == last_user) 

250 # Own change 

251 or (change is not None and user == change.user) 

252 ): 

253 continue 

254 

255 last_user = user 

256 if subscription.frequency != frequency: 256 ↛ 258line 256 didn't jump to line 258 because the condition on line 256 was always true

257 continue 

258 if frequency == NotificationFrequency.FREQ_INSTANT and ( 

259 change is None or self.should_skip(user, change) 

260 ): 

261 continue 

262 last_user.current_subscription = subscription 

263 yield last_user 

264 

265 def send( 

266 self, address: str, subject: str, body: str, headers: dict[str, str] 

267 ) -> None: 

268 is_blocked, reason = rate_limit_notify(address) 

269 

270 if is_blocked: 

271 LOGGER.info( 

272 "discarding notification %s to %s due to rate limit: %s", 

273 self.get_name(), 

274 address, 

275 reason, 

276 ) 

277 else: 

278 self.outgoing.append( 

279 { 

280 "address": address, 

281 "subject": subject, 

282 "body": body, 

283 "headers": headers, 

284 } 

285 ) 

286 # Avoid building huge queue of notifications in memory 

287 if len(self.outgoing) > EMAIL_BATCH_SIZE: 287 ↛ 288line 287 didn't jump to line 288 because the condition on line 287 was never true

288 queue_mails(self.outgoing) 

289 self.outgoing.clear() 

290 

291 def render_template(self, suffix: str, context: dict, digest: bool = False) -> str: 

292 """Render single mail template with given context.""" 

293 base_name = self.digest_template if digest else self.template_name 

294 template_name = f"mail/{base_name}{suffix}" 

295 return render_to_string(template_name, context).strip() 

296 

297 def get_context( 

298 self, 

299 change: Change | None = None, 

300 subscription: Subscription | None = None, 

301 extracontext: dict | None = None, 

302 *, 

303 changes: QuerySet[Change] | list[Change] | list[dict[str, Any]] | None = None, 

304 summaries: list[dict[str, Any]] | None = None, 

305 ) -> dict[str, Any]: 

306 """Return context for rendering mail.""" 

307 result = { 

308 "LANGUAGE_CODE": get_language(), 

309 "LANGUAGE_BIDI": get_language_bidi(), 

310 "current_site_url": get_site_url(), 

311 "site_title": settings.SITE_TITLE, 

312 "notification_name": self.verbose, 

313 } 

314 if changes is not None: 314 ↛ 315line 314 didn't jump to line 315 because the condition on line 314 was never true

315 result["changes"] = changes 

316 elif summaries is not None: 316 ↛ 317line 316 didn't jump to line 317 because the condition on line 316 was never true

317 result["changes"] = summaries 

318 if subscription is not None: 318 ↛ 319line 318 didn't jump to line 319 because the condition on line 318 was never true

319 result["unsubscribe_url"] = get_site_url( 

320 "{}?i={}".format( 

321 reverse("unsubscribe"), TimestampSigner().sign(f"{subscription.pk}") 

322 ) 

323 ) 

324 result["subscription_user"] = subscription.user 

325 else: 

326 result["subscription_user"] = None 

327 if extracontext: 327 ↛ 328line 327 didn't jump to line 328 because the condition on line 327 was never true

328 result.update(extracontext) 

329 if change: 329 ↛ 352line 329 didn't jump to line 352 because the condition on line 329 was always true

330 result["change"] = change 

331 # Extract change attributes 

332 attribs = ( 

333 "unit", 

334 "translation", 

335 "component", 

336 "project", 

337 "comment", 

338 "suggestion", 

339 "announcement", 

340 "alert", 

341 "user", 

342 "target", 

343 "old", 

344 "details", 

345 ) 

346 for attrib in attribs: 

347 result[attrib] = getattr(change, attrib) 

348 if change.translation: 348 ↛ 352line 348 didn't jump to line 352 because the condition on line 348 was always true

349 result["translation_url"] = get_site_url( 

350 change.translation.get_absolute_url() 

351 ) 

352 return result 

353 

354 def get_headers(self, context: dict[str, Any]) -> dict[str, str]: 

355 headers = get_email_headers(self.get_name()) 

356 

357 # Set From header to contain user full name 

358 if user := context.get("user"): 358 ↛ 361line 358 didn't jump to line 361 because the condition on line 358 was always true

359 from_name = user.get_visible_name() 

360 else: 

361 from_name = settings.SITE_TITLE 

362 headers["From"] = formataddr((from_name, settings.DEFAULT_FROM_EMAIL)) 

363 

364 # References for unit events 

365 references = None 

366 unit = context.get("unit") 

367 if unit: 367 ↛ 371line 367 didn't jump to line 371 because the condition on line 367 was always true

368 translation = unit.translation 

369 component = translation.component 

370 references = f"{component.project.slug}/{component.slug}/{translation.language.code}/{unit.id}" 

371 if references is not None: 371 ↛ 375line 371 didn't jump to line 375 because the condition on line 371 was always true

372 references = f"<{references}@{get_site_domain()}>" 

373 headers["In-Reply-To"] = references 

374 headers["References"] = references 

375 if unsubscribe_url := context.get("unsubscribe_url"): 375 ↛ 376line 375 didn't jump to line 376 because the condition on line 375 was never true

376 headers["List-Unsubscribe"] = unsubscribe_url 

377 return headers 

378 

379 def send_immediate( 

380 self, 

381 language: str | None, 

382 email: str, 

383 change: Change, 

384 extracontext: dict | None = None, 

385 subscription: Subscription | None = None, 

386 ) -> None: 

387 with override("en" if language is None else language): 

388 context = self.get_context(change, subscription, extracontext) 

389 subject = self.render_template("_subject.txt", context) 

390 context["subject"] = subject 

391 LOGGER.info( 

392 "sending notification %s on %s to %s", 

393 self.get_name(), 

394 context["component"], 

395 email, 

396 ) 

397 self.send( 

398 email, 

399 subject, 

400 self.render_template(".html", context), 

401 self.get_headers(context), 

402 ) 

403 

404 def _convert_change_skip(self, change: Change) -> Change: 

405 return change 

406 

407 def should_skip(self, user: User, change: Change) -> bool: 

408 if not self.skip_when_notify: 

409 return False 

410 if self.child_notify is None: 

411 self.child_notify = [ 

412 notify_class([]) for notify_class in self.skip_when_notify 

413 ] 

414 converted_change = self._convert_change_skip(change) 

415 return any( 

416 list( 

417 child_notify.get_users( 

418 NotificationFrequency.FREQ_INSTANT, 

419 converted_change, 

420 users=[user.pk], 

421 ) 

422 ) 

423 for child_notify in self.child_notify 

424 ) 

425 

426 def notify_immediate(self, change: Change) -> None: 

427 for user in self.get_users(NotificationFrequency.FREQ_INSTANT, change): 427 ↛ 428line 427 didn't jump to line 428 because the loop on line 427 never started

428 if change.project is None or user.can_access_project(change.project): 

429 self.send_immediate( 

430 user.profile.language, 

431 user.email, 

432 change, 

433 subscription=user.current_subscription, 

434 ) 

435 # Delete onetime subscription 

436 current_subscription = cast("Subscription", user.current_subscription) 

437 if current_subscription.onetime: 

438 current_subscription.delete() 

439 

440 def send_digest( 

441 self, 

442 language: str, 

443 email: str, 

444 *, 

445 changes: QuerySet[Change] | list[Change] | list[dict[str, Any]] | None = None, 

446 summaries: list[dict[str, Any]] | None = None, 

447 subscription: Subscription | None = None, 

448 overlimit: bool = False, 

449 ) -> None: 

450 with override("en" if language is None else language): 

451 context = self.get_context( 

452 subscription=subscription, 

453 changes=changes, 

454 summaries=summaries, 

455 extracontext={"overlimit": overlimit}, 

456 ) 

457 subject = self.render_template("_subject.txt", context, digest=True) 

458 context["subject"] = subject 

459 length = 0 

460 if changes: 

461 length = len(changes) 

462 elif summaries: 

463 length = len(summaries) 

464 try: 

465 body = self.render_template(".html", context, digest=True) 

466 except Exception: 

467 report_error("Could not render changes", level="critical") 

468 LOGGER.exception( 

469 "sending digest notification %s on %d changes to %s failed", 

470 self.get_name(), 

471 length, 

472 email, 

473 ) 

474 else: 

475 LOGGER.info( 

476 "sending digest notification %s on %d changes to %s", 

477 self.get_name(), 

478 length, 

479 email, 

480 ) 

481 self.send(email, subject, body, self.get_headers(context)) 

482 

483 def notify_digest( 

484 self, 

485 frequency: NotificationFrequency, 

486 changes: QuerySet[Change], 

487 ) -> None: 

488 notifications: dict[int, list[Change]] = defaultdict(list) 

489 users = {} 

490 for change in changes: 

491 for user in self.get_users(frequency, change): 

492 if change.project is None or user.can_access_project(change.project): 

493 notifications[user.pk].append(change) 

494 users[user.pk] = user 

495 for user in users.values(): 

496 user_changes = notifications[user.pk] 

497 overlimit = False 

498 if len(user_changes) > 100: 

499 user_changes = user_changes[:100] 

500 overlimit = True 

501 

502 self.send_digest( 

503 user.profile.language, 

504 user.email, 

505 changes=user_changes, 

506 subscription=user.current_subscription, 

507 overlimit=overlimit, 

508 ) 

509 

510 def filter_changes( 

511 self, days: int = 0, weeks: int = 0, months: int = 0 

512 ) -> QuerySet[Change]: 

513 return Change.objects.filter( 

514 action__in=self.actions, 

515 timestamp__gte=timezone.now() 

516 - relativedelta(days=days, weeks=weeks, months=months), 

517 ).prefetch_for_render() 

518 

519 def notify_daily(self) -> None: 

520 self.notify_digest( 

521 NotificationFrequency.FREQ_DAILY, self.filter_changes(days=1) 

522 ) 

523 

524 def notify_weekly(self) -> None: 

525 self.notify_digest( 

526 NotificationFrequency.FREQ_WEEKLY, self.filter_changes(weeks=1) 

527 ) 

528 

529 def notify_monthly(self) -> None: 

530 self.notify_digest( 

531 NotificationFrequency.FREQ_MONTHLY, self.filter_changes(months=1) 

532 ) 

533 

534 

535@register_notification 

536class RepositoryNotification(Notification): 

537 actions = ( 

538 ActionEvents.COMMIT, 

539 ActionEvents.PUSH, 

540 ActionEvents.RESET, 

541 ActionEvents.REBASE, 

542 ActionEvents.MERGE, 

543 ) 

544 verbose = pgettext_lazy( 

545 "Notification name", "Operation was performed in the repository" 

546 ) 

547 template_name = "repository_operation" 

548 

549 

550@register_notification 

551class LockNotification(Notification): 

552 actions = ( 

553 ActionEvents.LOCK, 

554 ActionEvents.UNLOCK, 

555 ) 

556 verbose = pgettext_lazy("Notification name", "Component was locked or unlocked") 

557 template_name = "component_lock" 

558 

559 

560@register_notification 

561class LicenseNotification(Notification): 

562 actions = ( 

563 ActionEvents.LICENSE_CHANGE, 

564 ActionEvents.AGREEMENT_CHANGE, 

565 ) 

566 verbose = pgettext_lazy("Notification name", "License was changed") 

567 template_name = "component_license" 

568 

569 

570@register_notification 

571class ParseErrorNotification(Notification): 

572 actions = (ActionEvents.PARSE_ERROR,) 

573 verbose = pgettext_lazy("Notification name", "Parse error occurred") 

574 template_name = "parse_error" 

575 

576 def get_context( 

577 self, 

578 change: Change | None = None, 

579 subscription: Subscription | None = None, 

580 extracontext: dict | None = None, 

581 *, 

582 changes: QuerySet[Change] | list[Change] | list[dict[str, Any]] | None = None, 

583 summaries: list[dict[str, Any]] | None = None, 

584 ) -> dict[str, Any]: 

585 context = super().get_context( 

586 change, subscription, extracontext, changes=changes, summaries=summaries 

587 ) 

588 if change and change.component: 

589 context["details"]["filelink"] = change.component.get_repoweb_link( 

590 change.details.get("filename"), "1", user=context["user"] 

591 ) 

592 return context 

593 

594 

595@register_notification 

596class NewStringNotificaton(Notification): 

597 actions = ( 

598 ActionEvents.NEW_UNIT, 

599 ActionEvents.NEW_UNIT_REPO, 

600 ActionEvents.NEW_UNIT_UPLOAD, 

601 ActionEvents.MARKED_EDIT, 

602 ActionEvents.SOURCE_CHANGE, 

603 ) 

604 verbose = pgettext_lazy("Notification name", "String is available for translation") 

605 template_name = "new_string" 

606 filter_languages = True 

607 required_attr = "unit" 

608 

609 

610@register_notification 

611class NewContributorNotificaton(Notification): 

612 actions = (ActionEvents.NEW_CONTRIBUTOR,) 

613 verbose = pgettext_lazy( 

614 "Notification name", "Contributor made their first translation" 

615 ) 

616 template_name = "new_contributor" 

617 filter_languages = True 

618 

619 

620@register_notification 

621class NewSuggestionNotificaton(Notification): 

622 actions = (ActionEvents.SUGGESTION,) 

623 verbose = pgettext_lazy("Notification name", "Suggestion was added") 

624 template_name = "new_suggestion" 

625 filter_languages = True 

626 required_attr = "suggestion" 

627 

628 

629@register_notification 

630class LanguageTranslatedNotificaton(Notification): 

631 actions = (ActionEvents.COMPLETE,) 

632 verbose = pgettext_lazy("Notification name", "Language was translated") 

633 template_name = "translated_language" 

634 required_attr = "translation" 

635 

636 

637@register_notification 

638class ComponentTranslatedNotificaton(Notification): 

639 actions = (ActionEvents.COMPLETED_COMPONENT,) 

640 verbose = pgettext_lazy("Notification name", "Component was translated") 

641 template_name = "translated_component" 

642 required_attr = "component" 

643 

644 

645@register_notification 

646class NewCommentNotificaton(Notification): 

647 actions = (ActionEvents.COMMENT,) 

648 verbose = pgettext_lazy("Notification name", "Comment was added") 

649 template_name = "new_comment" 

650 filter_languages = True 

651 required_attr = "comment" 

652 

653 def get_language_filter( 

654 self, change: Change | None, translation: Translation | None 

655 ) -> Language | None: 

656 if ( 

657 translation is not None 

658 and change is not None 

659 and not cast("Unit", change.unit).is_source 

660 ): 

661 return translation.language 

662 return None 

663 

664 def notify_immediate(self, change: Change) -> None: 

665 super().notify_immediate(change) 

666 

667 # Notify upstream 

668 report_source_bugs = cast("Component", change.component).report_source_bugs 

669 if change.comment and change.comment.unit.is_source and report_source_bugs: 

670 self.send_immediate("en", report_source_bugs, change) 

671 

672 

673@register_notification 

674class MentionCommentNotificaton(Notification): 

675 actions = (ActionEvents.COMMENT,) 

676 verbose = pgettext_lazy("Notification name", "You were mentioned in a comment") 

677 template_name = "new_comment" 

678 ignore_watched = True 

679 required_attr = "comment" 

680 skip_when_notify: ClassVar[set[type[Notification]]] = {NewCommentNotificaton} 

681 

682 def get_users( 

683 self, 

684 frequency: NotificationFrequency, 

685 change: Change | None = None, 

686 project: Project | None = None, 

687 component: Component | None = None, 

688 translation: Translation | None = None, 

689 users: list[int] | None = None, 

690 ) -> Iterable[User]: 

691 if change is None or self.missing_required_attrs(change): 691 ↛ 692line 691 didn't jump to line 692 because the condition on line 691 was never true

692 return [] 

693 return super().get_users( 

694 frequency, 

695 change, 

696 project, 

697 component, 

698 translation, 

699 list( 

700 get_mention_users(cast("Comment", change.comment).comment).values_list( 

701 "id", flat=True 

702 ) 

703 ), 

704 ) 

705 

706 

707@register_notification 

708class LastAuthorCommentNotificaton(Notification): 

709 actions = (ActionEvents.COMMENT,) 

710 verbose = pgettext_lazy("Notification name", "Your translation received a comment") 

711 template_name = "new_comment" 

712 ignore_watched = True 

713 required_attr = "comment" 

714 skip_when_notify: ClassVar[set[type[Notification]]] = {MentionCommentNotificaton} 

715 

716 def get_users( 

717 self, 

718 frequency: NotificationFrequency, 

719 change: Change | None = None, 

720 project: Project | None = None, 

721 component: Component | None = None, 

722 translation: Translation | None = None, 

723 users: list[int] | None = None, 

724 ) -> Iterable[User]: 

725 change_users: list[int] = [] 

726 if change is not None: 726 ↛ 730line 726 didn't jump to line 730 because the condition on line 726 was always true

727 last_author = cast("Unit", change.unit).get_last_content_change()[0] 

728 if not last_author.is_anonymous: 728 ↛ 729line 728 didn't jump to line 729 because the condition on line 728 was never true

729 change_users.append(last_author.pk) 

730 return super().get_users( 

731 frequency, change, project, component, translation, change_users 

732 ) 

733 

734 

735@register_notification 

736class TranslatedStringNotificaton(Notification): 

737 actions = ( 

738 ActionEvents.CHANGE, 

739 ActionEvents.NEW, 

740 ActionEvents.ACCEPT, 

741 ) 

742 verbose = pgettext_lazy("Notification name", "String was edited by user") 

743 template_name = "translated_string" 

744 filter_languages = True 

745 

746 

747@register_notification 

748class ApprovedStringNotificaton(Notification): 

749 actions = (ActionEvents.APPROVE,) 

750 verbose = pgettext_lazy("Notification name", "String was approved") 

751 template_name = "approved_string" 

752 filter_languages = True 

753 

754 

755@register_notification 

756class ChangedStringNotificaton(Notification): 

757 actions = Change.ACTIONS_CONTENT 

758 verbose = pgettext_lazy("Notification name", "String was changed") 

759 template_name = "changed_translation" 

760 filter_languages = True 

761 skip_when_notify: ClassVar[set[type[Notification]]] = { 

762 TranslatedStringNotificaton, 

763 ApprovedStringNotificaton, 

764 } 

765 

766 

767@register_notification 

768class NewTranslationNotificaton(Notification): 

769 actions = ( 

770 ActionEvents.ADDED_LANGUAGE, 

771 ActionEvents.REQUESTED_LANGUAGE, 

772 ) 

773 verbose = pgettext_lazy("Notification name", "New language was added or requested") 

774 template_name = "new_language" 

775 

776 def get_context( 

777 self, 

778 change: Change | None = None, 

779 subscription: Subscription | None = None, 

780 extracontext: dict | None = None, 

781 *, 

782 changes: QuerySet[Change] | list[Change] | list[dict[str, Any]] | None = None, 

783 summaries: list[dict[str, Any]] | None = None, 

784 ) -> dict[str, Any]: 

785 context = super().get_context( 

786 change, subscription, extracontext, changes=changes, summaries=summaries 

787 ) 

788 if change: