借助Cloud Function将数据加载至Google Cloud Storage与BigQuery时遇到缩进错误求助

首先，你遇到的直接错误是缩进错误：在upload_file函数中，bucket_name = "BUCKET_NAME"这一行的缩进不符合Python语法规范，它应该和函数内的其他代码保持一致的缩进层级（通常是4个空格）。不过除了这个问题，你的代码还有一些适配Cloud Function环境的问题需要调整，以下是完整的解决方案：

一、关键问题修复与适配调整 #

1. 缩进错误修复 #

把upload_file函数里的代码调整到正确的缩进层级，同时优化函数参数让它更灵活：

def upload_to_gcs(bucket_name, file_path):
    """上传文件到GCS存储桶"""
    storage_client = storage.Client()
    bucket = storage_client.get_bucket(bucket_name)
    blob_name = os.path.basename(file_path)
    blob = bucket.blob(blob_name)
    blob.upload_from_filename(file_path)

2. 适配Cloud Function的文件存储 #

Cloud Function的运行环境只有/tmp目录有写权限，所以生成CSV时要把路径指定到/tmp下，不能用当前目录：

download_dir = f'/tmp/pagespeed-results-{date.today()}.csv'

3. 添加Cloud Function入口函数 #

Cloud Function需要一个明确的入口函数（比如main），部署时要指定这个函数作为触发点。整个流程（生成CSV→上传GCS→加载BigQuery）要放在这个入口函数里执行。

4. 修复BigQuery加载逻辑 #

你生成的是CSV文件，但原代码里指定了NEWLINE_DELIMITED_JSON格式，要改成CSV；同时需要设置跳过表头、自动检测表结构，用job.result()替代未定义的wait_for_job函数来等待任务完成。

5. 安全优化：使用环境变量替代硬编码 #

不要把API_KEY、BUCKET_NAME这些敏感信息硬编码在代码里，Cloud Function可以通过环境变量配置这些值，代码里用os.getenv()获取：

API_KEY = os.getenv('PAGESPEED_API_KEY')
BUCKET_NAME = os.getenv('GCS_BUCKET_NAME')

6. 其他小问题修复 #

• 导入缺失的uuid库
• 使用with语句处理文件操作，避免手动调用close()导致资源泄漏
• 初始化变量默认值，避免KeyError后出现NameError
• 修复MAXFID的变量赋值错误（原代码里重复用了Cumulative Layout Shift的描述）

二、完整优化后的代码 #

import requests
import os
import datetime
import uuid
from google.cloud import storage
from google.cloud import bigquery

def generate_pagespeed_csv():
    # 从环境变量获取API密钥
    API_KEY = os.getenv('PAGESPEED_API_KEY')
    if not API_KEY:
        raise ValueError("请设置PAGESPEED_API_KEY环境变量")
    
    # 读取URL列表（部署时pagespeed.txt需和main.py放在同一目录）
    with open('pagespeed.txt') as pagespeedurls:
        date = datetime.datetime.now()
        # 使用/tmp目录存储CSV，Cloud Function只有这个目录可写
        download_dir = f'/tmp/pagespeed-results-{date.today()}.csv'
        
        with open(download_dir, 'w') as file:
            columnTitleRow = "URL, Cumulative Layout Shift Result, Largest Contentful Paint Result, First Input Delay Result, First Contentful Paint, First Interactive, Largest Contentful Paint, Cumulative Layout Shift, Max Potential FID\n"
            file.write(columnTitleRow)
            
            content = pagespeedurls.readlines()
            content = [line.rstrip('\n') for line in content]
            
            for line in content:
                pagespeed = f'https://www.googleapis.com/pagespeedonline/v5/runPagespeed?url={line}&key={API_KEY}&strategy=mobile'
                print(f'Requesting {pagespeed}...')
                request = requests.get(pagespeed)
                final = request.json()
                
                # 初始化默认值，避免KeyError后出现NameError
                urlid = line
                cruxcls = 'N/A'
                cruxlcp = 'N/A'
                cruxfid = 'N/A'
                urlfcp = 'N/A'
                urlfi = 'N/A'
                urllcp = 'N/A'
                urlcls = 'N/A'
                urlmaxfid = 'N/A'
                
                try:
                    urlid = final['id'].split('?')[0]
                    cruxcls = final['loadingExperience']['metrics']['CUMULATIVE_LAYOUT_SHIFT_SCORE']['category']
                    cruxlcp = final['loadingExperience']['metrics']['LARGEST_CONTENTFUL_PAINT_MS']['category']
                    cruxfid = final['loadingExperience']['metrics']['FIRST_INPUT_DELAY_MS']['category']
                    urlfcp = final['lighthouseResult']['audits']['first-contentful-paint']['displayValue']
                    urlfi = final['lighthouseResult']['audits']['interactive']['displayValue']
                    urllcp = final['lighthouseResult']['audits']['largest-contentful-paint']['displayValue']
                    urlcls = final['lighthouseResult']['audits']['cumulative-layout-shift']['displayValue']
                    urlmaxfid = final['lighthouseResult']['audits']['max-potential-fid']['numericValue']
                except KeyError as e:
                    print(f'<KeyError> 未找到键 {e}，URL: {line}')
                
                # 写入行数据
                row = f'{urlid},{cruxcls},{cruxlcp},{cruxfid},{urlfcp},{urlfi},{urllcp},{urlcls},{urlmaxfid}\n'
                file.write(row)
                print(f'处理完成: {urlid}')
    
    return download_dir

