Notes

from pyspark.sql import DataFrame
from pyspark_archetype.common.dicts import LazyDict
from pyspark import StorageLevel
from pyspark_archetype.parameters.process_parameter import Joins
from pyspark_archetype.transformation.reader import Reader
from pyspark_archetype.transformation.writer import Writer
from pyspark.sql.functions import col, lit, concat, when, current_timestamp, date_format, row_number,
broadcast, coalesce, trim, row_number, sum, substring, max, min, count,
length, instr, lpad, regexp_replace,avg,monotonically_increasing_id
from pyspark.sql.window import Window
from functools import reduce

from utils.utils import Utils


class Transformation(Joins, Utils):
"""
Transformations associated with the generation of .
"""

def __init__(self) -> None:
"""
Instance from Reader and Writer
"""
self._reader = Reader()
self._writer = Writer()
self._local_parameters = self._reader.config.local_parameters
self.spark = self._reader.spark

"""
params
"""
self.prm_spark_environment = self._local_parameters['PRM_SPARK_ENVIRONMENT']
self.prm_spark_fecha_rutina = self._local_parameters['PRM_SPARK_FECHA_RUTINA']
self.prm_spark_catalog = self._local_parameters['PRM_SPARK_CATALOG']
self.ruta = self._local_parameters["PRM_TEMP_PATH"]

"""
ESQUEMAS
"""
self.prm_spark_schema_udv = self._local_parameters['PRM_SPARK_SCHEMA_TABLE_UDV_BASE']
self.prm_spark_schema_ddv = self._local_parameters['PRM_SPARK_SCHEMA_TABLE_DDV_BASE']
"""
TABLAS
"""
self.HM_CONSOLIDADOOPERACIONSOSPECHOSA_CONS = "hm_consolidadooperacionsospechosa_ro"
self.ME_CUENTAFIDEICOMISOOPERACIONSOSPECHOSA_CONS = "me_cuentafideicomisooperacionsospechosa"
self.ME_TIPOPERACIONSOSPECHOSA_CONS = "me_tipoperacionsospechosa_ro"
self.ME_RELACIONDOCUMENTOBCPOPERACIONSOSPECHOSA_CONS = "me_relaciondocumentobcpoperacionsospechosa_ro"
self.ME_RELACIONOFICINABCPOPERACIONSOSPECHOSA_CONS = "me_relacionoficinabcpoperacionsospechosa_ro"
self.ME_RELACIONBANCOENTIDADOPERACIONSOSPECHOSA_CONS = "me_relacionbancoentidadoperacionsospechosa_ro"
self.HE_DETALLECLIENTEOPERACIONSOSPECHOSA_CONS = "he_detalleclienteoperacionsospechosa_ro"
self.H_INDIVIDUO_CONS = "h_individuo_ro"
self.H_CLIENTE_CONS = "h_cliente_ro"
self.M_DESPAIS_CONS = "m_despais_ro"
self.M_DEPARTAMENTO_CONS = "m_departamento_ro"
self.M_PROVINCIA_CONS = "m_provincia_ro"
self.M_DISTRITO_CONS = "m_distrito_ro"
self.M_UBIGEO_CONS = "m_ubigeo_ro"
self.M_DESCANAL_CONS = "m_descanal_ro"
self.M_DESMONEDA_CONS = "m_desmoneda_ro"
self.M_BANCOEXTERIOR_CONS = "m_bancoexterior_ro"
self.M_DESTIPOPROFESION_CONS = "m_destipoprofesion_ro"
self.M_AGENCIA_CONS = "m_agencia_ro"
self.H_TIPOCAMBIO_CONS = "h_tipocambio_ro"
self.H_PARTYDIRECCION_CONS = "h_partydireccion_llave_ro"
self.H_PARTYDAC_CONS = "h_partydac_llave_ro"
self.H_PARTY_CONS = "h_party_ro"
self.H_RELACIONCUENTAFINANCIERACLIENTE_CONS = "h_relacioncuentafinancieracliente_ro"
self.M_DESTIPOCLASIFICACIONINTERNAPARTY_CONS = "m_destipoclasificacioninternaparty_ro"
self.MM_DESTIPCUENTAOPERACIONSOSPECHOSA_CONS = "mm_destipcuentaoperacionsospechosa"
self.ME_INFORMACIONAPLICATIVOOPERACIONSOSPECHOSA_CONS = "me_informacionaplicativooperacionsospechosa_ro"

