腾讯云函数对接失败

经过老师指导问题已解决，谢谢老师。

测试没问题的云函数和后端部分的代码如下：

# 云函数：
import json
import sys
import base64
import datetime
import hashlib
import math
from urllib import parse as urlparse

def main_handler(event, context):
    """
    主处理函数，用于处理传入的事件并执行相应逻辑
    :param event: 包含请求信息的事件对象，通常包含请求体等内容
    :param context: 上下文对象，包含执行环境等相关信息（此处未实际用到）
    :return: 包含状态码和响应体的字典，用于返回处理结果
    """
    try:
        # 从 event 中获取请求体，若不存在则用空字典
        body = json.loads(event.get("body", "{}"))
        # 从请求体中提取相关参数
        code = body.get("code", "")
        func_name = body.get("func_name", "")
        params = body.get("params", [])
        print("code:", code)
        print("func_name:", func_name)
        print("params:", params)

        # 创建自定义全局命名空间
        exec_globals = {
            '__builtins__': __builtins__,
            'sys': sys,
            'json': json,
            'math': math,
            'datetime': datetime,
            'hashlib': hashlib,
            'base64': base64,
            'urlparse': urlparse  # 使用别名
        }

        # 执行传入的代码，限定内置函数环境，将执行后的变量存入 exec_globals
        exec(code, exec_globals)

        # 检查指定函数是否在执行后的全局变量中且可调用
        if func_name in exec_globals and callable(exec_globals[func_name]):
            # 调用函数并获取结果
            result = exec_globals[func_name](params)
            # 返回成功响应，包含状态码 200 和执行结果
            return {
                "statusCode": 200,
                "body": json.dumps({"result": result})
            }
        else:
            # 函数不存在或不可调用时返回的错误响应
            return {
                "statusCode": 400,
                "body": json.dumps({"error": "Function not found or not callable"})
            }
    except Exception as e:
        # 捕获异常时返回的错误响应，包含状态码 500 和异常信息
        return {
            "statusCode": 500,
            "body": json.dumps({"error": str(e)})
        }

# 后端增加对接云函数的代码
def _request_cloud_function(self, code: str, params: dict):
    """将函数发送给腾讯云函数去执行，并获取到结果"""
    url = os.getenv('CLOUD_FUNC_URL')
    # 1.组装请求数据
    payload = {
        "code": code,
        "func_name": "main",
        "params": params,
    }
    headers = {"Content-Type": "application/json"}
    # 2.发送请求
    # resp = requests.request('POST', url=url, json=payload, headers=headers)
    resp = requests.request('POST', url=url, data=json.dumps(payload), headers=headers)
    # resp = requests.post(url=url, json=payload, headers=headers)
    resp.raise_for_status()

    # 3.处理响应
    return resp.json()

然后再invoke中引用该函数

整理分享几个要点：

1、最上方的问题主要是_request_cloud_function的headers中设置错误，应该是设置Content-Type为json，表示发送的负载数据是json格式，这是是必须的，否则云函数那边会解析错误，而不是Accept json数据（表示接受的响应数据是json）：

更正后，简单测试下加减法的函数已经成功了

2、但如果是类似于课程之前的计算快递查询的签名一样的函数，其中import了几个包，这类函数在执行的时候，云函数就会报错__import__ not found：

经过沟通发现，云函数沙箱等环境有安全保护措施，exec执行的函数不能随便import。可以预先导入常用的模块，方式如下：

3.关于urllib.parse的导入需要使用别名

在云函数中：

from urllib import parse as urlparse

# 创建自定义全局命名空间
exec_globals = {
    '__builtins__': __builtins__,
    'sys': sys,
    'json': json,
    'math': math,
    'datetime': datetime,
    'hashlib': hashlib,
    'base64': base64,
    'urlparse': urlparse  # 使用别名
}

在llmops平台中Python代码节点就直接使用urlparse.quote，举例之前的快递鸟签名函数：

def main(params):
    # 构造快递鸟签名
    # 1.将RequestData和ApiKey拼接
    request_data = {'LogisticCode': f'{params.get("logistic_code")}'}
    api_key = '00edcf20-97e3-42e8-90ba-fec4b7be796b'
    combined_data = json.dumps(request_data) + api_key

    # 2.MD5加密并转换成小写
    md5_hash = hashlib.md5(combined_data.encode('utf-8')).hexdigest()

    # 3.Base64编码
    base64_encode = base64.b64encode(md5_hash.encode('utf-8')).decode('utf-8')

    # 4.URL编码------------------------这里直接使用urlparse.quote
    url_encoded = urlparse.quote(base64_encode)

    return {
        "RequestType": 8002,
        "EBusinessID": 1891677,
        "DataSign": url_encoded,
        "RequestData": urlparse.quote(json.dumps(request_data))
    }