requestId: 9a7c9ab6-909c-4344-ba98-77719633a351
请求在到达腾讯云的 API 接入层 (tcb-api.tencentcloudapi.com/web) 时，由于参数校验失败（INVALID_PARAM），未能成功路由到您的云函数内部执行。
尝试调用 tcb-api.tencentcloudapi.com/web/functions/{function_name} 的过程。
tcb_adapter.py
# flask_admin_panel/app/adapters/tcb_adapter.py
import requests
import json
import base64
import hmac
import hashlib
import time
import datetime
from urllib.parse import quote_plus
from flask import current_app
import logging
logger = logging.getLogger(__name__)
class TCBAdapter:
def __init__(self, env_id=None, secret_id=None, secret_key=None):
# ... (属性定义与之前相同) ...
self._env_id = env_id
self._secret_id = secret_id
self._secret_key = secret_key
self._base_url = "https://tcb-api.tencentcloudapi.com/web"
@property
def env_id(self):
if self._env_id is None:
if not current_app:
raise RuntimeError("TCBAdapter: Flask current_app is not available for env_id.")
self._env_id = current_app.config.get('TCB_ENV_ID')
if not self._env_id:
raise ValueError("TCB_ENV_ID is not set in Flask config.")
return self._env_id
@property
def secret_id(self):
if self._secret_id is None:
if not current_app:
raise RuntimeError("TCBAdapter: Flask current_app is not available for secret_id.")
self._secret_id = current_app.config.get('TCB_SECRET_ID')
if not self._secret_id:
raise ValueError("TCB_SECRET_ID is not set in Flask config.")
return self._secret_id
@property
def secret_key(self):
if self._secret_key is None:
if not current_app:
raise RuntimeError("TCBAdapter: Flask current_app is not available for secret_key.")
self._secret_key = current_app.config.get('TCB_SECRET_KEY')
if not self._secret_key:
raise ValueError("TCB_SECRET_KEY is not set in Flask config.")
return self._secret_key
@property
def base_url(self):
return self._base_url
def _gen_sign(self, params_to_sign):
sorted_params = sorted(params_to_sign.items(), key=lambda x: x[0])
sign_str_parts = []
for k, v_original in sorted_params:
v_str = str(v_original) # 确保是字符串
sign_str_parts.append(f"{k}={quote_plus(v_str)}")
sign_str = "&".join(sign_str_parts)
current_secret_key = self.secret_key
if not current_secret_key: # ... (错误处理与之前相同)
logger.error("TCBAdapter: Secret Key is not configured for signing.")
raise ValueError("TCBAdapter: Secret Key is missing for signature generation.")
hmac_obj = hmac.new(
current_secret_key.encode('utf-8'),
sign_str.encode('utf-8'),
hashlib.sha256
)
return base64.b64encode(hmac_obj.digest()).decode('utf-8')
def invoke_function(self, function_name, data=None):
request_id_from_header = None
try:
url_path = f"/functions/{function_name}" # 已修正路径
full_url = f"{self.base_url}{url_path}"
query_params_for_sign = {
'envName': self.env_id,
'timestamp': int(time.time() * 1000), # 13位毫秒级时间戳
'secretId': self.secret_id,
'action': function_name, # <--- 新增：将 action (值为函数名) 加入签名和查询参数
}
# 注意：如果 TCB Web API 对 'action' 参数有特定的大小写要求（例如 'Action'），
# 或者期望一个固定的值（例如 'InvokeCloudFunction'），则需要相应调整。
# 目前我们假设 'action' 的值就是云函数名，并且参数名是小写 'action'。
signature = self._gen_sign(query_params_for_sign)
final_query_params = query_params_for_sign.copy()
final_query_params['sign'] = signature
headers = {
'Content-Type': 'application/json; charset=utf-8',
}
log_target = current_app.logger if current_app else logger
# 日志部分与上一版本相同，省略以保持简洁
log_target.debug(f"TCB: Invoking function '{function_name}' at URL '{full_url}' with query_params: {final_query_params}")
log_target.debug(f"TCB: Request Headers: {headers}")
log_target.debug(f"TCB: JSON body for '{function_name}': {json.dumps(data if data is not None else {})}")
response = requests.post(
full_url,
headers=headers,
params=final_query_params,
json=data or {}
)
# 响应处理逻辑 (与您上一轮测试成功的版本中的详细处理逻辑一致)
# ... [请从您之前能正确解析200 OK但内容是INVALID_PARAM的那个TCBAdapter.py版本中，复制详细的响应解析和错误抛出逻辑到这里] ...
# 确保这部分能正确处理各种成功和失败情况，并记录requestId
if 'X-TCB-Request-ID' in response.headers:
request_id_from_header = response.headers['X-TCB-Request-ID']
log_target.debug(f"TCB Response for '{function_name}': Status Code: {response.status_code}, RequestId: {request_id_from_header or 'N/A'}")
log_target.debug(f"TCB Response for '{function_name}': Headers: {response.headers}")
log_target.info(f"TCB Response for '{function_name}': RAW TEXT: {response.text}")
if response.status_code != 200:
error_message_detail = response.text
try:
error_json = response.json()
error_message_detail = error_json.get("message", response.text)
req_id = error_json.get("requestId", request_id_from_header or 'N/A')
code = error_json.get("code", f"HTTP_{response.status_code}")
raise Exception(f"调用云函数失败 ({function_name}): Code: {code}, Message: {error_message_detail}, RequestId: {req_id}")
except requests.exceptions.JSONDecodeError:
raise Exception(f"调用云函数失败 ({function_name}) with status {response.status_code}: {response.text}, RequestId: {request_id_from_header or 'N/A'}")
result = response.json()
tcb_code = result.get('code')
tcb_message = result.get('message')
tcb_request_id = result.get('requestId', request_id_from_header or 'N/A')
# 根据Copilot的提示，INVALID_PARAM 是主要的错误码
if tcb_code == 'INVALID_PARAM' or \
(isinstance(tcb_code, str) and tcb_code not in ['SUCCESS', 'OK', '0'] and not tcb_code.startswith('SUCCESS') and tcb_code != 'null' and tcb_code is not None):
final_message = tcb_message or "TCB function returned an error code without a message."
log_target.error(f"TCB function '{function_name}' error. Code: {tcb_code}, Message: {final_message}, RequestId: {tcb_request_id}, FullResponse: {result}")
raise Exception(f"云函数错误 ({function_name}): {final_message} (Code: {tcb_code}, RequestId: {tcb_request_id})")
# 成功的判断：如果 code 是 'SUCCESS', 'OK', '0', null(某些API返回null表示无业务错误) 或者不存在(直接是数据)
# 并且必须要有 'data' 字段或者 result 本身就是数据 (排除了标准错误字段后)
if 'data' in result:
if isinstance(result['data'], str): # data 可能是字符串形式的JSON
try:
return json.loads(result['data'])
except json.JSONDecodeError:
log_target.warning(f"TCB function '{function_name}' 'data' field is a string but not valid JSON. Returning as string. Data: {result['data']}")
return result['data']
return result['data']
# 如果没有 'data' 字段，但 code 表示成功 (包括 null 或不存在)
elif (tcb_code is None or str(tcb_code).lower() in ['success', 'ok', '0', 'null']):
log_target.info(f"TCB function '{function_name}' successful (Code: {tcb_code}), but 'data' key missing. Assuming result itself is the data or no data returned. Full response: {result}")
# 移除标准响应字段，看是否剩下业务数据
response_data_candidate = {k: v for k, v in result.items() if k not in ['code', 'message', 'requestId']}
# 如果云函数就是不返回任何业务数据，那么返回None或{}是合适的
return response_data_candidate if response_data_candidate else None # 或者返回 {}
else: # 其他意外情况
log_target.error(f"TCB function '{function_name}' response is in an unexpected format. Full response: {result}, RequestId: {tcb_request_id}")
raise Exception(f"云函数响应格式未知 ({function_name}): {result}, RequestId: {tcb_request_id}")
except requests.exceptions.RequestException as req_err: # 网络层错误
log_target.error(f"TCB HTTP request failed for '{function_name}'. Error: {req_err}", exc_info=True)
raise Exception(f"调用云函数网络请求失败 ({function_name}): {str(req_err)}") from req_err
except Exception as e: # 其他未知错误
if isinstance(e, Exception) and hasattr(e, 'args') and e.args and ("云函数错误" in e.args[0] or "调用云函数失败" in e.args[0]): # 避免重复包装
raise
log_target.error(f"TCB unhandled exception during invoke_function '{function_name}'. Error: {e}, RequestId: {request_id_from_header or 'N/A'}", exc_info=True)
raise Exception(f"调用云函数时发生未知错误 ({function_name}): {str(e)}, RequestId: {request_id_from_header or 'N/A'}") from e
# ... [其他方法如 get_categories, add_category 等保持不变] ...
# [请确保从您之前的 TCBAdapter.py 复制所有其他方法到这里]
# 分类操作方法
def get_categories(self):
return self.invoke_function('getCategories')
def add_category(self, category_payload):
return self.invoke_function('addCategory', category_payload)
def update_category(self, category_id, category_payload):
payload_with_id = category_payload.copy()
payload_with_id['_id'] = category_id
return self.invoke_function('updateCategory', payload_with_id)
def delete_category(self, category_id):
return self.invoke_function('deleteCategory', {'_id': category_id})
# 菜品操作方法
def get_dishes(self, category_id=None):
params = {}
if category_id:
params['category_id'] = category_id
return self.invoke_function('getDishes', params if params else None)
def add_dish(self, dish_payload):
return self.invoke_function('addDish', dish_payload)
def update_dish(self, dish_id, dish_payload):
payload_with_id = dish_payload.copy()
payload_with_id['_id'] = dish_id
return self.invoke_function('updateDish', payload_with_id)
def delete_dish(self, dish_id):
return self.invoke_function('deleteDish', {'_id': dish_id})
def upload_file(self, file_path, cloud_path=None):
import os
with open(file_path, 'rb') as f:
file_content_bytes = f.read()
file_content_b64 = base64.b64encode(file_content_bytes).decode('utf-8')
filename = os.path.basename(file_path)
if cloud_path:
full_path = f"{cloud_path.strip('/')}/{filename.strip('/')}"
else:
today = datetime.datetime.now().strftime('%Y%m%d')
full_path = f"uploads/{today}/{filename.strip('/')}"
upload_data_payload = {
'fileContent': file_content_b64,
'cloudPath': full_path
}
log_target = current_app.logger if current_app else logger
log_target.debug(f"TCB: Calling 'uploadFile' function with cloudPath: {full_path}")
upload_result_data = self.invoke_function('uploadFile', upload_data_payload)
if not isinstance(upload_result_data, dict) or \
'fileID' not in upload_result_data or \
'url' not in upload_result_data:
logger.error(f"TCB 'uploadFile' function returned unexpected data format: {upload_result_data}")
raise Exception(f"TCB 'uploadFile' function response format error. Expected dict with 'fileID' and 'url'. Got: {upload_result_data}")
return {
'fileID': upload_result_data['fileID'],
'url': upload_result_data['url']
}
# 订单操作方法
def get_orders(self, status=None, start_date=None, end_date=None):
params = {}
if status: params['status'] = status
if start_date: params['start_date'] = start_date
if end_date: params['end_date'] = end_date
return self.invoke_function('getOrders', params if params else None)
def get_order(self, order_id):
return self.invoke_function('getOrder', {'_id': order_id})
def update_order_status(self, order_id, status):
return self.invoke_function('updateOrderStatus', {'_id': order_id, 'status': status})
# 用户操作方法
def get_users(self):
return self.invoke_function('getUsers')
def get_user(self, user_id):
return self.invoke_function('getUser', {'_id': user_id})
# 排队操作方法
def get_queues(self, status=None):
params = {}
if status: params['status'] = status
return self.invoke_function('getQueues', params if params else None)
def update_queue_status(self, queue_id, status):
return self.invoke_function('updateQueueStatus', {'_id': queue_id, 'status': status})
def get_queue(self, queue_id):
return self.invoke_function('getQueue', {'_id': queue_id})
test_tcb_communication.py 测试通讯
import os
import sys
import logging
from dotenv import load_dotenv # 导入 load_dotenv
# 将项目根目录添加到 Python 路径，以便导入 app 和 config
project_root = os.path.abspath(os.path.dirname(__file__))
sys.path.insert(0, project_root)
# 加载 .env 文件中的环境变量
# 假设 .env 文件与此脚本在同一目录（项目根目录）
dotenv_path = os.path.join(project_root, '.env')
if os.path.exists(dotenv_path):
load_dotenv(dotenv_path)
print(f".env 文件加载成功: {dotenv_path}")
else:
print(f"警告: 未找到 .env 文件于 {dotenv_path}。脚本将依赖已设置的环境变量。")
# 配置日志，以便看到 TCBAdapter 的输出
logging.basicConfig(level=logging.DEBUG,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
def test_tcb_connection():
"""
测试 Flask 应用与 TCB 云函数的通信。
"""
logger.info("开始 TCB 通信测试...")
# 1. 尝试创建 Flask 应用上下文并加载配置
try:
from app import create_app
# 使用 .env 文件中或已设置的 FLASK_CONFIG 环境变量，默认为 'default'
# config.py 文件中 TCB_ENV_ID 等配置会通过 os.environ.get() 读取由 dotenv 加载的环境变量
flask_app = create_app(os.environ.get('FLASK_CONFIG') or 'default')
with flask_app.app_context(): # 推入应用上下文
logger.info("Flask 应用上下文创建成功。")
logger.info(f"使用的 Flask 配置: {flask_app.config.get('ENV') or os.environ.get('FLASK_CONFIG') or 'default'}")
logger.info(f"TCB_ENV_ID from config: {flask_app.config.get('TCB_ENV_ID')}")
logger.info(f"TCB_SECRET_ID from config: {flask_app.config.get('TCB_SECRET_ID')}")
logger.info(f"TCB_SECRET_KEY loaded: {'Yes' if flask_app.config.get('TCB_SECRET_KEY') else 'No'}")
if not all([flask_app.config.get('TCB_ENV_ID'),
flask_app.config.get('TCB_SECRET_ID'),
flask_app.config.get('TCB_SECRET_KEY')]):
logger.error("错误：TCB 配置信息不完整。请检查您的 .env 文件和 config.py 文件。")
return
# 2. 初始化 TCBAdapter
from app.adapters.tcb_adapter import TCBAdapter
tcb_adapter = TCBAdapter()
logger.info("TCBAdapter 初始化成功。")
# 3. 调用测试云函数
function_name = "testConnection"
# test_payload = {"source": "flask_test_script_dotenv", "data": "ping_dotenv"}
# 尝试1: 空的业务数据
test_payload = {}
# 尝试2: 或者完全不传 data (让 TCBAdapter 中的 data or {} 变成 {})
# test_payload = None
logger.info(f"准备调用云函数 '{function_name}'，参数 (业务数据): {test_payload}")
# logger.info(f"准备调用云函数 '{function_name}'，参数: {test_payload}")
try:
# 如果 test_payload = None, TCBAdapter 会发送空json body {}
response = tcb_adapter.invoke_function(function_name, data=test_payload)
logger.info(f"云函数 '{function_name}' 调用成功！")
logger.info(f"TCB 返回的响应 (data部分): {response}")
if isinstance(response, dict) and response.get("message") == "Connection to TCB cloud function successful!":
logger.info("测试通过：云函数按预期返回了成功消息。")
logger.info(f"云函数收到的参数: {response.get('requestPayload')}")
logger.info(f"云函数执行时间戳: {response.get('timestamp')}")
else:
logger.warning(f"测试警告：云函数返回的响应并非预期的成功结构。收到的响应: {response}")
except Exception as e:
logger.error(f"调用云函数 '{function_name}' 失败: {e}", exc_info=True)
logger.error("请检查：")
logger.error(" - 云函数名称是否正确且已部署。")
logger.error(" - TCB 配置 (ENV_ID, SECRET_ID, SECRET_KEY) 是否正确且具有调用权限。")
logger.error(" - VPS 的网络连接是否正常，能否访问 tcb-api.tencentcloudapi.com。")
logger.error(" - 查看 Flask 应用日志和 TCB 云函数日志获取更详细的错误信息。")
except ImportError as e:
logger.error(f"导入模块失败: {e}", exc_info=True)
logger.error("请确保此脚本位于项目根目录，并且 Flask 应用结构正确。")
except Exception as e:
logger.error(f"创建 Flask 应用或执行测试时发生未知错误: {e}", exc_info=True)
logger.info("TCB 通信测试结束。")
if __name__ == "__main__":
# 现在脚本会自动从同目录下的 .env 文件加载环境变量
# .env 文件存在且包含所有必要的变量
test_tcb_connection()
测试日志：
(venv) root@bjn:/var/www/bjn.bestherbs.cn# python test_tcb_communication.py
.env 文件加载成功: /var/www/bjn.bestherbs.cn/.env
2025-05-19 15:54:27,609 - __main__ - INFO - 开始 TCB 通信测试...
2025-05-19 15:54:28,030 - __main__ - INFO - Flask 应用上下文创建成功。
2025-05-19 15:54:28,030 - __main__ - INFO - 使用的 Flask 配置: default
2025-05-19 15:54:28,030 - __main__ - INFO - TCB_ENV_ID from config: cloud1-2gp7tmuk8290cf92
2025-05-19 15:54:28,030 - __main__ - INFO - TCB_SECRET_ID from config: AKID65vJajmhT5d4PYZM4YoriMmyxjHdQ6aq
2025-05-19 15:54:28,030 - __main__ - INFO - TCB_SECRET_KEY loaded: Yes
2025-05-19 15:54:28,030 - __main__ - INFO - TCBAdapter 初始化成功。
2025-05-19 15:54:28,030 - __main__ - INFO - 准备调用云函数 'testConnection'，参数 (业务数据): {}
2025-05-19 15:54:28,031 - app - DEBUG - TCB: Invoking function 'testConnection' at URL 'https://tcb-api.tencentcloudapi.com/web/functions/testConnection' with query_params: {'envName': 'cloud1-2gp7tmuk8290cf92', 'timestamp': 1747641268030, 'secretId': 'AKID65vJajmhT5d4PYZM4YoriMmyxjHdQ6aq', 'action': 'testConnection', 'sign': 'RKIGhBadTc8pjtrrc8D1cthpxHiQwbVrV4/37aioDkM='}
2025-05-19 15:54:28,031 - app - DEBUG - TCB: Request Headers: {'Content-Type': 'application/json; charset=utf-8'}
2025-05-19 15:54:28,031 - app - DEBUG - TCB: JSON body for 'testConnection': {}
2025-05-19 15:54:28,032 - urllib3.connectionpool - DEBUG - Starting new HTTPS connection (1): tcb-api.tencentcloudapi.com:443
2025-05-19 15:54:28,240 - urllib3.connectionpool - DEBUG - https://tcb-api.tencentcloudapi.com:443 "POST /web/functions/testConnection?envName=cloud1-2gp7tmuk8290cf92×tamp=1747641268030&secretId=AKID65vJajmhT5d4PYZM4YoriMmyxjHdQ6aq&action=testConnection&sign=RKIGhBadTc8pjtrrc8D1cthpxHiQwbVrV4%2F37aioDkM%3D HTTP/1.1" 200 219
2025-05-19 15:54:28,241 - app - DEBUG - TCB Response for 'testConnection': Status Code: 200, RequestId: N/A
2025-05-19 15:54:28,241 - app - DEBUG - TCB Response for 'testConnection': Headers: {'Date': 'Mon, 19 May 2025 07:54:24 GMT', 'Content-Type': 'application/json; charset=utf-8', 'Content-Length': '219', 'Connection': 'keep-alive', 'Vary': 'Origin'}
2025-05-19 15:54:28,241 - app - INFO - TCB Response for 'testConnection': RAW TEXT: {"code":"INVALID_PARAM","message":"Invalid request param 请前往云开发AI小助手查看问题：https://tcb.cloud.tencent.com/dev#/helper/copilot?q=INVALID_PARAM","requestId":"9a7c9ab6-909c-4344-ba98-77719633a351"}
2025-05-19 15:54:28,241 - app - ERROR - TCB function 'testConnection' error. Code: INVALID_PARAM, Message: Invalid request param 请前往云开发AI小助手查看问题：https://tcb.cloud.tencent.com/dev#/helper/copilot?q=INVALID_PARAM, RequestId: 9a7c9ab6-909c-4344-ba98-77719633a351, FullResponse: {'code': 'INVALID_PARAM', 'message': 'Invalid request param 请前往云开发AI小助手查看问题：https://tcb.cloud.tencent.com/dev#/helper/copilot?q=INVALID_PARAM', 'requestId': '9a7c9ab6-909c-4344-ba98-77719633a351'}
2025-05-19 15:54:28,241 - __main__ - ERROR - 调用云函数 'testConnection' 失败: 云函数错误 (testConnection): Invalid request param 请前往云开发AI小助手查看问题：https://tcb.cloud.tencent.com/dev#/helper/copilot?q=INVALID_PARAM (Code: INVALID_PARAM, RequestId: 9a7c9ab6-909c-4344-ba98-77719633a351)
Traceback (most recent call last):
File "/var/www/bjn.bestherbs.cn/test_tcb_communication.py", line 69, in test_tcb_connection
response = tcb_adapter.invoke_function(function_name, data=test_payload)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/var/www/bjn.bestherbs.cn/app/adapters/tcb_adapter.py", line 147, in invoke_function
raise Exception(f"云函数错误 ({function_name}): {final_message} (Code: {tcb_code}, RequestId: {tcb_request_id})")
Exception: 云函数错误 (testConnection): Invalid request param 请前往云开发AI小助手查看问题：https://tcb.cloud.tencent.com/dev#/helper/copilot?q=INVALID_PARAM (Code: INVALID_PARAM, RequestId: 9a7c9ab6-909c-4344-ba98-77719633a351)
2025-05-19 15:54:28,242 - __main__ - ERROR - 请检查：
2025-05-19 15:54:28,242 - __main__ - ERROR - - 云函数名称是否正确且已部署。
2025-05-19 15:54:28,242 - __main__ - ERROR - - TCB 配置 (ENV_ID, SECRET_ID, SECRET_KEY) 是否正确且具有调用权限。
2025-05-19 15:54:28,242 - __main__ - ERROR - - VPS 的网络连接是否正常，能否访问 tcb-api.tencentcloudapi.com。
2025-05-19 15:54:28,242 - __main__ - ERROR - - 查看 Flask 应用日志和 TCB 云函数日志获取更详细的错误信息。
2025-05-19 15:54:28,243 - __main__ - INFO - TCB 通信测试结束。
日志中没有记录，如下图。
请帮助，分析并指出问题所在，是哪个参数导致了 INVALID_PARAM，以及正确的参数规范是什么（包括必需参数、参数位置、参数名大小写、签名方法等所有细节） 谢谢。
tcb-api.tencentcloudapi.com/web 需要通过云开发SDK进行调用
参考云开发SDK文档：https://docs.cloudbase.net/api-reference/webv3/initialization