"""
CONSTANTES
"""
self.FECDIA_CONS = "fecdia"
self.FECRUTINA_CONS = "fecrutina"
self.CODMES_CONS = "codmes"
self.CINCUENTA_MIL_CONS = 50000
self.DIEZ_MIL_CONS = 10000
self.N_CONS = "N"
self.S_CONS = "S"
self.TIPOPE_FIL_CONS = ['24', '25', '26', '27', '28', '29', '30', '32']
self.DESORIGENDINERO_CONS = "ERROR EN TRANSMISIÓN DE ORIGEN DE FONDOS DESDE TELLER"

self.FECTIPCAMBIO_CONS = "fectipcambio"
self.LLAVES_CONS = ["codsecuencial", "codclavectaordenante", "codmes", "flgopemultiple"]
self.PARTICION_CONS = ["codmes", "codappros", "flgopemultiple"]
self.NONE_CONS = None
self.COLS_TEMP_SEGMENTOS = [
"reg_id", "codofisbs", "numregistroopebcp", "codmes", "codsecuencial", "codclavectaordenante", "codcanal",
"descanal", "codruc_beneficiario", "codruc_ordenante",
"codruc_solicitante", "codclaveunicoclibeneficiario", "codclaveunicocliordenante",
"codclaveunicoclisolicitante", "codclavectabeneficiario", "codclavectasolicitante",
"codinternocomputacionalbeneficiario", "codinternocomputacionalordenante",
"codinternocomputacionalsolicitante", "tiprolclibeneficiario", "tiprolcliordenante", "tiprolclisolicitante",
"codclavepartyclibeneficiario", "codclavepartycliordenante", "codclavepartyclisolicitante", "codmoneda",
"codubigeo", "codappros", "codtrxbcp", "fecdia",
"flgtrxcompensacionelectronica", "flg_cce_intermediario", "flgtrxinternacional",
"flg_internacional_alcance", "flgumbraltrx", "hortrx", "mtocambiodolarros", "mtotrx", "mtotrxdol",
"mtodestino", "numtrxapp", "tipopereportesbs", "codmonedadestino", "codcargoclibeneficiario",
"codcargocliordenante",
"codcargoclisolicitante", "codclavepartycliperbeneficiario", "codclavepartycliperordenante",
"codclavepartyclipersolicitante",
"coddistritobeneficiario", "coddistritoordenante", "coddistritosolicitante", "codpaisdestinosbs",
"codpaisemisorsbs", "codswiftbcoexterioremisor", "codswiftbcoexteriordestino", "numcheque",
"desorigendinero", "nbrocupacionclibeneficiario", "nbrocupacioncliordenante", "nbrocupacionclisolicitante",
"flgumbralbeneficiario", "flgumbralordenante", "flgumbralsolicitante",
"codclavepartyage", "tipclibeneficiario_mod", "tipclibeneficiario", "tipcliordenante_mod",
"tipcliordenante",
"tipclisolicitante_mod", "tipclisolicitante", "tippartyidentificacionbeneficiario",
"tippartyidentificacionordenante",
"tippartyidentificacionsolicitante", "tipindicadorlavadoactivo",
"codpais_beneficiario", "codpais_ordenante", "codpais_solicitante", "mtotrx_sbs", "codmoneda_sbs",
"mtotrx_sbs_dol", "mtocambioalnuevosol", "codprofesion_beneficiario", "codprofesion_ordenante",
"codprofesion_solicitante", "codsucage", "flgtrxefectivo", "flgresidentesolicitante",
"flgresidenteordenante", "flgresidentebeneficiario", "tiprelpersonabcosolicitante",
"tiprelpersonabcoordenante",
"tiprelpersonabcobeneficiario", "desorigenfondos", "m_tipmancomuno", "m_origenro", "m_codclavecta",
"m_numcodclavepartycli_o", "m_numcodclavepartycli_b", "m_codclavepartycli_ori", "m_codclaveunicocli_ori",
"key_codclavepartyclisolicitante", "key_codclavepartycliordenante", "key_codclavepartyclibeneficiario",
"key_codclaveunicoclisolicitante", "key_codclaveunicocliordenante", "key_codclaveunicoclibeneficiario",
"tipctaordinario", "tipcargoabono", "tipctabeneficiario", "codbco_ord", "codbco_ben",
"codprovinciasolicitante", "coddepartamentosolicitante", "codprovinciaordenante",
"coddepartamentoordenante",
"codprovinciabeneficiario", "coddepartamentobeneficiario", "tipctabeneficiario_cal", "tipctaordenante",
"tipctasolicitante", "codcanal_mod", "descanal_mod", "nroparche", "codcuentas_orden_1",
"codcuentas_orden_2", "codcuentas_orden_4", "codcuentas_benef_1", "codcuentas_benef_2",
"codcuentas_benef_4", "codciiu_beneficiario", "codciiu_ordenante",
"codciiu_solicitante", "umbralminimo", "tip_rel", "cond_res", "ocup", "cod_ciiu", "cargo_per", "dept_per",
"prov_per", "distrit_per", "modalidad_ope", "tipocliente", "numregistroopesbs"]

