# ---
# jupyter:
#   jupytext:
#     text_representation:
#       extension: .py
#       format_name: light
#       format_version: '1.5'
#       jupytext_version: 1.14.4
#   kernelspec:
#     display_name: Python 3
#     language: python
#     name: python3
# ---

import pandas as pd
import sqlalchemy 
import psycopg2
import os
import numpy as np
import requests
import re
import copy
from pandas.api.types import is_numeric_dtype
from sklearn.linear_model import LinearRegression
from sqlalchemy import create_engine
import datetime
from datetime import datetime as dt
import sys
import importlib

sys.path.append('..')

import modules.transforms

importlib.reload(modules.transforms)

from modules.transforms import *

DBname=os.environ['DB_NAME']
postgres_psswd=os.environ['POSTGRES_PASSWORD']
postgres_user=os.environ['POSTGRES_USER']
postgres_port=str(os.environ['POSTGRES_PORT'])

# A long string that contains the necessary Postgres login information
postgres_str = ('postgresql://'+postgres_user+':'+postgres_psswd+'@'+DBname+':'+postgres_port+'/superset')
# Create the connection
cnx = create_engine(postgres_str)

#guardar paso en csv
path='/data/ETLcache/'
now = dt.now()
timestamp = now.strftime("_%d%m%Y_%H%M%S")

# +
#OBTENER RUTAS DE CARPETAS DISPONIBLES EN REPO COVID DE MINISTERIO DE CIENCIAS DE CHILE
url="https://github.com/MinCiencia/Datos-COVID19/tree/master/output/"
request=requests.get(url)
root=url.split('github.com')[1]
prefix=url.split('/Min')[0]
folders=request.text.split(root)

#product folders
for idx in range(len(folders)):
    f=folders[idx]
    f=f.split('">')[0]
    folders[idx]=f
folders=[prefix+root+f for f in folders if 'producto' in f]

# +
#DEFINIR LISTADO DE PRODUCTOS QUE SE VAN A CONSULTAR DE REPO DE MINISTERIO DE CIENCIAS

productos=['producto5','producto3','producto10','producto13','producto25','producto14','producto19','producto37','producto53','producto54','producto52','producto50','producto48','producto24'
    
]
# -

#LISTADO DE CARPETAS (RUTAS) FILTRADO, SEGÚN PRODUCTOS A CONSULTAR
fs=[]
for p in productos:
    fs.extend([ x for x in folders if p==x.rsplit('/',1)[1]])

#guardar listado de productos consultados en DF (para posteriormente guardar en BD POstgres)
productos=pd.DataFrame(productos)

# +
#crear esquema "tracking" si es que no existe en BD Postgres (para llevar registro de los productos consultados)
k='tracking'
if not cnx.dialect.has_schema(cnx, k):
    print('schema '+k+' does not exist, creating it')
    cnx.execute(sqlalchemy.schema.CreateSchema(k))
else:
    print('schema '+k+' exists, will not be created')

productos.to_sql('min_ciencias_products',con=cnx,schema=k,if_exists='replace')
# -

#LEER ARCHIVOS DENTRO DE RUTAS CORRESPONDIENTES A PRODUCTOS, GUARDARLOS EN DICCIONARIO "TABLES"
tables={}
for f in fs:
    files_folder=f.rsplit('/')[-1]
    files_folder_prefix='/MinCiencia/Datos-COVID19/blob/master/output/'+files_folder+'/'
    tables[files_folder]={}
    raw_path='https://raw.githubusercontent.com/MinCiencia/Datos-COVID19/master/output/'+files_folder+'/'
    req=requests.get(f)
    files=req.text.split(files_folder_prefix)
    for idx in range(len(files)):
        files[idx]=files[idx].split('">')[0]
    files=[raw_path+f for f in files if '.csv' in f]
    for fl in files:
        try:
            print('reading file: '+fl)
            table=pd.read_csv(fl)
            table_name='_'.join(fl.rsplit('/',2)[-2:]).strip('.csv')
            tables[files_folder][table_name]=table
        except Exception as e:
            print(str(e))
            pass

