| 36 | |
| 37 | |
| 38 | class UnsubscribeHandler: |
| 39 | def _extract_unsub_info_from_message( |
| 40 | self, message: Message |
| 41 | ) -> Optional[UnsubscribeData]: |
| 42 | header_value = message[headers.SUBJECT] |
| 43 | if not header_value: |
| 44 | return None |
| 45 | return UnsubscribeEncoder.decode_subject(header_value) |
| 46 | |
| 47 | def handle_unsubscribe_from_message(self, envelope: Envelope, msg: Message) -> str: |
| 48 | unsub_data = self._extract_unsub_info_from_message(msg) |
| 49 | if not unsub_data: |
| 50 | LOG.w("Wrong format subject %s", msg[headers.SUBJECT]) |
| 51 | return status.E507 |
| 52 | mailbox = Mailbox.get_by(email=envelope.mail_from) |
| 53 | if not mailbox: |
| 54 | LOG.w("Unknown mailbox %s", envelope.mail_from) |
| 55 | return status.E507 |
| 56 | |
| 57 | if unsub_data.action == UnsubscribeAction.DisableAlias: |
| 58 | return self._disable_alias(unsub_data.data, mailbox.user, mailbox) |
| 59 | elif unsub_data.action == UnsubscribeAction.DisableContact: |
| 60 | return self._disable_contact(unsub_data.data, mailbox.user, mailbox) |
| 61 | elif unsub_data.action == UnsubscribeAction.UnsubscribeNewsletter: |
| 62 | return self._unsubscribe_user_from_newsletter(unsub_data.data, mailbox.user) |
| 63 | elif unsub_data.action == UnsubscribeAction.OriginalUnsubscribeMailto: |
| 64 | return self._unsubscribe_original_behaviour(unsub_data.data, mailbox.user) |
| 65 | else: |
| 66 | raise Exception(f"Unknown unsubscribe action {unsub_data.action}") |
| 67 | |
| 68 | def handle_unsubscribe_from_request( |
| 69 | self, user: User, unsub_request: str |
| 70 | ) -> Optional[UnsubscribeData]: |
| 71 | unsub_data = UnsubscribeEncoder.decode_subject(unsub_request) |
| 72 | if not unsub_data: |
| 73 | LOG.w("Wrong request %s", unsub_request) |
| 74 | return None |
| 75 | if unsub_data.action == UnsubscribeAction.DisableAlias: |
| 76 | response_code = self._disable_alias(unsub_data.data, user) |
| 77 | elif unsub_data.action == UnsubscribeAction.DisableContact: |
| 78 | response_code = self._disable_contact(unsub_data.data, user) |
| 79 | elif unsub_data.action == UnsubscribeAction.UnsubscribeNewsletter: |
| 80 | response_code = self._unsubscribe_user_from_newsletter( |
| 81 | unsub_data.data, user |
| 82 | ) |
| 83 | elif unsub_data.action == UnsubscribeAction.OriginalUnsubscribeMailto: |
| 84 | response_code = self._unsubscribe_original_behaviour(unsub_data.data, user) |
| 85 | else: |
| 86 | raise Exception(f"Unknown unsubscribe action {unsub_data.action}") |
| 87 | if response_code == status.E202: |
| 88 | return unsub_data |
| 89 | return None |
| 90 | |
| 91 | def _disable_alias( |
| 92 | self, alias_id: int, user: User, mailbox: Optional[Mailbox] = None |
| 93 | ) -> str: |
| 94 | alias = Alias.get(alias_id) |
| 95 | if not alias: |
no outgoing calls