企业官网建设流程全解析

影刀RPA新手教程：MySQL数据库操作完全指南——连接配置、CRUD实战与批量插入优化

本文作者：林焱 | 转载请注明出处

开篇案例：MySQL连接池耗尽，整个RPA系统停了

去年维护一个电商RPA系统，10个机器人同时跑，每个机器人都要读写MySQL数据库。

某天上午10点，所有机器人同时报错：Too many connections。

数据库拒绝新连接，整个自动化系统停了15分钟。

排查后发现：每个机器人在循环里每次操作都新建连接，操作完不关闭，导致连接数爆了。

MySQL默认最大连接数是151，10个机器人 x 每个机器人同时开十几个连接 = 超过151。

解决方案：用连接池，并且确保每次操作后正确关闭连接。

这次事故让我彻底搞懂了MySQL连接管理的重要性。

本文所有案例，围绕"电商订单MySQL读写"这条真实业务线展开。

模块一：安装与准备工作

影刀RPA操作MySQL，推荐用Python的pymysql库。

安装方式：在命令行运行pip install pymysql。

如果安装失败，可能是网络问题，换国内源：

pip install pymysql -i https://pypi.tuna.tsinghua.edu.cn/simple

安装完成后验证：

importpymysqlprint(pymysql.__version__)

环境配置详细步骤可以参考 home.linyan.cloud 上的教程。

新建流程，命名为"MySQL订单处理Demo"。

模块二：元素定位（从网页获取订单数据写入MySQL）

订单数据往往来自网页后台，先定位提取，再写入MySQL。

XPath提取订单列表

//table[@id='order-table']/tbody/tr

这个XPath匹配订单表格的所有行，配合循环逐行提取。

提取每行的单元格数据

./td[1]/text()

点号开头表示"从当前节点开始"，td[1]是第一列（订单号）。

在影刀里用"循环"指令遍历行，用"获取元素属性"或"获取元素文本"指令提取每列的值。

模块三：变量与数据类型（MySQL与Python类型对应）

MySQL字段类型和Python类型的对应：

| TEXT | str | 长文本 |
| DATETIME | datetime | 日期时间 |
| DATE | date | 日期 |
| FLOAT/DOUBLE | float | 浮点数（不推荐用于金额） |

金额字段一定要用DECIMAL

我当时踩过这个坑：用FLOAT存金额，结果19.98存进去变成19.979999，对账时对不上。

MySQL建表时：price DECIMAL(10,2)表示最多10位数字，其中2位小数。

模块四：流程控制（订单处理主流程）

订单从网页到MySQL的完整流程：

1. 连接MySQL数据库 2. 从网页抓取订单列表（或在本地CSV里读取） 3. 对每一个订单： 3.1 检查订单号是否已存在于MySQL（防重复） 3.2 如果不存在，插入新订单 3.3 如果存在，比较状态是否有变化，有变化则更新 4. 关闭数据库连接 5. 输出处理结果统计

在影刀里，步骤1和4放在"Python代码"指令里，步骤2用网页自动化指令，步骤3用循环+条件判断。

模块五：网页自动化（结合MySQL检查订单是否已存在）

在写入订单前，先查MySQL里是否已有此订单。

如果已有，比较状态；如果状态相同，跳过；如果不同，更新。

完整示例

importpymysqldefis_order_exists(conn,order_id):""" 检查订单是否已存在 """cursor=conn.cursor()cursor.execute("SELECT status FROM orders WHERE order_id = %s",(order_id,))row=cursor.fetchone()cursor.close()returnrow[0]ifrowelseNone# 返回已有状态，或Nonedefinsert_order(conn,order_data):cursor=conn.cursor()cursor.execute(""" INSERT INTO orders (order_id, buyer, amount, status, create_time) ![在这里插入图片描述](https://i-blog.csdnimg.cn/direct/34776bb71eb7485d8621f2b00ed4c287.png#pic_center) VALUES (%s, %s, %s, %s, %s) """,(order_data["order_id"],order_data["buyer"],order_data["amount"],order_data["status"],order_data["create_time"]))conn.commit()cursor.close()defupdate_order_status(conn,order_id,new_status):cursor=conn.cursor()cursor.execute(""" UPDATE orders SET status = %s, update_time = NOW() WHERE order_id = %s """,(new_status,order_id))conn.commit()cursor.close()

注意：用%s作为占位符，不要用Python的f-string拼接SQL，否则有SQL注入风险。

模块六：数据处理——连接配置

MySQL连接配置是关键，配置错了连不上，配置不对会运行时报错。

基础连接配置

