Mejoras a logger y limpieza de email

common/helper/logger.py

@@ -5,6 +5,12 @@ from threading import Thread
 import queue
 
 
+def get_today(tz):
+    today = datetime.datetime.now(tz=tz)
+    locale.setlocale(locale.LC_TIME, 'es_ES')
+    return today
+
+
 class Logger:
     """
     Clase que lleva el diario de actividades de la secretaria
@@ -45,25 +51,31 @@ class Logger:
         self.log(msg)
 
     def log_action(self, action):
-        today = datetime.datetime.now(tz=self.tz)
+        msg = 'he realizado {0}'.format(action)
-        locale.setlocale(locale.LC_TIME, 'es_ES')
+        self.log_msg(msg)
-        msg = 'A las {0} del {1}, he realizado {2}'.format(today.strftime('%H:%M:%S'),
-                                                           today.strftime('%d de %B de %Y'), action)
-        self.log(msg)
 
     def log_not_action(self, action):
-        today = datetime.datetime.now(tz=self.tz)
+        msg = 'no he podido realizar {0}'.format(action)
-        locale.setlocale(locale.LC_TIME, 'es_ES')
+        self.log_msg(msg)
-        msg = 'A las {0} del {1}, no he podido realizar {2}'.format(today.strftime('%H:%M:%S'),
+
-                                                                    today.strftime('%d de %B de %Y'), action)
+    def start_turn(self, action):
-        self.log(msg)
+        self.log_msg('Inicio de turno de {0}'.format(action))
+
+    def end_turn(self, action):
+        self.log_msg('Termino de turno de {0}'.format(action))
+
+    def log_msg(self, msg):
+        today = get_today(self.tz)
+        line = 'A las {0} del {1}, {2}'.format(today.strftime('%H:%M:%S'), today.strftime('%d de %B de %Y'), msg)
+        self.log(line)
 
     def log(self, message):
-        line = message.rstrip('.') + '.'
+        line = message
+        if line[-1] != '.' and line != '--------':
+            line = line.rstrip('.') + '.'
         self.messages.append(line)
         if len(self.messages) > 1000:
             self.start_new()
-            self.load_last()
         with open(self.get_filename(), 'a') as f:
             f.write(line + "\n")
 
@@ -78,21 +90,40 @@ class Worker(Thread):
         self.logging = params['logging']
         self.queue.put({'is_start': True})
 
+    def parse_message(self, message):
+        if 'is_start' in message and message['is_start']:
+            self.logger.start_log()
+            return
+        if 'action' in message:
+            if 'not' in message and message['not']:
+                self.logger.log_not_action(message['action'])
+                return
+            self.logger.log_action(message['action'])
+            return
+        if 'start_turn' in message:
+            self.logger.start_turn(message['start_turn'])
+            return
+        if 'end_turn' in message:
+            self.logger.end_turn(message['end_turn'])
+            return
+        self.logger.log_msg(message['message'])
+
     def run(self):
         self.logging.log('Starting', caller=type(self))
         while not self.event.is_set():
             try:
                 message = self.queue.get(timeout=self.wait)
                 self.logging.log('Logger received message', caller=type(self))
-                if 'is_start' in message and message['is_start']:
+                self.parse_message(message)
-                    self.logger.start_log()
-                    continue
-                if 'not' in message and message['not']:
-                    self.logger.log_not_action(message['action'])
-                    continue
-                self.logger.log_action(message['action'])
             except queue.Empty:
-                pass
+                continue
+        while True:
+            try:
+                message = self.queue.get(timeout=self.wait)
+                self.logging.log('Logger received message', caller=type(self))
+                self.parse_message(message)
+            except queue.Empty:
+                break
         self.logger.stop_log()
         self.logging.log('Exiting', caller=type(self))
         return

config/email.json

@@ -17,6 +17,7 @@
     },
     "ssl": true
   },
+  "max": 5,
   "server": "imap.yandex.com",
   "port": 993,
   "user": {

src/email/__init__.py

@@ -1,3 +1 @@
-from .email_interpreter import EmailInterpreter
+from .workers import Obtenedor, Validador, Confirmador
-from .revisor_worker import RevisorEmailWorker
-from .workers import Obtenedor, Validador, Confirmador

src/email/email_interpreter.py

@@ -1,10 +0,0 @@
-from threading import Thread
-
-
-class EmailInterpreter(Thread):
-    def __init__(self, configs, params):
-        super().__init__()
-        self.logging = params['logging']
-
-    def run(self):
-        self.logging.log('Starting', type(self))

src/email/revisor_worker.py

@@ -1,90 +0,0 @@
-from threading import Thread
-import re
-import time
-import email.utils
-from bs4 import BeautifulSoup
-from entry.email.inbox import connect, check_inbox
-from src.communication.message import Message
-
-
-class RevisorEmailWorker(Thread):
-    def __init__(self, configs, params):
-        super().__init__()
-        self._url = configs.get('email.server')
-        self._port = configs.get('email.port')
-        self._username = configs.get('email.username')
-        self._password = configs.get('email.password')
-        self._ssl = configs.get('email.ssl')
-        self.queue = params['queues']['emails']
-        self.questions = params['queues']['questions']
-        self.event = params['events']['stop']
-        self._wait = configs.get('supervisor.wait')
-        self._bosses = params['bosses']
-        self._logger = params['queues']['log']
-        self.logging = params['logging']
-        self.revisados = []
-
-    def revisado(self, uid):
-        if not self.check_revisado(uid):
-            self.revisados.append(uid)
-
-    def check_revisado(self, uid):
-        if uid in self.revisados:
-            return True
-        return False
-
-    def run(self):
-        self.logging.log('Starting', type(self))
-        self._logger.put({'action': 'Inicio jornada trabajador Revisor Email'})
-        while not self.event.is_set():
-            self.logging.log('Looping status {0}'.format(not self.event.is_set()), type(self))
-            self.logging.log('Connecting to Email Server', type(self))
-            imap = connect(imap_url=self._url, port=self._port, username=self._username, password=self._password,
-                           ssl=self._ssl)
-            self.logging.log('Getting emails', type(self))
-            emails = check_inbox(imap)
-            if emails is not None:
-                c = 0
-                p = 0
-                for em in emails:
-                    if self.check_revisado(em.uid):
-                        continue
-                    sender = em.message['from']
-                    text = ' '.join([em.message['subject'] + '.'] + self.build_message(em.message))
-                    msg = Message('email', text=text, original=em, sender=sender,
-                                  datetime=email.utils.parsedate_to_datetime(em.message['Date']))
-                    if not self._bosses.is_boss(sender):
-                        self.logging.log('Sender {0} is not a boss'.format(sender), type(self))
-                        self.revisado(em.uid)
-                        self.questions.put(msg)
-                        p += 1
-                        continue
-                    self.queue.put(msg)
-                    self.revisado(em.uid)
-                    c += 1
-                self.logging.log('{0} emails checked'.format(c), type(self))
-                if c > 0:
-                    self._logger.put({'action': 'Revise {0} nuevos correos'.format(c)})
-                self.logging.log('{0} emails pending'.format(p), type(self))
-                if p > 0:
-                    self._logger.put({'action': 'Tengo dudas en {0} correos'.format(p)})
-            imap.close()
-            time.sleep(self._wait)
-        self.logging.log('Exiting', type(self))
-        return
-
-    def build_message(self, email_part):
-        output = []
-        if email_part.is_multipart():
-            for part in email_part.get_payload():
-                output.append(self.build_message(part))
-        else:
-            html = email_part.get_payload(decode=True)
-            bs = BeautifulSoup(html, 'html.parser')
-            if bs.body:
-                html = bs.body.get_text()
-            else:
-                html = bs.get_text()
-            html = re.sub(' +', ' ', re.sub("\n+", ' ', html)).strip(' ')
-            output.append(html)
-        return output

src/email/workers.py

@@ -55,7 +55,7 @@ class Obtenedor(Worker):
 
     def run(self) -> None:
         self.logger.log('Starting', type(self))
-        self.diary.put({'action': 'Inicio de turno de Obtenedor'})
+        self.diary.put({'start_turn': 'Obtenedor'})
         while not self.stop.is_set():
             e = 0
             with connect(self.url, self.port, self.user.name, self.user.password, self.ssl) as imap:
@@ -75,15 +75,15 @@ class Obtenedor(Worker):
                     self.add_revisado(em.uid)
                     e += 1
             self.logger.log('{0} new emails found'.format(e), type(self))
-            self.diary.put({'action': 'Obtenidos{0} correos nuevos'.format(e)})
+            self.diary.put({'message': 'Obtenidos {0} correos nuevos'.format(e)})
             time.sleep(self.frec)
+        self.diary.put({'end_turn': 'Obtenedor'})
         self.logger.log('Exiting', type(self))
-        self.diary.put({'action': 'Terminando el turno de Obtenedor'})
 
 
 class Validador(Worker):
     def __init__(self, configs, params):
-        super(Validador, self).__init__(configs, params)
+        super(Validador, self).__init__(configs=configs, params=params)
 
         self.emails = params['queues']['emails']
         self.validos = params['queues']['valid']
@@ -94,7 +94,7 @@ class Validador(Worker):
 
     def run(self):
         self.logger.log('Starting', type(self))
-        self.diary.put({'action': 'Inicio de turno de Validador'})
+        self.diary.put({'start_turn': 'Validador'})
         while not self.stop.is_set():
             try:
                 em = self.emails.get(timeout=self.frec)
@@ -106,7 +106,7 @@ class Validador(Worker):
                 continue
             self.validos.put(em)
         self.logger.log('Exiting', type(self))
-        self.diary.put({'action': 'Terminando el turno de Validador'})
+        self.diary.put({'end_turn': 'Validador'})
 
 
 class Confirmador(Worker):
@@ -115,15 +115,37 @@ class Confirmador(Worker):
 
         self.invalidos = params['queues']['invalid']
         self.frec = configs.get('supervisor.wait')
+        self.max = configs.get('email.max')
+        self.mensajes = []
 
-    def crear_mega_mensaje(self):
+    def mandar(self):
-        pass
+        self.logger.log('Sending {0} strange emails'.format(len(self.mensajes)), type(self))
+        print(self.mensajes)
+
+    def crear_mega_mensaje(self, msg):
+        if len(self.mensajes) >= self.max:
+            self.mandar()
+            self.mensajes = []
+        self.mensajes.append(msg)
 
     def run(self) -> None:
         self.logger.log('Starting', type(self))
+        self.diary.put({'start_turn': 'Confirmador'})
         while not self.stop.is_set():
             try:
                 em = self.invalidos.get(self.frec)
+                self.crear_mega_mensaje(em)
             except queue.Empty:
-                pass
+                continue
         self.logger.log('Exiting', type(self))
+        self.diary.put({'end_turn': 'Confirmador'})
+
+
+class Procesador(Worker):
+    def __init__(self, configs, params):
+        super(Procesador, self).__init__(configs=configs, params=params)
+
+    def run(self) -> None:
+        self.diary.put({'start_turn': 'Procesador'})
+        self.diary.put({'end_turn': 'Procesador'})
+        pass