# Lectura de fuentes

def __get_hm_consolidadooperacionsospechosa(self):
"""
Obtiene consolidado de operaciones sospechosas del mes de proceso.

Returns:
tuple: (DataFrame con datos, DataFrame vacío de error)
"""
consolidadooperacionsospechosa_df, empty_df = Utils.get_dataframe_max_mes(
self,
self.HM_CONSOLIDADOOPERACIONSOSPECHOSA_CONS,
self.prm_spark_schema_ddv,
self.prm_spark_catalog,
self.CODMES_CONS,
self.prm_spark_fecha_rutina
)

return consolidadooperacionsospechosa_df, empty_df

def __get_h_cliente(self):
"""
Obtiene información histórica de clientes del mes anterior.

Returns:
tuple: (DataFrame con datos, DataFrame vacío de error)
"""
cliente_df, empty_df = Utils.get_dataframe_max_mes_anterior(
self,
self.H_CLIENTE_CONS,
self.prm_spark_schema_udv,
self.prm_spark_catalog,
self.FECDIA_CONS,
self.prm_spark_fecha_rutina
)

return cliente_df, empty_df

def __get_h_individuo(self):
"""
Obtiene información histórica de individuos del mes anterior.

Returns:
tuple: (DataFrame con datos, DataFrame vacío de error)
"""
individuo_df, empty_df = Utils.get_dataframe_max_mes_anterior(
self,
self.H_INDIVIDUO_CONS,
self.prm_spark_schema_udv,
self.prm_spark_catalog,
self.FECDIA_CONS,
self.prm_spark_fecha_rutina
)

return individuo_df, empty_df

def __get_h_tipocambio(self):
"""
Obtiene tipos de cambio del mes anterior filtrados por clasificación '0002'.

Returns:
tuple: (DataFrame con tipos de cambio filtrados, DataFrame vacío de error)
"""
tipocambio_df, empty_df = Utils.get_dataframe_mes_anterior(
self,
self.H_TIPOCAMBIO_CONS,
self.prm_spark_schema_udv,
self.prm_spark_catalog,
self.FECTIPCAMBIO_CONS,
self.prm_spark_fecha_rutina
)

return tipocambio_df.filter(col("tipclasifinternatipcambio") == '0002'), empty_df

def __get_h_partydireccion(self):
"""
Obtiene direcciones de party del mes anterior.

Returns:
tuple: (DataFrame con direcciones, DataFrame vacío de error)
"""
partydireccion_df, empty_df = Utils.get_dataframe_max_mes_anterior(
self,
self.H_PARTYDIRECCION_CONS,
self.prm_spark_schema_udv,
self.prm_spark_catalog,
self.FECDIA_CONS,
self.prm_spark_fecha_rutina
)

