尧图网站建设 
简介这是一份面向短视频矩阵运营与自动化发布场景的系统设计源码借助Playwright实现自动登录、上传与分发视频到抖音、快手、视频号、小红书等主流平台。资源适合正在做课程设计、期末大作业或毕业设计的计算机相关专业学生也适合对浏览器自动化、爬虫与多平台API封装感兴趣的中高级学习者。压缩包共28个文件以19个Python源码文件为核心覆盖各平台上传器、队列登录、缓存配置与工具函数另含2个JavaScript文件含stealth.min.js反检测脚本、SQL建表脚本、视频示例、二维码图片及README说明整体约5.09MB。目录按douyin_uploader、ks_uploader、tencent_uploader、xhs_uploader等模块组织结构清晰方便逐一拆解自动发布流程与平台适配逻辑。目前已有187人学习下载借助完整源码、数据库设计和多媒体演示可快速理解短视频分发系统从登录鉴权到上传发布的全链路实现。1. 短视频矩阵内容分发系统为什么最后落在 playwright 上做矩阵号的团队基本都经历过这个阶段注册了一批账号剪辑了成批视频然后打开网页、登录、点上传、填标题、选封面、点发布六个平台来一轮就是半小时。等账号到十个以上每天光重复操作就吃掉一个全职人力。市面上不是没有批量分发工具但要么绑定平台内部接口、要么依赖模拟器平台前端一改版就失效账号风险也压不住。最后选型兜兜转转还是会落回网页自动化这条老路而 playwright 是目前把这条路走得最稳的框架。playwright 的价值不是“能模拟点击”而是它给了你一套可编排的浏览器状态管理能力多上下文隔离、持久化登录态、网络拦截、自动等待和重试机制都内置了。相比早期直接用 requests 模拟接口playwright 跑的是真实浏览器平台的 JS 环境、指纹采集、验证码弹出都能原样触发反而更容易让系统按“真人操作”的节奏运行。这套系统设计核心就是围绕 playwright 的 BrowserContext 隔离模型和 locator 自动等待机制把上传、分发、调度拆成可独立运维的任务流。对 5 年以上后端或自动化测试背景的人来说难点不在语法而在浏览器指纹对抗、平台差异适配和发布节奏控制这三个地方本文会逐个讲透。2. 系统架构第一关浏览器实例池与多账号会话隔离的设计2.1 为什么 BrowserContext 是矩阵系统的地基很多人第一次写 playwright 脚本习惯直接 browser.new_page() 单开页面操作。这在单账号场景没问题但矩阵系统同时持有几十个账号如果共用一个浏览器上下文登录态会互相覆盖Cookie 串号平台风控瞬间就会标记异常。playwright 里 BrowserContext 是独立的浏览器会话容器每个 Context 拥有自己的 Cookie、缓存、localStorage 和 UserAgent只要新建一个 Context 并加载每个账号的持久化登录态就能实现真正的账号隔离。我的做法是维护一个“连接池”固定启动 2 到 3 个 Browser 实例按服务器内存决定每个 Chromium 实例空载大约占 200MB 内存每个 Browser 下按需创建多个 Context。任务队列来一个视频发布任务就从池里借一个空闲 Context执行完归还。这样避免每条任务都冷启动浏览器发布单条视频的平均耗时能降低 40% 左右。from playwright.async_api import async_playwright async def get_browser_context(pw, user_data_dir: str, proxy: dict | None None): browser await pw.chromium.launch( headlessTrue, args[ --disable-blink-featuresAutomationControlled, --start-maximized, ], ) context await browser.new_context( storage_statef{user_data_dir}/state.json, viewport{width: 1440, height: 900}, localezh-CN, proxyproxy, # 模拟真实设备指纹 user_agentMozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36, ) return browser, context # 每个账号 一个独立 Context互不干扰 # storage_state 保存的是该账号的 cookies 和 localStorage 快照storage_state是这套设计的核心首次登录后把 Context 的context.storage_state()序列化成 JSON 保存到账号目录后续任务直接加载免去重复扫码登录。disable-blink-featuresAutomationControlled用于隐藏 webdriver 标记但要注意这只是第一层对抗后面会细讲指纹怎么处理。2.2 登录态持久化扫码登录只做一次每个平台首次接入时肯定躲不开一次人工登录。这块不能省但可以设计成“半自动”程序检测到登录页出现二维码就暂停流程并推送二维码图片到管理后台运营人员用手机扫一次之后 7 到 15 天内都不需要再扫码。async def ensure_login(page, platform: str, state_path: str): # 检测是否出现二维码元素 qr_locator page.locator(img[src*qr], canvas, div.login-qrcode) if await qr_locator.count() 0: # 截图保存推送到运维群 await page.screenshot(pathfqrcode_{platform}.png) # 轮询等待扫码成功最多等 120 秒 await page.wait_for_url(**/feed**, timeout120_000) # 保存最新登录态 context page.context state await context.storage_state() with open(state_path, w, encodingutf-8) as f: json.dump(state, f, ensure_asciiFalse) # 参数说明 # page.wait_for_url 的 timeout 设置的是“等待跳转到登录后页面的超时” # 这里用 feed 路径作为登录成功的标志不同平台需要换成自己的首页路径模式。这里有个细节不同平台的二维码元素结构差异很大抖音是 canvas 动态绘制视频号是 iframe 内嵌小红书是 img 标签。写选择器时不能用一套规则吃所有平台建议针对平台做映射表后面第 4 章的适配层会给出完整设计。登录成功前的状态触发也建议用条件等待而不是固定 sleep因为扫码时间不可控固定等待要么浪费要么超时。2.3 并发量控制不是线程越多越好多账号发布最常见的翻车点是并发设置过高。同一 IP 下 5 个账号同时上传视频平台服务端看到的流量特征瞬间集中加快视频上传秒传反而被认为是机器行为。我一般会将全局并发控制在 2 到 3 个任务同时执行并且每个任务之间的启动时间人为错开 3 到 8 秒。这个数字不是拍脑袋短视频平台的上传接口通常有单账号每分钟请求数限制且同 IP 短时高并发会触发滑块验证。如果你的脚本大面积触发滑块先检查并发不要急着换代理。import asyncio from semaphore_sim import AsyncSemaphore # 实际用 asyncio.Semaphore async def publisher_worker(task_queue, sem): async with sem: # 任务的完整生命周期取账号状态 - 打开平台 - 上传 - 发布 task await task_queue.get() try: await execute_publish(task) finally: task_queue.task_done() async def main(): sem asyncio.Semaphore(3) # 全局最多 3 个并发发布任务 task_queue asyncio.Queue() workers [asyncio.create_task(publisher_worker(task_queue, sem)) for _ in range(3)] # 每个任务入队前随机 sleep 3~8 秒打散起点异步并发用asyncio.Semaphore(3)限制同时执行的上传任务数队列消费者固定 3 个。任务入队前随机 sleep 的实现是为了让每个平台看到的网络请求之间有时间间隔更接近真实运营人员的手工操作间隔。二维码、验证码、视频上传进度这些步骤本来就耗时不需要过度担心吞吐稳定比速度优先。3. 核心链路playwright 实现视频上传的可执行方案3.1 统一任务数据模型设计无论目标平台是抖音、快手、视频号还是小红书发布操作的本质都是三件事选文件、填信息、点发布。区别只是页面上这些控件的类型input 标签、富文本编辑器、拖拽区域和交互顺序。所以任务模型需要抽象成中立的 JSON 结构平台适配层只负责把通用字段映射为每个平台的页面操作这也是让系统可持续扩展的关键。{ task_id: T20240611_001, account: { platform: douyin, username: matrix_001, state_file: /data/accounts/douyin_001.json }, video: { file_path: /data/videos/001/搞笑合集_no_watermark.mp4, cover_path: /data/videos/001/cover.png, title: 今天玩点新的你绝对没见过 #搞笑 #挑战, topic_tags: [搞笑, 挑战], allow_download: false, allow_comment: true, allow_share: true }, publish_time: 2024-06-11 18:30:00, retry_count: 0, status: pending }video.file_path建议统一使用绝对路径因为 playwright 的set_input_files在部分平台上要求完整路径。cover_path是可选项因为抖音和快手可以自动截取视频首帧但视频号和部分专业领域账号建议设定自制封面后续适配层会区分处理。publish_time是计划发布时间实际取到当前时间大于该值才执行发布这是定时发布的核心字段。3.2 文件上传的三种方式与取舍短视频平台的上传控件表面上都是input typefile但实际行为差异极大尤其是不允许多平台统一用同一套定位逻辑。我总结下来它们的区别主要在于平台上传入口交互特征推荐方式抖音顶部“发布”按钮打开后是弹窗页面上传是拖拽区点击发布按钮等待上传区出现用set_input_files挂在 file input 上快手首页右上角摄像机按钮进入创作中心需要先进入页面再操作导航 URL 直达 /creator选上传视频号移动端和 PC 后台不统一PC 端需要先进入视频号助手后台直接用 URL 进入助手点“发布视频”小红书左侧“发布”按钮支持图文和视频切换上传区定位不稳定切到视频 tab定位 file input底层统一用set_input_files它不会触发平台对剪贴板的限制也不会模拟拖拽带来的诸多判定性问题。为什么不推荐模拟拖拽或剪贴板粘贴因为平台上传组件大多绑定了事件的dataTransfer属性合成拖拽事件很容易因为drop事件校验不通过而失败。set_input_files是 CDP 层面的文件注入稳定性最高。async def upload_video(page, file_path: str, cover_path: str | None None): # 先找 input[typefile]部分平台是隐藏的Playwright 不需要可见即可操作 file_input page.locator(input[typefile]).first await file_input.set_input_files(file_path) # 等待上传进度条消失表示视频处理完成 await page.wait_for_function( () { const bars document.querySelectorAll(.upload-progress, .ant-progress); return bars.length 0 || [...bars].every(el el.style.display none); }, timeout180_000, ) if cover_path: cover_input page.locator(input[typefile][accept*image]).first await cover_input.set_input_files(cover_path) # 参数说明 # wait_for_function 传入的是浏览器内执行的 JS用轮询方式检查页面上是否存在进度条元素。 # 不同平台的进度条 class 不同实际使用需要在适配层配置这个选择器。代码里的进度条等待是一个关键技巧。视频上传分两个阶段先传文件到临时存储快然后服务端转码慢。进度条消失通常代表转码完成此时才能继续填标题和选择发布选项。如果这一步不做等待直接提交很多平台会报“视频正在处理中”。3.3 标题与话题标签的填写策略标题框在不同平台的类型差异很大抖音是标准的单行输入框textarea小红书是富文本contenteditable视频号后台则是普通input。针对富文本fill方法不生效必须用press_sequentially配合page.keyboard.insert_text方法逐字写入。async def fill_title(page, title: str, platform: str): title_map { douyin: (textarea.input-title, div.input-container textarea, fill), kuaishou: (input.title-input, .editor-title input, fill), shipinhao: (input.video-title, .title-content input, fill), xiaohongshu: (div.content-editor, .title-editor[contenteditabletrue], type), } selector, method title_map[platform] locator page.locator(selector).first await locator.wait_for(statevisible, timeout30_000) if method fill: await locator.fill(title) else: await locator.click() await page.keyboard.insert_text(title) # 参数说明 # xiaohongshu 是 contenteditable 组件fill 不会触发 input 事件 # 需要用 insert_text 逐字输入模拟真实键盘路径 # 同时保证 React/Vue 的受控组件能捕捉到值变更。话题标签# 话题的处理上不同平台策略也不同。抖音必须在标题内以#话题形式书写才会变成超链接快手则支持输入后用系统索引选择视频号后台有单独的话题添加按钮需要点击后单独输入。不能靠标题字符串统一解决要分别适配。3.4 发布按钮点击与结果确认点发布看似简单实际是失败率最高的环节。很多平台在点击发布后会有一层二次确认弹窗内容不是“确认发布”就是“补充定位/话题”如果脚本点完就结束任务经常实际未发布成功。我要求代码在点击发布按钮后必须等待页面跳回“作品列表”或出现“发布成功”Toast 再判定成功。async def click_publish(page, platform: str): publish_map { douyin: button:has-text(发布), .publish-btn, kuaishou: .submit-btn, button:has-text(发布), shipinhao: button:has-text(发表), .sender-btn, xiaohongshu: .submit, button:has-text(发布), } await page.locator(publish_map[platform]).first.click() # 处理二次确认弹窗弹窗文案通常包含“确认”或“确定” try: confirm_btn page.locator(div.ant-modal button:has-text(确认), div.modal button:has-text(确定)) if await confirm_btn.count() 0: await confirm_btn.first.click(timeout5_000) except Exception: pass # 关键等跳转等发布成功提示 await page.wait_for_load_state(networkidle, timeout60_000)4. 把平台差异封装成描述式适配层4.1 一个脚本跑四个平台的前置工作量用 playwright 同时写死四个平台的操作步骤代码会迅速膨胀到难以维护。我把每个平台的页面元素、操作顺序、等待条件、异常提示做成一份 JSON 描述文件业务代码根据平台名动态加载描述文件执行流程。这个设计早期会多花一些时间但当你遇到平台改版时只需更新 JSON 里的选择器配置不需要改动主逻辑。适配层至少要维护这些差异化配置{ platform: douyin, publish_entry: button:has-text(发布), upload: { file_input: input[typefile], progress_hidden_js: document.querySelectorAll(.upload-progress).length 0 }, title: { selector: textarea, method: fill }, topic_selector: .topic-input, input[placeholder*话题], confirm_button: button:has-text(发布), success_flag: .publish-result, .publish-success, error_keywords: [视频违规, 重复内容, 反向操作频繁, 不符合社区规范] }error_keywords用来在发布失败时快速判定失败类型。比如“重复内容”意味着这条视频此前发过不需要重试“反向操作频繁”则说明当前账号发布节奏过快应切换到另一个账号这些必须和重试逻辑解耦否则错误信息会被吞掉导致同一个账号连续触发风控。4.2 平台差异最小化的四个处理技巧四个平台最耗时的排错点集中在四个地方元素定位不稳定、文件上传控件尺寸小、富文本输入不生效、发布后异步回调慢。我给适配层加了四个约定解决这四类问题async def smart_wait_redirect(page, success_flag: str, timeout: float 60.0): # 1. 不依赖 URL 判断只依赖成功元素出现 try: await page.locator(success_flag).wait_for(statevisible, timeouttimeout * 1000) except Exception: # 2. 失败时尝试触发滚动部分手机端适配页面需要滚动才渲染成功状态 await page.mouse.wheel(0, 300) raise # 用法说明这套 wait 逻辑统一接管所有平台的发布后验证。 # 如果成功元素没出现就滚动页面强制刷新布局之后还找不到直接抛错由上层重试。已知的几个平台差异处理技巧可以沉淀成经验规则例如小红书的上传区域大就不用强制点击“上传视频”按钮直接定位 file input 更快视频号的登录页是 iframe 包裹等待时间要拉长快手的发布页会定期弹出“草稿”提示需要在发布前先关掉弹窗。把这些规则放进适配层配置后系统的每次发布任务失败率能控制在 5% 以内。4.3 playwright 自动等待与稳定性的边界playwright 的自动等待是很强的能力但前提是你正确使用 locator 而不是强等待。有人为了让脚本“更稳定”到处 sleep(10)这其实是反效果每个平台响应时间不同全局 sleep 会让短请求延迟过长长请求又覆盖不到反而更容易触发超时或误判。优先依赖locator.wait_for(statevisible)和expect(locator).to_be_visible()这类条件等待把兜底超时拉长到 30~60 秒适应弱网环境。from playwright.sync_api import expect # 发布后验证用 expect 轮询替代固定 sleep expect(page.locator(.publish-success)).to_be_visible(timeout90_000)expect的轮询机制是每 100ms 重新检查一次元素状态直到超时。相比wait_for_selector它直接支持图、文案、状态的一类断言debug 信息更友好。生产环境我会把默认超时调至PLAYWRIGHT_TIMEOUT60000再在具体步骤传入更精确的 timeout这样日志里能看到是“哪一步超时”而不是“整体超时”。5. 发布节奏、失败重试与冒烟验证的工程细节5.1 发布节奏控制与定时发布的调度实现内容分发系统一旦跑起来比“能不能发布”更重要的就是“什么时候发布”。短视频平台在特定时间段午休、晚间的推荐流量池放大效应很明显批量发布必须对准这些时间窗口。常见做法是在任务表里存publish_time轮询扫描到时间到期的任务才加入执行队列不让脚本定时全量扫描减少无效访问。async def schedule_loop(): while True: due_tasks await db.fetch_tasks_due() # SQL: WHERE statuspending AND publish_time NOW() for task in due_tasks: await task_queue.put(task) await db.mark_scheduled(task.task_id) await asyncio.sleep(10)任务进入队列后执行端加一个最小时间间隔判断如果当前时间比任务计划时间提前超过 30 秒就再等一段时间。四个平台的发布时间窗口可以配置成一张时间表比如抖音 12:00-13:00 和 18:00-21:00快手 07:00-09:00 和 19:00-22:00用于自动计算发布时间避免每天手工调。5.2 失败重试与降级策略平台的异常响应不会给全你的文案很多是抽象的错误码或空白页。我的重试策略分成三个级别页面上元素未出现重试 3 次每次间隔递增上传过程中断重新上传最多 2 次平台明确提示“重复”或“违规”不重试直接置为 failed 等待人工介入。指数退避重试间隔设置为 30 秒、60 秒、120 秒比固定重试更能减少对平台的瞬时冲击。import asyncio async def retry_with_backoff(func, max_retries: int 3): for attempt in range(max_retries): try: return await func() except Exception as e: if attempt max_retries - 1: raise # 30s * 2^attempt即 30s / 60s / 120s wait_time 30 * (2 ** attempt) await asyncio.sleep(wait_time) return None # 参数说明 # 2 ** attempt 是指数退避的基数是 2这比线性重试更科学 # 因为平台风控往往会在短时间内持续拦截间隔拉大后成功率显著提升。5.3 发布后的冒烟验证回读作品列表发布成功不代表万事大吉。抖音和快手这类平台的“发布成功”提示是前端的服务端可能还要经过审核、转码。最稳妥的闭环是发布完成后跳转到账号的作品列表页检查刚发布视频的标题是否出现在第一条。这个环节会额外花 5 到 10 秒但对矩阵号来说值得因为它能帮你确认视频真正进入了待审核队列而不是前端假成功。async def smoke_check(page, title_keyword: str, profile_url: str): await page.goto(profile_url, wait_untilnetworkidle) first_item_title page.locator(.video-item-title, .feed-card-title).first await first_item_title.wait_for(statevisible, timeout30_000) actual_title await first_item_title.text_content() if title_keyword not in actual_title: raise ValueError(f发布后回读校验失败, 期望包含: {title_keyword}, 实际: {actual_title})回读校验的profile_url同样要从适配层读取因为“作品列表页”的 URL 规则各平台完全不同。快手是https://www.kuaishou.com/profile/xxx抖音是https://www.douyin.com/user/xxx视频号直接把“视频号助手”里的作品列表页当作回读页面。如果这块没做成配置化新增平台时代码要动的地方就太多了。5.4 用 playwright codegen 加速选择器排查平台改版后适配层要更新的第一件事是选择器。手动打开 devtools 找元素太慢我习惯直接用 playwright codegen 临时打开目标平台页面在录制模式下手动点一遍上传流程codegen 会把生成的 locator 全部记录下来。虽然录制的代码不能直接用于生产但它非常高效地帮你定位到平台改版后的最新选择器结构和属性名差异。# 启动 codegen拖拽式生成选择器 playwright codegen --viewport-size1440,900 --langzh-CN https://creator.douyin.com/ # 在打开的 Chromium 窗口里手动执行一次上传操作 # 窗口右侧会自动生成对应的 python 代码包含最新的 locator 逻辑。codegen 生成的定位器经常带有#root div ...这类路径选择器建议提取其中属性较稳定的style="width:16px;margin-left:4px;vertical-align:text-bottom;cursor:text;" /> Playwright构建短视频矩阵分发系统：多账号隔离与自动化上传实战
简介这是一份面向短视频矩阵运营与自动化发布场景的系统设计源码借助Playwright实现自动登录、上传与分发视频到抖音、快手、视频号、小红书等主流平台。资源适合正在做课程设计、期末大作业或毕业设计的计算机相关专业学生也适合对浏览器自动化、爬虫与多平台API封装感兴趣的中高级学习者。压缩包共28个文件以19个Python源码文件为核心覆盖各平台上传器、队列登录、缓存配置与工具函数另含2个JavaScript文件含stealth.min.js反检测脚本、SQL建表脚本、视频示例、二维码图片及README说明整体约5.09MB。目录按douyin_uploader、ks_uploader、tencent_uploader、xhs_uploader等模块组织结构清晰方便逐一拆解自动发布流程与平台适配逻辑。目前已有187人学习下载借助完整源码、数据库设计和多媒体演示可快速理解短视频分发系统从登录鉴权到上传发布的全链路实现。1. 短视频矩阵内容分发系统为什么最后落在 playwright 上做矩阵号的团队基本都经历过这个阶段注册了一批账号剪辑了成批视频然后打开网页、登录、点上传、填标题、选封面、点发布六个平台来一轮就是半小时。等账号到十个以上每天光重复操作就吃掉一个全职人力。市面上不是没有批量分发工具但要么绑定平台内部接口、要么依赖模拟器平台前端一改版就失效账号风险也压不住。最后选型兜兜转转还是会落回网页自动化这条老路而 playwright 是目前把这条路走得最稳的框架。playwright 的价值不是“能模拟点击”而是它给了你一套可编排的浏览器状态管理能力多上下文隔离、持久化登录态、网络拦截、自动等待和重试机制都内置了。相比早期直接用 requests 模拟接口playwright 跑的是真实浏览器平台的 JS 环境、指纹采集、验证码弹出都能原样触发反而更容易让系统按“真人操作”的节奏运行。这套系统设计核心就是围绕 playwright 的 BrowserContext 隔离模型和 locator 自动等待机制把上传、分发、调度拆成可独立运维的任务流。对 5 年以上后端或自动化测试背景的人来说难点不在语法而在浏览器指纹对抗、平台差异适配和发布节奏控制这三个地方本文会逐个讲透。2. 系统架构第一关浏览器实例池与多账号会话隔离的设计2.1 为什么 BrowserContext 是矩阵系统的地基很多人第一次写 playwright 脚本习惯直接 browser.new_page() 单开页面操作。这在单账号场景没问题但矩阵系统同时持有几十个账号如果共用一个浏览器上下文登录态会互相覆盖Cookie 串号平台风控瞬间就会标记异常。playwright 里 BrowserContext 是独立的浏览器会话容器每个 Context 拥有自己的 Cookie、缓存、localStorage 和 UserAgent只要新建一个 Context 并加载每个账号的持久化登录态就能实现真正的账号隔离。我的做法是维护一个“连接池”固定启动 2 到 3 个 Browser 实例按服务器内存决定每个 Chromium 实例空载大约占 200MB 内存每个 Browser 下按需创建多个 Context。任务队列来一个视频发布任务就从池里借一个空闲 Context执行完归还。这样避免每条任务都冷启动浏览器发布单条视频的平均耗时能降低 40% 左右。from playwright.async_api import async_playwright async def get_browser_context(pw, user_data_dir: str, proxy: dict | None None): browser await pw.chromium.launch( headlessTrue, args[ --disable-blink-featuresAutomationControlled, --start-maximized, ], ) context await browser.new_context( storage_statef{user_data_dir}/state.json, viewport{width: 1440, height: 900}, localezh-CN, proxyproxy, # 模拟真实设备指纹 user_agentMozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36, ) return browser, context # 每个账号 一个独立 Context互不干扰 # storage_state 保存的是该账号的 cookies 和 localStorage 快照storage_state是这套设计的核心首次登录后把 Context 的context.storage_state()序列化成 JSON 保存到账号目录后续任务直接加载免去重复扫码登录。disable-blink-featuresAutomationControlled用于隐藏 webdriver 标记但要注意这只是第一层对抗后面会细讲指纹怎么处理。2.2 登录态持久化扫码登录只做一次每个平台首次接入时肯定躲不开一次人工登录。这块不能省但可以设计成“半自动”程序检测到登录页出现二维码就暂停流程并推送二维码图片到管理后台运营人员用手机扫一次之后 7 到 15 天内都不需要再扫码。async def ensure_login(page, platform: str, state_path: str): # 检测是否出现二维码元素 qr_locator page.locator(img[src*qr], canvas, div.login-qrcode) if await qr_locator.count() 0: # 截图保存推送到运维群 await page.screenshot(pathfqrcode_{platform}.png) # 轮询等待扫码成功最多等 120 秒 await page.wait_for_url(**/feed**, timeout120_000) # 保存最新登录态 context page.context state await context.storage_state() with open(state_path, w, encodingutf-8) as f: json.dump(state, f, ensure_asciiFalse) # 参数说明 # page.wait_for_url 的 timeout 设置的是“等待跳转到登录后页面的超时” # 这里用 feed 路径作为登录成功的标志不同平台需要换成自己的首页路径模式。这里有个细节不同平台的二维码元素结构差异很大抖音是 canvas 动态绘制视频号是 iframe 内嵌小红书是 img 标签。写选择器时不能用一套规则吃所有平台建议针对平台做映射表后面第 4 章的适配层会给出完整设计。登录成功前的状态触发也建议用条件等待而不是固定 sleep因为扫码时间不可控固定等待要么浪费要么超时。2.3 并发量控制不是线程越多越好多账号发布最常见的翻车点是并发设置过高。同一 IP 下 5 个账号同时上传视频平台服务端看到的流量特征瞬间集中加快视频上传秒传反而被认为是机器行为。我一般会将全局并发控制在 2 到 3 个任务同时执行并且每个任务之间的启动时间人为错开 3 到 8 秒。这个数字不是拍脑袋短视频平台的上传接口通常有单账号每分钟请求数限制且同 IP 短时高并发会触发滑块验证。如果你的脚本大面积触发滑块先检查并发不要急着换代理。import asyncio from semaphore_sim import AsyncSemaphore # 实际用 asyncio.Semaphore async def publisher_worker(task_queue, sem): async with sem: # 任务的完整生命周期取账号状态 - 打开平台 - 上传 - 发布 task await task_queue.get() try: await execute_publish(task) finally: task_queue.task_done() async def main(): sem asyncio.Semaphore(3) # 全局最多 3 个并发发布任务 task_queue asyncio.Queue() workers [asyncio.create_task(publisher_worker(task_queue, sem)) for _ in range(3)] # 每个任务入队前随机 sleep 3~8 秒打散起点异步并发用asyncio.Semaphore(3)限制同时执行的上传任务数队列消费者固定 3 个。任务入队前随机 sleep 的实现是为了让每个平台看到的网络请求之间有时间间隔更接近真实运营人员的手工操作间隔。二维码、验证码、视频上传进度这些步骤本来就耗时不需要过度担心吞吐稳定比速度优先。3. 核心链路playwright 实现视频上传的可执行方案3.1 统一任务数据模型设计无论目标平台是抖音、快手、视频号还是小红书发布操作的本质都是三件事选文件、填信息、点发布。区别只是页面上这些控件的类型input 标签、富文本编辑器、拖拽区域和交互顺序。所以任务模型需要抽象成中立的 JSON 结构平台适配层只负责把通用字段映射为每个平台的页面操作这也是让系统可持续扩展的关键。{ task_id: T20240611_001, account: { platform: douyin, username: matrix_001, state_file: /data/accounts/douyin_001.json }, video: { file_path: /data/videos/001/搞笑合集_no_watermark.mp4, cover_path: /data/videos/001/cover.png, title: 今天玩点新的你绝对没见过 #搞笑 #挑战, topic_tags: [搞笑, 挑战], allow_download: false, allow_comment: true, allow_share: true }, publish_time: 2024-06-11 18:30:00, retry_count: 0, status: pending }video.file_path建议统一使用绝对路径因为 playwright 的set_input_files在部分平台上要求完整路径。cover_path是可选项因为抖音和快手可以自动截取视频首帧但视频号和部分专业领域账号建议设定自制封面后续适配层会区分处理。publish_time是计划发布时间实际取到当前时间大于该值才执行发布这是定时发布的核心字段。3.2 文件上传的三种方式与取舍短视频平台的上传控件表面上都是input typefile但实际行为差异极大尤其是不允许多平台统一用同一套定位逻辑。我总结下来它们的区别主要在于平台上传入口交互特征推荐方式抖音顶部“发布”按钮打开后是弹窗页面上传是拖拽区点击发布按钮等待上传区出现用set_input_files挂在 file input 上快手首页右上角摄像机按钮进入创作中心需要先进入页面再操作导航 URL 直达 /creator选上传视频号移动端和 PC 后台不统一PC 端需要先进入视频号助手后台直接用 URL 进入助手点“发布视频”小红书左侧“发布”按钮支持图文和视频切换上传区定位不稳定切到视频 tab定位 file input底层统一用set_input_files它不会触发平台对剪贴板的限制也不会模拟拖拽带来的诸多判定性问题。为什么不推荐模拟拖拽或剪贴板粘贴因为平台上传组件大多绑定了事件的dataTransfer属性合成拖拽事件很容易因为drop事件校验不通过而失败。set_input_files是 CDP 层面的文件注入稳定性最高。async def upload_video(page, file_path: str, cover_path: str | None None): # 先找 input[typefile]部分平台是隐藏的Playwright 不需要可见即可操作 file_input page.locator(input[typefile]).first await file_input.set_input_files(file_path) # 等待上传进度条消失表示视频处理完成 await page.wait_for_function( () { const bars document.querySelectorAll(.upload-progress, .ant-progress); return bars.length 0 || [...bars].every(el el.style.display none); }, timeout180_000, ) if cover_path: cover_input page.locator(input[typefile][accept*image]).first await cover_input.set_input_files(cover_path) # 参数说明 # wait_for_function 传入的是浏览器内执行的 JS用轮询方式检查页面上是否存在进度条元素。 # 不同平台的进度条 class 不同实际使用需要在适配层配置这个选择器。代码里的进度条等待是一个关键技巧。视频上传分两个阶段先传文件到临时存储快然后服务端转码慢。进度条消失通常代表转码完成此时才能继续填标题和选择发布选项。如果这一步不做等待直接提交很多平台会报“视频正在处理中”。3.3 标题与话题标签的填写策略标题框在不同平台的类型差异很大抖音是标准的单行输入框textarea小红书是富文本contenteditable视频号后台则是普通input。针对富文本fill方法不生效必须用press_sequentially配合page.keyboard.insert_text方法逐字写入。async def fill_title(page, title: str, platform: str): title_map { douyin: (textarea.input-title, div.input-container textarea, fill), kuaishou: (input.title-input, .editor-title input, fill), shipinhao: (input.video-title, .title-content input, fill), xiaohongshu: (div.content-editor, .title-editor[contenteditabletrue], type), } selector, method title_map[platform] locator page.locator(selector).first await locator.wait_for(statevisible, timeout30_000) if method fill: await locator.fill(title) else: await locator.click() await page.keyboard.insert_text(title) # 参数说明 # xiaohongshu 是 contenteditable 组件fill 不会触发 input 事件 # 需要用 insert_text 逐字输入模拟真实键盘路径 # 同时保证 React/Vue 的受控组件能捕捉到值变更。话题标签# 话题的处理上不同平台策略也不同。抖音必须在标题内以#话题形式书写才会变成超链接快手则支持输入后用系统索引选择视频号后台有单独的话题添加按钮需要点击后单独输入。不能靠标题字符串统一解决要分别适配。3.4 发布按钮点击与结果确认点发布看似简单实际是失败率最高的环节。很多平台在点击发布后会有一层二次确认弹窗内容不是“确认发布”就是“补充定位/话题”如果脚本点完就结束任务经常实际未发布成功。我要求代码在点击发布按钮后必须等待页面跳回“作品列表”或出现“发布成功”Toast 再判定成功。async def click_publish(page, platform: str): publish_map { douyin: button:has-text(发布), .publish-btn, kuaishou: .submit-btn, button:has-text(发布), shipinhao: button:has-text(发表), .sender-btn, xiaohongshu: .submit, button:has-text(发布), } await page.locator(publish_map[platform]).first.click() # 处理二次确认弹窗弹窗文案通常包含“确认”或“确定” try: confirm_btn page.locator(div.ant-modal button:has-text(确认), div.modal button:has-text(确定)) if await confirm_btn.count() 0: await confirm_btn.first.click(timeout5_000) except Exception: pass # 关键等跳转等发布成功提示 await page.wait_for_load_state(networkidle, timeout60_000)4. 把平台差异封装成描述式适配层4.1 一个脚本跑四个平台的前置工作量用 playwright 同时写死四个平台的操作步骤代码会迅速膨胀到难以维护。我把每个平台的页面元素、操作顺序、等待条件、异常提示做成一份 JSON 描述文件业务代码根据平台名动态加载描述文件执行流程。这个设计早期会多花一些时间但当你遇到平台改版时只需更新 JSON 里的选择器配置不需要改动主逻辑。适配层至少要维护这些差异化配置{ platform: douyin, publish_entry: button:has-text(发布), upload: { file_input: input[typefile], progress_hidden_js: document.querySelectorAll(.upload-progress).length 0 }, title: { selector: textarea, method: fill }, topic_selector: .topic-input, input[placeholder*话题], confirm_button: button:has-text(发布), success_flag: .publish-result, .publish-success, error_keywords: [视频违规, 重复内容, 反向操作频繁, 不符合社区规范] }error_keywords用来在发布失败时快速判定失败类型。比如“重复内容”意味着这条视频此前发过不需要重试“反向操作频繁”则说明当前账号发布节奏过快应切换到另一个账号这些必须和重试逻辑解耦否则错误信息会被吞掉导致同一个账号连续触发风控。4.2 平台差异最小化的四个处理技巧四个平台最耗时的排错点集中在四个地方元素定位不稳定、文件上传控件尺寸小、富文本输入不生效、发布后异步回调慢。我给适配层加了四个约定解决这四类问题async def smart_wait_redirect(page, success_flag: str, timeout: float 60.0): # 1. 不依赖 URL 判断只依赖成功元素出现 try: await page.locator(success_flag).wait_for(statevisible, timeouttimeout * 1000) except Exception: # 2. 失败时尝试触发滚动部分手机端适配页面需要滚动才渲染成功状态 await page.mouse.wheel(0, 300) raise # 用法说明这套 wait 逻辑统一接管所有平台的发布后验证。 # 如果成功元素没出现就滚动页面强制刷新布局之后还找不到直接抛错由上层重试。已知的几个平台差异处理技巧可以沉淀成经验规则例如小红书的上传区域大就不用强制点击“上传视频”按钮直接定位 file input 更快视频号的登录页是 iframe 包裹等待时间要拉长快手的发布页会定期弹出“草稿”提示需要在发布前先关掉弹窗。把这些规则放进适配层配置后系统的每次发布任务失败率能控制在 5% 以内。4.3 playwright 自动等待与稳定性的边界playwright 的自动等待是很强的能力但前提是你正确使用 locator 而不是强等待。有人为了让脚本“更稳定”到处 sleep(10)这其实是反效果每个平台响应时间不同全局 sleep 会让短请求延迟过长长请求又覆盖不到反而更容易触发超时或误判。优先依赖locator.wait_for(statevisible)和expect(locator).to_be_visible()这类条件等待把兜底超时拉长到 30~60 秒适应弱网环境。from playwright.sync_api import expect # 发布后验证用 expect 轮询替代固定 sleep expect(page.locator(.publish-success)).to_be_visible(timeout90_000)expect的轮询机制是每 100ms 重新检查一次元素状态直到超时。相比wait_for_selector它直接支持图、文案、状态的一类断言debug 信息更友好。生产环境我会把默认超时调至PLAYWRIGHT_TIMEOUT60000再在具体步骤传入更精确的 timeout这样日志里能看到是“哪一步超时”而不是“整体超时”。5. 发布节奏、失败重试与冒烟验证的工程细节5.1 发布节奏控制与定时发布的调度实现内容分发系统一旦跑起来比“能不能发布”更重要的就是“什么时候发布”。短视频平台在特定时间段午休、晚间的推荐流量池放大效应很明显批量发布必须对准这些时间窗口。常见做法是在任务表里存publish_time轮询扫描到时间到期的任务才加入执行队列不让脚本定时全量扫描减少无效访问。async def schedule_loop(): while True: due_tasks await db.fetch_tasks_due() # SQL: WHERE statuspending AND publish_time NOW() for task in due_tasks: await task_queue.put(task) await db.mark_scheduled(task.task_id) await asyncio.sleep(10)任务进入队列后执行端加一个最小时间间隔判断如果当前时间比任务计划时间提前超过 30 秒就再等一段时间。四个平台的发布时间窗口可以配置成一张时间表比如抖音 12:00-13:00 和 18:00-21:00快手 07:00-09:00 和 19:00-22:00用于自动计算发布时间避免每天手工调。5.2 失败重试与降级策略平台的异常响应不会给全你的文案很多是抽象的错误码或空白页。我的重试策略分成三个级别页面上元素未出现重试 3 次每次间隔递增上传过程中断重新上传最多 2 次平台明确提示“重复”或“违规”不重试直接置为 failed 等待人工介入。指数退避重试间隔设置为 30 秒、60 秒、120 秒比固定重试更能减少对平台的瞬时冲击。import asyncio async def retry_with_backoff(func, max_retries: int 3): for attempt in range(max_retries): try: return await func() except Exception as e: if attempt max_retries - 1: raise # 30s * 2^attempt即 30s / 60s / 120s wait_time 30 * (2 ** attempt) await asyncio.sleep(wait_time) return None # 参数说明 # 2 ** attempt 是指数退避的基数是 2这比线性重试更科学 # 因为平台风控往往会在短时间内持续拦截间隔拉大后成功率显著提升。5.3 发布后的冒烟验证回读作品列表发布成功不代表万事大吉。抖音和快手这类平台的“发布成功”提示是前端的服务端可能还要经过审核、转码。最稳妥的闭环是发布完成后跳转到账号的作品列表页检查刚发布视频的标题是否出现在第一条。这个环节会额外花 5 到 10 秒但对矩阵号来说值得因为它能帮你确认视频真正进入了待审核队列而不是前端假成功。async def smoke_check(page, title_keyword: str, profile_url: str): await page.goto(profile_url, wait_untilnetworkidle) first_item_title page.locator(.video-item-title, .feed-card-title).first await first_item_title.wait_for(statevisible, timeout30_000) actual_title await first_item_title.text_content() if title_keyword not in actual_title: raise ValueError(f发布后回读校验失败, 期望包含: {title_keyword}, 实际: {actual_title})回读校验的profile_url同样要从适配层读取因为“作品列表页”的 URL 规则各平台完全不同。快手是https://www.kuaishou.com/profile/xxx抖音是https://www.douyin.com/user/xxx视频号直接把“视频号助手”里的作品列表页当作回读页面。如果这块没做成配置化新增平台时代码要动的地方就太多了。5.4 用 playwright codegen 加速选择器排查平台改版后适配层要更新的第一件事是选择器。手动打开 devtools 找元素太慢我习惯直接用 playwright codegen 临时打开目标平台页面在录制模式下手动点一遍上传流程codegen 会把生成的 locator 全部记录下来。虽然录制的代码不能直接用于生产但它非常高效地帮你定位到平台改版后的最新选择器结构和属性名差异。# 启动 codegen拖拽式生成选择器 playwright codegen --viewport-size1440,900 --langzh-CN https://creator.douyin.com/ # 在打开的 Chromium 窗口里手动执行一次上传操作 # 窗口右侧会自动生成对应的 python 代码包含最新的 locator 逻辑。codegen 生成的定位器经常带有#root div ...这类路径选择器建议提取其中属性较稳定的style="width:16px;margin-left:4px;vertical-align:text-bottom;cursor:text;" />