moving risk_data function to a new file and tests for Escribir_Datos_Osiris

Dockerfile

@@ -8,6 +8,8 @@ COPY Pipfile Pipfile.lock ./
 
 RUN pipenv install --system --deploy --ignore-pipfile
 
+ENV file_directory="Subida Osiris"
+
 COPY . .
 
 CMD ["python3", "src/subida.py"]

Pipfile

@@ -4,12 +4,13 @@ verify_ssl = true
 name = "pypi"
 
 [packages]
-pytest = "7.2.2"
 pandas = "2.0.0"
 numpy = "1.24.2"
-ipdb = "0.13.13"
 
 [dev-packages]
+pytest = "7.2.2"
+ipdb = "0.13.13"
+freezegun = "1.2.2"
 
 [requires]
 python_version = "3.10"

src/risk_data.py

@@ -0,0 +1,64 @@
+import pandas as pd
+from constants.constants import NUMBER_OF_COLUMNS
+from write_data_osiris import Escribir_Datos_Osiris
+
+
+def get_phone_risk(df_risk: pd.DataFrame) -> pd.DataFrame:
+    frames = list()
+    num_tel = len([col_name for col_name in list(df_risk.columns) if 'tel_riesgo' in col_name])
+    for i in range(1, num_tel + 1):
+        col_riesgo = f'tel_riesgo_{i}'
+        df = df_risk.loc[df_risk[col_riesgo].notnull(), ['Cuenta', col_riesgo]]\
+            .rename(columns={col_riesgo: 'TEL'}).copy()
+        df['OBS'] = f'RIESGO {i}'
+        df['ID_FONO'] = '8'
+        frames.append(df)
+    df_phone_risk = pd.concat(frames)
+    return df_phone_risk
+
+
+def read_osiris_accounts() -> pd.DataFrame:
+    try:
+        uploaded_accounts = pd.read_csv('cuentas.csv', encoding='latin_1', sep=';', dtype=str)
+    except Exception:
+        uploaded_accounts = pd.read_csv('cuentas.csv', encoding='ANSI', sep=';', dtype=str)
+    uploaded_accounts = uploaded_accounts[['Cuenta', 'Mat. Unica']].rename(
+        columns={
+            'Mat. Unica': 'DNI'
+            },
+        inplace=False
+        )
+    return uploaded_accounts
+
+
+def risk_data():
+    'NECESARIOS: tener el archivos.csv de riesgo y el archivo de las cuentas de osiris como cuentas.csv'
+    col_numbers = {
+        'Número.1': 'tel_riesgo_1',
+        'Número.2': 'tel_riesgo_2',
+        'Número.3': 'tel_riesgo_3',
+        'Número.4': 'tel_riesgo_4'
+    }
+
+    col_list = [x for x in range(NUMBER_OF_COLUMNS)]
+    try:
+        risk = pd.read_csv('riesgo.csv', sep=';', encoding='latin_1', dtype=str, index_col=False, usecols=col_list)
+    except Exception:
+        risk = pd.read_csv('riesgo.csv', sep=';', encoding='ANSI', dtype=str, index_col=False, usecols=col_list)
+    df_risk = risk[['DNI'] + list(col_numbers.keys()) + ['NSE']]
+
+    df_risk = df_risk.rename(columns=col_numbers, inplace=False)
+
+    uploaded_accounts = read_osiris_accounts()
+
+    df_risk = pd.merge(uploaded_accounts, df_risk, how="inner", on="DNI")
+
+    df_phone_risk = get_phone_risk(df_risk=df_risk)
+
+    Escribir_Datos_Osiris(
+        df_phone_risk,
+        'RIESGO_telefonos.csv',
+        ['Cuenta', 'ID_FONO', 'TEL', 'OBS'],
+        ["ID Cuenta o Nro. de Asig. (0)", "ID Tipo de Teléfono (17)", "Nro. de Teléfono (18)", "Obs. de Teléfono (19)"]
+    )
+    print('Planilla de Teléfonos de RIESGO escrita')

src/subida.py

@@ -3,24 +3,23 @@
 import shutil
 import pandas as pd
 import numpy as np
-import datetime
-import time
+from datetime import datetime
 import traceback
 
 from driver_email import enviar_mail_con_adjuntos
 from constants.constants import (
     PROGRAMMER,
     USER,
     PASSWORD,
-    DATA_UPLOADER_HEADER,
     PROVINCES,
     UTIL_COLS_COMAFI,
     ACCOUNT_PREP_COL,
-    NUMBER_OF_COLUMNS,
     DATA_PREP_COLUMNS,
     DATA_INFO_COLUMNS,
     DATA_INFO_COLUMNS_RENAME,
 )