return partydireccion_df, empty_df

def __get_h_partydac(self):
"""
Obtiene información de party DAC del mes anterior.

Returns:
tuple: (DataFrame con party DAC, DataFrame vacío de error)
"""
partydac_df, empty_df = Utils.get_dataframe_max_mes_anterior(
self,
self.H_PARTYDAC_CONS,
self.prm_spark_schema_udv,
self.prm_spark_catalog,
self.FECDIA_CONS,
self.prm_spark_fecha_rutina
)

return partydac_df, empty_df

def __get_h_party(self):
"""
Obtiene información histórica de party del mes anterior.

Returns:
tuple: (DataFrame con party, DataFrame vacío de error)
"""
party_df, empty_df = Utils.get_dataframe_max_mes_anterior(
self,
self.H_PARTY_CONS,
self.prm_spark_schema_udv,
self.prm_spark_catalog,
self.FECDIA_CONS,
self.prm_spark_fecha_rutina
)

return party_df, empty_df

def __get_h_relacioncuentafinancieracliente(self):
"""
Obtiene relaciones cuenta financiera-cliente del mes anterior.

Returns:
tuple: (DataFrame con relaciones, DataFrame vacío de error)
"""
relacioncuentafinancieracliente_df, empty_df = Utils.get_dataframe_max_mes_anterior(
self,
self.H_RELACIONCUENTAFINANCIERACLIENTE_CONS,
self.prm_spark_schema_udv,
self.prm_spark_catalog,
self.FECRUTINA_CONS,
self.prm_spark_fecha_rutina
)

return relacioncuentafinancieracliente_df, empty_df

def __get_he_detalleclienteoperacionsospechosa(self):
"""
Obtiene detalle de clientes en operaciones sospechosas del mes anterior.

Returns:
tuple: (DataFrame con detalle, DataFrame vacío de error)
"""
detalleclienteoperacionsospechosa_df, empty_df = Utils.get_dataframe_max_mes_anterior(
self,
self.HE_DETALLECLIENTEOPERACIONSOSPECHOSA_CONS,
self.prm_spark_schema_ddv,
self.prm_spark_catalog,
self.FECDIA_CONS,
self.prm_spark_fecha_rutina
)

return detalleclienteoperacionsospechosa_df, empty_df

def __get_me_informacionaplicativooperacionsospechosa(self):
"""
Obtiene la maestra parametrica de RO.

Returns:
tuple: (DataFrame con parametrica, DataFrame vacío de error)
"""
informacionaplicativooperacionsospechosa_df, df_empty = Utils.get_dataframe_maestra(self,
self.ME_INFORMACIONAPLICATIVOOPERACIONSOSPECHOSA_CONS,
self.prm_spark_schema_ddv,
self.prm_spark_catalog)

