影刀RPA店群自动化教程：Python协同浏览器标签页工作区与多店铺并行隔离实战
一个店铺开了八个标签页，三个店铺就是二十四个标签页。
浏览器在八十个标签页面前瑟瑟发抖，而你还在手动切来切去找那个“已付款”的订单。
店群矩阵自动化突破运营极限！
店群运营的日常有一个很容易被忽略的场景：每个店铺后台，通常需要同时打开好几个页面——商品列表、订单管理、客服聊天、活动报名、数据看板。
当店铺数量增多，自动化执行时标签页管理就会失控。即使我们已经用浏览器实例池隔离了店铺进程，但同一个店铺内的多页面操作仍然混乱。
我们曾经遇到过这样的线上事故：一个店铺的自动回复流程正在操作客服页面，另一个采集流程却在同一个浏览器里把当前标签页切走了，导致回复消息发到了商品编辑框里。
于是我们设计了一套基于Chrome DevTools Protocol的标签页工作区管理系统，把每个店铺的标签页都装进独立的“工作区”，互不干扰。
这篇文章就展开这套标签页分组、工作区快照、崩溃隔离与并行操作的工程实践。
一、标签页工作区的核心概念
我们把“工作区”定义为一个店铺在浏览器实例内打开的一组标签页的集合，包括它们的URL、激活状态、滚动位置、表单草稿等上下文信息。
temu店群自动化报活动案例
每个店铺的工作区逻辑上相互隔离：自动化流程在操作时，只在自己工作区内的标签页之间切换，不会误入其他店铺的页面。
工作区由Python调度代理通过CDP协议管理，影刀RPA流程只需要关心当前聚焦的标签页。
二、工作区的创建与标签页分组
浏览器启动后，Python代理通过CDP创建标签页，并将它们归入工作区。利用Chrome的targetId和标签页事件进行管理。
importasynciofromdataclassesimportdataclass,fieldfromtypingimportDict,List,Optional@dataclassclassTabInfo:tab_id:str# CDP targetIdurl:str="about:blank"title:str=""scroll_x:float=0.0scroll_y:float=0.0form_data:dict=field(default_factory=dict)is_active:bool=Falselast_accessed:float=0.0@dataclassclassWorkspace:shop_id:strplatform:strtabs:Dict[str,TabInfo]=field(default_factory=dict)active_tab_id:Optional[str]=NoneclassWorkspaceManager:def__init__(self,cdp_session):self.cdp=cdp_session self.workspaces:Dict[str,Workspace]={}asyncdefcreate_workspace(self,shop_id:str,platform:str,urls:List[str])->Workspace:ws=Workspace(shop_id=shop_id,platform=platform)forurlinurls:tab_info=awaitself._create_tab(url)ws.tabs[tab_info.tab_id]=tab_info self.workspaces[shop_id]=ws# 默认激活第一个标签页first_tab=list(ws.tabs.values())[0]ifws.tabselseNoneiffirst_tab:awaitself.cdp.activate_tab(first_tab.tab_id)ws.active_tab_id=first_tab.tab_id first_tab.is_active=Truereturnwsasyncdef_create_tab(self,url:str)->TabInfo:target_id=awaitself.cdp.create_target(url)tab=TabInfo(tab_id=target_id,url=url)returntabasyncdefswitch_tab(self,shop_id:str,tab_id:str):ws=self.workspaces.get(shop_id)ifnotwsortab_idnotinws.tabs:raiseTabNotFoundError# 保存离开的标签页状态ifws.active_tab_id:awaitself._save_tab_state(ws,ws.active_tab_id)awaitself.cdp.activate_tab(tab_id)ws.tabs[tab_id].is_active=Truews.tabs[tab_id].last_accessed=asyncio.get_event_loop().time()ifws.active_tab_idandws.active_tab_id!=tab_id:ws.tabs[ws.active_tab_id].is_active=Falsews.active_tab_id=tab_id ``` 每个店铺的工作区在创建时就可定义所需的URL集合，如：`["https://seller.pdd.com/order/list","https://seller.pdd.com/product/list","https://seller.pdd.com/im"]`。 后续任务中，影刀流程需要切换到哪个功能页面，只需调用`switch_tab`，Python代理会静默完成切换并恢复页面上下文。---## 三、工作区状态快照与自动恢复标签页的页面状态在流程运行中可能丢失：浏览器崩溃、进程重启、标签页被意外关闭。 我们实现了工作区状态的自动保存和恢复。每当标签页切换、或定时每隔30秒，都会捕获当前标签页的URL、滚动位置和表单草稿。 ```pythonasyncdef_save_tab_state(self,ws:Workspace,tab_id:str):tab=ws.tabs.get(tab_id)ifnottab:return# 获取当前URL和滚动位置tab.url=awaitself.cdp.evaluate("window.location.href",tab_id=tab_id)scroll=awaitself.cdp.evaluate("JSON.stringify({x: window.scrollX, y: window.scrollY})",tab_id=tab_id)scroll_data=json.loads(scroll)tab.scroll_x=scroll_data["x"]tab.scroll_y=scroll_data["y"]# 获取表单数据（简易示例：只保存可见输入框的值）form_json=awaitself.cdp.evaluate(""" JSON.stringify( Array.from(document.querySelectorAll('input:not([type="hidden"]), textarea')) .reduce((acc, el) => { acc[el.name || el.id || el.className] = el.value; return acc; }, {}) ) """,tab_id=tab_id)tab.form_data=json.loads(form_json)ifform_jsonelse{}asyncdefrestore_workspace(self,shop_id:str):ws=self.workspaces.get(shop_id)ifnotws:returnfortab_id,tabinws.tabs.items():# 检查标签页是否仍存在ifnotawaitself.cdp.target_exists(tab_id):# 重新创建标签页并导航到保存的URLnew_id=awaitself.cdp.create_target(tab.url)tab.tab_id=new_id ws.tabs[new_id]=tabdelws.tabs[tab_id]else:# 如果URL发生变化（用户可能手动导航），则重新加载保存的URLcurrent_url=awaitself.cdp.evaluate("window.location.href",tab_id=tab.tab_id)ifcurrent_url!=tab.url:awaitself.cdp.navigate(tab.tab_id,tab.url)# 恢复滚动位置awaitself.cdp.evaluate(f"window.scrollTo({tab.scroll_x},{tab.scroll_y})",tab_id=tab.tab_id)# 恢复表单数据forselector,valueintab.form_data.items():try:awaitself.cdp.evaluate(f"""document.querySelector('[name="{selector}"], [id="{selector}"]).value ={json.dumps(value)}""",tab_id=tab.tab_id)except:pass``` 当Worker重启或浏览器进程崩溃后，工作区恢复可以让店铺操作继续进行，而不需要从登录开始重新走流程。 我们曾经在一次内存泄漏引发的浏览器重启中，用这套机制在30秒内恢复了四个店铺的工作区，避免了上货任务中断。---## 四、标签页隔离与崩溃隔离在同一个浏览器实例内，多个标签页共享进程资源，一个页面的JS死循环或内存泄漏可能影响整个浏览器。 我们采用Chrome的“Site Isolation”和标签页崩溃检测来做隔离。-启用 `--site-per-process` 启动参数，强制每个站点独立进程。--Python代理定期检测每个标签页的健康状态：通过CDP发送心跳请求，如果无响应则标记为崩溃。--崩溃的标签页自动重载，并尝试恢复之前的状态。 ```pythonasyncdefhealth_check_loop(self):whileTrue:forshop_id,wsinself.workspaces.items():fortab_id,tabinws.tabs.items():try:awaitasyncio.wait_for(self.cdp.send("Runtime.evaluate",{"expression":"1+1"},tab_id=tab_id),timeout=5.0)except:logger.warning(f"Tab{tab_id}(shop{shop_id}) appears crashed, reloading...")awaitself._recover_tab(ws,tab_id)awaitasyncio.sleep(15)asyncdef_recover_tab(self,ws:Workspace,tab_id:str):tab=ws.tabs.get(tab_id)ifnottab:returnawaitself.cdp.close_target(tab_id)new_id=awaitself.cdp.create_target(tab.url)tab.tab_id=new_id ws.tabs[new_id]=tabdelws.tabs[tab_id]# 如果崩溃的是活跃标签页，重新激活ifws.active_tab_id==tab_id:ws.active_tab_id=new_idawaitself.cdp.activate_tab(new_id)``` 这样，一个店铺的客服页面崩溃，不会导致同浏览器里其他店铺的商品列表页面挂掉。---## 五、与影刀RPA操作的协同影刀流程在工作区模式下，操作的是当前激活的标签页。Python代理负责在影刀指令执行前，确保焦点正确的标签页。 我们为指令执行器增加了工作区感知： ```pythonclassWorkspaceAwareExecutor:def__init__(self,workspace_manager,browser):self.ws_manager=workspace_manager self.browser=browserasyncdefexecute_step(self,shop_id:str,step:dict):# 如果步骤指定了目标功能页面，先切换标签页if"target_page"instep:target_tab_id=awaitself.ws_manager.get_tab_for_page(shop_id,step["target_page"])iftarget_tab_id:awaitself.ws_manager.switch_tab(shop_id,target_tab_id)# 执行具体操作ifstep["action"]=="click":awaitself.browser.click(step["locator"])elifstep["action"]=="type_text":awaitself.browser.type_text(step["locator"],step["value_from"])# ...``` 影刀流程录制时，可以为每个操作选择“所属功能页面”，在指令配置里标记`target_page`。 运行时，标签页切换对影刀透明，它始终感觉自己在一个单一页面里操作。---## 六、并行操作：同一店铺多个标签页各司其职有了工作区，我们可以在一个店铺内利用不同标签页并行执行无依赖的任务。 比如：标签页A进行“商品上架”，标签页B同时“查看订单物流状态”。它们通过不同的标签页操作，互不干扰。 Python代理使用`asyncio`并发控制这两个指令序列，每个序列绑定各自的目标标签页。 ```pythonasyncdefrun_parallel_steps(shop_id:str,steps_a:list,steps_b:list):task_a=asyncio.create_task(execute_step_sequence(shop_id,steps_a,page="upload"))task_b=asyncio.create_task(execute_step_sequence(shop_id,steps_b,page="orders"))awaitasyncio.gather(task_a,task_b)``` 这样，单个店铺的任务吞吐量翻倍，浏览器资源利用更充分。---## 七、标签页休眠与内存回收几十个标签页同时活跃，内存会迅速攀升。我们对非活跃标签页启用Chromium的自动丢弃（discard）机制。 ```pythonasyncdefdiscard_idle_tabs(self,idle_seconds=300):now=time.time()forwsinself.workspaces.values():fortab_id,tabinws.tabs.items():iftab_id!=ws.active_tab_idand(now-tab.last_accessed)>idle_seconds:awaitself.cdp.send("Page.discard",{},tab_id=tab_id)logger.info(f"Discarded tab{tab_id}for shop{ws.shop_id}")``` 被丢弃的标签页在下次切换时会自动重新加载，但URL和表单状态已保存在快照中，恢复体验平顺。---## 八、监控与度量-各店铺工作区标签页数量--标签页崩溃恢复次数--工作区切换延迟--休眠标签页占比--表单恢复成功率 这些指标呈现在Grafana面板中，帮助判断标签页管理的健康度。---## 九、踩坑记录**CDP并发限制。**同一时间向同一个标签页发送多个CDP命令可能导致`Target closed`错误。我们为每个标签页操作加了`asyncio.Lock`，串行化CDP命令，避免竞争。**页面自动跳转破坏工作区。**某些平台页面在闲置时会自动跳转到登录页或首页。我们在恢复时检测URL是否剧烈变化，如果是则重新导航到目标页，并记录异常。**标签页被用户手动关闭。**虽然执行机不直接交互，但偶尔运维远程桌面时误操作关闭了标签页。我们通过崩溃检测与恢复机制一并覆盖。---## 十、写在最后浏览器标签页的精细化管理，是店群自动化执行层从“可用”走向“精良”的必经之路。 工作区让每个店铺在浏览器中拥有一块独立、稳定、可恢复的操作空间，即使页面崩溃也能自动站起来继续工作。>自动化最难的不是写出能点击的脚本，而是让几百个页面在无人值守的深夜里，依然井然有序。---*作者：林焱*
