项目概述
一个通过浏览器自动化模拟人工投放操作的全栈平台，实现多渠道并发执行、智能任务调度和全链路可观测。 从桌面脚本起步，演进为「引擎 + 后端 + 前端 + AI Agent + 机器人」的完整系统，支持 PyInstaller 一键打包交付。
Web UI
Vue 3 + Element Plus 管理界面，WebSocket 实时日志推送，可视化配置与调试
飞书群聊
群里发自然语言命令即可管理账号、渠道、任务队列与定时任务，全程无需打开浏览器
AI 对话
自研 Brain Agent：意图识别 → 任务规划 → 工具执行，35+ 注册工具覆盖全部业务操作
调用链路
无论从哪个入口进入，最终都汇聚到同一条执行链路：指令解析 → API 调度 → 引擎执行 → 浏览器操作 → 结果回传。
功能演示
全流程自动化运行实录：任务导入 → 渠道分配 → 并发执行 → 结果统计，完整闭环。
系统架构
引擎、后端、前端、AI、数据五层分离，各自独立演进，通过 REST 与 WebSocket 协作。
核心能力
CDP 直连 Chrome
connect_over_cdp 接管已登录的浏览器，自动化与人工操作同屏共存，页面崩溃自动恢复
步骤注册系统
装饰器注册操作步骤，流程定义在 JSON 配置中，支持热重载，改流程不用重启
智能错误处理
指数退避 + 随机抖动重试，6 大类 18 子类异常层次，失败自动截图留存现场
崩溃可恢复
执行进度实时持久化到 JSON，进程重启后从断点继续，任务不丢失
飞书机器人
60+ 聊天命令覆盖账号、渠道、任务、队列、定时、数据工具等全部管理操作
一键打包交付
PyInstaller 打包为 Windows 单 exe，配合一键启停脚本，非技术人员也能部署
技术栈
关键设计决策
几个决定系统形态的核心取舍，以及背后的原因。
CDP 接管，而非新开浏览器
通过 remote-debugging 端口连接已经在用、已登录的 Chrome：免登录态迁移，自动化与人工互不干扰，可随时人工接管。
适配器模式抽象渠道
已实现创量（字节小程序流量）、创编（端原生流量）两个适配器；新渠道只需实现 PlatformAdapter 接口并注册，核心逻辑零改动。
把失败当作常态设计
页面元素多变是浏览器自动化的宿命：可配置重试策略 + 异常分类体系 + 自动截图，让失败可诊断、可恢复、可追溯。
并发有界，状态可查
Semaphore 控制并发上限（默认 3），支持暂停 / 恢复 / 全部停止；进度持久化到 JSON，崩溃后从断点续跑。
部分源代码
三个最能体现设计思路的真实代码片段，完整源码已开源在 GitHub。
平台适配器 — 新渠道即插即用
抽象基类定义流程执行、页面校验、错误处理三个接口，各渠道继承实现，经 PlatformRegistry 注册后即用。
class PlatformAdapter(ABC): """各广告平台适配器基类""" @abstractmethod async def execute_flow(self, flow: FlowDefinition) -> FlowResult: """执行渠道操作流程""" ... @abstractmethod async def validate_page(self) -> bool: """校验当前页面状态是否可执行""" ... @abstractmethod async def handle_error(self, error: Exception) -> ErrorAction: """根据错误类型决定重试 / 跳过 / 终止""" ...
步骤注册 — 流程可配置、可热重载
每个页面操作是一个继承 BaseStep 的步骤类，用装饰器注册到指定流程与分组；流程顺序由 JSON 配置决定，改流程不动代码。
@step_registry.register("click_change_button", "standard", "account") class ClickChangeButton(BaseStep): async def execute(self, ctx: StepContext) -> StepResult: element = await self.find_element(ctx.page, "change_button") await element.click() return StepResult.success()
数据访问 — Repository 模式
每个聚合对应一个 Manager，业务层不直接写 SQL；查询、进度更新等语义化方法集中管理。
class DramaTaskDBManager: async def get_by_status(self, status: TaskStatus) -> List[DramaTask]: ... async def update_progress(self, task_id: int, progress: float) -> None: ... class ChannelDBManager: async def get_ordered(self) -> List[Channel]: ... async def reorder(self, ids: List[int]) -> None: ...