4. 数据采集基础：数据源类型与 Python 连接示例

在催收触达波动监控中，数据采集是脚本运行的起点。你需要从不同数据源获取催收记录、触达结果和客户状态。本章将梳理三种主流数据源类型，并给出 Python 连接 MySQL 与 PostgreSQL 的完整示例。

4.1 数据源类型概览

催收监控脚本通常对接以下三类数据源：

4.2 数据源架构关系

下图展示了监控脚本与三种数据源之间的采集关系：

4.3 Python 连接 MySQL 示例

使用 pymysql 驱动连接 MySQL，执行催收记录查询：

import pymysql
from pymysql.cursors import DictCursor

# 数据库连接配置
DB_CONFIG = {
    "host": "192.168.1.100",
    "port": 3306,
    "user": "monitor_user",
    "password": "secure_pass_2024",
    "database": "collection_db",
    "charset": "utf8mb4"
}

def fetch_collection_records(start_time, end_time):
    """采集指定时间段的催收触达记录"""
    connection = pymysql.connect(**DB_CONFIG)
    try:
        with connection.cursor(DictCursor) as cursor:
            sql = """
                SELECT 
                    id, case_id, collector_name,
                    contact_channel, contact_result,
                    contact_time, duration_seconds
                FROM collection_contact_log
                WHERE contact_time BETWEEN %s AND %s
                ORDER BY contact_time ASC
            """
            cursor.execute(sql, (start_time, end_time))
            records = cursor.fetchall()
            return records
    finally:
        connection.close()

# 使用示例
if __name__ == "__main__":
    from datetime import datetime, timedelta
    end = datetime.now()
    start = end - timedelta(hours=1)
    data = fetch_collection_records(start, end)
    print(f"采集到 {len(data)} 条催收记录")

4.4 Python 连接 PostgreSQL 示例

使用 psycopg2 驱动连接 PostgreSQL，适用于更复杂的催收数据仓库：

import psycopg2
import psycopg2.extras

PG_CONFIG = {
    "host": "192.168.1.101",
    "port": 5432,
    "user": "pg_monitor",
    "password": "pg_strong_pass",
    "database": "collection_warehouse"
}

def fetch_touch_wave_data(threshold_minutes=30):
    """采集触达波动异常数据"""
    conn = psycopg2.connect(**PG_CONFIG)
    try:
        with conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur:
            query = """
                SELECT 
                    channel_type,
                    COUNT(*) AS total_touches,
                    COUNT(*) FILTER (WHERE result = 'SUCCESS') AS success_count,
                    ROUND(
                        COUNT(*) FILTER (WHERE result = 'SUCCESS') * 100.0 / NULLIF(COUNT(*), 0),
                        2
                    ) AS success_rate
                FROM touch_wave_log
                WHERE log_time >= NOW() - INTERVAL '1 hour'
                GROUP BY channel_type
                HAVING COUNT(*) FILTER (WHERE result = 'SUCCESS') * 100.0 / NULLIF(COUNT(*), 0) < %s
            """
            cur.execute(query, (threshold_minutes,))
            results = cur.fetchall()
            return results
    finally:
        conn.close()

# 使用示例
if __name__ == "__main__":
    anomalies = fetch_touch_wave_data(threshold_minutes=30)
    for row in anomalies:
        print(f"渠道: {row['channel_type']}, 成功率: {row['success_rate']}%")

4.5 两种数据库连接对比

4.6 连接管理最佳实践

• 使用连接池： 避免每次查询都创建新连接，推荐 SQLAlchemy 统一管理。
• 设置超时： 数据库连接与查询都要设置超时，防止脚本卡死。
• 参数化查询： 永远使用占位符传参，防止 SQL 注入。
• 异常捕获： 区分连接异常与查询异常，分别记录日志。
• 敏感信息保护： 数据库密码通过环境变量或密钥管理服务注入，不硬编码。