#crear "producto11" como caso especial de producto 4
tables['producto11']={}
tables['producto11']['producto4']=pd.read_csv('https://raw.githubusercontent.com/MinCiencia/Datos-COVID19/master/output/producto11/bulk/producto4.csv')
tables['producto11']['producto4'].columns=[x.replace('*','x').replace('%','porcentaje_') for x in tables['producto11']['producto4'].columns]

ortables=copy.deepcopy(tables)

# +
#normalizar tablas transpuestas 
for s in tables.keys():
    for n in tables[s].keys():
        #print(n)
        if '_T' in n[-2:]:
            print('fixing transposed table: '+n)
            tables[s][n]=transpose(tables[s][n])
#y cambiar campos "fecha" (donde haya) a tipo datetime           
for s in tables.keys():
    print('working schema: '+s)
    for n in tables[s].keys():
        if 'fecha' in [str(x).lower() for x in tables[s][n].columns]:
            print('converting to date in table: '+n)
            tables[s][n]=to_datetime(tables[s][n])
        elif 'Fecha' in [str(x).lower() for x in tables[s][n].columns]:
            print('converting to date in table: '+n)
            tables[s][n]=to_datetime(tables[s][n])

            
            
for s in tables.keys():
    print('working schema: '+s)
    for n in tables[s].keys():
        print(tables[s][n].columns)
        if 'fecha' in [str(x) for x in tables[s][n].columns]:
            print('adding reverse date and index in table: '+n)
            tables[s][n]=reverse_idx(tables[s][n],date_field='fecha')
        elif 'Fecha' in [str(x) for x in tables[s][n].columns]:
            print('adding reverse date and index in table: '+n)
            tables[s][n]=reverse_idx(tables[s][n],date_field='Fecha')
            

# +
#tables['producto5']['producto5_TotalesNacionales_T']['Fecha']=pd.to_datetime(tables['producto5']['producto5_TotalesNacionales_T']['Fecha'])

# +
#calcular índice y fechas invertidos para tablas en que se desea extraer último dato

#to_reverse_idx=[{'producto54':'producto54_r.nacional'},{'producto53':'producto53_confirmados_nacional'},
#                {'producto53':'producto53_confirmados_regionale'}]

#for n in to_reverse_idx:
#    k,i=list(n.items())[0]
#    print('calculating reverse index and date field in table: '+str(k)+' : '+str(i))
#    tables[k][i]=reverse_idx(tables[k][i],date_field='fecha')
    
#to_reverse_idx=[{'producto5':'producto5_TotalesNacionales_T'},
#                {'producto14':'producto14_FallecidosCumulativo_T'},
#                {'producto14':'producto14_FallecidosCumulativo_std'}]

#for n in to_reverse_idx:
#    k,i=list(n.items())[0]
#    print('calculating reverse index and date field in table: '+str(k)+' : '+str(i))
#    tables[k][i]=reverse_idx(tables[k][i],date_field='Fecha')
# -

#agregar variable dummy para región metropolitana
for s in tables.keys():
    for n in tables[s].keys():
        #print(n)
        if 'region' in [x.lower() for x in tables[s][n].columns]:
            print('creando variable dummy para región metropolitana en tabla: '+n)
            tables[s][n]=make_dummies(tables[s][n],fields=['region'],keep_strings=['metropolitana'],value_map={'metropolitana':{0:'Resto de Chile',1:'Región Metropolitana'}})

for k in tables.keys():
    if not cnx.dialect.has_schema(cnx, k):
        print('schema '+k+' does not exist, creating it')
        cnx.execute(sqlalchemy.schema.CreateSchema(k))
    else:
        print('schema '+k+' exists, will not be created')

for schema in tables.keys():
    for name in tables[schema].keys():
        try:
            df=tables[schema][name]
            name=name.replace('-','_')
            print("creating table "+name+' ,schema: '+schema)
            df.to_sql(name, schema=schema,con=cnx,if_exists='replace')
            print("saving table"+path+name+timestamp+'.csv in cache')
            df.to_csv(path+name+timestamp+'.csv',encoding='utf-8')
        except Exception as e: 
            print(str(e))
            pass

os.system('jupyter nbconvert --output /home/jovyan/work/ETLdocs/' + 'ETL_covid-chile.html' + ' --to html ' + 
          '/home/jovyan/work/ETL/covid-chile.ipynb')

StarcoderdataJupyterScriptsDedupFiltered