影刀RPA新手教程MySQL数据库操作完全指南——连接配置、CRUD实战与批量插入优化本文作者林焱 | 转载请注明出处开篇案例MySQL连接池耗尽整个RPA系统停了去年维护一个电商RPA系统10个机器人同时跑每个机器人都要读写MySQL数据库。某天上午10点所有机器人同时报错Too many connections。数据库拒绝新连接整个自动化系统停了15分钟。排查后发现每个机器人在循环里每次操作都新建连接操作完不关闭导致连接数爆了。MySQL默认最大连接数是15110个机器人 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[idorder-table]/tbody/tr这个XPath匹配订单表格的所有行配合循环逐行提取。提取每行的单元格数据./td[1]/text()点号开头表示从当前节点开始td[1]是第一列订单号。在影刀里用循环指令遍历行用获取元素属性或获取元素文本指令提取每列的值。模块三变量与数据类型MySQL与Python类型对应MySQL字段类型和Python类型的对应MySQL类型Python类型说明INTint整数BIGINTint大整数DECIMALDecimal精确小数金额必用VARCHARstr字符串拼多多店群自动化报活动上架| 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): 检查订单是否已存在 cursorconn.cursor()cursor.execute(SELECT status FROM orders WHERE order_id %s,(order_id,))rowcursor.fetchone()cursor.close()returnrow[0]ifrowelseNone# 返回已有状态或Nonedefinsert_order(conn,order_data):cursorconn.cursor()cursor.execute( INSERT INTO orders (order_id, buyer, amount, status, create_time)  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):cursorconn.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:connpymysql.connect(hostlocalhost,# 数据库服务器地址port3306,# 端口默认3306userrpa_user,# 用户名passwordyour_password,# 密码databaseorder_db,# 数据库名charsetutf8mb4,# 字符集支持emoji用utf8mb4connect_timeout10,# 连接超时秒read_timeout30,# 读取超时秒write_timeout30,# 写入超时秒)print(MySQL连接成功)returnconnexceptpymysql.MySQLErrorase:print(fMySQL连接失败{e})returnNone连接超时问题参考素材里的解决方案素材里提到连接MySQL时报2013, Lost connection to MySQL server during query。解决方法增加read_timeout参数的值。connpymysql.connect(# ...其他参数read_timeout600,# 改为600秒10分钟)我当时也遇到过这个问题查询一张800万行的表MySQL需要20秒才能返回结果但read_timeout默认是30秒所以超时断开了。模块七数据处理——建库建表在MySQL里创建订单表importpymysqldefinit_mysql_table(conn): 初始化MySQL订单表 cursorconn.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) ) ENGINEInnoDB DEFAULT CHARSETutf8mb4 )conn.commit()cursor.close()print(MySQL表初始化完成)建表语句说明ENGINEInnoDB支持事务和外键必须用InnoDB不要用MyISAMCHARSETutf8mb4支持所有Unicode字符包括emojiorder_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 cursorconn.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最快的方式 cursorconn.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找拼图缺口位置 bgcv2.imread(bg_path)piececv2.imread(piece_path)# 转灰度bg_graycv2.cvtColor(bg,cv2.COLOR_BGR2GRAY)piece_graycv2.cvtColor(piece,cv2.COLOR_BGR2GRAY)# 模板匹配resultcv2.matchTemplate(bg_gray,piece_gray,cv2.TM_CCOEFF_NORMED)_,_,_,max_loccv2.minMaxLoc(result)gap_xmax_loc[0]returngap_x模块十进阶技能技能一连接池管理多机器人必备importpymysqlfromdbutils.pooled_dbimportPooledDB# 创建连接池安装pip install dbutilspoolPooledDB(creatorpymysql,maxconnections10,# 连接池最大连接数mincached2,# 初始化时创建的空闲连接数hostlocalhost,port3306,userrpa_user,passwordpassword,databaseorder_db,charsetutf8mb4)defget_conn_from_pool():returnpool.connection()用连接池后每次操作用get_conn_from_pool()获取连接用完关闭还回池里。技能二事务与回滚批量操作要用事务要么全部成功要么全部失败。deftransfer_order_status(conn,order_ids,new_status): 批量更新订单状态用事务 cursorconn.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_size1000): 分批处理大量查询结果 cursorconn.cursor()cursor.execute(SELECT * FROM orders WHERE status pending)whileTrue:rowscursor.fetchmany(batch_size)ifnotrows:breakforrowinrows:process_row(row)cursor.close()模块十一平台实战把MySQL操作流程部署到影刀控制台时注意以下几点。要点一数据库密码不能明文存流程里用影刀的凭据管理功能存数据库密码流程里只引用凭据名。或者把密码存在服务器的环境变量里importos db_passwordos.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_retry3foriinrange(max_retry):try:returnoperation_func(*args,**kwargs)except(pymysql.OperationalError,pymysql.InterfaceError):print(f数据库连接断开第{i1}次重连...)time.sleep(2)# 重新获取连接connget_db_connection()args(conn,)args[1:]raiseException(数据库操作失败已达到最大重试次数)要点三用控制台查看数据库操作日志在MySQL里开启慢查询日志可以找出哪些SQL语句执行慢SETGLOBALslow_query_logON;SETGLOBALlong_query_time2;执行时间超过2秒的SQL会被记录方便优化。模块十二系统联动与工程化规范工程化规范一数据库配置统一管理创建db_config.json{mysql:{host:localhost,port:3306,user:rpa_user,password_env:RPA_MYSQL_PASSWORD,database:order_db,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连接参数参数推荐值说明charsetutf8mb4支持所有字符connect_timeout10连接超时秒read_timeout30-600读取超时慢查询要调大autocommitFalse手动提交事务更安全cursorclassDictCursor返回字典而不是元组报错排查指南报错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实战作者林焱
)
