Procesado de cdr moviles fijos xtratelecom realizado

.env

@@ -11,10 +11,10 @@ CDR_SFTP_PORT=22
 
 SQLSERVER_SERVER=192.168.2.200
 SQLSERVER_DATABASE=bdd_wifi
-SQLSERVER_USERNAME=apicdr
-SQLSERVER_PASSWORD=cdr5485
+SQLSERVER_USERNAME=bdbackupservice
+SQLSERVER_PASSWORD=bdbs2k21
 
-SQLSERVER_SERVER_TEST=192.168.2.200
-SQLSERVER_DATABASE_TEST=cdr_test
-SQLSERVER_USERNAME_TEST=apicdr
-SQLSERVER_PASSWORD_TEST=cdr5485
+SQLSERVER_SERVER_TEST=192.168.2.93
+SQLSERVER_DATABASE_TEST=bdd_wifi
+SQLSERVER_USERNAME_TEST=bdbackupservice
+SQLSERVER_PASSWORD_TEST=bdbs2k21
\ No newline at end of file

Readme.md

-# Importación de CDRs y Facturas de MásMóvil
+# Importación de CDRs de fijos xtratelecom
 Descarga CDRs fijas desde el sFPT de MásMóvil para añadirlos a los sistemas de Rentel.
 
-
 ## Introducción
 ### Funcionalidad
-Se ejecuta de forma periodica para conectarse a dos sFTP diferentes desde donde extrae el consumo de las líneas fijas (de clientes con fibra indirecta). 
+Se ejecuta de forma periodica para conectarse al sFTP desde donde extrae el consumo de las líneas fijas 
 
 ### Lenguajes de Programación
 Le herramienta está desarrollada con Python. Se utilizan librerías de python para el acceso a los servidores SFTP.
@@ -12,15 +11,13 @@ Le herramienta está desarrollada con Python. Se utilizan librerías de python p
 ### Credenciales
 | Servicio    | Descripción                                                                           | Producción | Dirección                          | Base de datos | Usuario | Contraseña | 
 |-------------|---------------------------------------------------------------------------------------|------------|------------------------------------|---------------|---------|------------|
-| CDR Fijos y móviles | Servidor FTP de CableMóvil con los CDR de los consumos de las líneas Fijos y móviles          | Si         | sftp.xtratelecom.es                |               | cF33028 | 9BRW78sf   |
-| SQL Server  | Servidor Principal SQL Server                                                         | Si         | 10.58.4.5 / 192.168.2.200          | bdd_wifi      | apicdr  | cdr5485    |
+| CDR Fijos   | Servidor FTP de CableMóvil con los CDR de los consumos de las líneas Fijos y móviles          | No        | sftp.xtratelecom.es                |               | cF33028 | 9BRW78sf   |
+| SQL Server  | Servidor Principal SQL Server                                                         | Si         |  192.168.2.200          | bdd_wifi      | apicdr  | cdr5485    |
 
 
 ### Importador MásMóvil
-Está compuesto de un script, `setup.py`,  se encarga de conectarse al sFTP de MásMóvil, y descargar los archivos de CDRs nuevos.
-Posteriormente procederemos al procesado de dichos ficheros, guardando cada línea de estos archivos previamente descargados en las tablas `TB_Cdr`, y `TB_Cdr_Fijos`. 
+Está compuesto de un script, `main.py`,  se encarga de conectarse al sFTP de MásMóvil, y descargar los archivos de CDRs nuevos y guarda un registro en TB_CDR_Ficheros (tipo = 2) .
+Posteriormente procederemos al procesado de dichos ficheros, guardando cada línea de estos archivos previamente descargados en las tablas `TB_Cdr_Fijos`. 
 En cada línea aparece la información de cada llamada, el teléfono que realiza la llamada, destino de llamada, duración, entre otros.
 
-
 El script se encuentra alojado en la máquina `debian.utils` con la IP `192.168.2.63`.
-Dicho script se ejecutará según la configuración de un crontab.
\ No newline at end of file

app/conector_db_wifi.py

@@ -2,6 +2,8 @@ import os
 from log import logging
 import platform
 import pyodbc