"""
informacionaplicativooperacionsospechosa_01_df = informacionaplicativooperacionsospechosa_df.select(
*[col(c) for c in informacionaplicativooperacionsospechosa_df.columns if c != "codappros"],
when(trim(col("codappros")) == 'ALS', lit('ALS'))
.when(trim(col("codappros")) == 'APOB', lit('APOB'))
.when(trim(col("codappros")) == 'ATM', lit('ATM'))
.when(trim(col("codappros")) == 'BAC', lit('BAC'))
.when(trim(col("codappros")) == 'BKTD', lit('BKTD'))
.when(trim(col("codappros")) == 'CCFM', lit('CCFM'))
.when(trim(col("codappros")) == 'COB', lit('COB'))
.when(trim(col("codappros")) == 'COBT', lit('COBT'))
.when(trim(col("codappros")) == 'CSCO', lit('CSCO'))
.when(trim(col("codappros")) == 'CSLT', lit('CSLT'))
.when(trim(col("codappros")) == 'CYL', lit('CYLM'))
.when(trim(col("codappros")) == 'FED', lit('FED'))
.when(trim(col("codappros")) == 'FUE', lit('FUE'))
.when(trim(col("codappros")) == 'GT', lit('GGTT'))
.when(trim(col("codappros")) == 'IMP', lit('IMP'))
.when(trim(col("codappros")) == 'IO', lit('IO'))
.when(trim(col("codappros")) == 'MTN', lit('MTN'))
.when(trim(col("codappros")) == 'NHB', lit('NHBK'))
.when(trim(col("codappros")) == 'NHBK', lit('NHBK'))
.when(trim(col("codappros")) == 'NTLC', lit('NTLC'))
.when(trim(col("codappros")) == 'PIA', lit('PIA'))
.when(trim(col("codappros")) == 'POR', lit('POR'))
.when(trim(col("codappros")) == 'POSS', lit('POSS'))
.when(trim(col("codappros")) == 'REMI', lit('REMI'))
.when(trim(col("codappros")) == 'SAV', lit('SAV'))
.when(trim(col("codappros")) == 'SIO', lit('BSIO'))
.when(trim(col("codappros")) == 'TCAC', lit('TCAC'))
.when(trim(col("codappros")) == 'TTI', lit('TTIB'))
.when(trim(col("codappros")) == 'VP', lit('VPLU'))
.when(trim(col("codappros")) == 'WLEA', lit('WLEA'))
.when(trim(col("codappros")) == 'WRDA', lit('WRDA'))
.when(trim(col("codappros")) == 'YAP', lit('YAP'))
.otherwise(trim(col("codappros")))
)
"""

return informacionaplicativooperacionsospechosa_df, df_empty

def __filter_informacionaplicativooperacionsospechosa(self, informacionaplicativooperacionsospechosa_df):
"""
Obtiene la maestra parametrica de RO filtrada.

Args:
informacionaplicativooperacionsospechosa_df (DataFrame): Información total parametrica

Returns:
Dataframe: DataFrame con parametrica
"""
informacionaplicativooperacionsospechosa_fil_df = (informacionaplicativooperacionsospechosa_df.
filter(col("flgregeliminado") == self.N_CONS))

return informacionaplicativooperacionsospechosa_fil_df

def __get_m_agencia(self):
"""
Obtiene catálogo maestro de agencias activas.

Returns:
tuple: (DataFrame con agencias activas, DataFrame vacío de error)
"""
agencia_df, empty_df = Utils.get_dataframe_maestra(
self,
self.M_AGENCIA_CONS,
self.prm_spark_schema_udv,
self.prm_spark_catalog
)

return agencia_df, empty_df

def __filter_agencia(self, agencia_df):
"""
Obtiene catálogo maestro de agencias activas filtradas.

Args:
agencia_df (DataFrame): Información total de agencias

Returns:
tuple: DataFrame con agencias activas
"""
agencia_fil_df = agencia_df.filter(col("flgregeliminadofuente") == self.N_CONS)

return agencia_fil_df

def __get_m_despais(self):
"""
Obtiene catálogo maestro de países activos.

Returns:
tuple: (DataFrame con países activos, DataFrame vacío de error)
"""
despais_df, empty_df = Utils.get_dataframe_maestra(
self,
self.M_DESPAIS_CONS,
self.prm_spark_schema_udv,
self.prm_spark_catalog
)

return despais_df, empty_df

def __filter_despais(self, despais_df):
"""
Obtiene catálogo maestro de países activos filtrados.

Args:
despais_df (DataFrame): Información total de paises

Returns:
Dataframe: (DataFrame con países activos)
"""
despais_fil_df = despais_df.filter(col("flgregeliminadofuente") == self.N_CONS)

return despais_fil_df

def __get_m_departamento(self):
"""
Obtiene catálogo maestro de departamentos activos.

Returns:
tuple: (DataFrame con departamentos activos, DataFrame vacío de error)
"""
departamento_df, empty_df = Utils.get_dataframe_maestra(
self,
self.M_DEPARTAMENTO_CONS,
self.prm_spark_schema_udv,
self.prm_spark_catalog
)

