4. 数据采集进阶：通过API接口获取预警处理日志

在贷后预警监控体系中，预警处理日志是判断延迟尖峰是否已传导至人工处置环节的核心数据源。本章将深入讲解如何通过API接口高效、稳定地采集这些日志，涵盖JSON响应解析、异常处理与重试机制，确保数据采集层的健壮性。

4.1 预警处理日志API接口规范

典型的预警处理日志API遵循RESTful设计，返回结构化的JSON数据。以下为接口定义示例：

标准JSON响应结构：

{
  "code": 200,
  "message": "success",
  "data": {
    "total": 1250,
    "page": 1,
    "page_size": 100,
    "records": [
      {
        "log_id": "LOG202504010001",
        "warning_id": "WARN202504010001",
        "event_type": "延迟尖峰",
        "severity": "high",
        "status": "已处理",
        "handler": "张三",
        "process_time": "2025-04-01T10:30:00Z",
        "process_duration_sec": 320,
        "remark": "触发自动核销规则"
      }
    ]
  }
}

4.2 Python请求封装与JSON解析

使用 requests 库封装API调用，并利用 json 模块解析响应数据。以下为完整的数据采集函数：

import requests
import json
from typing import List, Dict, Optional

def fetch_process_logs(
    api_url: str,
    token: str,
    start_time: str,
    end_time: str,
    page: int = 1,
    page_size: int = 100
) -> Optional[List[Dict]]:
    """
    从API获取预警处理日志
    """
    headers = {
        "Authorization": f"Bearer {token}",
        "Content-Type": "application/json"
    }
    params = {
        "start_time": start_time,
        "end_time": end_time,
        "page": page,
        "page_size": page_size
    }

    try:
        response = requests.get(api_url, headers=headers, params=params, timeout=10)
        response.raise_for_status()  # 触发HTTP错误异常

        # 解析JSON响应
        resp_data = response.json()

        # 业务状态码校验
        if resp_data.get("code") != 200:
            raise ValueError(f"API业务错误: {resp_data.get('message')}")

        # 提取记录列表
        records = resp_data.get("data", {}).get("records", [])
        return records

    except requests.exceptions.RequestException as e:
        print(f"网络请求异常: {e}")
        return None
    except json.JSONDecodeError as e:
        print(f"JSON解析失败: {e}")
        return None
    except ValueError as e:
        print(f"业务逻辑错误: {e}")
        return None

4.3 异常重试机制：retry装饰器

网络抖动、服务临时不可用等场景下，直接失败会导致数据缺失。使用 tenacity 库的 @retry 装饰器实现智能重试，显著提升采集成功率。

安装依赖：

pip install tenacity

retry装饰器核心参数：

带重试机制的采集函数：

from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type, before_sleep_log
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=1, max=10),
    retry=retry_if_exception_type(requests.exceptions.RequestException),
    before_sleep=before_sleep_log(logger, logging.WARNING)
)
def fetch_process_logs_with_retry(
    api_url: str,
    token: str,
    start_time: str,
    end_time: str,
    page: int = 1,
    page_size: int = 100
) -> List[Dict]:
    """
    带自动重试的预警处理日志采集
    """
    headers = {
        "Authorization": f"Bearer {token}",
        "Content-Type": "application/json"
    }
    params = {
        "start_time": start_time,
        "end_time": end_time,
        "page": page,
        "page_size": page_size
    }

    response = requests.get(api_url, headers=headers, params=params, timeout=10)
    response.raise_for_status()

    resp_data = response.json()
    if resp_data.get("code") != 200:
        raise ValueError(f"API业务错误: {resp_data.get('message')}")

    records = resp_data.get("data", {}).get("records", [])
    return records

4.4 数据采集架构总览

以下SVG图展示了从API请求到数据落地的完整采集链路，以及重试机制在其中的位置：

4.5 分页采集与全量数据拉取

预警处理日志通常数据量较大，需通过分页逐批采集。以下为全量拉取实现：

def fetch_all_process_logs(
    api_url: str,
    token: str,
    start_time: str,
    end_time: str,
    page_size: int = 100
) -> List[Dict]:
    """
    全量拉取预警处理日志（自动处理分页）
    """
    all_records = []
    page = 1
    total_pages = None

    while True:
        records = fetch_process_logs_with_retry(
            api_url, token, start_time, end_time, page, page_size
        )
        if records is None:
            break

        all_records.extend(records)

        # 首次请求后获取总页数
        if total_pages is None:
            # 假设API在响应头或data中返回total
            # 此处简化处理：若返回记录数小于page_size，则说明已到最后一页
            if len(records) < page_size:
                break
        else:
            if page >= total_pages:
                break

        page += 1

    return all_records

4.6 异常场景与处理策略对比

4.7 实战要点总结

• 认证管理： Token应定期轮换，建议从环境变量或密钥管理服务读取，避免硬编码。
• 超时设置： 根据API平均响应时间设置合理的超时值（如10秒），避免长时间阻塞。
• 重试幂等性： 确保API接口是幂等的，多次请求不会产生重复数据或副作用。
• 日志记录： 每次重试都应记录日志，便于事后排查延迟尖峰的根本原因。
• 熔断机制： 当连续重试失败次数超过阈值（如10次），应触发熔断，暂停采集并告警。