diff --git a/src/risk_data.py b/src/risk_data.py new file mode 100644 index 0000000..31b90f8 --- /dev/null +++ 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 index 916f4a7..0e2a0ba 100644 --- 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 new file mode 100644 index 0000000..ce32b72 --- /dev/null +++ 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 b/tests/test_risk_data.py similarity index 94% rename from tests/test_get_phone_numbers.py rename to tests/test_risk_data.py index bf7644a..1b412d3 100644 --- a/tests/test_get_phone_numbers.py +++ b/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,