自动化系统做到一定规模，一定会遇到一个瓶颈：
系统跑得挺好，但运营和管理系统的人，根本不知道里面发生了什么。
任务有没有在跑？哪些店铺在排队？昨天失败了多少任务？
这些问题，技术团队可以通过查日志、翻数据库来回答。
但运营团队、业务负责人呢？他们只能一次次在群里问。
我们最开始也没有管理后台。
运营要查任务状态，得找人。运营想手动触发一个任务，得找人。
运营想知道今天上架了多少商品，还是得找人。
技术团队每天都在回答重复的问题，真正该写的代码反而没时间写。
这篇文章要写的，就是如何从零搭建一个店群自动化的任务管理后台和调度中控台。
不追求花哨的 UI，只追求让所有该看见的信息被看见，该操作的动作能操作。
一、管理后台的定位：不是监控，是控制
监控系统（Grafana、Kibana）是给技术人员看的。
管理后台，是给使用系统的人用的。
两者职责不同。
监控告诉你系统健康不健康。
管理后台让你能操作系统里的事——创建任务、查看进度、重跑失败、管理店铺配置。
我们把管理后台定位成：店群自动化系统的操作入口，而非只读看板。
它的用户是运营人员，而不是 SRE。
二、技术选型：轻量优先，不搞大而全
管理后台的技术选型，我们定了一条原则：
在满足需求的前提下，选最轻的方案。
最终用的是 FastAPI + Vue3 + Redis + PostgreSQL。
没有用 Django Admin，虽然它开箱即用，但定制起来反而更累。
FastAPI 的异步支持和 API 文档自动生成，让前后端联调效率很高。
前端没有用重型框架，一个简单的 Vue3 SPA，打包后不到 400KB。
UI 组件库选了 Element Plus，中后台场景足够用。
整个后台部署在一个 2C4G 的小机器上，撑起了每天数千次的操作请求，毫无压力。
三、核心功能设计：运营每天都要用的东西
和运营团队聊了几轮之后，我们确定了管理后台必须有的几个核心模块。
3.1 任务管理面板
这是使用频率最高的页面。
运营需要看到：
哪些任务在排队、哪些在执行、哪些失败了、为什么失败。
我们设计的任务列表，支持按平台、店铺、状态、时间范围筛选。
每条任务显示它的创建时间、开始执行时间、耗时、当前状态、错误概要。
列表背后是一个简单的查询 API：
@router.get("/api/tasks")
async def list_tasks(
platform: str = None,
shop_id: str = None,
status: str = None,
page: int = 1,
page_size: int = 20,
db: AsyncSession = Depends(get_db),
):
query = select(Task).order_by(Task.created_at.desc())
if platform:
query = query.where(Task.platform == platform)
if shop_id:
query = query.where(Task.shop_id == shop_id)
if status:
query = query.where(Task.status == status)
total = await db.scalar(select(func.count()).select_from(query.subquery()))
tasks = await db.execute(query.offset((page - 1) * page_size).limit(page_size))
return {
"total": total,
"items": [task.to_dict() for task in tasks.scalars().all()],
}
运营最满意的功能之一，是任务详情页。
点进去能看到一条任务从创建到结束的完整时间线——
什么时候入队、什么时候被节点拉取、浏览器什么时候准备好、流程什么时候开始、每一步耗时多少。
这个时间线数据，来自我们前面文章里埋的那些耗时打点。
出了问题不用再找技术查日志。
运营自己点开详情页，截图发群里，直接进入排查环节。
3.2 手动任务创建
大部分任务由数据管道自动生成，但运营偶尔需要手动触发。
比如某个店铺临时要做一次全店改价，或者新品要立刻上架。
后台提供了一个任务创建表单：选择平台、选择店铺、选择动作、填写参数。
表单提交后，任务直接入队，和自动生成的任务走完全相同的调度链路。
@router.post("/api/tasks")
async def create_task(
req: TaskCreateRequest,
user: User = Depends(get_current_user),
db: AsyncSession = Depends(get_db),
queue: TaskQueue = Depends(get_queue),
):
task = Task(
platform=req.platform,
shop_id=req.shop_id,
action=req.action,
params=req.params,
created_by=user.username,
status=TaskStatus.CREATED,
)
db.add(task)
await db.commit()
await queue.enqueue(task)
return {"task_id": task.id, "status": task.status}
表单里做了前端和后端双重校验。
比如拼多多的商品价格不能超过 999999，TEMU 的图片链接必须是 HTTPS。
这些校验规则维护在配置里，不用改代码。
3.3 店铺配置管理
随着店铺数增长，店铺的配置管理本身也成了一个问题。
哪些店铺在自动运行、哪些暂停了、它们的代理 IP 是什么、流程版本用的是哪个——
这些信息散落在不同的配置文件里，改一次要上服务器。
管理后台的店铺配置页面，把这些信息集中在一起。
运营可以看到每个店铺的自动化状态（运行中 / 暂停 / 异常），
可以手动暂停某个店铺的自动化（比如店铺临时需要人工处理），
也可以查看店铺的代理信息、浏览器环境标识。
对技术的价值是，店铺配置的变更被纳入了审计日志。
谁在什么时间改了哪个配置，全部可查。
3.4 任务重跑与批量操作
失败的任务，运营需要能重跑。
但重跑不是简单的“再来一次”。
重跑前要检查：
这个任务上次失败的原因是什么？
如果是业务错误（比如商品数据本身有问题），重跑没有意义。
如果是网络超时，现在网络恢复了，重跑才有价值。
后台在重跑按钮上，会根据失败原因给出提示：
“该任务因网络超时失败，建议重跑” 或 “该任务因商品数据异常失败，请检查数据后重试”。
批量操作也是一个高频需求。
运营经常需要一次性重跑所有失败的拼多多上架任务。
后台支持按条件批量选择和重跑，但要输入确认文字，防止误操作。
四、中控台的职责：系统状态一目了然
管理后台是给运营用的。
中控台，是给技术团队自己用的。
它不需要太复杂，但必须在打开的第一屏，就能看出系统有没有问题。
我们中控台的首页只有四个模块：
- 各平台任务流量图：实时显示每个平台队列的深度、执行中任务数、完成速率。
- 节点状态网格：所有执行节点以卡片排列，绿色正常、黄色繁忙、红色离线，一目了然。
- 最近告警列表：过去 1 小时内的告警，按级别排列。
- 关键指标趋势：过去 24 小时的吞吐量、延迟 P99、失败率曲线。
这些数据的实时性靠 WebSocket 推送，而不是页面轮询。
FastAPI 的 WebSocket 支持很原生，几行代码就能把 Redis Pub/Sub 的消息推到前端。
@router.websocket("/ws/dashboard")
async def dashboard_ws(websocket: WebSocket):
await websocket.accept()
pubsub = redis.pubsub()
await pubsub.subscribe("rpa:stats:*")
async for message in pubsub.listen():
if message["type"] == "message":
await websocket.send_text(message["data"])
前端收到数据后更新图表组件。
我们用了 ECharts 做曲线图和仪表盘，配置简单，性能也好。
五、权限与安全：不是所有人都能做任何事
管理后台直接操作着上百个店铺的自动化流程。
权限控制不到位，一个误操作就可能批量下架商品。
我们实现了三级权限：
- 查看者：只能看任务列表和仪表盘，不能做任何操作。
- 操作者：可以创建任务、重跑任务，但不能改配置、不能管理节点。
- 管理员：全部权限。
权限校验在 API 层做，用 FastAPI 的 Depends 注入：
async def require_role(role: str, user: User = Depends(get_current_user)):
if user.role not in [role, "admin"]:
raise HTTPException(status_code=403, detail=f"需要 {role} 权限")
return user
@router.post("/api/tasks")
async def create_task(
req: TaskCreateRequest,
user: User = Depends(require_role("operator")),
...
):
...
所有的敏感操作——修改店铺配置、发布流程版本、删除任务——
都在后端记录了操作日志，带上操作人和时间戳。
有一次一个运营误操作，批量重跑了 50 个已完成的历史任务。
幸亏有操作日志，我们几分钟内就定位到了是谁、在什么时候、做了什么。
没有操作日志的话，排查可能要大半天。
六、日志与追踪的界面化
技术团队习惯在 Kibana 里查日志，但不能要求运营也学会写 ES 查询。
管理后台提供了一个轻量的日志查看器。
输入任务 ID，就能看到该任务的所有关联日志，按时间线排列。
输入店铺 ID 和时间范围，就能看到该店铺在这段时间内所有自动化操作的摘要。
实现原理很简单：
后端接收查询参数，拼成 ES 查询语句，返回格式化后的结果。
对运营来说，就是一个搜索框，完全不需要懂底层技术。
七、前端交互的几个工程细节
管理后台的前端，有一些细节值得一说。
任务状态的实时更新：
任务列表页面，任务状态是动态变化的（排队中 → 执行中 → 完成）。
我们用 WebSocket 推送状态变更事件，前端监听后局部更新对应行的状态标签。
不需要整页刷新，运营可以一直开着页面，状态自动变。
长任务的进度展示：
有些离线批量任务可能跑几十分钟。
任务详情页里有一个进度条，显示“已处理 45/120 店铺”。
进度数据由执行节点在处理子任务时通过 Redis 更新，前端每秒拉取一次。
表单的智能提示：
创建任务时，如果选择了 TEMU 平台和商品上架动作，表单会自动多出几个 TEMU 特有的字段——
比如“是否需要提供 CPC 认证”、“商品类目选择”。
这些字段的显示逻辑维护在一份配置里，不同平台不同动作的字段定义不同。
@router.get("/api/forms/{platform}/{action}")
async def get_form_schema(platform: str, action: str):
schema = FORM_REGISTRY.get(platform, {}).get(action)
if not schema:
raise HTTPException(status_code=404, detail="未找到对应表单")
return schema
八、真实教训：一个并发引发的脏读
管理后台上线初期，出过一次诡异的 bug。
运营在任务列表里看到某个任务状态是“执行中”，但点进去详情显示“已完成”。
排查后发现，任务列表查询的是主库，任务详情查询的是只读副本。
主从同步有几百毫秒的延迟，任务状态恰好在两个查询之间发生了变化。
修法很简单：把任务详情页的状态查询也切到主库。
但更深层的教训是：对于用户操作触发的查询，不要依赖只读副本。
只读副本适合跑报表，不适合做即时查询。
九、管理后台的可扩展性
管理后台的 API 设计之初就预留了扩展点。
新增一个功能模块，不需要改动已有代码，只需注册新的路由。
class ModuleRegistry:
_modules = {}
@classmethod
def register(cls, name: str, router):
cls._modules[name] = router
# 各模块注册自己的路由
ModuleRegistry.register("tasks", task_router)
ModuleRegistry.register("shops", shop_router)
ModuleRegistry.register("flows", flow_router)
后来增加了“数据看板”模块、“费用统计”模块，都是在不改核心代码的情况下插入的。
这种插件化的组织方式，让后台能跟着业务需求逐步生长，而不是一开始就设计一个庞大的蓝图。
十、写在最后
管理后台和中控台，是自动化系统“从后台走到前台”的一步。
它让非技术人员也能看得见自动化在干什么、也能参与其中的操作。
很多技术团队看不起管理后台，觉得 CRUD 没有技术含量。
但管理后台的质量，直接决定了运营团队对自动化系统的信任度。
一个任务状态不更新的后台，一个错误信息不明确的后台，一个点错按钮没有确认的后台——
会让运营宁愿回到手工操作，也不愿意用你写的系统。
我们把管理后台当一个正经产品来做。
先问清楚运营每天要做什么，再设计功能；
先保证状态准确、操作安全，再考虑好不好看。
这套后台跑了一年多，从最初只有任务列表，到现在涵盖了任务管理、店铺配置、流程版本、数据看板、费用统计。
它不是一次设计出来的，是跟着业务一起长出来的。
如果你也在搭类似的系统，希望这篇复盘能给你一些后台设计的参考。
别把它当附属品，它是自动化系统的门面。
作者：林焱