importpymysqldefget_db_connection():""" 获取MySQL连接 """try:conn=pymysql.connect(host="localhost",# 数据库服务器地址port=3306,# 端口，默认3306user="rpa_user",# 用户名password="your_password",# 密码database="order_db",# 数据库名charset="utf8mb4",# 字符集（支持emoji用utf8mb4）connect_timeout=10,# 连接超时（秒）read_timeout=30,# 读取超时（秒）write_timeout=30,# 写入超时（秒）)print("MySQL连接成功")returnconnexceptpymysql.MySQLErrorase:print(f"MySQL连接失败：{e}")returnNone

连接超时问题（参考素材里的解决方案）

素材里提到：连接MySQL时报2013, 'Lost connection to MySQL server during query'。

解决方法：增加read_timeout参数的值。

conn=pymysql.connect(# ...其他参数read_timeout=600,# 改为600秒（10分钟）)

我当时也遇到过这个问题：查询一张800万行的表，MySQL需要20秒才能返回结果，但read_timeout默认是30秒，所以超时断开了。

模块七：数据处理——建库建表

在MySQL里创建订单表：

importpymysqldefinit_mysql_table(conn):""" 初始化MySQL订单表 """cursor=conn.cursor()cursor.execute(""" CREATE TABLE IF NOT EXISTS orders ( id BIGINT AUTO_INCREMENT PRIMARY KEY, order_id VARCHAR(64) NOT NULL UNIQUE, buyer VARCHAR(128) NOT NULL, amount DECIMAL(10, 2) NOT NULL DEFAULT 0.00, status VARCHAR(32) NOT NULL DEFAULT 'pending', status_code INT NOT NULL DEFAULT 0, create_time DATETIME NOT NULL, update_time DATETIME ON UPDATE CURRENT_TIMESTAMP, raw_data TEXT, INDEX idx_order_id (order_id), INDEX idx_status (status), INDEX idx_create_time (create_time) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 """)conn.commit()cursor.close()print("MySQL表初始化完成")

建表语句说明

• ENGINE=InnoDB：支持事务和外键（必须用InnoDB，不要用MyISAM）
• CHARSET=utf8mb4：支持所有Unicode字符（包括emoji）
• order_id VARCHAR(64) NOT NULL UNIQUE：订单号唯一
• INDEX idx_xxx：建索引加速查询
• ON UPDATE CURRENT_TIMESTAMP：更新时自动刷新时间

模块八：数据处理——批量插入优化

逐条插入MySQL很慢，1000条记录可能要花10秒。

批量插入可以把速度提升100倍。

错误示范：逐条插入

# 慢！不要这样写fororderinorders:cursor.execute("INSERT INTO orders VALUES (...)",(...))conn.commit()# 每条都提交，极慢

正确示范：批量插入

defbatch_insert_orders(conn,orders):""" 批量插入订单，用executemany """cursor=conn.cursor()sql=""" INSERT INTO orders (order_id, buyer, amount, status, create_time) VALUES (%s, %s, %s, %s, %s) """# 把orders列表转成元组列表values=[(o["order_id"],o["buyer"],o["amount"],o["status"],o["create_time"])foroinorders]cursor.executemany(sql,values)conn.commit()cursor.close()print(f"批量插入了{len(orders)}条订单")

executemany比循环execute快很多，因为它把多条插入合并成一次网络传输。

更进一步：用 LOAD DATA（超大数据量）

如果一次要插入10万条以上，用LOAD DATA INFILE最快：

defload_data_from_csv(conn,csv_path):""" 用LOAD DATA把CSV导入MySQL（最快的方式） """cursor=conn.cursor()sql=""" LOAD DATA INFILE %s INTO TABLE orders FIELDS TERMINATED BY ',' ENCLOSED BY '"' LINES TERMINATED BY '\\n' IGNORE 1 ROWS (order_id, buyer, amount, status, create_time) """cursor.execute(sql,(csv_path,))conn.commit()cursor.close()

注意：需要MySQL开启LOCAL INFILE权限，并且CSV文件要在MySQL服务器上（或者用LOCAL关键字从客户端加载）。

模块九：鼠标键盘与图像操作（验证码处理）

从网页抓取订单时遇到验证码，参考前面的文章用截图+OCR或者打码平台。

这里补充一个技巧：有些验证码是拖拽拼图，可以用图像识别计算拼图缺口位置，然后模拟拖拽。

# 伪代码，核心思路# 1. 截图验证码背景图和拼图# 2. 用OpenCV计算拼图应该放的位置# 3. 用影刀的"拖拽元素"指令，从起点拖到计算出的位置importcv2importnumpyasnpdeffind_puzzle_gap(bg_path,piece_path):""" 用OpenCV找拼图缺口位置 """bg=cv2.imread(bg_path)piece=cv2.imread(piece_path)# 转灰度bg_gray=cv2.cvtColor(bg,cv2.COLOR_BGR2GRAY)piece_gray=cv2.cvtColor(piece,cv2.COLOR_BGR2GRAY)# 模板匹配result=cv2.matchTemplate(bg_gray,piece_gray,cv2.TM_CCOEFF_NORMED)_,_,_,max_loc=cv2.minMaxLoc(result)gap_x=max_loc[0]returngap_x

模块十：进阶技能

技能一：连接池管理（多机器人必备）

importpymysqlfromdbutils.pooled_dbimportPooledDB# 创建连接池（安装：pip install dbutils）pool=PooledDB(creator=pymysql,maxconnections=10,# 连接池最大连接数mincached=2,# 初始化时创建的空闲连接数host="localhost",port=3306,user="rpa_user",password="password",database="order_db",charset="utf8mb4")defget_conn_from_pool():returnpool.connection()

用连接池后，每次操作用get_conn_from_pool()获取连接，用完关闭（还回池里）。

技能二：事务与回滚

批量操作要用事务，要么全部成功，要么全部失败。

deftransfer_order_status(conn,order_ids,new_status):""" 批量更新订单状态，用事务 """cursor=conn.cursor()try:conn.begin()fororder_idinorder_ids:cursor.execute("UPDATE orders SET status = %s WHERE order_id = %s",(new_status,order_id))conn.commit()print(f"成功更新{len(order_ids)}条")exceptExceptionase:conn.rollback()print(f"更新失败，已回滚：{e}")finally:cursor.close()

技能三：查询结果与内存管理

查询大量数据时，不要用fetchall()，会占用大量内存。

用fetchone()逐条处理，或者用stream模式：

defprocess_large_query(conn,batch_size=1000):""" 分批处理大量查询结果 """cursor=conn.cursor()cursor.execute("SELECT * FROM orders WHERE status = 'pending'")whileTrue:rows=cursor.fetchmany(batch_size)ifnotrows:breakforrowinrows:process_row(row)cursor.close()

模块十一：平台实战

把MySQL操作流程部署到影刀控制台时，注意以下几点。

要点一：数据库密码不能明文存流程里

用影刀的"凭据管理"功能存数据库密码，流程里只引用凭据名。

或者把密码存在服务器的环境变量里：

importos db_password=os.environ.get("RPA_MYSQL_PASSWORD","")[video(video-IIsTvo1x-1782551085140)(type-csdn)(url-https://live.csdn.net/v/embed/526817)(image-https://v-blog.csdnimg.cn/asset/1d3c3709da119dd8c13ab01e9b282520/cover/Cover0.jpg)(title-TEMU店群矩阵自动化运营核价报活动)]

要点二：MySQL连接异常自动重连

网络抖动可能导致MySQL连接断开，流程要能自动重连：

defsafe_db_operation(operation_func,*args,**kwargs):""" 包装数据库操作，连接断开时自动重连 """max_retry=3foriinrange(max_retry):try:returnoperation_func(*args,**kwargs)except(pymysql.OperationalError,pymysql.InterfaceError):print(f"数据库连接断开，第{i+1}次重连...")time.sleep(2)# 重新获取连接conn=get_db_connection()args=(conn,)+args[1:]raiseException("数据库操作失败，已达到最大重试次数")

要点三：用控制台查看数据库操作日志

在MySQL里开启慢查询日志，可以找出哪些SQL语句执行慢：

SETGLOBALslow_query_log='ON';SETGLOBALlong_query_time=2;

执行时间超过2秒的SQL会被记录，方便优化。

模块十二：系统联动与工程化规范

工程化规范一：数据库配置统一管理

创建db_config.json：

{"mysql":{"host":"localhost","port":3306,"user":"rpa_user","password_env":"RPA_MYSQL_PASSWORD","database":"order_db",![在这里插入图片描述](https://i-blog.csdnimg.cn/direct/c8dede6f2c7043a282f264efa8222b13.png#pic_center)"charset":"utf8mb4","pool_size":10}}

流程启动时读取这个文件，密码从环境变量读取。

工程化规范二：SQL语句统一存放

不要把SQL语句散落在代码各处。

创建一个sql_templates.py：

# sql_templates.pyORDERS_INSERT=""" INSERT INTO orders (order_id, buyer, amount, status, create_time) VALUES (%s, %s, %s, %s, %s) """ORDERS_BY_STATUS=""" SELECT * FROM orders WHERE status = %s ORDER BY create_time DESC LIMIT %s """# 在流程里fromsql_templatesimportORDERS_INSERT cursor.execute(ORDERS_INSERT,(...))

工程化规范三：数据库变更版本管理

数据库表结构变更时，用版本号管理：

-- migration_001_create_orders.sqlCREATETABLEorders(...);-- migration_002_add_index.sqlCREATEINDEXidx_statusONorders(status);-- 在流程里检查当前版本，自动执行未运行的迁移

速查表：MySQL连接参数

报错排查指南

报错：pymysql.err.OperationalError: (2003, “Can’t connect to MySQL server”)

原因：MySQL服务器地址不对，或者服务器没启动，或者防火墙拦了。

解决：检查host和port，用telnet host port测试连通性。

报错：pymysql.err.ProgrammingError: (1146, “Table doesn’t exist”)

原因：表名写错了，或者数据库选错了。

解决：检查database参数，确认表名大小写（Linux上MySQL表名区分大小写）。

总结

MySQL操作的核心要点：用连接池、批量操作要用executemany、事务保证原子性、密码不明文存储。

把这四个做好，你的RPA系统就能稳定可靠地操作MySQL。

更多MySQL优化技巧和完整代码，访问 home.linyan.cloud 获取。

#影刀RPA #RPA教程 #MySQL #数据库自动化 #RPA实战

作者：林焱