基于Google Cloud Functions从JSON生成CSV并上传GCS报错排查

错误原因排查 #

1. 数据格式不匹配：你传递给upload_from_string的file_data是Python列表套元组的原生结构，而该方法要求传入字符串格式的CSV文本内容，无法直接将原生Python数据转为字节流，因此触发could not be converted to bytes错误。
2. 键名拼写错误：输入JSON的file2中字段是stutName，但代码中写的是stuName，运行时会抛出KeyError。
3. 缺失JSON读取逻辑：原代码未体现如何读取两个JSON文件的内容，需要补充读取逻辑。
4. CSV生成逻辑错误：原代码的行列组织不符合CSV规范格式，需用标准工具生成合法内容。

修正后的完整代码 #

import os
import json
import csv
from io import StringIO
from google.cloud import storage

def process_json_to_csv(event, context):
    # 初始化GCS客户端
    project_id = os.environ['project_id']
    bucket_name = os.environ['bucket_name']
    storage_client = storage.Client(project_id)
    bucket = storage_client.get_bucket(bucket_name)

    # 读取file1.json（替换为你的JSON文件路径）
    blob_file1 = bucket.blob("path/to/file1.json")
    file1_content = blob_file1.download_as_text()
    REC_KEYS = json.loads(file1_content)

    # 读取file2.json（替换为你的JSON文件路径）
    blob_file2 = bucket.blob("path/to/file2.json")
    file2_content = blob_file2.download_as_text()
    json_data = json.loads(file2_content)

    # 按预期格式组织CSV行数据
    csv_rows = []
    csv_rows.append(["Key", "Value"])  # 表头
    # 从file1提取数据
    csv_rows.append([REC_KEYS['k1']['key1'], REC_KEYS['k1']['key_val']])
    # 从file2提取数据（修正键名）
    csv_rows.append(["stuid", json_data['stu']['stuid']])
    csv_rows.append(["stutName", json_data['stu']['stutName']])

    # 生成标准CSV字符串
    output = StringIO()
    writer = csv.writer(output)
    writer.writerows(csv_rows)
    csv_content = output.getvalue()

    # 上传CSV到GCS
    folder_name = "your-target-folder"
    file_name = "output.csv"
    content_type = "text/csv"
    try:
        upload_file(project_id, bucket_name, folder_name, file_name, csv_content, content_type)
        print("CSV文件上传成功")
    except Exception as e:
        print(f"上传失败: {str(e)}")

def upload_file(project_id, bucket_name, bucket_folder, file_name, document, content_type):
    storage_client = storage.Client(project_id)
    bucket = storage_client.get_bucket(bucket_name)
    # 优化路径拼接，避免斜杠问题
    blob_path = f"{bucket_folder.rstrip('/')}/{file_name.lstrip('/')}"
    blob = bucket.blob(blob_path)
    blob.upload_from_string(data=document, content_type=content_type)

关键说明 #

• 使用csv.writer和StringIO生成标准CSV字符串，确保格式合法且能被upload_from_string正确处理。
• 补充了从GCS读取JSON文件的逻辑，若你的JSON来源是HTTP请求体等其他渠道，可替换对应读取代码。
• 修正了键名拼写错误，确保能正确获取JSON字段值。
• 优化了GCS路径拼接逻辑，避免首尾斜杠导致的路径异常。
• 增加了错误详情打印，便于排查问题。

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