巨量广告计划 自动化搭建系统

📅 2025 📂 Python + Playwright + FastAPI ⏱ 阅读约 8 分钟

📝 项目概述

本系统是一个基于 Python asyncio + Playwright + FastAPI 的全栈自动化工具， 专注于 巨量引擎（字节系）广告计划 的批量搭建与优化。 解决投放优化师日常工作中「重复建计划、人工效率低、无法 7×24h 运行」的核心痛点。

支持多账户并发调度、自动填充计划配置、批量上传素材、定时执行。从 Tkinter 桌面工具起步， 演进为 FastAPI + Vue 3 的 Web 全栈应用，并接入 Hermes AI Agent 框架 + 飞书机器人， 实现「发条消息就能建计划」的无人值守体验。

🔁 系统流程设计

用户通过飞书发送自然语言指令，系统自动解析意图、调度资源、执行操作并回传结果，形成完整闭环。

💬飞书指令自然语言
→
🧠Hermes意图识别
→
⚡FastAPIREST+WS
→
🤖Playwright并发执行
→
🎯巨量引擎计划搭建
→
📊数据闭环复盘优化

🎥 功能演示

全流程自动化运行效果：任务导入 → 渠道分配 → 并发执行 → 结果统计，完整闭环。

🏗️ 系统架构

采用经典的四层架构设计，各层职责清晰、独立演进。

🎨
前端展示层
Vue 3 + Element Plus SPA，WebSocket 实时日志展示
Vue 3WebSocket
⚡
API 服务层
FastAPI 后端 · 飞书机器人 · AI Agent · APScheduler 定时调度
FastAPI飞书 Bot
🤖
自动化引擎层
Playwright + Chrome CDP · 多平台适配器 · asyncio 并发控制
PlaywrightAsyncio
🗄️
数据持久层
SQLAlchemy ORM + SQLite · 15 张业务表 · Repository 模式
SQLAlchemySQLite

✨ 核心能力

🚀

全栈架构

FastAPI + Vue 3 + SQLite，从桌面工具升级为 Web 全栈应用

🤖

AI Agent

35+ 注册工具，自然语言意图识别，自动编排执行步骤

💬

飞书机器人

60+ 聊天命令，飞书即可完成所有管理操作

🔌

平台适配器

适配器模式抽象各平台，新渠道即插即用

🗓️

定时调度

APScheduler 支持 cron / 每日 / 每周定时执行

🎯

步骤调试

拖拽式单步调试 + 实时日志 WebSocket

🛠️ 技术栈

Python 3.13 Playwright FastAPI Vue 3 SQLAlchemy Hermes AI APScheduler Asyncio WebSocket WSL 飞书 API Claude Code DeepSeek

📋 PRD — 产品需求文档

核心产品需求定义，明确系统边界与关键设计决策。

用户故事

关键设计决策

🎯 选型：Playwright vs Selenium

Playwright 基于 CDP 协议，速度比 Selenium 快 3-5 倍，原生支持 asyncio 和 BrowserContext 隔离，多账户并行更安全。

🎯 专注：巨量引擎计划搭建

当前聚焦巨量引擎（字节系）广告计划搭建，覆盖批量建计划、素材上传、配置填充等核心环节，后续可扩展至更多渠道。

🧠 AI：Hermes Agent 编排

将核心功能封装为 skills + JSON Schema 描述，LLM 自然语言识别意图后自动调用对应 skill，无需硬编码对话逻辑。

⏱ 调度：APScheduler

使用 AsyncIOScheduler 集成到 FastAPI 中，支持 cron 表达式、每日定时、周期性执行，与异步引擎无缝配合。

📄 部分源代码

项目核心模块的关键代码片段，完整源码已开源在 GitHub。

1. 自动化引擎 — 多账户并发调度

核心是 asyncio.Semaphore 控制并发数，每个任务通过 async with self._semaphore 获取 Playwright BrowserContext 实例执行，用完自动释放。

🐍 Pythoncore/automation_engine.py
# 自动化引擎核心 — 多账户并发调度
class AutomationEngine:
    """自动化引擎：管理 Playwright 浏览器实例与任务队列"""

    def __init__(self, max_concurrency: int = 5):
        self._semaphore = asyncio.Semaphore(max_concurrency)
        self._browsers: dict[str, BrowserContext] = {}
        self._task_queue = asyncio.Queue()
        self._running = False

    async def dispatch(self, task: Task):
        """分配任务到可用浏览器上下文"""
        async with self._semaphore:
            ctx = await self._get_context(task.platform)
            try:
                result = await self._execute(ctx, task)
                await task.callback(True, result)
            except Exception as e:
                await task.callback(False, str(e))
                logger.error(f"任务失败: {task.id} - {e}")
            finally:
                self._task_queue.task_done()

2. 平台适配器模式 — 新渠道即插即用

抽象基类定义三个核心接口，快手、抖音等平台各自实现，新增渠道只需继承 PlatformAdapter(ABC)。

🐍 Pythoncore/platform_adapter.py
# 平台适配器模式 — 新渠道即插即用
class PlatformAdapter(ABC):
    """各广告平台适配器基类"""

    @abstractmethod
    async def login(self, credentials: dict) -> bool:
        """平台登录"""
        ...

    @abstractmethod
    async def create_plan(self, plan_data: dict) -> str:
        """创建广告计划"""
        ...

    @abstractmethod
    async def upload_material(self, file_path: str) -> str:
        """上传素材"""
        ...

class KSAdapter(PlatformAdapter):
    """快手渠道适配器"""
    async def login(self, creds):
        logger.info("快手渠道登录...")
        return True

    async def create_plan(self, data):
        plan_id = await self.page.evaluate("createPlan", data)
        return plan_id

3. FastAPI 任务管理 API

异步路由处理任务创建与状态查询，创建后自动加入自动化队列执行。

⚡ FastAPIapi/routes/tasks.py
router = APIRouter(prefix="/api/tasks")

@router.post("/create")
async def create_task(
    task: TaskCreate,
    db: Session = Depends(get_db)
):
    """创建自动化投放任务"""
    db_task = TaskModel(
        platform=task.platform, plan_name=task.plan_name,
        budget=task.budget, schedule_type=task.schedule_type,
        status="pending"
    )
    db.add(db_task); db.commit(); db.refresh(db_task)
    await engine.dispatch(Task(id=db_task.id, platform=db_task.platform, data=task.dict()))
    return {"success": True, "task_id": db_task.id}

@router.get("/status/{task_id}")
async def get_task_status(task_id: int, db: Session = Depends(get_db)):
    task = db.query(TaskModel).get(task_id)
    if not task: raise HTTPException(404, "任务不存在")
    return {
        "id": task.id, "platform": task.platform,
        "status": task.status, "progress": task.progress,
        "logs": task.logs[-50:]
    }

4. SQLAlchemy ORM 模型

15 张业务表覆盖任务、账户、平台配置、执行日志等，Repository 模式封装数据访问。

🗄️ SQLAlchemymodels/models.py
class TaskModel(Base):
    __tablename__ = "tasks"
    id = Column(Integer, primary_key=True)
    platform = Column(String(32), nullable=False)
    plan_name = Column(String(128))
    budget = Column(Float, default=0)
    status = Column(String(16), default="pending")
    progress = Column(Integer, default=0)
    logs = Column(Text, default="")
    created_at = Column(DateTime, default=datetime.utcnow)
    account_id = Column(Integer, ForeignKey("accounts.id"))
    account = relationship("AccountModel", back_populates="tasks")
查看完整 GitHub 源码 →