AutoScraperX/core/utils/request_preparer.py

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': ''}