影刀RPA店群自动化教程：Python协同浏览器标签页工作区与多店铺并行隔离实战
·
影刀RPA店群自动化教程：Python协同浏览器标签页工作区与多店铺并行隔离实战
一个店铺开了八个标签页，三个店铺就是二十四个标签页。
浏览器在八十个标签页面前瑟瑟发抖，而你还在手动切来切去找那个“已付款”的订单。
店群矩阵自动化突破运营极限！
店群运营的日常有一个很容易被忽略的场景：每个店铺后台，通常需要同时打开好几个页面——商品列表、订单管理、客服聊天、活动报名、数据看板。
当店铺数量增多，自动化执行时标签页管理就会失控。即使我们已经用浏览器实例池隔离了店铺进程，但同一个店铺内的多页面操作仍然混乱。
我们曾经遇到过这样的线上事故：一个店铺的自动回复流程正在操作客服页面，另一个采集流程却在同一个浏览器里把当前标签页切走了，导致回复消息发到了商品编辑框里。
于是我们设计了一套基于Chrome DevTools Protocol的标签页工作区管理系统，把每个店铺的标签页都装进独立的“工作区”，互不干扰。
这篇文章就展开这套标签页分组、工作区快照、崩溃隔离与并行操作的工程实践。
一、标签页工作区的核心概念
我们把“工作区”定义为一个店铺在浏览器实例内打开的一组标签页的集合，包括它们的URL、激活状态、滚动位置、表单草稿等上下文信息。
temu店群自动化报活动案例
每个店铺的工作区逻辑上相互隔离：自动化流程在操作时，只在自己工作区内的标签页之间切换，不会误入其他店铺的页面。
工作区由Python调度代理通过CDP协议管理，影刀RPA流程只需要关心当前聚焦的标签页。
二、工作区的创建与标签页分组
浏览器启动后，Python代理通过CDP创建标签页，并将它们归入工作区。利用Chrome的targetId和标签页事件进行管理。
import asyncio
from dataclasses import dataclass, field
from typing import Dict, List, Optional
@dataclass
class TabInfo:
tab_id: str # CDP targetId
url: str = "about:blank"
title: str = ""
scroll_x: float = 0.0
scroll_y: float = 0.0
form_data: dict = field(default_factory=dict)
is_active: bool = False
last_accessed: float = 0.0
@dataclass
class Workspace:
shop_id: str
platform: str
tabs: Dict[str, TabInfo] = field(default_factory=dict)
active_tab_id: Optional[str] = None
class WorkspaceManager:
def __init__(self, cdp_session):
self.cdp = cdp_session
self.workspaces: Dict[str, Workspace] = {}
async def create_workspace(self, shop_id: str, platform: str, urls: List[str]) -> Workspace:
ws = Workspace(shop_id=shop_id, platform=platform)
for url in urls:
tab_info = await self._create_tab(url)
ws.tabs[tab_info.tab_id] = tab_info
self.workspaces[shop_id] = ws
# 默认激活第一个标签页
first_tab = list(ws.tabs.values())[0] if ws.tabs else None
if first_tab:
await self.cdp.activate_tab(first_tab.tab_id)
ws.active_tab_id = first_tab.tab_id
first_tab.is_active = True
return ws
async def _create_tab(self, url: str) -> TabInfo:
target_id = await self.cdp.create_target(url)
tab = TabInfo(tab_id=target_id, url=url)
return tab
async def switch_tab(self, shop_id: str, tab_id: str):
ws = self.workspaces.get(shop_id)
if not ws or tab_id not in ws.tabs:
raise TabNotFoundError
# 保存离开的标签页状态
if ws.active_tab_id:
await self._save_tab_state(ws, ws.active_tab_id)
await self.cdp.activate_tab(tab_id)
ws.tabs[tab_id].is_active = True
ws.tabs[tab_id].last_accessed = asyncio.get_event_loop().time()
if ws.active_tab_id and ws.active_tab_id != tab_id:
ws.tabs[ws.active_tab_id].is_active = False
ws.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、滚动位置和表单草稿。
```python
async def _save_tab_state(self, ws: Workspace, tab_id: str):
tab = ws.tabs.get(tab_id)
if not tab:
return
# 获取当前URL和滚动位置
tab.url = await self.cdp.evaluate("window.location.href", tab_id=tab_id)
scroll = await self.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 = await self.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) if form_json else {}
async def restore_workspace(self, shop_id: str):
ws = self.workspaces.get(shop_id)
if not ws:
return
for tab_id, tab in ws.tabs.items():
# 检查标签页是否仍存在
if not await self.cdp.target_exists(tab_id):
# 重新创建标签页并导航到保存的URL
new_id = await self.cdp.create_target(tab.url)
tab.tab_id = new_id
ws.tabs[new_id] = tab
del ws.tabs[tab_id]
else:
# 如果URL发生变化（用户可能手动导航），则重新加载保存的URL
current_url = await self.cdp.evaluate("window.location.href", tab_id=tab.tab_id)
if current_url != tab.url:
await self.cdp.navigate(tab.tab_id, tab.url)
# 恢复滚动位置
await self.cdp.evaluate(
f"window.scrollTo({tab.scroll_x}, {tab.scroll_y})", tab_id=tab.tab_id
)
# 恢复表单数据
for selector, value in tab.form_data.items():
try:
await self.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发送心跳请求，如果无响应则标记为崩溃。
- - 崩溃的标签页自动重载，并尝试恢复之前的状态。
```python
async def health_check_loop(self):
while True:
for shop_id, ws in self.workspaces.items():
for tab_id, tab in ws.tabs.items():
try:
await asyncio.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...")
await self._recover_tab(ws, tab_id)
await asyncio.sleep(15)
async def _recover_tab(self, ws: Workspace, tab_id: str):
tab = ws.tabs.get(tab_id)
if not tab:
return
await self.cdp.close_target(tab_id)
new_id = await self.cdp.create_target(tab.url)
tab.tab_id = new_id
ws.tabs[new_id] = tab
del ws.tabs[tab_id]
# 如果崩溃的是活跃标签页，重新激活
if ws.active_tab_id == tab_id:
ws.active_tab_id = new_id
await self.cdp.activate_tab(new_id)
```
这样，一个店铺的客服页面崩溃，不会导致同浏览器里其他店铺的商品列表页面挂掉。
---
## 五、与影刀RPA操作的协同
影刀流程在工作区模式下，操作的是当前激活的标签页。Python代理负责在影刀指令执行前，确保焦点正确的标签页。
我们为指令执行器增加了工作区感知：
```python
class WorkspaceAwareExecutor:
def __init__(self, workspace_manager, browser):
self.ws_manager = workspace_manager
self.browser = browser
async def execute_step(self, shop_id: str, step: dict):
# 如果步骤指定了目标功能页面，先切换标签页
if "target_page" in step:
target_tab_id = await self.ws_manager.get_tab_for_page(shop_id, step["target_page"])
if target_tab_id:
await self.ws_manager.switch_tab(shop_id, target_tab_id)
# 执行具体操作
if step["action"] == "click":
await self.browser.click(step["locator"])
elif step["action"] == "type_text":
await self.browser.type_text(step["locator"], step["value_from"])
# ...
```
影刀流程录制时，可以为每个操作选择“所属功能页面”，在指令配置里标记`target_page`。
运行时，标签页切换对影刀透明，它始终感觉自己在一个单一页面里操作。
---
## 六、并行操作：同一店铺多个标签页各司其职
有了工作区，我们可以在一个店铺内利用不同标签页并行执行无依赖的任务。
比如：标签页A进行“商品上架”，标签页B同时“查看订单物流状态”。它们通过不同的标签页操作，互不干扰。
Python代理使用`asyncio`并发控制这两个指令序列，每个序列绑定各自的目标标签页。
```python
async def run_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"))
await asyncio.gather(task_a, task_b)
```
这样，单个店铺的任务吞吐量翻倍，浏览器资源利用更充分。
---
## 七、标签页休眠与内存回收
几十个标签页同时活跃，内存会迅速攀升。我们对非活跃标签页启用Chromium的自动丢弃（discard）机制。
```python
async def discard_idle_tabs(self, idle_seconds=300):
now = time.time()
for ws in self.workspaces.values():
for tab_id, tab in ws.tabs.items():
if tab_id != ws.active_tab_id and (now - tab.last_accessed) > idle_seconds:
await self.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是否剧烈变化，如果是则重新导航到目标页，并记录异常。
**标签页被用户手动关闭。** 虽然执行机不直接交互，但偶尔运维远程桌面时误操作关闭了标签页。我们通过崩溃检测与恢复机制一并覆盖。
---
## 十、写在最后
浏览器标签页的精细化管理，是店群自动化执行层从“可用”走向“精良”的必经之路。
工作区让每个店铺在浏览器中拥有一块独立、稳定、可恢复的操作空间，即使页面崩溃也能自动站起来继续工作。
> 自动化最难的不是写出能点击的脚本，而是让几百个页面在无人值守的深夜里，依然井然有序。
---
*作者：林焱*
更多推荐
11
0- 0
已为社区贡献44条内容
所有评论(0)