影刀RPA剪贴板操作：跨应用数据传递的5种方式
作者：林焱
RPA最大的价值之一就是打通不同应用之间的数据壁垒。本文详解剪贴板、文件、API、数据库、消息队列5种跨应用数据传递方式，让你的机器人真正实现"万物互联"。
前言：为什么跨应用数据传递这么重要？
真实工作中，数据很少只停留在一个应用里：
- 从网页采集数据 → 粘贴到Excel
- 从ERP导出数据 → 导入CRM
- 从邮件提取附件 → 上传到网盘
- 从数据库查询结果 → 发送到微信群
这些跨应用的数据传递，就是RPA最核心的使用场景。
方式一：剪贴板——最直接的跨应用传递
1.1 基本剪贴板操作
店群矩阵自动化突破运营极限！
importpyperclip# 复制到剪贴板pyperclip.copy("要传递的数据")# 从剪贴板粘贴data=pyperclip.paste()print(data)1.2 网页数据 → Excel（剪贴板方式）
importpyperclipimporttimedefweb_to_excel_via_clipboard(page,excel_app):"""通过剪贴板从网页传递数据到Excel"""# 1. 从网页复制数据# 选中表格内容page.click("#data-table")page.keyboard.press("Control+A")# 全选page.keyboard.press("Control+C")# 复制time.sleep(0.5)# 2. 切换到Excelexcel_app.activate()time.sleep(1)# 3. 粘贴数据excel_app.type("^v")# Ctrl+Vtime.sleep(1)# 4. 获取剪贴板内容（用于验证）clipboard_data=pyperclip.paste()print(f"已传递{len(clipboard_data)}个字符")1.3 剪贴板监控
importpyperclipimporttimefromdatetimeimportdatetimeclassClipboardMonitor:"""剪贴板监控器"""def__init__(self,callback=None):self.callback=callback self.last_content=""self.is_running=Falsedefstart(self):"""开始监控"""self.is_running=Trueself.last_content=pyperclip.paste()print("剪贴板监控已启动")whileself.is_running:current=pyperclip.paste()ifcurrent!=self.last_content:self.last_content=currentprint(f"[{datetime.now()}] 剪贴板内容已变更")ifself.callback:self.callback(current)time.sleep(0.5)defstop(self):self.is_running=False# 使用：监控剪贴板变化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]}...")monitor=ClipboardMonitor(callback=on_clipboard_change)方式二：中间文件——最通用的数据传递
2.1 CSV作为中间格式
importcsvimportosclassCSVBridge:"""CSV中间文件数据桥接"""def__init__(self,bridge_dir="bridge"):self.bridge_dir=bridge_dir os.makedirs(bridge_dir,exist_ok=True)defwrite(self,data,filename="transfer.csv"):"""写入中间文件"""filepath=os.path.join(self.bridge_dir,filename)withopen(filepath,"w",newline="",encoding="utf-8-sig")asf:ifdataandisinstance(data,list)andisinstance(data[0],dict):writer=csv.DictWriter(f,fieldnames=data[0].keys())writer.writeheader()writer.writerows(data)elifdataandisinstance(data[0],list):writer=csv.writer(f)writer.writerows(data)print(f"中间文件已写入:{filepath}")returnfilepathdefread(self,filename="transfer.csv"):"""读取中间文件"""filepath=os.path.join(self.bridge_dir,filename)data=[]withopen(filepath,"r",encoding="utf-8-sig")asf:reader=csv.DictReader(f)forrowinreader:data.append(dict(row))print(f"已读取{len(data)}条数据")returndatadefis_ready(self,filename="transfer.csv"):"""检查中间文件是否存在"""returnos.path.exists(os.path.join(self.bridge_dir,filename))# 使用bridge=CSVBridge()# 应用A：写入数据bridge.write([{"name":"张三","amount":1000,"date":"2024-03-15"},{"name":"李四","amount":2000,"date":"2024-03-16"},])# 应用B：读取数据data=bridge.read()2.2 JSON中间文件
importjsonclassJSONBridge:"""JSON中间文件桥接，支持复杂数据结构"""def__init__(self,bridge_dir="bridge"):self.bridge_dir=bridge_dir os.makedirs(bridge_dir,exist_ok=True)defwrite(self,data,filename="transfer.json"):filepath=os.path.join(self.bridge_dir,filename)withopen(filepath,"w",encoding="utf-8")asf:json.dump(data,f,ensure_ascii=False,indent=2)returnfilepathdefread(self,filename="transfer.json"):filepath=os.path.join(self.bridge_dir,filename)withopen(filepath,"r",encoding="utf-8")asf:returnjson.load(f)# 传递复杂数据结构bridge=JSONBridge()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,headers=None):self.base_url=base_url self.headers=headersor{}self.session=requests.Session()defget(self,endpoint,params=None):"""GET请求"""response=self.session.get(f"{self.base_url}{endpoint}",params=params,headers=self.headers)returnresponse.json()defpost(self,endpoint,data=None):"""POST请求"""response=self.session.post(f"{self.base_url}{endpoint}",json=data,headers=self.headers)returnresponse.json()# 使用：从CRM获取客户数据crm=APIClient(base_url="https://api.crm.com/v1",headers={"Authorization":"Bearer your_token"})# 查询客户customers=crm.get("/customers",params={"page":1,"size":100})# 写入到ERPerp=APIClient(base_url="https://api.erp.com/v1",headers={"Authorization":"Bearer erp_token"})forcustomerincustomers["data"]:erp.post("/customers",data=customer)3.2 RPA流程间API通信
fromhttp.serverimportHTTPServer,BaseHTTPRequestHandlerimportjsonimportthreadingclassFlowAPIHandler(BaseHTTPRequestHandler):"""流程间API通信处理器"""defdo_POST(self):content_length=int(self.headers['Content-Length'])post_data=self.rfile.read(content_length)try:data=json.loads(post_data)# 根据路径分发ifself.path=="/api/process":result=self._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(port=8888):"""启动API服务"""server=HTTPServer(('localhost',port),FlowAPIHandler)thread=threading.Thread(target=server.serve_forever,daemon=True)thread.start()print(f"API服务已启动: http://localhost:{port}")方式四：数据库——最持久的数据传递
temu店群自动化报活动案例
4.1 数据库作为中间层
importsqlite3classDataBridge:"""数据库中间层"""def__init__(self,db_path="bridge.db"):self.conn=sqlite3.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_ascii=False)))self.conn.commit()defreceive(self,target,data_type=None):"""接收数据"""condition="target = ? AND status = 'pending'"params=[target]ifdata_type:condition+=" AND data_type = ?"params.append(data_type)cursor=self.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,content=row messages.append({"id":msg_id,"source":source,"data_type":dtype,"content":json.loads(content)})# 标记为已处理self.conn.execute(""" UPDATE data_transfer SET status='processed', processed_at=CURRENT_TIMESTAMP WHERE id=? """,(msg_id,))self.conn.commit()returnmessages# 流程A：发送数据bridge=DataBridge()bridge.send("采集流程","录入流程","customer_data",[{"name":"张三","phone":"13812345678"},{"name":"李四","phone":"15987654321"},])# 流程B：接收数据data=bridge.receive("录入流程","customer_data")foritemindata:print(f"来自{item['source']}:{len(item['content'])}条数据")方式五：消息队列——最高效的异步传递
5.1 基于Redis的简单消息队列
importredisimportjsonclassMessageQueue:"""Redis消息队列"""def__init__(self,host="localhost",port=6379):self.redis=redis.Redis(host=host,port=port,decode_responses=True)defpublish(self,channel,message):"""发布消息"""self.redis.publish(channel,json.dumps(message,ensure_ascii=False))defsubscribe(self,channel,callback):"""订阅消息"""pubsub=self.redis.pubsub()pubsub.subscribe(channel)formessageinpubsub.listen():ifmessage["type"]=="message":data=json.loads(message["data"])callback(data)defpush(self,queue_name,data):"""推入队列"""self.redis.lpush(queue_name,json.dumps(data,ensure_ascii=False))defpop(self,queue_name,timeout=30):"""弹出队列"""result=self.redis.brpop(queue_name,timeout=timeout)ifresult:_,data=resultreturnjson.loads(data)returnNone# 使用mq=MessageQueue()# 流程A：生产数据mq.push("order_queue",{"order_id":"ORD001","amount":1000})# 流程B：消费数据whileTrue:data=mq.pop("order_queue",timeout=30)ifdata:print(f"处理订单:{data['order_id']}")else:print("队列为空，等待中...")五种方式对比
| 方式 | 复杂度 | 实时性 | 数据量 | 可靠性 | 适用场景 |
|---|---|---|---|---|---|
| 剪贴板 | ⭐ | 实时 | 小 | 低 | 简单快速传递 |
| 中间文件 | ⭐⭐ | 近实时 | 大 | 中 | 离线批处理 |
| API | ⭐⭐⭐ | 实时 | 中 | 高 | 规范化集成 |
| 数据库 | ⭐⭐⭐ | 近实时 | 大 | 高 | 持久化传递 |
| 消息队列 | ⭐⭐⭐⭐ | 实时 | 大 | 高 | 高并发异步 |
结语
跨应用数据传递是RPA的核心能力。选择原则：
- 简单场景：剪贴板/中间文件就够了
- 规范集成：优先用API
- 持久化需求：用数据库中间层
- 高性能异步：上消息队列
从简单开始，按需升级。好的架构不是最复杂的，而是刚好够用的。
)
)
