from datetime import datetime,timedelta
import pandas as pd
pd.set_option('display.max_rows', 10000)
------------------------------------------------------------------------------------------------------------------------------------------------------
hoy = datetime.now().strftime('%Y%m%d')
ayer = datetime.strftime(datetime.now() - timedelta(1), '%Y%m%d')
preDF = []
rutas = []
## NOTA: Descomentar al menos una de las siguientes rutas
## Rutas de STAGING de los proyectos vigentes de LIVE
## rutas += ['/in/staging/ratransmit/algorithmics/tepr','/in/staging/ratransmit/analitica/xexc','/in/staging/ratransmit/clan/tedt','/in/staging/ratransmit/deltaonsite/xexc','/in/staging/ratransmit/digital_units/kexc','/in/staging/ratransmit/host/rrrg','/in/staging/ratransmit/host/tlsm','/in/staging/ratransmit/intranet_es/ewvp','/in/staging/ratransmit/ldap/peng','/in/staging/ratransmit/marketing_cs/kexc','/in/staging/ratransmit/mcg/kwue','/in/staging/ratransmit/mentor/eezt','/in/staging/ratransmit/mentor_es/eezt','/in/staging/ratransmit/Metakninght/kmkn','/in/staging/ratransmit/monitoring/xexc','/in/staging/ratransmit/pfd/ezwf','/in/staging/ratransmit/piva/eyvr','/in/staging/ratransmit/pricing/xexc','/in/staging/ratransmit/ptcib_es/endf','/in/staging/ratransmit/ptcib_es/endg','/in/staging/ratransmit/ptcib_es/endt','/in/staging/ratransmit/ptcib_es/endv','/in/staging/ratransmit/ptcib_es/enwf','/in/staging/ratransmit/ptcib_es/enyd','/in/staging/ratransmit/quality_rules/xexc','/in/staging/ratransmit/quotepad/kyum','/in/staging/ratransmit/rdr/kytl','/in/staging/ratransmit/smart_data/xexc','/in/staging/ratransmit/swift_es/sxsx']
## Rutas de RAW de los proyectos vigentes de LIVE
rutas += ['/data/raw/eezt/data','/data/raw/endg/data','/data/raw/endv/data','/data/raw/ewvp/data','/data/raw/ezwf/data','/data/raw/kexc/kexc/data','/data/raw/kwue/data','/data/raw/peng/data','/data/raw/rrrg/data','/data/raw/tlsm/data','/data/raw/xcsf/eezt/data','/data/raw/xcsf/endf/data','/data/raw/xcsf/endt/data','/data/raw/xcsf/enwf/data','/data/raw/xcsf/enyd/data','/data/raw/xcsf/eyvr/data','/data/raw/xcsf/kytl/data','/data/raw/xcsf/kyum/data','/data/raw/xcsf/sxsx/data','/data/raw/xcsf/tedt/data','/data/raw/xcsf/tepr/data','/data/raw/xexc/ebdm/data','/data/raw/xexc/xexc/data']
## Rutas de MASTER de los proyectos vigentes de LIVE
rutas += ['/data/master/cib/master_data/xdrd/data','/data/master/cib/reference_data/xrda/data','/data/master/cib/xdip/data','/data/master/commercialdelivery/kcha/data','/data/master/commercialdelivery/kdtm/data','/data/master/corporatefunctions/xpng/datatmp','/data/master/ebdm/data','/data/master/finance/investments_cost/xivg/data','/data/master/finance/xbfp/data','/data/master/finance/xfmd/data','/data/master/finance/xrro/data','/data/master/finance/xyat/data','/data/master/kcdo/data','/data/master/kdtm/kdtm/data','/data/master/people/xsag/data','/data/master/productsservices/agreements_services/xase/data','/data/master/productsservices/settlements/xset/data','/data/master/productsservices/xgug/data','/data/master/productsservices/xmol/data','/data/master/productsservices/xpan/data','/data/master/risk/market_risk/xmqr/data','/data/master/risk/xctk/data','/data/master/structuraldata/corporate_catalogs/xcca/data']
## Rutas de OUT de los proyectos vigentes de LIVE
## rutas += ['/out/staging/ratransmit/escd']
## Bucle que se repite para cada una de las rutas de las que están definidas arriba
for i in rutas:

    ## Conexión a la ruta

    hadoop = sc._jvm.org.apache.hadoop

    fs = hadoop.fs.FileSystem
    conf = hadoop.conf.Configuration()
    path = hadoop.fs.Path(i)
    f = hadoop.fs.FileStatus

    ## El TRY/EXCEPT es para evitar el error cuando el usuario no tiene permisos en la ruta

    try:

        ## Añadir a una lista los subdirectorios (tablas) de todas las rutas definidas arriba

        for f in fs.get(conf).listStatus(path):
            existeCarpetaLog = 'No'

            ficheroNombre = f.getPath().toString().replace('hdfs://gldaaslive.sces1p100.isi','')
            ficheroFecha = datetime.fromtimestamp(int(f.getModificationTime())/1000.0).strftime('%Y%m%d')

            ruta1 = hadoop.fs.Path(ficheroNombre)
            existeSuccess = 0
            fechaActualizacionSuccess = 'N/A'

            ## Buscar el fichero '_SUCCESS' dentro de la tabla

            for g in fs.get(conf).listStatus(ruta1):
                nombreArchivo = g.getPath().toString().replace('hdfs://gldaaslive.sces1p100.isi','').replace(ficheroNombre + '/','')
                fechaArchivo = datetime.fromtimestamp(int(g.getModificationTime())/1000.0).strftime('%Y%m%d')
                if nombreArchivo == '_SUCCESS':
                    existeSuccess = 1
                    fechaActualizacionSuccess = fechaArchivo

            ## Si no existe fichero '_SUCCESS' en principio no buscar nada más, y añadir registro en el DataFrame

            if existeSuccess == 0:
                ## print(ficheroNombre, ': No existe el fichero "_SUCCESS" en este directorio')
                preDF += [[i,ficheroNombre.replace(i + '/',''),'N/A','N/A','Sí','No','N/A','N/A','N/A','N/A']]

            ## Si existe fichero '_SUCCESS', hacer lo siguiente

            else:

                lista = [ficheroNombre]

                numBytesTabla = 0

                ## Iterar en las subcarpetas de las tablas

                while lista != []:

                    primerElemento = lista[0]
                    lista.pop(0)

                    ruta2 = hadoop.fs.Path(primerElemento)

                    numBytes = 0

                    for h in fs.get(conf).listStatus(ruta2):
                        subFicheroNombre = h.getPath().toString().replace('hdfs://gldaaslive.sces1p100.isi','')
                        subFicheroFecha = datetime.fromtimestamp(int(h.getModificationTime())/1000.0).strftime('%d/%m/%Y %H:%M:%S')
                        ##print(subFicheroNombre)
                        ## Si la "subcarpeta" es un directorio, se añade a la lista de carpetas a revisar y se comprueba si contiene la carpeta LOG
                        if h.isDirectory():
                            lista += [subFicheroNombre]
                            if subFicheroNombre == ficheroNombre + '/log':
                                existeCarpetaLog = 'Sí'
                        ##elif subFicheroFecha == hoy and h.isFile():
                        ## Si la "subcarpeta" es un archivo, se suma el tamaño del archivo al resumen de la subcarpeta
                        elif h.isFile():
                            numBytes += h.getLen()
                    ## print(ficheroNombre, ': SE HA MODIFICADO HOY', ': ' + str(numBytes) + 'bytes añadidos al HDFS')
                    ficheroNombreLimpio = ficheroNombre.replace(i + '/','')
                    ## print(ficheroNombreLimpio)

                    ## Se añade información al DataFrame dependiendo de si la subcarpeta es carpeta raíz o no
                    subRuta = primerElemento.replace(ficheroNombre + '/','').replace(primerElemento,'(Carpeta Raíz)')
                    if subRuta == '(Carpeta Raíz)':
                        preDF += [[i,ficheroNombreLimpio,subRuta,subFicheroFecha,'Sí','Sí',fechaActualizacionSuccess,numBytes,existeCarpetaLog,'N/A']]
                    else:
                        preDF += [[i,ficheroNombreLimpio,subRuta,subFicheroFecha,'Sí','Sí',fechaActualizacionSuccess,numBytes,'N/A','N/A']]

    ## Si no se tienen permisos para acceder a la carpeta, se muestra un registro en el DataFrame
    except:
        ## print('ERROR : ' + i + ' : puede que no tengas permisos suficientes')
        preDF += [[i,'(Toda La Ruta)','N/A','N/A','No','N/A','N/A','N/A','N/A','N/A']]
