CloudFunction.py

import asyncio
import os
import zipfile
import pandas as pd
import aiohttp
from aiohttp import ClientConnectorError, TCPConnector
from google.cloud import storage
from google.oauth2 import service_account
from datetime import datetime

async def obtener_url_descarga(request):
    try:
        package_id = request.args.get('package_id', default="5e8bb1f8-f0a5-4719-a877-38543545505b", type=str)
        url_api = f"https://datos.gob.cl/api/action/package_show?id={package_id}"
        response = requests.get(url_api)
        response.raise_for_status()
        data = response.json()
        if data["success"]:
            return data["result"]["resources"][0]["url"]
        else:
            raise ValueError(f"Error en la solicitud API: {data}")
    except requests.exceptions.RequestException as e:
        return str(e)

async def descargar_y_descomprimir(request):
    try:
        url_descarga = await obtener_url_descarga(request)
        destino = "/tmp/datos_historicos.zip"
        response = requests.get(url_descarga)
        response.raise_for_status()
        
        with open(destino, 'wb') as archivo:
            archivo.write(response.content)

        fecha_str = url_descarga.split('/')[-1].split('.')[0][-8:]

        with zipfile.ZipFile(destino, 'r') as zip_ref:
            csv_folder = f"/tmp/Archivos_CSV/{fecha_str}"
            diarios_folder = os.path.join(csv_folder, 'Diarios')
            os.makedirs(diarios_folder, exist_ok=True)
            originales_folder = f"/tmp/Historicos_originales/{fecha_str}"
            os.makedirs(originales_folder, exist_ok=True)

            for archivo in zip_ref.namelist():
                ruta_archivo = os.path.join(os.path.dirname(destino), archivo)

                if archivo.endswith('.txt'):
                    destino_txt = os.path.join(originales_folder, f'{os.path.splitext(archivo)[0]}_{fecha_str}.txt')
                    with open(destino_txt, 'wb') as f:
                        f.write(zip_ref.read(archivo))
                elif archivo.endswith('.csv'):
                    destino_csv = os.path.join(csv_folder, f'{os.path.splitext(archivo)[0]}_{fecha_str}.csv')
                    with open(destino_csv, 'wb') as f:
                        f.write(zip_ref.read(archivo))

            convertir_a_csv(originales_folder, csv_folder, fecha_str)
            convertir_a_csv(diarios_folder, csv_folder, fecha_str)

        return f'Descarga y descompresión exitosas para fecha: {fecha_str}'

    except requests.exceptions.RequestException as e:
        return f'Error en la descarga y descompresión: {e}'
    except zipfile.BadZipFile:
        return 'Error: El archivo descargado no es un archivo ZIP válido.'

def convertir_a_csv(directorio_originales, directorio_csv, fecha_str):
    archivos_txt = [archivo for archivo in os.listdir(directorio_originales) if archivo.endswith('.txt')]
    
    for archivo in archivos_txt:
        ruta_archivo_txt = os.path.join(directorio_originales, archivo)
        dataframe = pd.read_csv(ruta_archivo_txt, delimiter='\t')
        ruta_csv = os.path.join(directorio_csv, f'{os.path.splitext(archivo)[0]}.csv')
        dataframe.to_csv(ruta_csv, index=False)

async def obtener_codigos_servicio(request):
    try:
        url = "https://www.red.cl/restservice_v2/rest/getservicios/all"
        async with aiohttp.ClientSession() as session:
            async with session.get(url) as response:
                response.raise_for_status()
                return await response.json()
    except aiohttp.ClientError as e:
        return None

async def descargar_json_y_convertir_a_csv(request):
    try:
        codigos_servicio = await obtener_codigos_servicio(request)

        if codigos_servicio:
            códigos_fallidos = set(codigos_servicio)

            while códigos_fallidos:
                codigo = códigos_fallidos.pop()
                url = f"https://www.red.cl/restservice_v2/rest/conocerecorrido?codsint={codigo}"

                connector = TCPConnector(limit_per_host=10)
                async with aiohttp.ClientSession(connector=connector) as session:
                    async with session.get(url) as response:
                        response.raise_for_status()
                        dataframe = pd.json_normalize(await response.json())
                        diarios_folder = f"/tmp/Archivos_CSV/Diarios"
                        os.makedirs(diarios_folder, exist_ok=True)
                        ruta_csv = os.path.join(diarios_folder, f'datos_diarios_{codigo}.csv')
                        dataframe.to_csv(ruta_csv, index=False)

        return 'Proceso de descarga y conversión de datos diarios completado'

    except (aiohttp.ClientError, ClientConnectorError) as e:
        return f'Error en la descarga de datos diarios: {e}'

def subir_datos_a_bucket(local_folder, bucket):
    archivos_nuevos = 0
    archivos_actualizados = 0

    def upload_file(local_file, blob_name):
        nonlocal archivos_nuevos, archivos_actualizados
        blob = bucket.blob(blob_name)

        if blob.exists():
            archivos_actualizados += 1
            accion = "actualizado"
        else:
            archivos_nuevos += 1
            accion = "agregado"

        blob.upload_from_filename(local_file)
        print(f"Archivo {local_file} {accion} en {blob_name}.")

    for root, dirs, files in os.walk(local_folder):
        for filename in files:
            local_file_path = os.path.join(root, filename)
            upload_file(local_file_path, filename)

    print(f"Total de archivos nuevos agregados: {archivos_nuevos}")
    print(f"Total de archivos actualizados: {archivos_actualizados}")

def cloud_function_handler(request):
    try:
        loop = asyncio.get_event_loop()
        result = loop.run_until_complete(descargar_y_descomprimir(request))
        return result

    except Exception as e:
        return f'Error en la ejecución principal: {e}'