📚 课程目录第4章 Python自动化脚本编写

4. Python自动化脚本编写：连接数据库、执行SQL、写入DataFrame

好，咱们进入实战环节。前面几章讲了审计数据的来源、表结构、告警规则设计，但数据不会自己从数据库里跑出来。你得写脚本去拉数据，然后交给下游处理。

这一章，我就带你手写一个Python脚本。核心就三件事：连上数据库、执行SQL、把结果塞进DataFrame。听起来简单，但坑不少。我做了这么多年审计自动化，光连接串写错就折腾过好几个通宵。

4.1 环境准备：装好你的工具箱

先别急着写代码。你得确认Python环境里装了这几个库：

库名	用途	安装命令
pymysql	连接MySQL数据库	`pip install pymysql`
pandas	数据处理与DataFrame	`pip install pandas`
sqlalchemy	ORM与连接池管理	`pip install sqlalchemy`
cryptography	密码加密传输（可选）	`pip install cryptography`

我个人习惯用 sqlalchemy 做连接管理，而不是直接用 pymysql 的裸连接。为什么？因为 sqlalchemy 自带连接池，审计脚本通常要跑几十张表，频繁建连断连会把数据库搞崩。你想想看，凌晨跑批的时候数据库连接数爆了，DBA 的电话打到你手机上……嗯，我经历过。

我的建议： 生产环境一定要用连接池。单次查询用裸连接没问题，但批量跑报表时，连接池能救你的命。

4.2 连接数据库：别把密码写死在代码里

先看一个最基础的连接示例。注意，这里我用的是 sqlalchemy 的 create_engine：

from sqlalchemy import create_engine
import pandas as pd

# 数据库配置
DB_CONFIG = {
    "host": "192.168.1.100",
    "port": 3306,
    "user": "audit_user",
    "password": "your_password_here",
    "database": "audit_db",
    "charset": "utf8mb4"
}

# 创建连接引擎
engine = create_engine(
    f"mysql+pymysql://{DB_CONFIG['user']}:{DB_CONFIG['password']}"
    f"@{DB_CONFIG['host']}:{DB_CONFIG['port']}"
    f"/{DB_CONFIG['database']}?charset={DB_CONFIG['charset']}",
    pool_size=5,
    max_overflow=10,
    pool_recycle=3600
)

# 测试连接
with engine.connect() as conn:
    result = conn.execute("SELECT 1")
    print("数据库连接成功！")

我曾经踩过的坑： 密码里如果有特殊字符（比如 @、#、$），直接拼在连接串里会报错。解决办法是用 urllib.parse.quote 对密码做URL编码。另外，千万别把密码硬编码提交到Git仓库——我见过不止一次，整个公司的数据库密码被传到GitHub上。

4.3 执行SQL查询：从简单到复杂

连接建好了，接下来就是执行SQL。审计报表的SQL通常分三类：

全量查询：比如拉取某张表的全部记录
带参数查询：比如按日期范围筛选
复杂聚合查询：比如多表JOIN + GROUP BY

咱们一个一个来。先看全量查询：

# 全量查询
sql_all = "SELECT * FROM transaction_log"
df_all = pd.read_sql(sql_all, engine)
print(f"共查询到 {len(df_all)} 条记录")

带参数查询要小心SQL注入。虽然审计系统是内部使用，但好习惯要养成：

# 带参数查询（安全写法）
sql_param = """
    SELECT trade_id, amount, trade_date, status
    FROM transaction_log
    WHERE trade_date >= %(start_date)s
      AND trade_date < %(end_date)s
      AND status IN %(status_list)s
"""
params = {
    "start_date": "2024-01-01",
    "end_date": "2024-01-31",
    "status_list": ("FAILED", "PENDING")
}
df_param = pd.read_sql(sql_param, engine, params=params)

复杂聚合查询，我建议先在数据库客户端里调好SQL，再粘到Python里。别在代码里反复调试SQL，效率太低：

# 复杂聚合：按天统计交易金额与笔数
sql_agg = """
    SELECT 
        DATE(trade_date) AS trade_day,
        COUNT(*) AS trade_count,
        SUM(amount) AS total_amount,
        AVG(amount) AS avg_amount
    FROM transaction_log
    WHERE status = 'SUCCESS'
    GROUP BY DATE(trade_date)
    ORDER BY trade_day
"""
df_agg = pd.read_sql(sql_agg, engine)

  核心原则： 能在SQL里做的聚合，就别拉到DataFrame里做。数据库是专门干这个的，比pandas快一个数量级。我在项目中见过有人把几百万行数据拉到本地再用groupby，结果内存爆了——其实一条SQL就搞定。

4.4 写入DataFrame：数据清洗与类型转换

数据从数据库出来，进了DataFrame，但还没完。数据库里的字段类型和Python的类型不一定一一对应。我总结了几种常见问题：