return departamento_df, empty_df

def __filter_departamento(self, departamento_df):
"""
Obtiene catálogo maestro de departamentos activos filtrados.

Args:
departamento_df (DataFrame): Información total de departamentos

Returns:
Dataframe: (DataFrame con departamentos activos)
"""
departamento_fil_df = departamento_df.filter(col("flgregeliminadofuente") == self.N_CONS)

return departamento_fil_df

def __get_m_provincia(self):
"""
Obtiene catálogo maestro de provincias activas.

Returns:
tuple: (DataFrame con provincias activas, DataFrame vacío de error)
"""
provincia_df, empty_df = Utils.get_dataframe_maestra(
self,
self.M_PROVINCIA_CONS,
self.prm_spark_schema_udv,
self.prm_spark_catalog
)

return provincia_df, empty_df

def __filter_provincia(self, provincia_df):
"""
Obtiene catálogo maestro de provincias activas filtrados.

Args:
provincia_df (DataFrame): Información total de provincias

Returns:
Dataframe: (DataFrame con provincias activos)
"""
provincia_fil_df = provincia_df.filter(col("flgregeliminadofuente") == self.N_CONS)

return provincia_fil_df

def __get_m_distrito(self):
"""
Obtiene catálogo maestro de distritos activos.

Returns:
tuple: (DataFrame con distritos activos, DataFrame vacío de error)
"""
distrito_df, empty_df = Utils.get_dataframe_maestra(
self,
self.M_DISTRITO_CONS,
self.prm_spark_schema_udv,
self.prm_spark_catalog
)

return distrito_df, empty_df

def __filter_distrito(self, distrito_df):
"""
Obtiene catálogo maestro de distritos activos filtrados.

Args:
distrito_df (DataFrame): Información total de distritos

Returns:
Dataframe: (DataFrame con distritos activos)
"""
distrito_fil_df = distrito_df.filter(col("flgregeliminadofuente") == self.N_CONS)

return distrito_fil_df

def __get_m_ubigeo(self):
"""
Obtiene catálogo maestro de ubigeos activos.

Returns:
tuple: (DataFrame con ubigeos activos, DataFrame vacío de error)
"""
ubigeo_df, empty_df = Utils.get_dataframe_maestra(
self,
self.M_UBIGEO_CONS,
self.prm_spark_schema_udv,
self.prm_spark_catalog
)

return ubigeo_df, empty_df

def __filter_ubigeo(self, ubigeo_df):
"""
Obtiene catálogo maestro de ubigeos activos filtrados.

Args:
ubigeo_df (DataFrame): Información total de ubigeos

Returns:
Dataframe: (DataFrame con ubigeos activos)
"""
ubigeo_fil_df = ubigeo_df.filter(col("flgregeliminadofuente") == self.N_CONS)

return ubigeo_fil_df

def __get_m_descanal(self):
"""
Obtiene catálogo maestro de canales activos.

Returns:
tuple: (DataFrame con canales activos, DataFrame vacío de error)
"""
descanal_df, empty_df = Utils.get_dataframe_maestra(
self,
self.M_DESCANAL_CONS,
self.prm_spark_schema_udv,
self.prm_spark_catalog
)

return descanal_df, empty_df

def __filter_descanal(self, descanal_df):
"""
Obtiene catálogo maestro de canales activos filtrados.

Args:
descanal_df (DataFrame): Información total de canales

Returns:
Dataframe: (DataFrame con canales activos)
"""
descanal_fil_df = descanal_df.filter(col("flgregeliminadofuente") == self.N_CONS)

return descanal_fil_df

def __get_m_desmoneda(self):
"""
Obtiene catálogo maestro de monedas activas.

Returns:
tuple: (DataFrame con monedas activas, DataFrame vacío de error)
"""
desmoneda_df, empty_df = Utils.get_dataframe_maestra(
self,
self.M_DESMONEDA_CONS,
self.prm_spark_schema_udv,
self.prm_spark_catalog
)

