codigod_prefec

# https://cloud.prefect.io/marciogsantana-yahoo-cm-br-s-account/welcome/resources

#1 -pip install prefect
#2- prefect backend cloud 
#3- token = BqoyEZNV9o1IWaW5_esEMQ  (token gerado na pagina)
#4 -prefect auth login -t BqoyEZNV9o1IWaW5_esEMQ  (token user)
#5- prefect auth create-token -n my-runner-token -s RUNNER (Create a Runner Token) gerou este token YGMMBXn8NJr5QZF8EHCCjQ
#6- set PREFECT__CLOUD__AGENT__AUTH_TOKEN=YGMMBXn8NJr5QZF8EHCCjQ   (set para windows e export para linux)

# sequencia para autenticar e conectar no prefect cloud

# 1 prefect backend cloud 
# 2 prefect auth login -t(token pagina)
# 3 prefect auth create-token -n my-runner-token -s RUNNER
# 4 set PREFECT__CLOUD__AGENT__AUTH_TOKEN = (token gerado comando anterior)



################helo word.py###############################
mport prefect
from prefect import task, Flow

@task
def hello_word():
    logger =  prefect.context.get("logger")
    logger.info("hello word do prefect em clound")

with Flow("Hello Word") as flow:
    hello_word()


flow.register(project_name= "hello_word")
flow.run_agent()

############################pref01.py######################################

from datetime import datetime, timedelta
import prefect
from prefect import task, Flow
from prefect.schedules import IntervalSchedule
import pandas as pd

retry_delay = timedelta(minutes= 1)
schedule = IntervalSchedule(interval=timedelta(minutes = 2))

@task

def get_data():
    df = pd.read_csv("https://raw.githubusercontent.com/A3Data/hermione/master/hermione/file_text/train.csv")
    return df

@task

def calcula_media_idade(df):
    return df.Age.mean()

@task
def exibi_media_calculada(m):
    logger = prefect.context.get("logger")
    logger.info(f"A media de idade calculada  foi {m}")

@task

def exibe_dataset(df):
    logger = prefect.context.get("logger")
    logger.info(df.head(5).to_json())

with Flow("Titanic01", schedule = schedule) as flow:
    df = get_data()
    med = calcula_media_idade(df)
    e = exibi_media_calculada(med)
    ed = exibe_dataset(df)

flow.register(project_name= 'IGTI', idempotency_key= flow.serialized_hash())
flow.run_agent(token="8dHEbM4P3f3j4CBh11s_rg")


############################pref02.py############################################

from datetime import datetime, timedelta
import pendulum
import prefect
from prefect import task, Flow
from prefect.schedules import CronSchedule
import pandas as pd
from io import BytesIO
import zipfile
import requests

schedule = CronSchedule(

    cron = "*/10 * * * *",   # executar a cada 10 minutos
    start_date=pendulum.datetime(2020, 11, 30, 15, 30, tz='America/Sao_Paulo')  #30/11/2020 15:30
)

@task

# fazer download dados
def get_raw_data():
    url = "http://download.inep.gov.br/microdados/Enade_Microdados/microdados_enade_2019.zip"
    filebytes = BytesIO(requests.get(url).content)

    # Extrair o conteudo do zipfile
    myzip = zipfile.ZipFile(filebytes)
    myzip.extractall()
    path = './microdados_enade_2019/2019/3.DADOS/'
    return path

@task

# filtros

def aplica_filtros(path):
    cols = ['CO_GRUPO', 'TP_SEXO', 'NU_IDADE', 'NT_GER', 'NT_FG', 'NT_CE', 'QE_I01', 'QE_I02',
            'QE_I04', 'QE_I05', 'QE_I08']
    enade = pd.read_csv(path + 'microdados_enade_2019.txt', sep=';', decimal=',', usecols=cols)
    enade = enade.loc[(enade.NU_IDADE > 20) & (enade.NU_IDADE < 40) & (enade.NT_GER > 0)]
    return enade

@task

def constroi_idade_centralizada(df):
    idade = df[['NU_IDADE']]
    idade['idadecent'] = idade.NU_IDADE - idade.NU_IDADE.mean()
    return idade[['idadecent']]

