moving risk_data function to a new file and moving Escribir_Datos_Osi…

…ris to a new file to avoid circular import between risk_data file and subida.py
alquimistas-org · Apr 8, 2023 · 88e347f · 88e347f
1 parent 94b4dc6
commit 88e347f
Show file tree

Hide file tree

Showing 4 changed files with 103 additions and 96 deletions.
diff --git a/src/risk_data.py b/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')
diff --git a/src/subida.py b/src/subida.py
@@ -7,20 +7,20 @@
 import time
 import traceback
 
-from driver_email import enviar_mail_con_adjuntos
+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 +61,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"
@@ -375,67 +347,6 @@ def Preparacion_Datos_Comafi():
     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)

diff --git a/src/write_data_osiris.py b/src/write_data_osiris.py
@@ -0,0 +1,32 @@
+
+import pandas as pd
+import time
+from .constants.constants import DATA_UPLOADER_HEADER
+
+
+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
diff --git a/tests/test_get_phone_numbers.py → tests/test_risk_data.py b/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,