7ueugtJd

1"""
22 объекта для работы с почтой по IMAP4
3EmailRead - создает подключение к ящику и позволяет управлять письмами
4Message - само письмо
5Протестировано на yandex.ru
6"""
7import imaplib
8import email
9from typing import List
10
11
12class Message:
13
14    def __init__(self, email_id, body):
15        self._body = body
16        self._email_id = email_id
17
18    def _get_payload(self, part):
19        body = part.get_payload()
20        if isinstance(body, list):
21            part_list: list = []
22            for i in body:
23                part_list.append(self._get_payload(i))
24            return '\n\n'.join(part_list)
25        elif isinstance(body, str):
26            return body
27        else:
28            return body.get_payload()
29
30    def __str__(self):
31        return f'FROM: {self.from_message}\nTO: {self.to}\nSUBJECT: {self.subject}\n\nMESSAGE:\n{self.message}'
32
33    def __repr__(self):
34        return self.__str__()
35
36    @property
37    def email_id(self):
38        return self._email_id
39
40    @staticmethod
41    def _clean_symbols(text: str) -> str:
42        return text.strip().replace('<', '').replace('>', '')
43
44    @property
45    def message(self) -> str:
46        """Возвращает тело письма"""
47        return self._get_payload(self._body)
48
49    @property
50    def subject(self) -> str:
51        """Возвращает тему письма"""
52        return email.header.decode_header(self._body['Subject'])[0][0].decode()
53
54    @property
55    def to(self) -> str:
56        """Возвращает адресата"""
57        text = email.header.decode_header(self._body['TO'])[0][0]
58        return self._clean_symbols(text)
59
60    @property
61    def from_message(self) -> str:
62        """Возвращает отправилтеля"""
63        text = email.header.decode_header(self._body['FROM'])[1][0].decode()
64        return self._clean_symbols(text)
65
66
67class EmailRead:
68    IMAP_HOST: str = 'imap.yandex.ru'
69    LOGIN: str = ''
70    PASSWORD: str = ''
71    FROM: str
72    MAILBOX: str
73    connect: imaplib
74
75    def __init__(self, host: str = None, login: str = None, password: str = None):
76        """
77        :param host:
78        :param login:
79        :param password:
80        """
81        self.IMAP_HOST = host if host is not None else self.IMAP_HOST
82        self.LOGIN = login if login is not None else self.LOGIN
83        self.PASSWORD = password if password is not None else self.PASSWORD
84
85        self.connect = imaplib.IMAP4_SSL(self.IMAP_HOST)
86        self.connect.login(user=self.LOGIN, password=self.PASSWORD)
87
88    def __del__(self):
89        pass
90
91    def search(self, from_mess: str, mailbox: str = 'INBOX') -> List[bytes]:
92        """
93        Возвращает список id писем по заданным параметрам
94        :param from_mess: Поле от кого
95        :param mailbox: Папка для поиска, по умолчанию INBOX - входящие, ALL - везда
96        :return: Список id писем
97        """
98        self.FROM = from_mess
99        self.MAILBOX = mailbox
100
101        self.connect.select(mailbox=self.MAILBOX)
102        status, mail_ids = self.connect.search(None, 'FROM', self.FROM)
103        assert status == 'OK', f'bad status {status}'
104        ids: list = mail_ids[0].split()
105        ids.reverse()
106        return ids
107
108    def id_delete(self, email_id: bytes or list) -> None:
109        """
110        Удаляет письма из ящика
111        :param email_id: список элементов или 1 элемент id писем возвращаемых self.search()
112        :return: None
113        """
114        ids: list = email_id if isinstance(email_id, list) else [email_id]
115        for i in ids:
116            self.connect.store(i, '+FLAGS', '\\Deleted')
117        self.connect.expunge()
118
119    def message_delete(self, message: Message) -> None:
120        """
121        Удаляет письмо из почтового ящака
122        :param message: Объект Message
123        :return: None
124        """
125        self.id_delete(message.email_id)
126
127    def read(self, email_id: bytes or list) -> List[Message]:
128        list_ids: list = email_id if isinstance(email_id, list) else [email_id]
129        messages: list = []
130        for ids in list_ids:
131            status, data = self.connect.fetch(ids, '(RFC822)')
132            assert status == 'OK', f'bad status {status}'
133            text = email.message_from_bytes(data[0][1])
134            messages.append(Message(ids, text))
135
136        return messages