影刀RPA MySQL连接与查询远程数据库操作署名林焱什么情况用SQLite适合单机小数据量但你的RPA流程要对接公司业务系统怎么办ERP、CRM、订单系统几乎都用MySQL。你需要从MySQL里查订单数据做处理或者把采集结果写回业务系统。典型场景从ERP系统拉取当日订单生成发货报表把采集的竞品价格写入MySQL的价格监控表从CRM查询客户联系方式批量发送通知RPA处理完的数据回写到业务系统的结果表怎么做1. 安装依赖并连接先在影刀的Python环境中安装pymysql拼多多店群自动化上架方案# 在【执行Python代码】组件中先安装依赖importsubprocess subprocess.check_call([pip,install,pymysql,-q])连接MySQLimportpymysql# 数据库连接配置db_config{host:192.168.1.100,port:3306,user:rpa_user,password:RPA2024#secure,database:erp_system,charset:utf8mb4,cursorclass:pymysql.cursors.DictCursor# 返回字典格式}# 建立连接connpymysql.connect(**db_config)cursorconn.cursor()# 测试查询cursor.execute(SELECT VERSION())versioncursor.fetchone()print(fMySQL连接成功版本{version})conn.close()影刀操作建议把数据库配置放在【设置变量】组件中用全局变量管理不要硬编码在代码里。2. 查询订单数据importpymysqlfromdatetimeimportdatetime,timedelta db_config{host:192.168.1.100,port:3306,user:rpa_user,password:RPA2024#secure,database:erp_system,charset:utf8mb4,cursorclass:pymysql.cursors.DictCursor}connpymysql.connect(**db_config)cursorconn.cursor()# 查询今日订单todaydatetime.now().strftime(%Y-%m-%d)cursor.execute( SELECT order_id, customer_name, product_name, quantity, total_price, status FROM orders WHERE create_time %s AND status pending ORDER BY create_time DESC ,(today,))orderscursor.fetchall()print(f今日待处理订单{len(orders)}条)fororderinorders[:5]:print(f{order[order_id]}-{order[customer_name]}-{order[product_name]}- ¥{order[total_price]})conn.close()3. 写入采集数据importpymysqlfromdatetimeimportdatetime db_config{host:192.168.1.100,port:3306,user:rpa_user,password:RPA2024#secure,database:price_monitor,charset:utf8mb4,cursorclass:pymysql.cursors.DictCursor}connpymysql.connect(**db_config)cursorconn.cursor()# 批量插入竞品价格price_data[(SKU-001,竞品A,99.9,天猫,datetime.now()),(SKU-002,竞品B,159.0,京东,datetime.now()),(SKU-003,竞品C,45.5,拼多多,datetime.now()),]# INSERT ... ON DUPLICATE KEY UPDATE有则更新无则插入cursor.executemany( INSERT INTO competitor_prices (sku, competitor_name, price, platform, check_time) VALUES (%s, %s, %s, %s, %s) ON DUPLICATE KEY UPDATE price VALUES(price), check_time VALUES(check_time) ,price_data)conn.commit()print(f写入{cursor.rowcount}条价格数据)conn.close()4. 封装数据库操作类importpymysqlfromdatetimeimportdatetimeclassMySQLHelper:MySQL操作封装类def__init__(self,host,port,user,password,database):self.config{host:host,port:port,user:user,password:password,database:database,charset:utf8mb4,cursorclass:pymysql.cursors.DictCursor}self.connNoneself.cursorNonedef__enter__(self):self.connpymysql.connect(**self.config)self.cursorself.conn.cursor()returnselfdef__exit__(self,exc_type,exc_val,exc_tb):ifexc_type:# 有异常则回滚self.conn.rollback()else:self.conn.commit()self.cursor.close()self.conn.close()defquery(self,sql,paramsNone):查询返回字典列表self.cursor.execute(sql,params)returnself.cursor.fetchall()defquery_one(self,sql,paramsNone):查询单条self.cursor.execute(sql,params)returnself.cursor.fetchone()defexecute(self,sql,paramsNone):执行单条SQLself.cursor.execute(sql,params)returnself.cursor.rowcountdefbatch_execute(self,sql,params_list):批量执行self.cursor.executemany(sql,params_list)returnself.cursor.rowcount# 使用示例withMySQLHelper(192.168.1.100,3306,rpa_user,RPA2024#secure,erp_system)asdb:# 查询ordersdb.query(SELECT * FROM orders WHERE status %s LIMIT 10,(pending,))# 批量更新状态order_ids[(o[order_id],)foroinorders]db.batch_execute(UPDATE orders SET status %s WHERE order_id %s,[(processing,oid[0])foroidinorder_ids])print(f处理了{len(orders)}条订单)5. 连接池管理多流程共用fromdbutils.pooled_dbimportPooledDBimportpymysql# 创建连接池最多5个连接poolPooledDB(creatorpymysql,maxconnections5,mincached2,maxcached3,host192.168.1.100,port3306,userrpa_user,passwordRPA2024#secure,databaseerp_system,charsetutf8mb4,cursorclasspymysql.cursors.DictCursor)# 从连接池获取连接connpool.connection()cursorconn.cursor()cursor.execute(SELECT COUNT(*) as total FROM orders)print(cursor.fetchone())conn.close()# 实际是归还连接池不是真正关闭有什么坑坑1连接超时断开MySQL默认wait_timeout是8小时如果RPA流程中间有大量等待操作比如等待网页加载、等待文件生成连接会超时断开再执行SQL就报MySQL server has gone away。解决办法# 方案1执行前ping一下conn.ping(reconnectTrue)# 方案2设置自动重连db_config[connect_timeout]10# 方案3每次操作都重新连接简单粗暴但可靠defsafe_query(sql,paramsNone):try:[video(video-apG1bpPw-1783577092277)(type-csdn)(url-https://live.csdn.net/v/embed/524993)(image-https://v-blog.csdnimg.cn/asset/a547123d88ad712dccba346c9217e237/cover/Cover0.jpg)(title-TEMU店群如何管理运营)]cursor.execute(sql,params)returncursor.fetchall()exceptpymysql.OperationalError:connpymysql.connect(**db_config)# 重连cursorconn.cursor()cursor.execute(sql,params)returncursor.fetchall()坑2密码含特殊字符导致连接失败如果密码里有、#、!等字符在拼接连接字符串时会出问题。用字典传参就没问题# 正确用字典connpymysql.connect(host10.0.0.1,userrpa,passwordPss#word!,databasedb)# 错误用URL格式拼接# mysql://rpa:Pss#word!10.0.0.1/db ← 这个会解析错误坑3DictCursor返回的字段名带表名前缀JOIN查询时如果两个表有同名字段如都有idDictCursor只保留最后一个前面的会被覆盖。解决办法是给字段起别名cursor.execute( SELECT o.id as order_id, o.customer_name, c.id as customer_id, c.phone FROM orders o JOIN customers c ON o.customer_id c.id )坑4批量插入数据量太大导致超时executemany一次插几万条数据可能触发MySQL的max_allowed_packet限制默认4MB。解决分批插入。defbatch_insert(cursor,sql,data,batch_size1000):分批插入foriinrange(0,len(data),batch_size):batchdata[i:ibatch_size]cursor.executemany(sql,batch)conn.commit()# 每批提交一次print(f已插入{ilen(batch)}/{len(data)})坑5SSH隧道连接内网MySQL公司MySQL在内网RPA运行在外网需要通过SSH跳板机连接fromsshtunnelimportSSHTunnelForwarderimportpymysql# 先建SSH隧道serverSSHTunnelForwarder((jump.host.com,22),ssh_usernamejump_user,ssh_passwordjump_pass,remote_bind_address(192.168.1.100,3306))server.start()# 通过本地转发端口连接MySQLconnpymysql.connect(host127.0.0.1,portserver.local_bind_port,# SSH隧道分配的本地端口userrpa_user,passwordRPA2024#secure,databaseerp_system)# 使用完毕后关闭conn.close()server.stop()别忘了在finally里关掉SSH隧道否则连接泄漏。
