Google Cloud Function Python后端CORS错误未解决，求非flask-cors方案

问题背景 #

在Google Cloud Function上部署了供前端调用的Python后端API，该API在Postman和GCP测试页可正常运行，但浏览器调用时触发CORS错误。尝试过flask-cors但无效，需无需flask-cors的解决方案并排查代码问题。

浏览器报错信息 #

search:1 Access to fetch at 'https://asia-south1-transcript-extension-lb.cloudfunctions.net/notes_capture' from origin 'https://www.google.com' has been blocked by CORS policy: Response to preflight request doesn't pass access control check: No 'Access-Control-Allow-Origin' header is present on the requested resource. If an opaque response serves your needs, set the request's mode to 'no-cors' to fetch the resource with CORS disabled.

代码问题排查 #

1. Flask与functions_framework混用冲突：Google Cloud Functions默认通过functions_framework处理请求，但你同时初始化了Flask App并使用@app.route装饰器，导致请求未经过Flask的CORS中间件，手动设置的Headers也无法正确生效。
2. 预检请求处理不彻底：虽然在路由中处理了OPTIONS请求，但由于框架混用，预检请求的响应Headers可能未被正确返回。

解决方案 #

方案1：移除Flask依赖，纯用functions_framework实现 #

直接使用Google Cloud Functions的functions_framework处理请求，手动添加CORS Headers，无需依赖Flask。

修改后的完整代码：

import os
import ast
import re
import requests
import numpy as np
import onnxruntime as rt
import torch
from google.cloud import storage
from transformers import DistilBertTokenizer
import openai
import sys
import functions_framework

model = None
tokenizer = None
dl_dir = "/tmp"
storage_client = storage.Client()
bucket = storage_client.get_bucket("python_punctuation")

def download_model():
    global model, tokenizer
    if model:
        return
    if not os.path.exists(dl_dir):
        os.makedirs(dl_dir)
    i = 0
    for blob in bucket.list_blobs(prefix="Notes_capture/", delimiter="/"):
        destination_uri = os.path.join(dl_dir, blob.name)

        if i==0 and not os.path.exists(destination_uri):  ### Create folder for /tmp/Notes_capture/
            os.makedirs(destination_uri)
            i = i+1
        else:  ### Else download all the model and tokenizers
            blob.download_to_filename(destination_uri)

    model_path = os.path.join(dl_dir,"Notes_capture", 'action_items.onnx')
    model = rt.InferenceSession(model_path, providers=['CUDAExecutionProvider', 'CPUExecutionProvider'])
    tokenizer = DistilBertTokenizer.from_pretrained(os.path.join(dl_dir,"Notes_capture"), do_lower_case=True)
    return model, tokenizer

class predict_actions:
    def __init__(self, model, tokenizer):
        self.model = model
        self.tokenizer = tokenizer

    def predict(self, input_text):
        text = input_text
        padded_sequence = self.tokenizer(text, padding='max_length', max_length=128, truncation=True, return_tensors="pt")
        input_id = padded_sequence['input_ids'].squeeze(1)
        mask = padded_sequence['attention_mask']
        batch_x = {
            'input_id': input_id.cpu().numpy(),
            'mask': mask.cpu().numpy()
        }
        output = self.model.run(None, batch_x)
        return np.argmax(output[0], axis=1)

    def gpt_response(self, sentence):
        api_key_file = open("/tmp/Notes_capture/Auth_key.txt","r")
        api_key = api_key_file.read()
        api_key = api_key.replace("\n","")
        openai.api_key = (api_key)
        example = {"action_items": "", "roadblocks": "", "conclusion": ""}
        completion = openai.ChatCompletion.create(
        model="gpt-3.5-turbo",
        messages=[
            {"role": "system", "content": "You are capturing notes from the given text paragraph, these all the sentences are the main topics from the converstaion."},
            {"role": "assistant", "content": f"""You are capturing notes from the given text paragraph, these all the sentences are the main topics from the converstaion. action_items, roadblocks(if the service related blockers/issues from the client or the user call conversation/ meeting, else give ""(empty string) in the output), conclusion(conclusion of the conversation), extract all this information from this paragraph only in the Python Dictionary format, please do not specify any kind of chind dictionary in the values of the key(only simple string values), and if in case the value is not detected give ""(empty string), if you can't able to detect anything respond only "" (Python Dictionary Response Example: {example}): {sentence}"""}
        ]
        )
        return completion.choices[0].message['content']

@functions_framework.http
def notes_capture(request):
    # 定义CORS Headers
    headers = {
        "Access-Control-Allow-Origin": "*",
        "Access-Control-Allow-Methods": "POST, OPTIONS",
        "Access-Control-Allow-Headers": "Content-Type",
        "Access-Control-Max-Age": "3600",
        "Access-Control-Allow-Credentials": "true"
    }

    # 处理预检OPTIONS请求
    if request.method == "OPTIONS":
        return ("", 204, headers)

    # 处理POST请求
    if request.method != "POST":
        return ({"error": "Only POST requests are allowed"}, 405, headers)

    try:
        text = request.get_json()
        if not text:
            return ({"error": "Invalid JSON input"}, 400, headers)

        split_text = text.split(".")
        words = len(re.findall(r'\w+', text))

        action_items = []
        roadblocks = []
        conclusion = []

        if words > 40:
            global model, tokenizer
            if not model:
                model, tokenizer = download_model()
            predict_class = predict_actions(model, tokenizer)
            action_text = []
            for sentence in split_text:
                action_pred = predict_class.predict(sentence)
                if action_pred == 1:
                    action_text.append(sentence)
            
            action_para = ". ".join(action_text)
            output = predict_class.gpt_response(action_para)
            output_dict = ast.literal_eval(output)

            action_items.append(output_dict['action_items'] if len(output_dict['action_items'])>0 else None)
            conclusion.append(output_dict['conclusion'] if len(output_dict['conclusion'])>0 else None)
            roadblocks.append(output_dict['roadblocks'] if len(output_dict['roadblocks'])>0 else None)

            final_dict = { 
                "action_items": action_items, 
                "conclusion": conclusion, 
                "roadblocks": roadblocks 
            }
            return (final_dict, 200, headers)
        else:
            return ({"error": "Give paragraph as input"}, 200, headers)
    except Exception as e:
        return ({"error": str(e)}, 500, headers)

方案2：保留Flask但正确适配Google Cloud Functions #

若必须使用Flask，需将Flask App交给functions_framework处理，避免路由冲突：

修改后的核心代码部分：

# 保留Flask导入和初始化
from flask import Flask, request
app = Flask(__name__)

# 原路由逻辑不变，去掉flask-cors相关代码
@app.route('/notes_capture', methods=['POST','OPTIONS'])
def notes_capture():
    headers = {
        "Access-Control-Allow-Origin": "*",
        "Access-Control-Allow-Methods": "POST, OPTIONS",
        "Access-Control-Allow-Headers": "Content-Type",
        "Access-Control-Max-Age": "3600",
        "Access-Control-Allow-Credentials": "true"
    }

    if request.method == "OPTIONS":
        return ("", 204, headers)

    # 原处理逻辑...

# 用functions_framework包装Flask App
@functions_framework.http
def entry_point(request):
    return app.handle_request(
        request.environ,
        start_response=None
    )

关键注意事项 #

• 所有响应（包括预检请求、错误响应）必须携带完整的CORS Headers。
• 确保预检OPTIONS请求返回204 No Content状态码。
• 避免在Google Cloud Functions中同时使用Flask路由和functions_framework装饰器，这会导致请求处理流程混乱。

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