import loguru from core.utils.extractors import safe_extract from typing import Dict, Any class RequestPreparer: """ 动态准备请求体: - 支持 {{var_name}} 动态占位符 - 自动从上次响应根据 response_parse 配置提取对应值 - 无值时使用空字符串 - 可选日志记录提取失败情况 """ def __init__(self, response_parse_config: Dict[str, str], logger=None, aliyun_log=None): """ :param response_parse_config: 如 {"next_cursor": "$.data.next_cursor"} :param logger: 可选 logger :param aliyun_log: 可选阿里云日志实例 """ self.response_parse_config = response_parse_config self.logger = logger or loguru.logger self.aliyun_log = aliyun_log def prepare(self, request_body_config: Dict[str, Any], response_data: Dict[str, Any]) -> Dict[str, Any]: """ 根据 request_body_config 和上次响应 response_data,返回可直接请求接口的 request_body """ prepared_body = {} for key, value in request_body_config.items(): if isinstance(value, str) and "{{" in value and "}}" in value: var_name = value.strip("{}").split("|")[0] # 支持后续扩展默认值 jsonpath_expr = self.response_parse_config.get(var_name) if jsonpath_expr: extracted_value = safe_extract(response_data, jsonpath_expr, default="") prepared_body[key] = extracted_value else: # response_parse_config 中未配置路径,默认空字符串 prepared_body[key] = "" else: prepared_body[key] = value return prepared_body if __name__ == "__main__": # ===== 示例运行观察效果 ===== request_body_config = { "cursor": "{{next_cursor}}", "category": "recommend", "flag": "{{flag}}" } response_parse_config = { "next_cursor": "$.data.next_cursor", "flag": "$.data.flag" } # 模拟响应 response_data = { "data": { "next_cursor": "abc123", "flag": "on" } } preparer = RequestPreparer(response_parse_config) prepared_body = preparer.prepare(request_body_config, response_data) print("准备好的请求体:", prepared_body) # 输出: {'cursor': 'abc123', 'category': 'recommend', 'flag': 'on'} # 测试首次请求情况 prepared_body_first = preparer.prepare(request_body_config, {}) print("首次请求时请求体:", prepared_body_first) # 输出: {'cursor': '', 'category': 'recommend', 'flag': ''}