AutoScraperX/core/utils/request_preparer.py

import loguru
from typing import Dict, Any

from core.utils.extractors import safe_extract


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 or {}
        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
        """
        if not request_body_config:
            return {}
            
        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]
                
                # 首先尝试直接从 response_data 中获取（处理扁平字典情况）
                if var_name in response_data:
                    prepared_body[key] = response_data[var_name]
                else:
                    # 否则使用 JSONPath 提取（处理嵌套对象情况）
                    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
                        
                        # 记录提取信息（仅在有日志记录器时）
                        if extracted_value == "" and self.logger:
                            self.logger.debug(f"变量 {var_name} 提取结果为空，使用默认值")
                    else:
                        # response_parse_config 中未配置路径，默认空字符串
                        prepared_body[key] = ""
                        
                        # 记录警告信息
                        if self.logger:
                            self.logger.warning(f"未在response_parse_config中找到变量 {var_name} 的路径配置")
            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': ''}
    
    # 测试扁平字典情况
    flat_data = {
        "next_cursor": "flat_abc123",
        "flag": "flat_on"
    }
    prepared_body_flat = preparer.prepare(request_body_config, flat_data)
    print("扁平字典请求体:", prepared_body_flat)
    # 输出: {'cursor': 'flat_abc123', 'category': 'recommend', 'flag': 'flat_on'}