@task

def constroi_idade_cent_quad(df):
    idadecent = df.copy()
    idadecent['idade2'] = idadecent.idadecent ** 2
    return idadecent[['idade2']]

@task

def constroi_est_civil(df):
    filtro = df[['QE_I01']]
    filtro['estcivil'] = filtro.QE_I01.replace({

        'A': 'Solteiro',
        'B': 'Casado',
        'C': 'Separado',
        'D': 'Viúvo',
        'E': 'Outro'
    })
    return filtro[['estcivil']]

@task

def constroi_cor(df):
    filtro = df[['QE_I02']]
    filtro['cor'] = filtro.QE_I02.replace({

        'A': 'Branca',
        'B': 'Preta',
        'C': 'Amarela',
        'D': 'Parda',
        'E': 'Indigena',
        'F': "",
        ' ': ""

    })
    return filtro[['cor']]

@task

def constroi_est_escopai(df):
    filtro = df[['QE_I04']]
    filtro['escopai'] = filtro.QE_I04.replace({

        'A': 0,
        'B': 1,
        'C': 2,
        'D': 3,
        'E': 4,
        'F': 5
    })
    return filtro[['escopai']]

@task

def constroi_est_escomae(df):
    filtro = df[['QE_I05']]
    filtro['escomae'] = filtro.QE_I05.replace({

        'A': 0,
        'B': 1,
        'C': 2,
        'D': 3,
        'E': 4,
        'F': 5
    })
    return filtro[['escomae']]

@task

def constroi_renda(df):
    filtro = df[['QE_I08']]
    filtro['renda'] = filtro.QE_I08.replace({

        'A': 0,
        'B': 1,
        'C': 2,
        'D': 3,
        'E': 4,
        'F': 5,
        'G': 6
    })
    return filtro[['renda']]

@task

def join_data(df, idadecent, idadequadrado, estcivil, cor, escopai, escomae, renda):
    final = pd.concat([df, idadecent, idadequadrado, estcivil, cor, escopai, escomae, renda],axis=1)
    final =final [['CO_GRUPO', 'TP_SEXO', 'idadecent', 'idade2', 'estcivil', 'cor', 'escopai',  'escomae', 'renda']]
    logger = prefect.context.get('logger')
    logger.info(final.head().to_json())
    final.to_csv('enade_tratado.csv', index= False)


with Flow('enade', schedule) as flow:
    path = get_raw_data()
    filtro = aplica_filtros(path)
    idadecent = constroi_idade_centralizada(filtro)
    idadequadrado = constroi_idade_cent_quad(filtro)
    estcivil = constroi_est_civil(filtro)
    cor = constroi_cor(filtro)
    escomae = constroi_est_escomae(filtro)
    escopai = constroi_est_escopai(filtro)
    renda = constroi_renda(filtro)

    j = join_data(filtro, idadecent, idadequadrado, estcivil, cor, escopai, escomae, renda)

flow.register(project_name='IGTI', idempotency_key=flow.serialized_hash())
flow.run_agent(token="8dHEbM4P3f3j4CBh11s_rg")

#####################################pre02.py gravando no banco de dados########################################################

from datetime import datetime, timedelta
import pendulum
import prefect
from prefect import task, Flow
from prefect.schedules import CronSchedule
import pandas as pd
from io import BytesIO
import zipfile
import requests
import sqlalchemy
import pyodbc

schedule = CronSchedule(

    cron = "*/10 * * * *",   # executar a cada 10 minutos
    start_date=pendulum.datetime(2020, 11, 30, 15, 30, tz='America/Sao_Paulo')  #30/11/2020 15:30
)

@task

# fazer download dados
def get_raw_data():
    url = "http://download.inep.gov.br/microdados/Enade_Microdados/microdados_enade_2019.zip"
    filebytes = BytesIO(requests.get(url).content)

    # Extrair o conteudo do zipfile
    myzip = zipfile.ZipFile(filebytes)
    myzip.extractall()
    path = './microdados_enade_2019/2019/3.DADOS/'
    return path

@task

# filtros

