AutoScraperX/core/utils/extractors.py

from typing import Dict, Any

from jsonpath_ng import parse


def safe_extract(json_obj: Any, path: str, default=None):
    """
    安全提取单个字段值，返回匹配到的第一个，否则返回默认值。

    :param json_obj: 输入的 JSON 对象
    :param path: JSONPath 表达式
    :param default: 提取失败时返回的默认值
    :return: 提取结果或默认值
    """
    # 处理空对象或None的情况
    if json_obj is None:
        return default
        
    try:
        jsonpath_expr = parse(path)
        match = jsonpath_expr.find(json_obj)
        if match:
            return match[0].value
    except Exception as e:
        # 记录错误但不中断程序执行
        print(f"[extractor] Error extracting {path}: {e}")
    return default


def extract_multiple(json_obj: Any, fields: dict) -> dict:
    """
    根据字段配置提取多个字段。

    :param json_obj: 输入的 JSON 对象
    :param fields: 字段配置，如 {"title": "$.title", "id": "$.id"}
    :return: 字段名 -> 提取值的字典
    """
    return {key: safe_extract(json_obj, path) for key, path in fields.items()}


def extract_fields(video: Dict, field_map: Dict, logger=None, trace_id=None, aliyun_log=None) -> Dict:
    result = {}
    for field, path in field_map.items():
        # 如果path不是字符串或不是以$开头的jsonpath，则直接使用值
        if not isinstance(path, str) or not path.startswith("$"):
            result[field] = path
            continue
        value = safe_extract(video, path)
        if value is None and logger:
            logger.warning(f"字段提取失败: {field} 路径: {path}")
            if aliyun_log:
                aliyun_log.logging(
                    code="9024",
                    message=f"字段提取失败: {field} 路径: {path}",
                    data={"video": video}
                )
        result[field] = value
    return result