影刀RPA Python操作MySQL数据库：从连接到增删改查完整实战
影刀RPA Python操作MySQL数据库：从连接到增删改查完整实战
作者：林焱
影刀内置的数据库指令只支持SQLite，但现实中大多数业务系统用的是MySQL。这篇讲如何在影刀的Python代码块里连接MySQL，完成增删改查。
准备工作
安装MySQL驱动：
pipinstallpymysqlpymysql是纯Python实现的MySQL驱动，不需要安装额外C库，Windows下兼容性最好。比mysql-connector-python轻量，比MySQLdb好安装。
连接模板
importpymysql# 连接配置config={'host':'192.168.1.100',# 数据库IP'port':3306,# 端口'user':'your_username',# 用户名'password':'your_password',# 密码'database':'your_database',# 库名[video(video-UCajC8p9-1783622550236)(type-csdn)(url-https://live.csdn.net/v/embed/525010)(image-https://v-blog.csdnimg.cn/asset/f4faa587144cb7070f19e8b36813806b/cover/Cover0.jpg)(title-店群矩阵自动化突破运营极限！)]'charset':'utf8mb4',# 字符集（支持emoji）'connect_timeout':10# 连接超时}connection=pymysql.connect(**config)cursor=connection.cursor()查询数据
# 查询所有商品sql="SELECT id, name, price, stock FROM products WHERE status = %s"cursor.execute(sql,('active',))results=cursor.fetchall()# results是一个元组列表：[ (1, '商品A', 99.00, 100), (2, '商品B', 199.00, 50) ]# 转成字典列表（方便后续处理）columns=[desc[0]fordescincursor.description]data=[dict(zip(columns,row))forrowinresults]foritemindata:print(f"{item['name']}: ¥{item['price']}，库存{item['stock']}")注意：%s是参数占位符，不是字符串格式化。用%s传入参数可以防止SQL注入（pymysql会自动转义），不要用f-string或%拼接SQL。
插入/更新/删除（同一模式）
try:# 插入sql="INSERT INTO products (name, price, stock) VALUES (%s, %s, %s)"cursor.execute(sql,('新产品',299.00,50))# 更新sql="UPDATE products SET price = %s WHERE id = %s"cursor.execute(sql,(259.00,1))# 删除sql="DELETE FROM products WHERE stock = 0"cursor.execute(sql)# 提交事务（查询不需要commit，增删改必须commit）connection.commit()print(f"影响行数:{cursor.rowcount}")exceptExceptionase:connection.rollback()# 出错回滚print(f"操作失败:{e}")批量操作（影刀数据写入MySQL）
典型场景：影刀采集到1000条商品数据，批量写入MySQL：
# 影刀采集到的数据（假设是一个列表）products=[('商品A',99.00,100),('商品B',199.00,50),# ... 更多数据]sql="INSERT INTO products (name, price, stock) VALUES (%s, %s, %s)"cursor.executemany(sql,products)# 一次插入多条connection.commit()print(f"成功插入{cursor.rowcount}条")executemany比循环execute快得多，因为它把多条数据打包成一次网络往返发送。
实战：影刀采集 + MySQL存储完整流程
影刀定时触发 → 影刀采集网页数据 → Python代码块写入MySQL ↓ 影刀定时触发 → Python代码块查询MySQL → 影刀生成报表 → 影刀发送邮件步骤1：写入（采集流程）
importpymysql# 从影刀变量拿到采集的数据collected_data=[...]# 影刀传入conn=pymysql.connect(**config)try:withconn.cursor()ascursor:# 先清空今日数据cursor.execute("DELETE FROM daily_sales WHERE date = CURDATE()")# 批量插入新数据sql="INSERT INTO daily_sales (date, product, sales) VALUES (CURDATE(), %s, %s)"cursor.executemany(sql,collected_data)conn.commit()finally:conn.close()步骤2：读取（报表流程）
conn=pymysql.connect(**config)try:withconn.cursor()ascursor:sql=""" SELECT product, SUM(sales) as total FROM daily_sales WHERE date >= DATE_SUB(CURDATE(), INTERVAL 7 DAY) GROUP BY product ORDER BY total DESC """cursor.execute(sql)week_data=cursor.fetchall()finally:conn.close()# week_data返回给影刀，生成周报踩坑实录
坑1：连接不关闭导致数据库连接数耗尽
MySQL默认最大连接数是151。如果每次查询都新建连接但不关闭，很快就会耗尽。务必用try/finally关闭连接，或者用with语句（pymysql的连接支持上下文管理器）：
temu店群自动化报活动案例
withpymysql.connect(**config)asconn:withconn.cursor()ascursor:cursor.execute(sql)坑2：中文乱码
两个地方都要设字符集：连接参数里设charset='utf8mb4'，建表时设DEFAULT CHARSET=utf8mb4。utf8mb4支持emoji，utf8只支持基本多语言平面（BMP）。
坑3：长时间空闲连接被MySQL断开
MySQL默认8小时无活动会自动断开连接。如果你的影刀流程两个数据库操作之间隔了很久，需要用conn.ping(reconnect=True)检查连接状态：
conn.ping(reconnect=True)# 如果断了自动重连cursor=conn.cursor()坑4：安全性——密码不要硬编码
不要把数据库密码写在Python代码里。用配置文件（上一篇文章讲的YAML/INI）管理连接参数，或者用环境变量：
importos password=os.environ.get('DB_PASSWORD','')写在最后
MySQL是RPA流程数据持久化的最佳方案。SQLite够轻量但不够强大，MySQL才能支撑起真正的业务数据流转。影刀负责从网页把数据拿回来，MySQL负责把数据存起来供后续分析——这是一个完整的自动化数据管道。