def aplica_filtros(path):
    cols = ['CO_GRUPO', 'TP_SEXO', 'NU_IDADE', 'NT_GER', 'NT_FG', 'NT_CE', 'QE_I01', 'QE_I02',
            'QE_I04', 'QE_I05', 'QE_I08']
    enade = pd.read_csv(path + 'microdados_enade_2019.txt', sep=';', decimal=',', usecols=cols)
    enade = enade.loc[(enade.NU_IDADE > 20) & (enade.NU_IDADE < 40) & (enade.NT_GER > 0)]
    return enade

@task

def constroi_idade_centralizada(df):
    idade = df[['NU_IDADE']]
    idade['idadecent'] = idade.NU_IDADE - idade.NU_IDADE.mean()
    return idade[['idadecent']]

@task

def constroi_idade_cent_quad(df):
    idadecent = df.copy()
    idadecent['idade2'] = idadecent.idadecent ** 2
    return idadecent[['idade2']]

@task

def constroi_est_civil(df):
    filtro = df[['QE_I01']]
    filtro['estcivil'] = filtro.QE_I01.replace({

        'A': 'Solteiro',
        'B': 'Casado',
        'C': 'Separado',
        'D': 'Viúvo',
        'E': 'Outro'
    })
    return filtro[['estcivil']]

@task

def constroi_cor(df):
    filtro = df[['QE_I02']]
    filtro['cor'] = filtro.QE_I02.replace({

        'A': 'Branca',
        'B': 'Preta',
        'C': 'Amarela',
        'D': 'Parda',
        'E': 'Indigena',
        'F': "",
        ' ': ""

    })
    return filtro[['cor']]

@task

def constroi_est_escopai(df):
    filtro = df[['QE_I04']]
    filtro['escopai'] = filtro.QE_I04.replace({

        'A': 0,
        'B': 1,
        'C': 2,
        'D': 3,
        'E': 4,
        'F': 5
    })
    return filtro[['escopai']]

@task

def constroi_est_escomae(df):
    filtro = df[['QE_I05']]
    filtro['escomae'] = filtro.QE_I05.replace({

        'A': 0,
        'B': 1,
        'C': 2,
        'D': 3,
        'E': 4,
        'F': 5
    })
    return filtro[['escomae']]

@task

def constroi_renda(df):
    filtro = df[['QE_I08']]
    filtro['renda'] = filtro.QE_I08.replace({

        'A': 0,
        'B': 1,
        'C': 2,
        'D': 3,
        'E': 4,
        'F': 5,
        'G': 6
    })
    return filtro[['renda']]

@task

def join_data(df, idadecent, idadequadrado, estcivil, cor, escopai, escomae, renda):
    final = pd.concat([df, idadecent, idadequadrado, estcivil, cor, escopai, escomae, renda],axis=1)
    final =final [['CO_GRUPO', 'TP_SEXO', 'idadecent', 'idade2', 'estcivil', 'cor', 'escopai',  'escomae', 'renda']]
    logger = prefect.context.get('logger')
    logger.info(final.head().to_json())
    #final.to_csv('enade_tratado.csv', index= False)   # esta linha gera o csv tratado
    return final

@task

def escreve_dw(df):
    engine = sqlalchemy.create_engine(
        'mssql+pyodbc://@localhost/enade?Integrated Security=True;driver=ODBC+Driver+17+for+SQL+Server'
    )
    df.to_sql("tratado", con = engine, index = False, if_exists= 'append')


with Flow('enade', schedule) as flow:
    path = get_raw_data()
    filtro = aplica_filtros(path)
    idadecent = constroi_idade_centralizada(filtro)
    idadequadrado = constroi_idade_cent_quad(idadecent)
    estcivil = constroi_est_civil(filtro)
    cor = constroi_cor(filtro)
    escomae = constroi_est_escomae(filtro)
    escopai = constroi_est_escopai(filtro)
    renda = constroi_renda(filtro)

    j = join_data(filtro, idadecent, idadequadrado, estcivil, cor, escopai, escomae, renda)

    w = escreve_dw(j)

flow.register(project_name='IGTI', idempotency_key=flow.serialized_hash())
flow.run_agent(token="jJIyAOceFtw1cXKcakLQhg")