+from dotenv import load_dotenv
+load_dotenv()
 
 class ConectorDbWifi():
     
@@ -27,12 +29,12 @@ class ConectorDbWifi():
                 )
             else:
                 self.connection = pyodbc.connect(
-                    f"DRIVER=FreeTDS;server={sql_server_hostname};PORT=1433;DATABASE={sql_server_database};UID={sql_server_username};PWD={sql_server_password};TDS_Version=7.0", encoding="utf-8", ansi=True
+                    f"DRIVER=FreeTDS;server={sql_server_hostname};PORT=1433;DATABASE={sql_server_database};UID={sql_server_username};PWD={sql_server_password};TDS_Version=7.0", ansi=True
                 )
         
             
         except Exception as e:
-            print('Error al conectarse con la base de datos dbo.wifi')
+            print(f'Error al conectarse con la base de datos dbo.wifi => {e}')
             logging.warning('Error al conectarse con la base de datos dbo.wifi')
             
     

app/manager_cdr_fijos.py

-from conector_sftp import SFTPConexion
-import os, time
+from app.conector_sftp import SFTPConexion
+import os, time, csv
 from datetime import datetime
 from paramiko.ssh_exception import SSHException
 from log import logging
-from conector_db_wifi import ConectorDbWifi
+from dotenv import load_dotenv
+from app.conector_db_wifi import ConectorDbWifi
+from mailer import send_email_warning 
+
+load_dotenv()
+
+APP_BASE_PATH: str = os.path.abspath(os.path.join(__file__, "../.."))
+DOWNLOAD_PATH = os.path.join(APP_BASE_PATH,'cdrs')
 
 class ManagerCdrFijos:
     
@@ -11,7 +18,7 @@ class ManagerCdrFijos:
         hostname_xtrtelecom = os.getenv('CDR_HOSTNAME')
         username_xtrtelecom = os.getenv('CDR_USERNAME')
         pwd_xtrtelecom = os.getenv('CDR_PASSWORD')
-        port_xtrtelecom = int(os.getenv("CDR_SFTP_PORT"))
+        port_xtrtelecom = int(os.getenv('CDR_SFTP_PORT'))
       
         # Guardo las rutas absolutas donde se almacena los ficheros que vamos a descargar
         self.filename_fijo_download_sftp_path = []
@@ -19,15 +26,15 @@ class ManagerCdrFijos:
         self.filename_fijo_download_sftp = []
         # Diccionario donde se almacenara nombre fichero: fecha sftp de la creación del mismo en el sftp
         self.filename_datetime_fijo = {}
-        
+    
         try:    
             self.cnx_sftp_xtrtelecom = SFTPConexion.connection(
                 hostname_xtrtelecom, username_xtrtelecom, pwd_xtrtelecom, port_xtrtelecom
                 )
         except SSHException as e:
             message = "No se ha podido establecer la conexión con el servidor de los CDR de móviles y fijos.. Error: {e}".format(e=e)
-            logging.info(message)
             print(message)
+            logging.info(message)
         
         
     def get_cdr_filename_download_fijos(self):
