3. 数据采集基础：从数据库读取预警任务表

贷后预警监控的第一步，是从业务数据库中可靠地读取待处理的任务。本章将围绕预警任务表的结构、任务状态字段的语义，以及如何使用 Python 连接 MySQL 与 PostgreSQL 进行数据采集展开。

3.1 预警任务表核心结构

预警任务表是贷后监控的“待办清单”，每条记录代表一笔需要评估的贷款或客户。典型表结构如下：

3.2 任务状态字段深度解析

任务状态是监控流程的“交通信号灯”，不同状态决定了采集模块的下一步动作：

• 0 - 待处理： 新生成的预警任务，等待被采集线程拉取。采集 SQL 应优先筛选此状态。
• 1 - 处理中： 已被某个 worker 锁定，正在执行模型评估或规则计算。需配合 update_time 判断是否超时。
• 2 - 已完成： 任务处理成功，结果已写入结果表。采集模块应跳过此状态。
• 3 - 异常： 处理过程中发生不可恢复的错误（如数据缺失、模型异常）。需人工介入或触发补偿流程。

3.3 使用 pymysql 连接 MySQL

对于 MySQL 数据库，pymysql 是 Python 生态中最常用的纯 Python 驱动。以下为连接与查询示例：

import pymysql
from datetime import datetime, timedelta

# 建立连接
conn = pymysql.connect(
    host='192.168.1.100',
    port=3306,
    user='monitor_user',
    password='SecurePass123!',
    database='loan_early_warning',
    charset='utf8mb4',
    cursorclass=pymysql.cursors.DictCursor
)

try:
    with conn.cursor() as cursor:
        # 查询最近5分钟内生成的待处理任务
        sql = """
            SELECT task_id, loan_id, rule_id, create_time
            FROM warning_task
            WHERE task_status = 0
              AND create_time >= %s
            ORDER BY create_time ASC
            LIMIT 100
        """
        cutoff_time = datetime.now() - timedelta(minutes=5)
        cursor.execute(sql, (cutoff_time,))
        tasks = cursor.fetchall()
        print(f"采集到 {len(tasks)} 条待处理任务")
finally:
    conn.close()

3.4 使用 psycopg2 连接 PostgreSQL

PostgreSQL 用户首选 psycopg2，它支持原生二进制协议，性能优于纯 Python 驱动。连接方式与 MySQL 类似，但参数名略有差异：

import psycopg2
from datetime import datetime, timedelta

# 建立连接
conn = psycopg2.connect(
    host='192.168.1.101',
    port=5432,
    user='monitor_user',
    password='SecurePass456!',
    dbname='loan_early_warning'
)

try:
    with conn.cursor() as cursor:
        # 查询待处理任务，PostgreSQL 使用 %s 占位符
        sql = """
            SELECT task_id, loan_id, rule_id, create_time
            FROM warning_task
            WHERE task_status = 0
              AND create_time >= %s
            ORDER BY create_time ASC
            LIMIT 100
        """
        cutoff_time = datetime.now() - timedelta(minutes=5)
        cursor.execute(sql, (cutoff_time,))
        tasks = cursor.fetchall()
        print(f"采集到 {len(tasks)} 条待处理任务")
finally:
    conn.close()

3.5 数据库连接参数对比

3.6 数据采集流程架构

以下 SVG 图展示了从数据库采集预警任务到状态解析的完整数据流：

3.7 状态字段解析逻辑

采集到任务数据后，需在 Python 侧对状态字段进行二次解析，以便下游模块快速决策。以下为推荐的状态解析函数：

from enum import IntEnum

class TaskStatus(IntEnum):
    PENDING = 0      # 待处理
    PROCESSING = 1   # 处理中
    COMPLETED = 2    # 已完成
    ERROR = 3        # 异常

def parse_task_status(row: dict) -> dict:
    """
    解析单条任务记录，返回增强后的状态信息
    """
    status_code = row.get('task_status', -1)
    status_name = TaskStatus(status_code).name if status_code in (0,1,2,3) else 'UNKNOWN'
    
    # 判断是否超时（处理中状态超过30分钟）
    is_timeout = False
    if status_code == TaskStatus.PROCESSING:
        from datetime import datetime
        update_time = row.get('update_time')
        if update_time and (datetime.now() - update_time).total_seconds() > 1800:
            is_timeout = True
    
    return {
        'task_id': row['task_id'],
        'status_code': status_code,
        'status_name': status_name,
        'is_timeout': is_timeout,
        'should_process': status_code in (TaskStatus.PENDING, TaskStatus.ERROR)
    }

3.8 连接异常处理最佳实践

数据库连接可能因网络抖动、连接池耗尽等原因失败。以下为健壮连接示例：

import time
from pymysql.err import OperationalError

def get_connection_with_retry(max_retries=3, retry_delay=2):
    """
    带重试机制的数据库连接获取
    """
    for attempt in range(max_retries):
        try:
            conn = pymysql.connect(
                host='192.168.1.100',
                user='monitor_user',
                password='SecurePass123!',
                database='loan_early_warning',
                connect_timeout=5,
                read_timeout=10
            )
            return conn
        except OperationalError as e:
            print(f"连接失败 (第{attempt+1}次): {e}")
            if attempt < max_retries - 1:
                time.sleep(retry_delay * (attempt + 1))
            else:
                raise  # 最后一次失败则向上抛出异常