Cloud Functions Excel处理函数开发及结果文件获取求助

修正后的Cloud Functions处理代码 #

你的代码存在几处关键错误，以下是修复后可直接使用的版本：

from google.cloud import storage
import pandas as pd
import datetime
import io

def procesar_excel(event, context):
    # 初始化Storage客户端并获取Bucket实例
    client = storage.Client()
    bucket_name = event['bucket']
    bucket = client.get_bucket(bucket_name)
    blob_name = event['name']

    # 下载目标文件的二进制内容
    blob = bucket.get_blob(blob_name)
    content = blob.download_as_bytes()

    # 读取Excel文件（用BytesIO包装二进制内容）
    df = pd.read_excel(io.BytesIO(content))

    # 定义列的目标顺序
    nuevo_orden = [
        'Fecha_Carga', 'Periodo_Carga', 'Email', 'Id_Nino', 'Id_Persona',
        'Nombre_Nino', 'Apellido_Paterno_Nino', 'Apellido_Materno_Nino',
        'Rut_Nino', 'Nombre_Persona', 'Apellido_Paterno_Persona',
        'Apellido_Materno_Persona', 'Rut_Persona', 'Relacion', 'Profesion',
        'Tipo', 'Fono', 'Celular', 'Id_Centro', 'Nombre_Centro', 'Zona',
        'Comuna_Centro', 'Nivel', 'Jornada', 'Comuna_Domicilio',
        'Comuna_Trabajo', 'Fecha_Ingreso_Vitamina', 'Fecha_Ingreso_Centro',
        'Fecha_Avance_Nivel', 'Fecha_Retiro', 'Estado', 'Canal', 'Empresa',
        'Rut_Empresa', 'Tipo_Cliente', 'Grupo_Empresa', 'Rut_Grupo_Empresa',
        'Id_Sala', 'TAG', 'Id_Nino_Centro', 'Estado_Aux'
    ]

    # 重新排列列顺序
    df = df[nuevo_orden]

    # 处理日期格式
    df["Fecha_Carga"] = pd.to_datetime(df["Fecha_Carga"])
    df["Fecha_Carga"] = df["Fecha_Carga"].dt.strftime("%d-%m-%y")

    # 清理数据：移除Email为空的行
    df.dropna(subset=['Email'], inplace=True)
    
    # 按Email排序并重置索引
    df = df.sort_values(by='Email', ascending=True).reset_index(drop=True)
    
    # 构建Email对应的多Nivel映射字典
    email_map = {}
    for index, row in df.iterrows():
        mail = row['Email']
        nivel = row['Nivel']
        
        if mail not in email_map:
            email_map[mail] = {'Nivel1': nivel}
        elif 'Nivel1' in email_map[mail] and 'Nivel2' not in email_map[mail]:
            email_map[mail]['Nivel2'] = nivel
        else:
            email_map[mail]['Nivel3'] = nivel

    # 转换映射为DataFrame并合并到原数据
    nuevo_df = pd.DataFrame.from_dict(email_map, orient='index').reset_index().rename(columns={'index': 'Email'})
    df = df.drop_duplicates(subset='Email')
    df = pd.merge(df, nuevo_df, on='Email').reset_index(drop=True)
    
    # 生成UTC时间戳作为文件名后缀（避免时区问题）
    timestamp = datetime.datetime.utcnow().strftime("%Y%m%d_%H%M%S")

    # 将处理后的DataFrame写入内存中的Excel文件
    with io.BytesIO() as output:
        writer = pd.ExcelWriter(output, engine='xlsxwriter')
        df.to_excel(writer, index=False)
        writer.close()  # 替换writer.save()，适配新版pandas
        output.seek(0)
        
        # 上传处理后的文件到原Bucket
        nuevo_blob = bucket.blob(f'resultado_{timestamp}.xlsx')
        nuevo_blob.upload_from_file(output, content_type='application/vnd.openxmlformats-officedocument.spreadsheetml.sheet')

    return f'Archivo procesado y guardado como: resultado_{timestamp}.xlsx'

你的疑问解答 #

1. Cloud Storage是否会自动检测上传的文件？ #

Cloud Storage本身不会自动触发函数，需要给Cloud Functions配置Cloud Storage触发器：

• 创建/编辑Cloud Functions时，选择触发器类型为「Cloud Storage」
• 指定你已创建的Bucket，触发事件选择「最终创建（finalize）」
• 配置完成后，每当有新文件上传到该Bucket，系统会自动调用你的procesar_excel函数处理文件

2. 如何获取处理后的结果文件？ #

你的代码已将处理后的文件上传到同一个Bucket，文件名格式为resultado_<时间戳>.xlsx，获取方式：

• 直接登录Cloud Storage控制台，找到对应Bucket即可看到生成的结果文件
• 使用gsutil命令行工具下载，示例：

  gsutil cp gs://tu-nombre-bucket/resultado_20240520_123456.xlsx ./local-destino.xlsx
  

• 通过Cloud Storage API编写代码下载（适用于自动化流程）

内容的提问来源于stack exchange，提问作者Alvaro Rodriguez