@@ -80,16 +87,15 @@ class ManagerCdrFijos:
             Descarga los archivos CDR del servidor SFTP almacenandolos en el directorio cdrs
         """
         try:
-            with ConectorDbWifi() as cursor:
-                if self.filename_fijo_download_sftp:
-                    for file in self.filename_fijo_download_sftp:
-                        self.cnx_sftp_xtrtelecom.get("cdrs/" + file,  "/home/rentel/procesado-cdr-fijos/cdrs/" + file)
-                        filepath = "/home/rentel/procesado-cdr-fijos/cdrs/" + file
-                        message = f'Fichero descargados correctamente en la carpeta cdrs {file}'
-                        logging.info(message)
-                else:
-                    message = "No hay fichero para descargar "
+            if self.filename_fijo_download_sftp:
+                for file in self.filename_fijo_download_sftp:
+                    self.cnx_sftp_xtrtelecom.get("cdrs/" + file,  "/home/rentel/procesado-cdr-fijos/cdrs/" + file)
+                    filepath = "/home/rentel/procesado-cdr-fijos/cdrs/" + file
+                    message = f'Fichero descargados correctamente en la carpeta cdrs {file}'
                     logging.info(message)
+            else:
+                message = "No hay fichero para descargar "
+                logging.info(message)
         except Exception as e:
             logging.warning(f'Error al descargar los archivos CDR del servidor SFTP: {e}')
             
@@ -150,6 +156,206 @@ class ManagerCdrFijos:
                     logging.info(message)
     
     
+
+    def save_cdr_fijos_into_database(self):
+        """
+            Procesa los cdr fijos(tipo=2) que esten sin procesar(precessed_at = null) en la tabla TB_CDR_Ficheros
+        """
+       
+        with ConectorDbWifi() as cursor:
+            try:
+                filenames = []
+                # Obtengo el nombre del fichero de fijo que necesito procesar
+                cursor.execute("SELECT filename FROM TB_CDR_Ficheros WHERE tipo = 2 AND processed_at is NULL ORDER BY fecha_origen")
+                           
+                fetch = cursor.fetchall()
+                if fetch == None:
+                    message = 'No existe ficheros de moviles para procesar en la tabla TB_CDR_Ficheros'
+                    logging.info(message)
+                else:
+                    for file in fetch:
+                        filenames.append(file[0])
+               
+                if len(filenames) != 0:
+                    # Obtengo una lista con las rutas de los ficheros a procesar
+                    path_filenames = cdr_list(filenames)
+                    
+                    #Crea el proceso de cdr de fijos introduciendolo en la tabla tb_cdr_fijos
+                    cdr_insert_update_cdr_fijos(path_filenames, cursor)    
+                          
+            except  Exception as e:
+                logging.error(f"An error occurred: {e}")        
+        
+
+def cdr_insert_update_cdr_fijos(files:list, cursor):
+    """
+        Inserto o actualizo en la tabla TB_Cdr_fijos los registros de fijos.
+        Args:
+            files:list -> Lista de rutas absolutas de los ficheros de fijos para procesar
+            cursor:cursor -> cursor de conexión a la base de datos ddd_wifi
+    """
+    print(files)
+    for cdr_file in files:
+        # En caso de que se halla producido un error en el procesado de un fichero 
+        # tendremos el num de registros procesados por el que se ha quedado guardado en la bd TB_CDR_ficheros
+        # con este proceso empezariamos desde el registro donde se quedo
+        
+        print(f'llega')
+        nombre_fichero = os.path.basename(cdr_file)
+        print(nombre_fichero)
+        cursor.execute("SELECT Num_Reg_Pro, Num_Reg_Dup FROM TB_CDR_Ficheros WHERE filename = ?",
+                        nombre_fichero,
+                        ) 
+        Num_Reg_Pro_d = cursor.fetchone()
+        print('pasa')
+        Num_Reg_Pro_db = int(Num_Reg_Pro_d[0]) + int(Num_Reg_Pro_d[1]) # Total de registros procesados de un fichero por si hay una excepción partir el procesado desde ese registro
+        time_init = datetime.now()
+        
+        print(cdr_file)
+        if "mvno_" not in cdr_file:
+                logging.info(f'Inicio procesado del fichero {cdr_file}')
+                logging.info(f'Número de registros procesados de ese fichero en la base de datos {Num_Reg_Pro_db}')
+                Num_Reg_Pro_file = 0
+                cdr_fijos_update_count = int(Num_Reg_Pro_d[1]) #Contador de registros actualizados
+                cdr_fijos_insert_count = int(Num_Reg_Pro_d[0]) #Contador de registros insertados
+                cdr_observaciones_duplicados = ''
+                line_process_actual = 0
+             
+                try:
+                    for csv_line in read_file(cdr_file):
+                        line_process_actual += 1
+                        if Num_Reg_Pro_db == Num_Reg_Pro_file: # En caso de que antes alla surgido una excepción empezaremos a partir de la linea donde causo la excepción
+                           
+                            fijo_ddi = csv_line[0]
+                            fijo_tarifa = csv_line[1]
+                            fijo_destino = str(csv_line[2])
+                            fijo_fecha = datetime.strptime(
+                                csv_line[3] + " " + csv_line[4], '%d-%m-%Y %H:%M:%S')
+                            fijo_duracion = csv_line[5]
+                            fijo_desc_destino = csv_line[6]
+                            fijo_coste = csv_line[7]
+                            fijo_info = csv_line[8]
+                            fijo_fichero = os.path.basename(cdr_file)
+                            
+                            cursor.execute("SELECT count(*) FROM TB_Cdr_Fijos WHERE ddi = ? AND fecha = ?", fijo_ddi, fijo_fecha,)
+                            fetch = cursor.fetchone()
+                            
+                            if fetch[0] != 0:
+                                id = str(fetch[0])
+                                
+                                cursor.execute("""UPDATE TB_Cdr_Fijos SET ddi = ?, tarifa = ?, destino = ?, fecha = ?, duracion = ?, desc_destino = ?, coste = ?, info = ? WHERE id = ?""",(
+                                    fijo_ddi, fijo_tarifa, fijo_destino, fijo_fecha, fijo_duracion, fijo_desc_destino, fijo_coste, fijo_info, str(fetch[0]))
+                                            )
+                                cursor.commit()
+                                cdr_fijos_update_count += 1
+                                cdr_observaciones_duplicados += f'{str(id)}, '
+                            else:
+
+                                cursor.execute(
+                                        "INSERT INTO TB_Cdr_Fijos (ddi, tarifa, destino, fecha, duracion, desc_destino, coste, info, fichero) "
+                                        "VALUES ( ?, ?, ?, ?, ?, ?, ?, ?, ? )",
+                                        (fijo_ddi, fijo_tarifa, fijo_destino, fijo_fecha, fijo_duracion, fijo_desc_destino, fijo_coste, fijo_info,fijo_fichero,),
+                                    )
+                                cursor.commit()
+                                
+                                cdr_fijos_insert_count += 1
+                                
+                                
+                        else:
+                            Num_Reg_Pro_file += 1   
+                            
+                except Exception as e:
+                    print(f'Excepcion: {e}')
+                    message = f'Se ha producido un error en la linea {csv_line} {e}'
+                    logging.info(message)
+                    total_process = cdr_fijos_insert_count + cdr_fijos_update_count
+                    message = f'Total registros procesados {total_process}.'
+                    logging.info(message)
+                    cursor.execute("UPDATE TB_CDR_Ficheros SET Observaciones=?, Num_Reg_Pro=?, Num_Reg_Dup=? WHERE filename=?",
+                                    message,
+                                    cdr_fijos_insert_count,
+                                    cdr_fijos_update_count,
+                                    nombre_fichero,
+                                )
+                    cursor.commit() 
+                    message_warning = f'Se ha producido un error en la linea {csv_line} {e}'
+                    send_email_warning(cursor, message_warning)
+                    quit() 
+                    
+        if cdr_observaciones_duplicados == '' :
+                cdr_observaciones = ''
+        else:
+            cdr_observaciones = f'Duplicados: {cdr_observaciones_duplicados}'
+
+        time_final = datetime.now()
+        time_procesado =  time_final - time_init 
+        time_procesado_fin = str(time_procesado)
+        Num_Reg_Txt = calcular_line_file(cdr_file)
+        cursor.execute(
+            "UPDATE TB_CDR_Ficheros SET processed_at=?, Num_Reg_Txt=?, Num_Reg_Pro=?, Num_Reg_Dup=?, Time_Procesado=?,Observaciones=? WHERE filename=?",
+            datetime.now(),
+            Num_Reg_Txt,
+            cdr_fijos_insert_count,
+            cdr_fijos_update_count,
+            time_procesado_fin,
+            cdr_observaciones,
+            nombre_fichero,
+        )
+        cursor.commit()
+        
+        print(nombre_fichero)
+        message = "Se ha procesado el fichero {fichero}.".format(
+        fichero=nombre_fichero
+        )
+        logging.info(message)
+        message = "Se han insertado  {registros} en la tabla TB_Cdr.".format(
+        registros=cdr_fijos_insert_count
+        )
+        logging.info(message)
+        message = "Se han actualizado {registros} en la tabla TB_Cdr.".format(
+        registros=cdr_fijos_update_count
+        )
+        logging.info(message)
+        message = "Número total de lineas que contiene el fichero {registros}".format(
+        registros=Num_Reg_Txt
+        )
+        logging.info(message)
+
+
+def read_file(file_path: str) -> list:
+        """
+            Genera una lista por cada archivo con los datos a introducir en la base de datos.
+
+        Args:
+            file_path (str): Ruta del archivo.
+
+        Returns:
+            list: Lista con las líneas de los archivos.
+        """
+        with open(file_path, newline="", encoding="latin-1") as file:
+            reader = csv.reader(file, delimiter=";")
+            lines = list(reader)
+        return lines
+       
+        
+def cdr_list(filenames:list) -> list:
+    """
+        Genera una lista con la ruta de los archivos que coincida con el nombre de los archivos pasados
+        Param:
+            filenames -> list: Lista de nombre de ficheros
+        Returns:
+            list: Lista con rutas de los archivos coincidentes
+    """
+    print(filenames)
+    print(DOWNLOAD_PATH)
+    list_files= []
+    for currentpath, folders, files in os.walk(DOWNLOAD_PATH):
+        for file in files:
+            if file in filenames:
+               list_files.append(os.path.join(currentpath, file)) 
+    return list_files
+
+
 def calcular_line_file(filepath):
     with open(filepath, "rb") as myfile:
         num_lines=myfile.readlines()

mailer.py

+from email.mime.text import MIMEText
+from email.mime.multipart import MIMEMultipart
+import smtplib
+import itertools
+   
+
+def send_email_warning(cursor, message:str):
+    """Envio de correo si falla el proceso de los cdr
+        Args:
+            cursor: cnx.
+            message(str): Mensaje de error.
+    """
+    #Obtengo credenciales para enviar el correo smtp
+    cursor.execute("SELECT Usuario, Password, Remitente, Servidor FROM TB_Servidores_De_Correos where IdServidorCorreo=2")
+    data_send_email = cursor.fetchone()
+   
+    if data_send_email is not None:
+        usuario = data_send_email[0]
+        contraseña = data_send_email[1]
+        remitente = data_send_email[2]
+        servidor = data_send_email[3]
+        destinatario = 'informatica@rentelwifi.com'
+       
+        # Datos del correo
+        asunto = f"Error procesado de cdr fijos"
+
+        # Crear el mensaje HTML
+        mensaje_html = f"""
+            <html>
+            <body>
+                <h3>Error procesado de cdr fijos</h3>
+                <p>{message}<p>
+            </body>
+            </html>    
+            """
+       
+        # Crea un mensaje MIME
+        mime_mensaje = MIMEMultipart()
+        mime_mensaje["From"] = remitente
+        mime_mensaje["To"] = destinatario
+        mime_mensaje["Subject"] = asunto
+
+        # Agregar el mensaje HTML al mensaje MIME
+        mensaje_html_mime = MIMEText(mensaje_html, "html")
+        mime_mensaje.attach(mensaje_html_mime)
+
+        # Conecta al servidor SMTP y envía el correo
+        try:
+            servidor = smtplib.SMTP(servidor, 587)
+            servidor.starttls()
+            servidor.login(usuario, contraseña)
+            servidor.sendmail(remitente, destinatario, mime_mensaje.as_string())
+            servidor.quit()
+            #print(f'Correo enviado a {destinatario}')
+        except Exception as e:
+            raise Exception(f'Fallo en el envio de email al comercial {e}')
+
+
+
+

main.py

+from app.manager_cdr_fijos import ManagerCdrFijos
+
+### -- Accede al SFTP de xtratelecom y descarga el fichero que le toca y crea registro en tb_cdr_ficheros -- ### 
+if __name__ == '__main__':
+    
+    x = ManagerCdrFijos()
+    # Obtengo nombre de los ficheros de fijos a descargar
+    x.get_cdr_filename_download_fijos()
+    # Descargo de sftp los cdr de fijos
+    x.download_cdr_fijos_xtratelecom()
+    # Inserto registro en la tb_cdr_ficheros
+    x.update_insert_date_fijo()
+    
+    # Procesa ficheros de fijos y los inserta o actualiza en tb_CDR_fijos
+    x.save_cdr_fijos_into_database()
\ No newline at end of file