+from risk_data import risk_data
+from write_data_osiris import Escribir_Datos_Osiris
 
 
 def limpiar_numeros(df_num):
@@ -61,34 +60,6 @@ def limpiar_numeros(df_num):
     return df_num
 
 
-def Escribir_Datos_Osiris(df, filename, cols_df, cols_osiris):
-
-    Control_Carpeta_Subida()
-
-    df_subida = pd.DataFrame(columns=DATA_UPLOADER_HEADER)
-    df_subida[cols_osiris] = df[cols_df]
-    try:
-        df_subida.to_csv(
-            f'Subida Osiris/{time.strftime("(%H.%M hs) -")} {filename}',
-            sep=';',
-            index=False,
-            encoding='latin_1'
-        )
-    except Exception:
-        df_subida.to_csv(
-            f'Subida Osiris/{time.strftime("(%H.%M hs) -")} {filename}',
-            sep=';',
-            index=False,
-            encoding='ANSI'
-        )
-
-
-def Control_Carpeta_Subida():
-    'COntrola que exista una carpeta llamada Subida Osiris y sino la crea'
-    pass
-    # no la estoy usando
-
-
 def Preparacion_Cuentas():
 
     "Condiciones"
@@ -141,14 +112,14 @@ def Preparacion_Cuentas():
         df_sub = df_sub.drop('riesgo', inplace=False, axis=1)
         try:
             df_sub.to_csv(
-                f'Subida Osiris/{time.strftime("(%H.%M hs) -")} subida_cartera_{name}.csv',
+                f'Subida Osiris/{datetime.now().strftime("(%H.%M hs) -")} subida_cartera_{name}.csv',
                 sep=';',
                 encoding='latin_1',
                 index=False
             )
         except Exception:
             df_sub.to_csv(
-                f'Subida Osiris/{time.strftime("(%H.%M hs) -")} subida_cartera_{name}.csv',
+                f'Subida Osiris/{datetime.now().strftime("(%H.%M hs) -")} subida_cartera_{name}.csv',
                 sep=';',
                 encoding='ANSI',
                 index=False
@@ -213,7 +184,7 @@ def Preparacion_Cuentas_Comafi():
     df_os['IDSucursal(17)'] = '1'
     df_os['subcliente'] = df['subcliente']
 
-    name_folder = f'Subida Osiris/{time.strftime("(%H.%M hs) -")} {nombre_cartera}'
+    name_folder = f'Subida Osiris/{datetime.now().strftime("(%H.%M hs) -")} {nombre_cartera}'
     if os.path.isdir(name_folder):
         shutil.rmtree(name_folder)
     os.mkdir(name_folder)
@@ -360,82 +331,21 @@ def Preparacion_Datos_Comafi():
     print('Guardando planilla subida...')
     try:
         df_subida.to_csv(
-            f'Subida Osiris/{time.strftime("(%H.%M hs) -")}DATOS_EMERIX_subida_telefono.csv',
+            f'Subida Osiris/{datetime.now().strftime("(%H.%M hs) -")}DATOS_EMERIX_subida_telefono.csv',
             sep=';',
             index=False,
             encoding='ANSI'
         )
     except Exception:
         df_subida.to_csv(
-            f'Subida Osiris/{time.strftime("(%H.%M hs) -")}DATOS_EMERIX_subida_telefono.csv',
+            f'Subida Osiris/{datetime.now().strftime("(%H.%M hs) -")}DATOS_EMERIX_subida_telefono.csv',
             sep=';',
             index=False,
             encoding='ANSI'
         )
     print('Guardado exitoso!')
 
 
-def get_phone_risk(df_risk: pd.DataFrame) -> pd.DataFrame:
-    frames = list()
-    num_tel = len([col_name for col_name in list(df_risk.columns) if 'tel_riesgo' in col_name])
-    for i in range(1, num_tel + 1):
-        col_riesgo = f'tel_riesgo_{i}'
-        df = df_risk.loc[df_risk[col_riesgo].notnull(), ['Cuenta', col_riesgo]]\
-            .rename(columns={col_riesgo: 'TEL'}).copy()
-        df['OBS'] = f'RIESGO {i}'
-        df['ID_FONO'] = '8'
-        frames.append(df)
-    df_phone_risk = pd.concat(frames)
-    return df_phone_risk
-
-
-def read_osiris_accounts() -> pd.DataFrame:
-    try:
-        uploaded_accounts = pd.read_csv('cuentas.csv', encoding='latin_1', sep=';', dtype=str)
-    except Exception:
-        uploaded_accounts = pd.read_csv('cuentas.csv', encoding='ANSI', sep=';', dtype=str)
-    uploaded_accounts = uploaded_accounts[['Cuenta', 'Mat. Unica']].rename(
-        columns={
-            'Mat. Unica': 'DNI'
-            },
-        inplace=False
-        )
-    return uploaded_accounts
-
-
-def risk_data():
-    'NECESARIOS: tener el archivos.csv de riesgo y el archivo de las cuentas de osiris como cuentas.csv'
-    col_numbers = {
-        'Número.1': 'tel_riesgo_1',
-        'Número.2': 'tel_riesgo_2',
-        'Número.3': 'tel_riesgo_3',
-        'Número.4': 'tel_riesgo_4'
-    }
-
-    col_list = [x for x in range(NUMBER_OF_COLUMNS)]
-    try:
-        risk = pd.read_csv('riesgo.csv', sep=';', encoding='latin_1', dtype=str, index_col=False, usecols=col_list)
-    except Exception:
-        risk = pd.read_csv('riesgo.csv', sep=';', encoding='ANSI', dtype=str, index_col=False, usecols=col_list)
-    df_risk = risk[['DNI'] + list(col_numbers.keys()) + ['NSE']]
-
-    df_risk = df_risk.rename(columns=col_numbers, inplace=False)
-
-    uploaded_accounts = read_osiris_accounts()
-
-    df_risk = pd.merge(uploaded_accounts, df_risk, how="inner", on="DNI")
-
-    df_phone_risk = get_phone_risk(df_risk=df_risk)
-
-    Escribir_Datos_Osiris(
-        df_phone_risk,
-        'RIESGO_telefonos.csv',
-        ['Cuenta', 'ID_FONO', 'TEL', 'OBS'],
-        ["ID Cuenta o Nro. de Asig. (0)", "ID Tipo de Teléfono (17)", "Nro. de Teléfono (18)", "Obs. de Teléfono (19)"]
-    )
-    print('Planilla de Teléfonos de RIESGO escrita')
-
-
 def Datos_Info():
     'NECESARIO cuentas de osiris como cuentas.csv, cuentas deinfo exporto como info.xlsx'
     cuentas_subidas = pd.read_csv('cuentas.csv', sep=';', dtype=str)

src/write_data_osiris.py

@@ -0,0 +1,34 @@
+
+from datetime import datetime
+import os
+import pandas as pd
+from constants.constants import DATA_UPLOADER_HEADER
+
+
+def Escribir_Datos_Osiris(df: pd.DataFrame, filename: str, cols_df: list[str], cols_osiris: list[str]):
+
+    Control_Carpeta_Subida()
+
+    df_subida = pd.DataFrame(columns=DATA_UPLOADER_HEADER)
+    df_subida[cols_osiris] = df[cols_df]
+    try:
+        df_subida.to_csv(
+            f'{os.getenv("file_directory", "Subida Osiris")}/{datetime.now().strftime("(%H.%M hs) -")} {filename}',
+            sep=';',
+            index=False,
+            encoding='latin_1'
+        )
+
+    except Exception:
+        df_subida.to_csv(
+            f'{os.getenv("file_directory", "Subida Osiris")}/{datetime.now().strftime("(%H.%M hs) -")} {filename}',
+            sep=';',
+            index=False,
+            encoding='ANSI'
+        )
+
+
+def Control_Carpeta_Subida():
+    'COntrola que exista una carpeta llamada Subida Osiris y sino la crea'
+    pass
+    # no la estoy usando

tests/test_get_phone_numbers.py → tests/test_risk_data.py

@@ -1,5 +1,5 @@
 import pandas as pd
-from src.subida import (
+from src.risk_data import (
     get_phone_risk,
     read_osiris_accounts,
     risk_data
@@ -51,9 +51,9 @@ def test_read_osiris_account_return(self, mock_read_csv):
 
     @mock.patch("pandas.read_csv")
     @mock.patch("pandas.merge")
-    @mock.patch("src.subida.read_osiris_accounts")
-    @mock.patch("src.subida.get_phone_risk")
-    @mock.patch("src.subida.Escribir_Datos_Osiris")
+    @mock.patch("src.risk_data.read_osiris_accounts")
+    @mock.patch("src.risk_data.get_phone_risk")
+    @mock.patch("src.risk_data.Escribir_Datos_Osiris")
     def test_risk_data(
         self,
         mock_Escribir_Datos_Osiris,