return desmoneda_df, empty_df

def __filter_desmoneda(self, desmoneda_df):
"""
Obtiene catálogo maestro de monedas activos filtrados.

Args:
desmoneda_df (DataFrame): Información total de monedas

Returns:
Dataframe: (DataFrame con monedas activos)
"""
desmoneda_fil_df = desmoneda_df.filter(col("flgregeliminadofuente") == self.N_CONS)

return desmoneda_fil_df

def __get_m_bancoexterior(self):
"""
Obtiene catálogo maestro de bancos exteriores activos.

Returns:
tuple: (DataFrame con bancos exteriores activos, DataFrame vacío de error)
"""
bancoexterior_df, empty_df = Utils.get_dataframe_maestra(
self,
self.M_BANCOEXTERIOR_CONS,
self.prm_spark_schema_udv,
self.prm_spark_catalog
)

return bancoexterior_df, empty_df

def __filter_bancoexterior(self, bancoexterior_df):
"""
Obtiene catálogo maestro de bancos exteriores activos filtrados.

Args:
bancoexterior_df (DataFrame): Información total de bancos exteriores

Returns:
Dataframe: (DataFrame con bancos exteriores activos)
"""
bancoexterior_fil_df = bancoexterior_df.filter(col("flgregeliminadofuente") == self.N_CONS)

return bancoexterior_fil_df

def __get_m_destipoprofesion(self):
"""
Obtiene catálogo maestro de tipos de profesión activos.

Returns:
tuple: (DataFrame con tipos de profesión activos, DataFrame vacío de error)
"""
destipoprofesion_df, empty_df = Utils.get_dataframe_maestra(
self,
self.M_DESTIPOPROFESION_CONS,
self.prm_spark_schema_udv,
self.prm_spark_catalog
)

return destipoprofesion_df, empty_df

def __filter_destipoprofesion(self, destipoprofesion_df):
"""
Obtiene catálogo maestro de profesiones activos filtrados.

Args:
destipoprofesion_df (DataFrame): Información total de profesiones

Returns:
Dataframe: (DataFrame con profesiones activos)
"""
bancoexterior_fil_df = destipoprofesion_df.filter(col("flgregeliminadofuente") == self.N_CONS)

return bancoexterior_fil_df

def __get_m_destipoclasificacioninternaparty(self):
"""
Obtiene catálogo maestro de clasificaciones internas de party activas.

Returns:
tuple: (DataFrame con clasificaciones activas, DataFrame vacío de error)
"""
destipoclasificacioninternaparty_df, empty_df = Utils.get_dataframe_maestra(
self,
self.M_DESTIPOCLASIFICACIONINTERNAPARTY_CONS,
self.prm_spark_schema_udv,
self.prm_spark_catalog
)

return destipoclasificacioninternaparty_df, empty_df

def __filter_destipoclasificacioninternaparty(self, destipoclasificacioninternaparty_df):
"""
Obtiene catálogo maestro de clasificaciones internas activos filtrados.

Args:
destipoclasificacioninternaparty_df (DataFrame): Información total de clasificaciones internas

Returns:
Dataframe: (DataFrame con clasificaciones internas activos)
"""
destipoclasificacioninternaparty_fil_df = destipoclasificacioninternaparty_df.filter(col("flgregeliminadofuente") == self.N_CONS)

return destipoclasificacioninternaparty_fil_df

def __get_me_relaciondocumentobcpoperacionsospechosa(self):
"""
Obtiene relación de documentos BCP para operaciones sospechosas activos.

Returns:
tuple: (DataFrame con relaciones activas, DataFrame vacío de error)
"""
relaciondocumentobcpoperacionsospechosa_df, empty_df = Utils.get_dataframe_maestra(
self,
self.ME_RELACIONDOCUMENTOBCPOPERACIONSOSPECHOSA_CONS,
self.prm_spark_schema_ddv,
self.prm_spark_catalog
)

return relaciondocumentobcpoperacionsospechosa_df, empty_df

