-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathapplication.py
162 lines (132 loc) · 6.23 KB
/
application.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
import email
import imaplib
import threading
from PySide2 import QtCore, QtWidgets, QtMultimedia
from telethon.sync import TelegramClient
from controllers.c_main import CMain
from models.config_data import ConfigData
from models.my_email import MyEmail
from windows.app_main_window import AppMainWindow
class App(QtWidgets.QApplication):
signal_new_emails_obtained = QtCore.Signal()
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.config_data = ConfigData()
self.gui = AppMainWindow()
self.c_main = CMain(self, self.gui)
self.client = None
self.telegram_client_connected = False
self.email_count = 0
self.old_emails = set()
self.new_emails = set()
self.get_emails_error = None
self.conected_with_email = False
self.timer_get_new_emails = QtCore.QTimer()
self.timer_get_new_emails.setSingleShot(False)
self.timer_get_new_emails.timeout.connect(self._get_emails)
self.signal_new_emails_obtained.connect(self._check_new_emails)
self.player = QtMultimedia.QMediaPlayer(self.gui, QtMultimedia.QMediaPlayer.LowLatency)
self.player.setMedia(QtCore.QUrl('resources/alarm.mp3'))
self.player.mediaStatusChanged.connect(self._repeat_alarm)
self.gui.connect_signals(self.c_main)
self._setup_app()
self.run_flananini()
def _check_new_emails(self):
if self.get_emails_error:
self.conected_with_email = False
self.gui.text_log.append(self.get_emails_error)
elif not self.conected_with_email:
self.conected_with_email = True
self.gui.text_log.append('Conectado con éxito con la cuenta de email.')
my_emails = {email for email in self.new_emails if email.from_.lower() in self.config_data.emails_to_check}
if not self.old_emails:
self.old_emails = my_emails
return
new_emails = my_emails - self.old_emails
self.old_emails = my_emails
for new_email in new_emails:
self.email_count += 1
if self.telegram_client_connected:
self.client.send_message(self.config_data.send_to, f'''
NUEVO CORREO
Despiertame cuando sea necesario, si es que es necesario.
=================================================
Asunto:
{new_email._subject}
Contenido:
{new_email._text or 'Correo vacío'}''', silent=False)
self.gui.text_log.append(f'Correo enviado a {self.config_data.send_to} ({self.email_count}).')
else:
self.gui.text_log.append(f'Correo nuevo ({self.email_count}).')
if new_emails and self.c_main.alarm_activated and self.player.state() != QtMultimedia.QMediaPlayer.PlayingState:
self.player.play()
def _get_emails(self):
def get_emails():
self.get_emails_error = None
imap = imaplib.IMAP4_SSL('imap.outlook.com')
try:
imap.login(self.config_data.email_name, self.config_data.email_pass)
except (AttributeError, imaplib.IMAP4.error) as e:
self.get_emails_error = 'Error: no se pudo conectar con la cuenta de email.'
self.signal_new_emails_obtained.emit()
return
status, n_messages = imap.select('INBOX')
n_messages = int(n_messages[0])
my_emails: set[MyEmail] = set()
for id_ in range(n_messages, n_messages - self.config_data.n_emails, -1):
res, msg = imap.fetch(str(id_), '(RFC822)')
for response in msg:
if not isinstance(response, tuple):
continue
my_email = MyEmail()
msg = email.message_from_bytes(response[1])
my_email.subject = email.header.decode_header(msg['Subject'])[0][0]
try:
my_email.from_ = msg['From'].split('<')[1].split('>')[0]
except:
self.get_emails_error = 'Error: "from" no tiene el formato "<direccion>"'
for part in msg.walk():
if part.get_content_type() == 'text/plain':
try:
my_email.text = part.get_payload(decode=True).decode()
except:
my_email.text = part.get_payload()
if my_email:
my_emails.add(my_email)
imap.close()
imap.logout()
self.new_emails = my_emails
self.signal_new_emails_obtained.emit()
thread = threading.Thread(target=get_emails)
thread.start()
def _repeat_alarm(self, media_status):
if media_status == QtMultimedia.QMediaPlayer.EndOfMedia:
self.player.play()
def _setup_app(self):
self.setStyle('fusion')
def run_flananini(self):
def get_code_from_user():
dialog = QtWidgets.QInputDialog(self.gui)
dialog.setInputMode(QtWidgets.QInputDialog.TextInput)
dialog.setWindowTitle('FlanaNini')
dialog.setLabelText('Introduce el código recibido en telegram')
dialog.setOkButtonText('Aceptar')
dialog.setCancelButtonText('Cancelar')
dialog.exec_()
return dialog.textValue()
try:
self.conected_with_email = False
self._get_emails()
self.timer_get_new_emails.start(self.config_data.period * 1000 * 60)
if self.client:
self.client.disconnect()
self.client = TelegramClient('session', self.config_data.api_id, self.config_data.api_hash)
self.client.start(self.config_data.phone, code_callback=get_code_from_user)
self.telegram_client_connected = True
self.gui.text_log.append('Conectado con éxito con telegram.')
self.email_count = 0
except (RuntimeError, ValueError):
self.telegram_client_connected = False
self.gui.text_log.append('Sin conectar con telegram. Modo solo alarma.')
except Exception as e:
self.gui.text_log.append(f'Error de configuración: {str(e)}')