影刀RPA剪贴板操作跨应用数据传递的5种方式作者林焱RPA最大的价值之一就是打通不同应用之间的数据壁垒。本文详解剪贴板、文件、API、数据库、消息队列5种跨应用数据传递方式让你的机器人真正实现万物互联。前言为什么跨应用数据传递这么重要真实工作中数据很少只停留在一个应用里从网页采集数据 → 粘贴到Excel从ERP导出数据 → 导入CRM从邮件提取附件 → 上传到网盘从数据库查询结果 → 发送到微信群这些跨应用的数据传递就是RPA最核心的使用场景。方式一剪贴板——最直接的跨应用传递1.1 基本剪贴板操作店群矩阵自动化突破运营极限importpyperclip# 复制到剪贴板pyperclip.copy(要传递的数据)# 从剪贴板粘贴datapyperclip.paste()print(data)1.2 网页数据 → Excel剪贴板方式importpyperclipimporttimedefweb_to_excel_via_clipboard(page,excel_app):通过剪贴板从网页传递数据到Excel# 1. 从网页复制数据# 选中表格内容page.click(#data-table)page.keyboard.press(ControlA)# 全选page.keyboard.press(ControlC)# 复制time.sleep(0.5)# 2. 切换到Excelexcel_app.activate()time.sleep(1)# 3. 粘贴数据excel_app.type(^v)# CtrlVtime.sleep(1)# 4. 获取剪贴板内容用于验证clipboard_datapyperclip.paste()print(f已传递{len(clipboard_data)}个字符)1.3 剪贴板监控importpyperclipimporttimefromdatetimeimportdatetimeclassClipboardMonitor:剪贴板监控器def__init__(self,callbackNone):self.callbackcallback self.last_contentself.is_runningFalsedefstart(self):开始监控self.is_runningTrueself.last_contentpyperclip.paste()print(剪贴板监控已启动)whileself.is_running:currentpyperclip.paste()ifcurrent!self.last_content:self.last_contentcurrentprint(f[{datetime.now()}] 剪贴板内容已变更)ifself.callback:self.callback(current)time.sleep(0.5)defstop(self):self.is_runningFalse# 使用监控剪贴板变化defon_clipboard_change(content):剪贴板内容变更回调# 自动识别内容类型importreifre.match(r^\d{11}$,content):print(f → 检测到手机号:{content})elifre.match(r^[\w.-][\w-]\.[\w.]$,content):print(f → 检测到邮箱:{content})elifre.match(r^https?://,content):print(f → 检测到URL:{content})else:print(f → 普通文本:{content[:50]}...)monitorClipboardMonitor(callbackon_clipboard_change)方式二中间文件——最通用的数据传递2.1 CSV作为中间格式importcsvimportosclassCSVBridge:CSV中间文件数据桥接def__init__(self,bridge_dirbridge):self.bridge_dirbridge_dir os.makedirs(bridge_dir,exist_okTrue)defwrite(self,data,filenametransfer.csv):写入中间文件filepathos.path.join(self.bridge_dir,filename)withopen(filepath,w,newline,encodingutf-8-sig)asf:ifdataandisinstance(data,list)andisinstance(data[0],dict):writercsv.DictWriter(f,fieldnamesdata[0].keys())writer.writeheader()writer.writerows(data)elifdataandisinstance(data[0],list):writercsv.writer(f)writer.writerows(data)print(f中间文件已写入:{filepath})returnfilepathdefread(self,filenametransfer.csv):读取中间文件filepathos.path.join(self.bridge_dir,filename)data[]withopen(filepath,r,encodingutf-8-sig)asf:readercsv.DictReader(f)forrowinreader:data.append(dict(row))print(f已读取{len(data)}条数据)returndatadefis_ready(self,filenametransfer.csv):检查中间文件是否存在returnos.path.exists(os.path.join(self.bridge_dir,filename))# 使用bridgeCSVBridge()# 应用A写入数据bridge.write([{name:张三,amount:1000,date:2024-03-15},{name:李四,amount:2000,date:2024-03-16},])# 应用B读取数据databridge.read()2.2 JSON中间文件importjsonclassJSONBridge:JSON中间文件桥接支持复杂数据结构def__init__(self,bridge_dirbridge):self.bridge_dirbridge_dir os.makedirs(bridge_dir,exist_okTrue)defwrite(self,data,filenametransfer.json):filepathos.path.join(self.bridge_dir,filename)withopen(filepath,w,encodingutf-8)asf:json.dump(data,f,ensure_asciiFalse,indent2)returnfilepathdefread(self,filenametransfer.json):filepathos.path.join(self.bridge_dir,filename)withopen(filepath,r,encodingutf-8)asf:returnjson.load(f)# 传递复杂数据结构bridgeJSONBridge()bridge.write({orders:[{id:ORD001,items:[{sku:A001,qty:2},{sku:B002,qty:1}]},{id:ORD002,items:[{sku:C003,qty:5}]},],metadata:{source:ERP,export_time:2024-03-15 10:30:00}})方式三API接口——最规范的集成方式3.1 影刀RPA作为API客户端importrequestsimportjsonclassAPIClient:通用API客户端def__init__(self,base_url,headersNone):self.base_urlbase_url self.headersheadersor{}self.sessionrequests.Session()defget(self,endpoint,paramsNone):GET请求responseself.session.get(f{self.base_url}{endpoint},paramsparams,headersself.headers)returnresponse.json()defpost(self,endpoint,dataNone):POST请求responseself.session.post(f{self.base_url}{endpoint},jsondata,headersself.headers)returnresponse.json()# 使用从CRM获取客户数据crmAPIClient(base_urlhttps://api.crm.com/v1,headers{Authorization:Bearer your_token})# 查询客户customerscrm.get(/customers,params{page:1,size:100})# 写入到ERPerpAPIClient(base_urlhttps://api.erp.com/v1,headers{Authorization:Bearer erp_token})forcustomerincustomers[data]:erp.post(/customers,datacustomer)3.2 RPA流程间API通信fromhttp.serverimportHTTPServer,BaseHTTPRequestHandlerimportjsonimportthreadingclassFlowAPIHandler(BaseHTTPRequestHandler):流程间API通信处理器defdo_POST(self):content_lengthint(self.headers[Content-Length])post_dataself.rfile.read(content_length)try:datajson.loads(post_data)# 根据路径分发ifself.path/api/process:resultself._process_data(data)self._send_response(200,result)elifself.path/api/status:self._send_response(200,{status:running})else:self._send_response(404,{error:Not found})exceptExceptionase:self._send_response(500,{error:str(e)})def_process_data(self,data):# 处理数据的逻辑return{processed:True,count:len(data)}def_send_response(self,code,data):self.send_response(code)self.send_header(Content-Type,application/json)self.end_headers()self.wfile.write(json.dumps(data).encode())defstart_api_server(port8888):启动API服务serverHTTPServer((localhost,port),FlowAPIHandler)threadthreading.Thread(targetserver.serve_forever,daemonTrue)thread.start()print(fAPI服务已启动: http://localhost:{port})方式四数据库——最持久的数据传递temu店群自动化报活动案例4.1 数据库作为中间层importsqlite3classDataBridge:数据库中间层def__init__(self,db_pathbridge.db):self.connsqlite3.connect(db_path)self._init_db()def_init_db(self):self.conn.execute( CREATE TABLE IF NOT EXISTS data_transfer ( id INTEGER PRIMARY KEY AUTOINCREMENT, source TEXT, target TEXT, data_type TEXT, content TEXT, status TEXT DEFAULT pending, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, processed_at TIMESTAMP ) )self.conn.commit()defsend(self,source,target,data_type,content):发送数据self.conn.execute( INSERT INTO data_transfer (source, target, data_type, content) VALUES (?, ?, ?, ?) ,(source,target,data_type,json.dumps(content,ensure_asciiFalse)))self.conn.commit()defreceive(self,target,data_typeNone):接收数据conditiontarget ? AND status pendingparams[target]ifdata_type:condition AND data_type ?params.append(data_type)cursorself.conn.execute(f SELECT id, source, data_type, content FROM data_transfer WHERE{condition}ORDER BY created_at ASC ,params)messages[]forrowincursor:msg_id,source,dtype,contentrow messages.append({id:msg_id,source:source,data_type:dtype,content:json.loads(content)})# 标记为已处理self.conn.execute( UPDATE data_transfer SET statusprocessed, processed_atCURRENT_TIMESTAMP WHERE id? ,(msg_id,))self.conn.commit()returnmessages# 流程A发送数据bridgeDataBridge()bridge.send(采集流程,录入流程,customer_data,[{name:张三,phone:13812345678},{name:李四,phone:15987654321},])# 流程B接收数据databridge.receive(录入流程,customer_data)foritemindata:print(f来自{item[source]}:{len(item[content])}条数据)方式五消息队列——最高效的异步传递5.1 基于Redis的简单消息队列importredisimportjsonclassMessageQueue:Redis消息队列def__init__(self,hostlocalhost,port6379):self.redisredis.Redis(hosthost,portport,decode_responsesTrue)defpublish(self,channel,message):发布消息self.redis.publish(channel,json.dumps(message,ensure_asciiFalse))defsubscribe(self,channel,callback):订阅消息pubsubself.redis.pubsub()pubsub.subscribe(channel)formessageinpubsub.listen():ifmessage[type]message:datajson.loads(message[data])callback(data)defpush(self,queue_name,data):推入队列self.redis.lpush(queue_name,json.dumps(data,ensure_asciiFalse))defpop(self,queue_name,timeout30):弹出队列resultself.redis.brpop(queue_name,timeouttimeout)ifresult:_,dataresultreturnjson.loads(data)returnNone# 使用mqMessageQueue()# 流程A生产数据mq.push(order_queue,{order_id:ORD001,amount:1000})# 流程B消费数据whileTrue:datamq.pop(order_queue,timeout30)ifdata:print(f处理订单:{data[order_id]})else:print(队列为空等待中...)五种方式对比方式复杂度实时性数据量可靠性适用场景剪贴板⭐实时小低简单快速传递中间文件⭐⭐近实时大中离线批处理API⭐⭐⭐实时中高规范化集成数据库⭐⭐⭐近实时大高持久化传递消息队列⭐⭐⭐⭐实时大高高并发异步结语跨应用数据传递是RPA的核心能力。选择原则简单场景剪贴板/中间文件就够了规范集成优先用API持久化需求用数据库中间层高性能异步上消息队列从简单开始按需升级。好的架构不是最复杂的而是刚好够用的。
影刀RPA剪贴板操作：跨应用数据传递的5种方式
影刀RPA剪贴板操作跨应用数据传递的5种方式作者林焱RPA最大的价值之一就是打通不同应用之间的数据壁垒。本文详解剪贴板、文件、API、数据库、消息队列5种跨应用数据传递方式让你的机器人真正实现万物互联。前言为什么跨应用数据传递这么重要真实工作中数据很少只停留在一个应用里从网页采集数据 → 粘贴到Excel从ERP导出数据 → 导入CRM从邮件提取附件 → 上传到网盘从数据库查询结果 → 发送到微信群这些跨应用的数据传递就是RPA最核心的使用场景。方式一剪贴板——最直接的跨应用传递1.1 基本剪贴板操作店群矩阵自动化突破运营极限importpyperclip# 复制到剪贴板pyperclip.copy(要传递的数据)# 从剪贴板粘贴datapyperclip.paste()print(data)1.2 网页数据 → Excel剪贴板方式importpyperclipimporttimedefweb_to_excel_via_clipboard(page,excel_app):通过剪贴板从网页传递数据到Excel# 1. 从网页复制数据# 选中表格内容page.click(#data-table)page.keyboard.press(ControlA)# 全选page.keyboard.press(ControlC)# 复制time.sleep(0.5)# 2. 切换到Excelexcel_app.activate()time.sleep(1)# 3. 粘贴数据excel_app.type(^v)# CtrlVtime.sleep(1)# 4. 获取剪贴板内容用于验证clipboard_datapyperclip.paste()print(f已传递{len(clipboard_data)}个字符)1.3 剪贴板监控importpyperclipimporttimefromdatetimeimportdatetimeclassClipboardMonitor:剪贴板监控器def__init__(self,callbackNone):self.callbackcallback self.last_contentself.is_runningFalsedefstart(self):开始监控self.is_runningTrueself.last_contentpyperclip.paste()print(剪贴板监控已启动)whileself.is_running:currentpyperclip.paste()ifcurrent!self.last_content:self.last_contentcurrentprint(f[{datetime.now()}] 剪贴板内容已变更)ifself.callback:self.callback(current)time.sleep(0.5)defstop(self):self.is_runningFalse# 使用监控剪贴板变化defon_clipboard_change(content):剪贴板内容变更回调# 自动识别内容类型importreifre.match(r^\d{11}$,content):print(f → 检测到手机号:{content})elifre.match(r^[\w.-][\w-]\.[\w.]$,content):print(f → 检测到邮箱:{content})elifre.match(r^https?://,content):print(f → 检测到URL:{content})else:print(f → 普通文本:{content[:50]}...)monitorClipboardMonitor(callbackon_clipboard_change)方式二中间文件——最通用的数据传递2.1 CSV作为中间格式importcsvimportosclassCSVBridge:CSV中间文件数据桥接def__init__(self,bridge_dirbridge):self.bridge_dirbridge_dir os.makedirs(bridge_dir,exist_okTrue)defwrite(self,data,filenametransfer.csv):写入中间文件filepathos.path.join(self.bridge_dir,filename)withopen(filepath,w,newline,encodingutf-8-sig)asf:ifdataandisinstance(data,list)andisinstance(data[0],dict):writercsv.DictWriter(f,fieldnamesdata[0].keys())writer.writeheader()writer.writerows(data)elifdataandisinstance(data[0],list):writercsv.writer(f)writer.writerows(data)print(f中间文件已写入:{filepath})returnfilepathdefread(self,filenametransfer.csv):读取中间文件filepathos.path.join(self.bridge_dir,filename)data[]withopen(filepath,r,encodingutf-8-sig)asf:readercsv.DictReader(f)forrowinreader:data.append(dict(row))print(f已读取{len(data)}条数据)returndatadefis_ready(self,filenametransfer.csv):检查中间文件是否存在returnos.path.exists(os.path.join(self.bridge_dir,filename))# 使用bridgeCSVBridge()# 应用A写入数据bridge.write([{name:张三,amount:1000,date:2024-03-15},{name:李四,amount:2000,date:2024-03-16},])# 应用B读取数据databridge.read()2.2 JSON中间文件importjsonclassJSONBridge:JSON中间文件桥接支持复杂数据结构def__init__(self,bridge_dirbridge):self.bridge_dirbridge_dir os.makedirs(bridge_dir,exist_okTrue)defwrite(self,data,filenametransfer.json):filepathos.path.join(self.bridge_dir,filename)withopen(filepath,w,encodingutf-8)asf:json.dump(data,f,ensure_asciiFalse,indent2)returnfilepathdefread(self,filenametransfer.json):filepathos.path.join(self.bridge_dir,filename)withopen(filepath,r,encodingutf-8)asf:returnjson.load(f)# 传递复杂数据结构bridgeJSONBridge()bridge.write({orders:[{id:ORD001,items:[{sku:A001,qty:2},{sku:B002,qty:1}]},{id:ORD002,items:[{sku:C003,qty:5}]},],metadata:{source:ERP,export_time:2024-03-15 10:30:00}})方式三API接口——最规范的集成方式3.1 影刀RPA作为API客户端importrequestsimportjsonclassAPIClient:通用API客户端def__init__(self,base_url,headersNone):self.base_urlbase_url self.headersheadersor{}self.sessionrequests.Session()defget(self,endpoint,paramsNone):GET请求responseself.session.get(f{self.base_url}{endpoint},paramsparams,headersself.headers)returnresponse.json()defpost(self,endpoint,dataNone):POST请求responseself.session.post(f{self.base_url}{endpoint},jsondata,headersself.headers)returnresponse.json()# 使用从CRM获取客户数据crmAPIClient(base_urlhttps://api.crm.com/v1,headers{Authorization:Bearer your_token})# 查询客户customerscrm.get(/customers,params{page:1,size:100})# 写入到ERPerpAPIClient(base_urlhttps://api.erp.com/v1,headers{Authorization:Bearer erp_token})forcustomerincustomers[data]:erp.post(/customers,datacustomer)3.2 RPA流程间API通信fromhttp.serverimportHTTPServer,BaseHTTPRequestHandlerimportjsonimportthreadingclassFlowAPIHandler(BaseHTTPRequestHandler):流程间API通信处理器defdo_POST(self):content_lengthint(self.headers[Content-Length])post_dataself.rfile.read(content_length)try:datajson.loads(post_data)# 根据路径分发ifself.path/api/process:resultself._process_data(data)self._send_response(200,result)elifself.path/api/status:self._send_response(200,{status:running})else:self._send_response(404,{error:Not found})exceptExceptionase:self._send_response(500,{error:str(e)})def_process_data(self,data):# 处理数据的逻辑return{processed:True,count:len(data)}def_send_response(self,code,data):self.send_response(code)self.send_header(Content-Type,application/json)self.end_headers()self.wfile.write(json.dumps(data).encode())defstart_api_server(port8888):启动API服务serverHTTPServer((localhost,port),FlowAPIHandler)threadthreading.Thread(targetserver.serve_forever,daemonTrue)thread.start()print(fAPI服务已启动: http://localhost:{port})方式四数据库——最持久的数据传递temu店群自动化报活动案例4.1 数据库作为中间层importsqlite3classDataBridge:数据库中间层def__init__(self,db_pathbridge.db):self.connsqlite3.connect(db_path)self._init_db()def_init_db(self):self.conn.execute( CREATE TABLE IF NOT EXISTS data_transfer ( id INTEGER PRIMARY KEY AUTOINCREMENT, source TEXT, target TEXT, data_type TEXT, content TEXT, status TEXT DEFAULT pending, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, processed_at TIMESTAMP ) )self.conn.commit()defsend(self,source,target,data_type,content):发送数据self.conn.execute( INSERT INTO data_transfer (source, target, data_type, content) VALUES (?, ?, ?, ?) ,(source,target,data_type,json.dumps(content,ensure_asciiFalse)))self.conn.commit()defreceive(self,target,data_typeNone):接收数据conditiontarget ? AND status pendingparams[target]ifdata_type:condition AND data_type ?params.append(data_type)cursorself.conn.execute(f SELECT id, source, data_type, content FROM data_transfer WHERE{condition}ORDER BY created_at ASC ,params)messages[]forrowincursor:msg_id,source,dtype,contentrow messages.append({id:msg_id,source:source,data_type:dtype,content:json.loads(content)})# 标记为已处理self.conn.execute( UPDATE data_transfer SET statusprocessed, processed_atCURRENT_TIMESTAMP WHERE id? ,(msg_id,))self.conn.commit()returnmessages# 流程A发送数据bridgeDataBridge()bridge.send(采集流程,录入流程,customer_data,[{name:张三,phone:13812345678},{name:李四,phone:15987654321},])# 流程B接收数据databridge.receive(录入流程,customer_data)foritemindata:print(f来自{item[source]}:{len(item[content])}条数据)方式五消息队列——最高效的异步传递5.1 基于Redis的简单消息队列importredisimportjsonclassMessageQueue:Redis消息队列def__init__(self,hostlocalhost,port6379):self.redisredis.Redis(hosthost,portport,decode_responsesTrue)defpublish(self,channel,message):发布消息self.redis.publish(channel,json.dumps(message,ensure_asciiFalse))defsubscribe(self,channel,callback):订阅消息pubsubself.redis.pubsub()pubsub.subscribe(channel)formessageinpubsub.listen():ifmessage[type]message:datajson.loads(message[data])callback(data)defpush(self,queue_name,data):推入队列self.redis.lpush(queue_name,json.dumps(data,ensure_asciiFalse))defpop(self,queue_name,timeout30):弹出队列resultself.redis.brpop(queue_name,timeouttimeout)ifresult:_,dataresultreturnjson.loads(data)returnNone# 使用mqMessageQueue()# 流程A生产数据mq.push(order_queue,{order_id:ORD001,amount:1000})# 流程B消费数据whileTrue:datamq.pop(order_queue,timeout30)ifdata:print(f处理订单:{data[order_id]})else:print(队列为空等待中...)五种方式对比方式复杂度实时性数据量可靠性适用场景剪贴板⭐实时小低简单快速传递中间文件⭐⭐近实时大中离线批处理API⭐⭐⭐实时中高规范化集成数据库⭐⭐⭐近实时大高持久化传递消息队列⭐⭐⭐⭐实时大高高并发异步结语跨应用数据传递是RPA的核心能力。选择原则简单场景剪贴板/中间文件就够了规范集成优先用API持久化需求用数据库中间层高性能异步上消息队列从简单开始按需升级。好的架构不是最复杂的而是刚好够用的。