def __filter_relaciondocumentobcpoperacionsospechosa(self, relaciondocumentobcpoperacionsospechosa_df):
"""
Obtiene catálogo maestro de rel documentos ro activos filtrados.

Args:
relaciondocumentobcpoperacionsospechosa_df (DataFrame): Información total de rel documentos ro

Returns:
Dataframe: (DataFrame con rel documentos ro activos)
"""
relaciondocumentobcpoperacionsospechosa_fil_df = relaciondocumentobcpoperacionsospechosa_df.filter(col("flgregeliminado") == self.N_CONS)

return relaciondocumentobcpoperacionsospechosa_fil_df

def __get_mm_destipcuentaoperacionsospechosa(self):
"""
Obtiene catálogo de tipos de cuenta para operaciones sospechosas activos.

Returns:
tuple: (DataFrame con tipos de cuenta activos, DataFrame vacío de error)
"""
destipcuentaoperacionsospechosa_df, empty_df = Utils.get_dataframe_maestra(
self,
self.MM_DESTIPCUENTAOPERACIONSOSPECHOSA_CONS,
self.prm_spark_schema_ddv,
self.prm_spark_catalog
)

return destipcuentaoperacionsospechosa_df, empty_df

def __filter_destipcuentaoperacionsospechosa(self, destipcuentaoperacionsospechosa_df):
"""
Obtiene catálogo maestro de cuentas ro filtrados.

Args:
destipcuentaoperacionsospechosa_df (DataFrame): Información total de cuentas ro

Returns:
Dataframe: (DataFrame con cuentas ro activos)
"""
destipcuentaoperacionsospechosa_fil_df = destipcuentaoperacionsospechosa_df.filter(col("flgregeliminado") == self.N_CONS)

return destipcuentaoperacionsospechosa_fil_df

def __get_me_cuentafideicomisooperacionsospechosa(self):
"""
Obtiene catálogo de cuentas fideicomiso para operaciones sospechosas activas.

Returns:
tuple: (DataFrame con cuentas fideicomiso activas, DataFrame vacío de error)
"""
cuentafideicomisooperacionsospechosa_df, empty_df = Utils.get_dataframe_maestra(
self,
self.ME_CUENTAFIDEICOMISOOPERACIONSOSPECHOSA_CONS,
self.prm_spark_schema_ddv,
self.prm_spark_catalog
)

return cuentafideicomisooperacionsospechosa_df, empty_df

def __filter_cuentafideicomisooperacionsospechosa(self, cuentafideicomisooperacionsospechosa_df):
"""
Obtiene catálogo maestro de cuentas fideicomiso ro filtrados.

Args:
cuentafideicomisooperacionsospechosa_df (DataFrame): Información total de cuentas fideicomiso ro

Returns:
Dataframe: (DataFrame con cuentas fideicomiso ro activos)
"""
cuentafideicomisooperacionsospechosa_fil_df = cuentafideicomisooperacionsospechosa_df.filter(col("flgregeliminado") == self.N_CONS)

return cuentafideicomisooperacionsospechosa_fil_df

def __get_me_tipooperacionsospechosa(self):
"""
Obtiene catálogo de tipos de operación sospechosa activos.

Returns:
tuple: (DataFrame con tipos de operación activos, DataFrame vacío de error)
"""
tipooperacionsospechosa_df, empty_df = Utils.get_dataframe_maestra(
self,
self.ME_TIPOPERACIONSOSPECHOSA_CONS,
self.prm_spark_schema_ddv,
self.prm_spark_catalog
)

return tipooperacionsospechosa_df, empty_df

def __filter_tipooperacionsospechosa(self, tipooperacionsospechosa_df):
"""
Obtiene catálogo maestro de tipo operaciones ro filtrados.

Args:
tipooperacionsospechosa_df (DataFrame): Información total de tipo operaciones ro

Returns:
Dataframe: (DataFrame con tipo operaciones ro activos)
"""
tipooperacionsospechosa_fil_df = tipooperacionsospechosa_df.filter(col("flgregeliminado") == self.N_CONS)

return tipooperacionsospechosa_fil_df