------------------------------------------------------------------------------------------------------------------------------------------------------
df1 = pd.DataFrame(preDF,columns = ['Ruta','Tabla','Sub-Ruta','Fecha Actualización','Se Ha Accedido','Existe _SUCCESS en Tabla','Fecha Actualización Success','Nº Bytes Añadidos','Log','Información Log'])
df1 = df1.loc[(df1['Se Ha Accedido'] == 'Sí') & (df1['Sub-Ruta'] == '(Carpeta Raíz)')][['Ruta','Tabla','Fecha Actualización Success','Log','Información Log']]
df1 = df1.set_index(['Ruta','Tabla'])
df1
------------------------------------------------------------------------------------------------------------------------------------------------------
df2 = pd.DataFrame(preDF,columns = ['Ruta','Tabla','Sub-Ruta','Fecha Actualización','Se Ha Accedido','Existe _SUCCESS en Tabla','Fecha Actualización Success','Nº Bytes Añadidos','Log','Información Log'])
df2['Fecha Actualización'] = df2['Fecha Actualización'].map(lambda x: datetime.strptime(x, '%d/%m/%Y %H:%M:%S') if x != 'N/A' else None)
dateTimeAyerAux = datetime.strftime(datetime.now() - timedelta(1), '%d/%m/%Y') + ' 18:00:00'
dateTimeAyer = datetime.strptime(dateTimeAyerAux,'%d/%m/%Y %H:%M:%S')
df2 = df2.loc[(df2['Se Ha Accedido'] == 'Sí') & (df2['Sub-Ruta'] != '(Carpeta Raíz)') & (df2['Fecha Actualización'] >= dateTimeAyer)][['Ruta','Tabla','Sub-Ruta','Fecha Actualización','Nº Bytes Añadidos']]
df2 = df2.set_index(['Ruta','Tabla','Sub-Ruta'])
df2