数据库类型	DataFrame默认类型	常见问题	处理方式
DECIMAL(18,2)	float64	精度丢失	用 `pd.to_numeric` 指定精度
DATETIME	object（字符串）	无法做时间运算	用 `pd.to_datetime` 转换
NULL	NaN	后续计算报错	用 `fillna` 或 `dropna`
VARCHAR 枚举值	object	占用内存大	转为 `category` 类型

来看一个完整的处理流程：

# 1. 查询数据
df_raw = pd.read_sql("SELECT * FROM transaction_log", engine)

# 2. 类型转换
df_raw['trade_date'] = pd.to_datetime(df_raw['trade_date'])
df_raw['amount'] = pd.to_numeric(df_raw['amount'], errors='coerce')
df_raw['status'] = df_raw['status'].astype('category')

# 3. 处理空值
df_raw['remark'] = df_raw['remark'].fillna('无备注')
df_raw = df_raw.dropna(subset=['trade_id'])  # 关键字段不能为空

# 4. 新增计算列
df_raw['amount_abs'] = df_raw['amount'].abs()
df_raw['is_abnormal'] = df_raw['amount_abs'] > 1000000

print(df_raw.dtypes)
print(df_raw.head())

我的经验： 日期字段一定要显式转换。我遇到过数据库里存的是 2024-01-01 00:00:00，但pandas读进来变成了字符串，后面做日期筛选全错了。加一行 pd.to_datetime 就能避免。

4.5 完整脚本：从连接到写入一条龙

下面是一个完整的脚本模板。你可以直接拿过去改改参数就能用：

"""
审计报表数据抽取脚本
功能：连接数据库 -> 执行SQL -> 写入DataFrame -> 保存为CSV
"""
import pandas as pd
from sqlalchemy import create_engine
from datetime import datetime, timedelta
import logging

# 配置日志
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)

# 数据库配置（建议从环境变量读取）
DB_CONFIG = {
    "host": "localhost",
    "port": 3306,
    "user": "audit_user",
    "password": "your_password",
    "database": "audit_db"
}

def get_engine():
    """创建数据库引擎"""
    conn_str = (
        f"mysql+pymysql://{DB_CONFIG['user']}:{DB_CONFIG['password']}"
        f"@{DB_CONFIG['host']}:{DB_CONFIG['port']}"
        f"/{DB_CONFIG['database']}?charset=utf8mb4"
    )
    return create_engine(conn_str, pool_size=5, pool_recycle=3600)

def extract_transactions(engine, days=7):
    """抽取最近N天的交易数据"""
    end_date = datetime.now()
    start_date = end_date - timedelta(days=days)
    
    sql = """
        SELECT 
            trade_id, account_id, trade_date,
            amount, status, channel
        FROM transaction_log
        WHERE trade_date BETWEEN %(start)s AND %(end)s
    """
    params = {"start": start_date, "end": end_date}
    
    logger.info(f"开始抽取数据：{start_date} 至 {end_date}")
    df = pd.read_sql(sql, engine, params=params)
    
    # 类型处理
    df['trade_date'] = pd.to_datetime(df['trade_date'])
    df['amount'] = pd.to_numeric(df['amount'], errors='coerce')
    df['status'] = df['status'].astype('category')
    
    logger.info(f"共抽取 {len(df)} 条记录")
    return df

def save_to_csv(df, filename=None):
    """保存为CSV文件"""
    if filename is None:
        filename = f"audit_data_{datetime.now().strftime('%Y%m%d_%H%M%S')}.csv"
    df.to_csv(filename, index=False, encoding='utf-8-sig')
    logger.info(f"数据已保存至：{filename}")

if __name__ == "__main__":
    engine = get_engine()
    df = extract_transactions(engine, days=7)
    save_to_csv(df)
    print("脚本执行完成！")

4.6 核心流程一览

我把这一章的知识点画成了一张流程图。你看一眼就能明白整个数据流的走向：

4.7 避坑指南：我踩过的那些坑

最后，分享几个实战中容易翻车的地方：

连接超时：审计脚本经常在凌晨跑，数据库可能在做备份。连接超时时间设长一点，或者加重试机制。
内存爆炸：一次查询几百万行，DataFrame直接占满内存。解决办法是分页查询，或者用 chunksize 参数分批读取。
字符集乱码：数据库用utf8，但Python默认可能不是。连接串里一定要指定 charset=utf8mb4，保存CSV时用 encoding='utf-8-sig'。
SQL注入：虽然是内部系统，但用 %s 拼接SQL的习惯要改。用 %(name)s 传参，安全又清晰。

我曾经翻过的一个大车： 有一次生产环境的数据库密码到期了，但脚本里写的是死密码，凌晨跑批全部失败，第二天早上业务部门炸了。从那以后，我所有脚本都从环境变量或密钥管理服务读密码，再也没出过类似问题。

好了，这一章的内容就这些。你照着上面的代码搭一个脚本，跑通从数据库到DataFrame的流程，后面几章咱们再往里面加告警逻辑和报表生成。