def upload_to_gcs(bucket_name, file_path):
    """上传文件到GCS存储桶"""
    storage_client = storage.Client()
    bucket = storage_client.get_bucket(bucket_name)
    # 提取文件名作为GCS中的对象名
    blob_name = os.path.basename(file_path)
    blob = bucket.blob(blob_name)
    blob.upload_from_filename(file_path)
    return f'gs://{bucket_name}/{blob_name}'

def load_to_bigquery(gcs_file_path, dataset_id, table_id):
    """从GCS加载CSV到BigQuery"""
    bigquery_client = bigquery.Client()
    dataset_ref = bigquery_client.dataset(dataset_id)
    table_ref = dataset_ref.table(table_id)
    
    job_config = bigquery.LoadJobConfig(
        source_format=bigquery.SourceFormat.CSV,
        skip_leading_rows=1,  # 跳过表头行
        autodetect=True,  # 自动检测表结构
        write_disposition=bigquery.WriteDisposition.WRITE_APPEND  # 追加数据到表中
    )
    
    load_job = bigquery_client.load_table_from_uri(
        gcs_file_path, table_ref, job_config=job_config
    )
    
    print(f'开始加载BigQuery任务: {load_job.job_id}')
    # 等待任务完成
    load_job.result()
    print(f'BigQuery加载完成，已加载 {load_job.output_rows} 行数据')

# Cloud Function入口函数，这里设置为HTTP触发，也可改成Pub/Sub等其他触发方式
def main(request):
    try:
        # 1. 生成Pagespeed CSV
        csv_path = generate_pagespeed_csv()
        print(f'CSV生成完成: {csv_path}')
        
        # 2. 上传到GCS
        BUCKET_NAME = os.getenv('GCS_BUCKET_NAME')
        if not BUCKET_NAME:
            raise ValueError("请设置GCS_BUCKET_NAME环境变量")
        gcs_file_url = upload_to_gcs(BUCKET_NAME, csv_path)
        print(f'已上传到GCS: {gcs_file_url}')
        
        # 3. 加载到BigQuery
        BQ_DATASET = os.getenv('BQ_DATASET')
        BQ_TABLE = os.getenv('BQ_TABLE')
        if not BQ_DATASET or not BQ_TABLE:
            raise ValueError("请设置BQ_DATASET和BQ_TABLE环境变量")
        load_to_bigquery(gcs_file_url, BQ_DATASET, BQ_TABLE)
        
        return "任务执行完成!", 200
    except Exception as e:
        print(f'执行出错: {str(e)}')
        return f'错误: {str(e)}', 500

三、部署准备事项 #

1. 创建requirements.txt文件
   在和main.py同目录下创建该文件，声明依赖包：

requests==2.31.0
google-cloud-storage==2.15.0
google-cloud-bigquery==3.12.0

2. 配置环境变量
   部署Cloud Function时，在配置页面添加以下环境变量：

• PAGESPEED_API_KEY: 你的Google Pagespeed API密钥
• GCS_BUCKET_NAME: 目标GCS存储桶名称
• BQ_DATASET: BigQuery数据集名称
• BQ_TABLE: BigQuery表名称

3. 上传pagespeed.txt文件
   把包含待检测URL的pagespeed.txt文件和main.py、requirements.txt放在同一目录，部署时一起上传到Cloud Function。

4. 设置触发方式
   部署时选择触发方式（比如HTTP触发），并指定入口函数为main。

5. 权限配置
   确保Cloud Function的服务账号拥有以下权限：

• GCS存储桶的写入权限（Storage Object Creator）
• BigQuery数据集的写入权限（BigQuery Data Editor）

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