尧图网页设计 
每次把最新一期视频剪辑好、配上标题和封面之后真正让人消耗精力的往往不是创作本身而是上线之后那一长串琐碎动作多平台登记、补发链接、整理“求关注”文案、提醒自己别忘了去更新状态。尤其是像“烩面大王DW”这类需要稳定更新的视频账号新视频发出后如果全靠手工记录很容易漏更、漏发、漏回复。这篇文章就从“【烩面大王DW】最新视频上线求关注”这个典型场景展开搭建一个基于 Python FastAPI SQLite 的本地视频发布管理助手。它可以把视频标题、平台、链接、发布时间统一登记自动生成“求关注”文案并支持定时巡检当一条新视频到了预设发布时间系统会自动推一条提醒把“该去发动态、该去引导关注”这件事固化成自动化流程。文章会给出完整项目结构、数据库建表语句、核心接口代码、页面模板、定时任务接入方式以及常见报错排查方案。1. 背景与核心概念1.1 视频上线后的“运营余波”很多内容创作者会有一个误区把所有时间放在视频剪辑上把发布当作最后一个步骤。实际上对一个想把账号做起来的新人而言“视频上传完成”只是运营的开始。视频上线后通常还伴随着几个环节把视频链接同步到各个粉丝群、朋友圈或动态里。在简介或评论区写出引导关注的话术。记录这条视频发布在哪个平台、发布时间是什么时候。第二天观察数据决定要不要投流或二次运营。如果这条视频有固定更新频率还要安排下一期的排期。对于一个人运营的账号来说手动做这些事情虽然也能完成但一旦视频数量增加加上多平台同时更新就容易出现记错状态、忘记提醒、同一条视频被重复推送的问题。更麻烦的是“求关注”这种话术如果每次都重新手打效率很低而且不同平台的评论区格式也不一样。如果把这个过程抽象成一个小型业务系统可以这样理解视频是一条一条的“任务”每个任务有自己的标题、所属平台、链接、状态和发布日期系统需要做的事就是登记任务、展示任务、生成运营文案并按计划触发提醒。这就是我们这篇文章要写的工具。1.2 技术选型为什么用 FastAPI SQLite APScheduler先说明一下方案背后的选型理由方便大家按自己的情况判断。Python适合快速写脚本和本地小工具生态成熟处理文本、时间、数据库都很方便。FastAPI它是当前很流行的 Python Web 框架优点是写接口简单、自带数据校验、自动生成接口文档。我们只需要提供几个页面和 API用 FastAPI 比 Flask 更“现代”后面想扩展成服务也容易。SQLite项目的第一阶段不需要单独安装 MySQLSQLite 是 Python 内置的轻量数据库数据保存在单个videos.db文件里方便拷贝和备份。对个人内容管理来说足够用。APScheduler解决定时任务问题。视频有“计划发布时间”如果希望系统到点自动提醒用 APScheduler 的 interval 或 cron 任务比较合适。Jinja2配合 FastAPI 渲染一个简单的 Web 页面方便在浏览器里查看视频状态。这套组合的特点是“本地优先、启动快、依赖少”。哪怕你还不熟悉 Web 开发也可以跟着项目结构一步步跑通。1.3 功能边界说明为了让文章不跑偏我们把功能范围控制在以下四个核心点上功能点说明视频登记新增一条视频记录包括标题、平台、链接、计划发布时间视频列表在页面和管理接口中查看所有视频记录文案生成输入视频信息自动生成带“求关注”引导语的内容定时巡检定时扫描已到发布时间但尚未提醒的视频并推送提醒至于视频播放量、粉丝数、评论数据采集等不属于本阶段的小工具范围。一方面这些指标往往依赖平台开放 API需要额外申请权限另一方面不同平台的数据规范也不一样。我们在第八节“最佳实践与工程建议”里会讲后续怎么扩展但第一步先把流程跑通。2. 环境准备与项目结构2.1 运行环境与版本说明本文示例基于以下环境操作系统Windows 10/11、macOS、Linux 均可Python 代码跨平台运行。Python 版本建议 3.10 及以上。代码中使用了str | None这种类型写法Python 3.10 开始原生支持如果你还在使用 Python 3.8/3.9可以把相关写法改成Optional[str]。Web 框架FastAPI依赖 Uvicorn 提供本地服务。数据库SQLitePython 自带不需要额外安装数据库服务。模板引擎Jinja2。定时任务APScheduler。不同版本的 FastAPI 对模板渲染接口的调用方式略有差异文中会采用当前版本常用的写法并在常见问题中说明报错时的调整方案。如果安装时提示版本冲突建议先创建虚拟环境再统一安装依赖。2.2 创建项目结构建议先创建一个目录例如video_manager。项目结构如下video_manager/ ├── app.py ├── database.py ├── push.py ├── requirements.txt ├── templates/ │ └── index.html └── videos.db过一遍各文件职责database.py负责 SQLite 连接、数据库初始化和建表。app.pyFastAPI 主程序包含接口、页面渲染、定时任务启动逻辑。push.py封装“生成关注文案”和“发送消息提醒”两个辅助函数。templates/index.html浏览器端管理页面。videos.dbSQLite 数据库文件第一次启动后自动生成。requirements.txt第三方依赖列表。2.3 安装依赖在项目目录下创建requirements.txtfastapi uvicorn[standard] jinja2 apscheduler python-dotenv然后在终端执行pip install -r requirements.txt如果你不想在当前全局 Python 环境里安装可以先创建虚拟环境python -m venv venvWindows 下激活venv\Scripts\activatemacOS 或 Linux 下激活source venv/bin/activate之后继续安装依赖即可。创建虚拟环境是一个推荐习惯可以避免把系统 Python 环境搞乱尤其是你电脑上还有其他项目的时候。3. 数据库设计与初始化3.1 表结构设计我们需要记录视频的基本信息和运营状态表名定为videos字段设计如下字段类型含义idINTEGER PRIMARY KEY AUTOINCREMENT视频记录 IDtitleTEXT NOT NULL视频标题platformTEXT NOT NULL发布平台名称video_urlTEXT NOT NULL视频链接publish_dateTEXT计划发布时间格式YYYY-MM-DD HH:MM:SSstatusTEXT DEFAULT pending记录状态pending 待提醒notified 已提醒remarkTEXT DEFAULT 备注信息created_atTEXT DEFAULT CURRENT_TIMESTAMP创建时间SQLite 自动写入状态字段没有设计得很复杂只区分pending和notified。这意味着一条视频刚录入时是待提醒状态已经在页面点击“生成提醒”或定时任务成功巡检后状态更新为已提醒避免重复推送。这里需要提醒一下publish_date在 SQLite 中存的是字符串所以代码里必须统一时间为YYYY-MM-DD HH:MM:SS格式否则后面publish_date 当前时间的字符串比较会不准确。这是一个很容易踩的坑我会在常见问题里再强调一次。3.2 数据库初始化代码在database.py中写入# 文件路径video_manager/database.py import sqlite3 from contextlib import contextmanager from pathlib import Path BASE_DIR Path(__file__).resolve().parent DB_PATH BASE_DIR / videos.db contextmanager def get_connection(): conn sqlite3.connect(DB_PATH) conn.row_factory sqlite3.Row try: yield conn conn.commit() except Exception: conn.rollback() raise finally: conn.close() def init_db(): with get_connection() as conn: conn.execute( CREATE TABLE IF NOT EXISTS videos ( id INTEGER PRIMARY KEY AUTOINCREMENT, title TEXT NOT NULL, platform TEXT NOT NULL, video_url TEXT NOT NULL, publish_date TEXT, status TEXT DEFAULT pending, remark TEXT DEFAULT , created_at TEXT DEFAULT CURRENT_TIMESTAMP ) ) conn.execute( CREATE INDEX IF NOT EXISTS idx_videos_status_publish ON videos(status, publish_date) )代码说明get_connection()使用contextmanager封装函数体内执行完数据库操作后自动提交事务异常时回滚最后关闭连接。设置row_factory sqlite3.Row这样查询出的每一行既能按列名取值也能转换成字典。CREATE TABLE IF NOT EXISTS保证重复启动不会报错。增加的索引用于优化后续定时任务的查询WHERE status pending AND publish_date ?是否走索引取决于数据量但提前建上不会有坏处。4. 基于 FastAPI 实现核心接口4.1 视频登记与查询接口接下来编写主程序app.py。先引入所需模块并初始化 FastAPI 和调度器。# 文件路径video_manager/app.py from contextlib import asynccontextmanager from datetime import datetime from apscheduler.schedulers.asyncio import AsyncIOScheduler from fastapi import FastAPI, HTTPException, Request from fastapi.responses import HTMLResponse from fastapi.templating import Jinja2Templates from pydantic import BaseModel, Field from database import get_connection, init_db from push import build_follow_copy, send_serverchan # 全局调度器 scheduler AsyncIOScheduler()然后定义视频新增接口的数据模型# 文件路径video_manager/app.py继续追加 class VideoCreate(BaseModel): title: str Field(..., min_length1, max_length200, description视频标题) platform: str Field(..., min_length1, max_length50, description发布平台) video_url: str Field(..., min_length1, description视频链接) publish_date: str | None Field(None, description计划发布时间如 2025-06-01 10:00:00) status: str Field(pending, description状态pending/notified) class VideoOut(VideoCreate): id: int created_at: str再实现注册接口和列表接口# 文件路径video_manager/app.py继续追加 app.post(/api/videos, response_modelVideoOut) def create_video(payload: VideoCreate): if payload.status not in (pending, notified): raise HTTPException(status_code400, detailstatus 只允许 pending 或 notified) publish_date payload.publish_date if not publish_date: publish_date datetime.now().strftime(%Y-%m-%d %H:%M:%S) with get_connection() as conn: cursor conn.execute( INSERT INTO videos (title, platform, video_url, publish_date, status) VALUES (?, ?, ?, ?, ?) , (payload.title, payload.platform, payload.video_url, publish_date, payload.status), ) row conn.execute( SELECT * FROM videos WHERE id ?, (cursor.lastrowid,) ).fetchone() return dict(row) app.get(/api/videos) def list_videos(): with get_connection() as conn: rows conn.execute(SELECT * FROM videos ORDER BY id DESC).fetchall() return [dict(row) for row in rows]创建视频接口的作用很直接接收 JSON 参数把记录写入 SQLite并返回包含自增 ID 的完整记录。业务上你可以在浏览器页面手工填写也可以后续写一个爬虫或脚本把各平台视频信息自动投递到这个新增接口。要注意cursor.lastrowid是 SQLite 插入后返回的自增主键用它再次查询可以拿到数据库自动补充的id、created_at等字段。4.2 页面渲染与生成“求关注”文案接口为了让非技术用户也可以操作我们增加一个首页路由用模板展示视频列表。# 文件路径video_manager/app.py继续追加 app FastAPI(title视频发布管理助手) templates Jinja2Templates(directorytemplates) app.get(/, response_classHTMLResponse) def index_page(request: Request): with get_connection() as conn: rows conn.execute(SELECT * FROM videos ORDER BY id DESC).fetchall() return templates.TemplateResponse( request, index.html, {videos: rows}, ) app.post(/api/videos/{video_id}/notify) def notify_video(video_id: int): with get_connection() as conn: row conn.execute(SELECT * FROM videos WHERE id ?, (video_id,)).fetchone() if row is None: raise HTTPException(status_code404, detail视频记录不存在) title row[title] platform row[platform] video_url row[video_url] copy build_follow_copy(title, platform, video_url) push_result send_serverchan(视频更新提醒, copy) conn.execute( UPDATE videos SET status notified WHERE id ?, (video_id,) ) return { id: video_id, copy: copy, push_result: push_result, }/api/videos/{video_id}/notify是两个核心动作的组合根据video_id查出视频信息。调用build_follow_copy生成关注文案。调用send_serverchan发送通知。把状态从pending改成notified防止下一次定时任务重复处理。这里并没有把文案生成逻辑塞到接口里而是放到了单独的push.py。这种拆分有利于代码复用当你以后要接入其他推送渠道时只需要改push.py而无需动接口逻辑。4.3 关注文案生成与推送模块在push.py中实现文案生成和消息推送。# 文件路径video_manager/push.py import os import urllib.parse import urllib.request SERVERCHAN_SENDKEY os.getenv(SERVERCHAN_SENDKEY, ) def build_follow_copy(title: str, platform: str, video_url: str) - str: return ( f【最新视频上线】《{title}》已经在 {platform} 发布啦。\n f观看链接{video_url}\n 如果你觉得这期内容对你有用欢迎点赞、评论、转发。\n 点击关注【烩面大王DW】后续更新不迷路最新视频第一时间推送给你 ) def send_serverchan(title: str, desp: str) - dict: if not SERVERCHAN_SENDKEY: return { success: False, message: 未配置 SERVERCHAN_SENDKEY跳过推送只返回文案, } api_url fhttps://sctapi.ftqq.com/{SERVERCHAN_SENDKEY}.send form_data urllib.parse.urlencode({title: title, desp: desp}).encode(utf-8) req urllib.request.Request(urlapi_url, dataform_data) try: with urllib.request.urlopen(req, timeout10) as resp: body resp.read().decode(utf-8) return {success: True, message: 推送成功, response: body} except Exception as exc: return {success: False, message: f推送失败{exc}}关于推送模块多说几句。这里以“Server酱”作为示例推送通道它通过一个 SendKey 将消息转发到微信。这类第三方推送服务适合个人项目不需要自己装 App、不需要维护移动端只需要在环境变量里配置SERVERCHAN_SENDKEY。如果你不方便使用 Server酱也可以替换成 Bark、PushPlus、邮件 SMTP 或企业微信机器人思路一致send_serverchan只是负责“把标题和正文发出去”对外暴露的接口参数不变。在未配置 SendKey 的情况下send_serverchan不会抛异常而是返回一个successFalse的结果。这保证你在本地测试接口时即使没有推送凭据也能正常拿到生成的“求关注”文案。5. 定时巡检与消息推送5.1 在 FastAPI 生命周期中启动调度器定时任务并不需要所有读者都用到但对一个视频账号运营者来说它的价值在于不需要每天记着“我几点发视频、几点要去互动”系统到点会自动提醒。在app.py中我们把调度器放入 FastAPI 的lifespan让 Web 服务启动时自动创建数据库并启动定时任务# 文件路径video_manager/app.py继续追加 asynccontextmanager async def lifespan(app: FastAPI): init_db() scheduler.start() yield scheduler.shutdown()然后把app的创建改成app FastAPI(title视频发布管理助手, lifespanlifespan)要注意顺序原来的代码里已经把app FastAPI(title视频发布管理助手)写在前面了请改成带lifespan的版本避免两处重复定义。5.2 定时扫描逻辑接下来定义巡检业务函数。它的任务是每隔一段时间扫描一次videos表找出已经到达发布时间、但状态还是pending的视频生成文案并推送然后把状态改成notified。# 文件路径video_manager/app.py继续追加 scheduler.scheduled_job(interval, minutes30, idscan_pending_videos_job) def scan_pending_videos_job(): now datetime.now().strftime(%Y-%m-%d %H:%M:%S) with get_connection() as conn: rows conn.execute( SELECT id, title, platform, video_url FROM videos WHERE status pending AND publish_date ? , (now,), ).fetchall() for row in rows: copy build_follow_copy( row[title], row[platform], row[video_url], ) result send_serverchan(视频更新提醒, copy) print(f[定时任务] 视频 {row[id]} 推送结果{result}) conn.execute( UPDATE videos SET status notified WHERE id ?, (row[id],), )代码整体思路比较清晰先取当前时间转成YYYY-MM-DD HH:MM:SS字符串。查询所有status pending且publish_date now的记录。对每一条记录生成文案并推送。推送完成后无论推送成功还是失败这里都先把状态置为notified避免定时任务每次启动都重复处理同一条历史记录。这里存在一个工程上的取舍把状态改成notified是在推送动作之后但如果推送失败这条视频可能就不再被巡检到你需要在页面或接口里手动重新推送。如果你希望失败自动重试可以把状态更新的条件改成“仅当推送成功时才置为 notified”并在push_result中读取success字段判断。为了演示职责边界代码保留了最简逻辑在实际项目中建议加上失败重试或告警。定时任务的触发间隔默认是 30 分钟。本地调试时不想等 30 分钟可以把scheduler.scheduled_job(interval, minutes30, ...)改成minutes1验证通过后再改回来。6. 完整页面与本地运行验证6.1 设计一个轻量的管理页面我们需要一个简单的操作页面方便在浏览器里“新增视频”和“点击生成关注文案”。在templates/index.html中写入以下代码!-- 文件路径video_manager/templates/index.html -- !DOCTYPE html html langzh-CN head meta charsetUTF-8 title视频发布管理助手/title style body { font-family: Microsoft YaHei, sans-serif; max-width: 960px; margin: 40px auto; padding: 0 16px; color: #333; } h1 { font-size: 22px; } form { background: #f7f8fa; padding: 16px 20px; border-radius: 8px; margin-bottom: 20px; } label { display: block; margin: 10px 0 4px; font-weight: 600; } input, select { box-sizing: border-box; width: 100%; padding: 8px; border: 1px solid #ccc; border-radius: 4px; } button { margin-top: 12px; padding: 8px 18px; border: 0; border-radius: 4px; background: #2d6cdf; color: #fff; cursor: pointer; } table { width: 100%; border-collapse: collapse; margin-top: 8px; } th, td { border: 1px solid #e3e3e3; padding: 8px; text-align: left; font-size: 14px; } .notice { padding: 10px 12px; margin: 16px 0; background: #eef6ff; border-left: 4px solid #2d6cdf; word-break: break-all; } /style /head body h1烩面大王DW · 视频发布管理助手/h1 form idvideoForm label fortitle视频标题/label input typetext idtitle nametitle required label forplatform发布平台/label select idplatform nameplatform option value视频号视频号/option option valueB站B站/option option value抖音抖音/option option value小红书小红书/option option value快手快手/option /select label forvideo_url视频链接/label input typeurl idvideo_url namevideo_url required label forpublish_date计划发布时间/label input typetext idpublish_date namepublish_date placeholder2025-06-01 10:00:00留空表示立即发布 button typesubmit立即登记视频/button /form div idnotice classnotice styledisplay: none;/div table thead tr thID/th th标题/th th平台/th th计划发布时间/th th状态/th th操作/th /tr /thead tbody {% for video in videos %} tr td{{ video[id] }}/td td{{ video[title] }}/td td{{ video[platform] }}/td td{{ video[publish_date] }}/td td {% if video[status] pending %} 待提醒 {% else %} 已提醒 {% endif %} /td td button onclicknotifyVideo({{ video[id] }})生成求关注文案/button /td /tr {% else %} tr td colspan6还没有视频记录先在上方表单中添加一条。/td /tr {% endfor %} /tbody /table script document.getElementById(videoForm).addEventListener(submit, async function (event) { event.preventDefault(); const formData new FormData(this); const payload { title: formData.get(title), platform: formData.get(platform), video_url: formData.get(video_url), publish_date: formData.get(publish_date) || null }; const res await fetch(/api/videos, { method: POST, headers: {Content-Type: application/json}, body: JSON.stringify(payload) }); const data await res.json(); showNotice(登记成功视频 ID data.id); setTimeout(() location.reload(), 800); }); async function notifyVideo(videoId) { const res await fetch(/api/videos/ videoId /notify, { method: POST }); const data await res.json(); showNotice(data.copy || JSON.stringify(data)); } function showNotice(text) { const box document.getElementById(notice); box.style.display block; box.innerText text; } /script /body /html页面功能不复杂核心逻辑有两点。一是表单提交时用fetch调用/api/videos接口把视频数据写入数据库二是点击表格中“生成求关注文案”按钮后调用/api/videos/{id}/notify接口页面上会显示生成的文案。为了避免小白操作时把时间格式写错页面里给出了明确的占位提示。模板中的{% for video in videos %}是 Jinja2 的循环语法它会把后端传入的视频列表逐行渲染到表格中。如果列表为空则走{% else %}分支。6.2 启动服务在video_manager目录下执行uvicorn app:app --reload --port 8000启动成功后终端会显示 Uvicorn 运行地址。打开浏览器访问http://127.0.0.1:8000你应该能看到标题为“视频发布管理助手”的页面。首次启动时会自动生成videos.db所以不用手动建库。需要特别说明的是--reload适合开发调试。它会在代码文件发生变化时自动重启服务。但在调试定时任务时如果代码频繁变更调度器会在重启后重新注册任务有可能出现重复日志建议验证定时任务时去掉--reload或者接受这种开发期小问题。真正部署时也不建议在systemd或 Docker 环境中使用--reload。6.3 用接口验证完整流程暂时不在页面操作先用curl测试接口更直观。第一步新增一条视频记录curl -X POST http://127.0.0.1:8000/api/videos \ -H Content-Type: application/json \ -d { title: 【烩面大王DW】最新视频上线求关注, platform: B站, video_url: https://www.bilibili.com/video/example, publish_date: 2025-06-01 10:00:00 }预期返回结果大致如下{ title: 【烩面大王DW】最新视频上线求关注, platform: B站, video_url: https://www.bilibili.com/video/example, publish_date: 2025-06-01 10:00:00, status: pending, id: 1, created_at: 2025-06-01 09:30:00 }第二步查询视频列表curl http://127.0.0.1:8000/api/videos第三步点击通知接口验证“生成求关注文案”逻辑curl -X POST http://127.0.0.1:8000/api/videos/1/notify接口会返回{ id: 1, copy: 【最新视频上线】《【烩面大王DW】最新视频上线求关注》已经在 B站 发布啦。\n观看链接...\n如果你觉得这期内容对你有用欢迎点赞、评论、转发。\n点击关注【烩面大王DW】后续更新不迷路最新视频第一时间推送给你, push_result: { success: false, message: 未配置 SERVERCHAN_SENDKEY跳过推送只返回文案 } }如果没有配置 SendKey你会看到push_result.successfalse但copy字段已经生成了完整文案。如果想要实际推送在启动服务前设置环境变量Windows PowerShell$env:SERVERCHAN_SENDKEY你的SendKeymacOS / Linuxexport SERVERCHAN_SENDKEY你的SendKey然后再启动服务。建议把 SendKey 放在.env文件或环境变量里不要写死在push.py源码中避免代码泄露后被人盗用推送通道。7. 常见问题与排查思路在运行这套代码时新手容易遇到下面几个问题。我把现象、可能原因和解决思路整理成一张表方便直接查阅。问题现象常见原因解决思路ModuleNotFoundError: No module named fastapi没有安装依赖或激活虚拟环境执行pip install -r requirements.txt并确认当前终端处于对应虚拟环境ModuleNotFoundError: No module named apschedulerrequirements 安装不完整单独执行pip install apscheduler或重新安装 requirements访问http://127.0.0.1:8000报 500数据库没有初始化或templates目录路径不对确认在video_manager目录下启动且lifespan中调用了init_db()sqlite3.OperationalError: no such table: videos启动时没有建表检查启动命令是否为uvicorn app:app --reload确保lifespan生效jinja2.exceptions.TemplateNotFound: index.html当前工作目录不是项目根目录在video_manager目录下执行命令不要在其他目录启动 uvicornTemplateResponse() missing ...Starlette 模板渲染写法与版本不匹配新版使用templates.TemplateResponse(request, index.html, {...})如果你的版本较旧改成templates.TemplateResponse(index.html, {request: request, ...})定时任务 30 分钟不触发间隔设置太长或者代码中还有语法错误把minutes30临时改为minutes1观察终端日志定时任务执行后数据库状态没变UPDATE语句没有提交确认用的是get_connection()上下文管理器里面包含conn.commit()Server酱返回推送失败SendKey 无效、网络不通或服务不稳定检查SERVERCHAN_SENDKEY环境变量确认 api_url 格式查看返回的response字段如果你遇到的报错不在表里最简单的方法是分三步排查先确认所有第三方依赖是否安装成功。再确认启动目录是不是video_manager根目录因为模板路径和数据库路径都用相对目录。最后看终端完整报错堆栈重点搜索File和Error两处定位是哪一行代码出错。8. 最佳实践与工程建议代码跑通只是第一步真正落到“长期使用”的时候以下几点建议会更重要。8.1 配置不要写死在代码里push.py里通过os.getenv(SERVERCHAN_SENDKEY)读取环境变量这个习惯值得保持。无论是推送 SendKey、数据库路径还是未来要接入的平台开放 API Secret都不要写死在源码里。配置文件方面可以使用python-dotenv在项目根目录创建.env文件SERVERCHAN_SENDKEY 基于FastAPI与SQLite的视频发布管理助手：自动登记与定时提醒
每次把最新一期视频剪辑好、配上标题和封面之后真正让人消耗精力的往往不是创作本身而是上线之后那一长串琐碎动作多平台登记、补发链接、整理“求关注”文案、提醒自己别忘了去更新状态。尤其是像“烩面大王DW”这类需要稳定更新的视频账号新视频发出后如果全靠手工记录很容易漏更、漏发、漏回复。这篇文章就从“【烩面大王DW】最新视频上线求关注”这个典型场景展开搭建一个基于 Python FastAPI SQLite 的本地视频发布管理助手。它可以把视频标题、平台、链接、发布时间统一登记自动生成“求关注”文案并支持定时巡检当一条新视频到了预设发布时间系统会自动推一条提醒把“该去发动态、该去引导关注”这件事固化成自动化流程。文章会给出完整项目结构、数据库建表语句、核心接口代码、页面模板、定时任务接入方式以及常见报错排查方案。1. 背景与核心概念1.1 视频上线后的“运营余波”很多内容创作者会有一个误区把所有时间放在视频剪辑上把发布当作最后一个步骤。实际上对一个想把账号做起来的新人而言“视频上传完成”只是运营的开始。视频上线后通常还伴随着几个环节把视频链接同步到各个粉丝群、朋友圈或动态里。在简介或评论区写出引导关注的话术。记录这条视频发布在哪个平台、发布时间是什么时候。第二天观察数据决定要不要投流或二次运营。如果这条视频有固定更新频率还要安排下一期的排期。对于一个人运营的账号来说手动做这些事情虽然也能完成但一旦视频数量增加加上多平台同时更新就容易出现记错状态、忘记提醒、同一条视频被重复推送的问题。更麻烦的是“求关注”这种话术如果每次都重新手打效率很低而且不同平台的评论区格式也不一样。如果把这个过程抽象成一个小型业务系统可以这样理解视频是一条一条的“任务”每个任务有自己的标题、所属平台、链接、状态和发布日期系统需要做的事就是登记任务、展示任务、生成运营文案并按计划触发提醒。这就是我们这篇文章要写的工具。1.2 技术选型为什么用 FastAPI SQLite APScheduler先说明一下方案背后的选型理由方便大家按自己的情况判断。Python适合快速写脚本和本地小工具生态成熟处理文本、时间、数据库都很方便。FastAPI它是当前很流行的 Python Web 框架优点是写接口简单、自带数据校验、自动生成接口文档。我们只需要提供几个页面和 API用 FastAPI 比 Flask 更“现代”后面想扩展成服务也容易。SQLite项目的第一阶段不需要单独安装 MySQLSQLite 是 Python 内置的轻量数据库数据保存在单个videos.db文件里方便拷贝和备份。对个人内容管理来说足够用。APScheduler解决定时任务问题。视频有“计划发布时间”如果希望系统到点自动提醒用 APScheduler 的 interval 或 cron 任务比较合适。Jinja2配合 FastAPI 渲染一个简单的 Web 页面方便在浏览器里查看视频状态。这套组合的特点是“本地优先、启动快、依赖少”。哪怕你还不熟悉 Web 开发也可以跟着项目结构一步步跑通。1.3 功能边界说明为了让文章不跑偏我们把功能范围控制在以下四个核心点上功能点说明视频登记新增一条视频记录包括标题、平台、链接、计划发布时间视频列表在页面和管理接口中查看所有视频记录文案生成输入视频信息自动生成带“求关注”引导语的内容定时巡检定时扫描已到发布时间但尚未提醒的视频并推送提醒至于视频播放量、粉丝数、评论数据采集等不属于本阶段的小工具范围。一方面这些指标往往依赖平台开放 API需要额外申请权限另一方面不同平台的数据规范也不一样。我们在第八节“最佳实践与工程建议”里会讲后续怎么扩展但第一步先把流程跑通。2. 环境准备与项目结构2.1 运行环境与版本说明本文示例基于以下环境操作系统Windows 10/11、macOS、Linux 均可Python 代码跨平台运行。Python 版本建议 3.10 及以上。代码中使用了str | None这种类型写法Python 3.10 开始原生支持如果你还在使用 Python 3.8/3.9可以把相关写法改成Optional[str]。Web 框架FastAPI依赖 Uvicorn 提供本地服务。数据库SQLitePython 自带不需要额外安装数据库服务。模板引擎Jinja2。定时任务APScheduler。不同版本的 FastAPI 对模板渲染接口的调用方式略有差异文中会采用当前版本常用的写法并在常见问题中说明报错时的调整方案。如果安装时提示版本冲突建议先创建虚拟环境再统一安装依赖。2.2 创建项目结构建议先创建一个目录例如video_manager。项目结构如下video_manager/ ├── app.py ├── database.py ├── push.py ├── requirements.txt ├── templates/ │ └── index.html └── videos.db过一遍各文件职责database.py负责 SQLite 连接、数据库初始化和建表。app.pyFastAPI 主程序包含接口、页面渲染、定时任务启动逻辑。push.py封装“生成关注文案”和“发送消息提醒”两个辅助函数。templates/index.html浏览器端管理页面。videos.dbSQLite 数据库文件第一次启动后自动生成。requirements.txt第三方依赖列表。2.3 安装依赖在项目目录下创建requirements.txtfastapi uvicorn[standard] jinja2 apscheduler python-dotenv然后在终端执行pip install -r requirements.txt如果你不想在当前全局 Python 环境里安装可以先创建虚拟环境python -m venv venvWindows 下激活venv\Scripts\activatemacOS 或 Linux 下激活source venv/bin/activate之后继续安装依赖即可。创建虚拟环境是一个推荐习惯可以避免把系统 Python 环境搞乱尤其是你电脑上还有其他项目的时候。3. 数据库设计与初始化3.1 表结构设计我们需要记录视频的基本信息和运营状态表名定为videos字段设计如下字段类型含义idINTEGER PRIMARY KEY AUTOINCREMENT视频记录 IDtitleTEXT NOT NULL视频标题platformTEXT NOT NULL发布平台名称video_urlTEXT NOT NULL视频链接publish_dateTEXT计划发布时间格式YYYY-MM-DD HH:MM:SSstatusTEXT DEFAULT pending记录状态pending 待提醒notified 已提醒remarkTEXT DEFAULT 备注信息created_atTEXT DEFAULT CURRENT_TIMESTAMP创建时间SQLite 自动写入状态字段没有设计得很复杂只区分pending和notified。这意味着一条视频刚录入时是待提醒状态已经在页面点击“生成提醒”或定时任务成功巡检后状态更新为已提醒避免重复推送。这里需要提醒一下publish_date在 SQLite 中存的是字符串所以代码里必须统一时间为YYYY-MM-DD HH:MM:SS格式否则后面publish_date 当前时间的字符串比较会不准确。这是一个很容易踩的坑我会在常见问题里再强调一次。3.2 数据库初始化代码在database.py中写入# 文件路径video_manager/database.py import sqlite3 from contextlib import contextmanager from pathlib import Path BASE_DIR Path(__file__).resolve().parent DB_PATH BASE_DIR / videos.db contextmanager def get_connection(): conn sqlite3.connect(DB_PATH) conn.row_factory sqlite3.Row try: yield conn conn.commit() except Exception: conn.rollback() raise finally: conn.close() def init_db(): with get_connection() as conn: conn.execute( CREATE TABLE IF NOT EXISTS videos ( id INTEGER PRIMARY KEY AUTOINCREMENT, title TEXT NOT NULL, platform TEXT NOT NULL, video_url TEXT NOT NULL, publish_date TEXT, status TEXT DEFAULT pending, remark TEXT DEFAULT , created_at TEXT DEFAULT CURRENT_TIMESTAMP ) ) conn.execute( CREATE INDEX IF NOT EXISTS idx_videos_status_publish ON videos(status, publish_date) )代码说明get_connection()使用contextmanager封装函数体内执行完数据库操作后自动提交事务异常时回滚最后关闭连接。设置row_factory sqlite3.Row这样查询出的每一行既能按列名取值也能转换成字典。CREATE TABLE IF NOT EXISTS保证重复启动不会报错。增加的索引用于优化后续定时任务的查询WHERE status pending AND publish_date ?是否走索引取决于数据量但提前建上不会有坏处。4. 基于 FastAPI 实现核心接口4.1 视频登记与查询接口接下来编写主程序app.py。先引入所需模块并初始化 FastAPI 和调度器。# 文件路径video_manager/app.py from contextlib import asynccontextmanager from datetime import datetime from apscheduler.schedulers.asyncio import AsyncIOScheduler from fastapi import FastAPI, HTTPException, Request from fastapi.responses import HTMLResponse from fastapi.templating import Jinja2Templates from pydantic import BaseModel, Field from database import get_connection, init_db from push import build_follow_copy, send_serverchan # 全局调度器 scheduler AsyncIOScheduler()然后定义视频新增接口的数据模型# 文件路径video_manager/app.py继续追加 class VideoCreate(BaseModel): title: str Field(..., min_length1, max_length200, description视频标题) platform: str Field(..., min_length1, max_length50, description发布平台) video_url: str Field(..., min_length1, description视频链接) publish_date: str | None Field(None, description计划发布时间如 2025-06-01 10:00:00) status: str Field(pending, description状态pending/notified) class VideoOut(VideoCreate): id: int created_at: str再实现注册接口和列表接口# 文件路径video_manager/app.py继续追加 app.post(/api/videos, response_modelVideoOut) def create_video(payload: VideoCreate): if payload.status not in (pending, notified): raise HTTPException(status_code400, detailstatus 只允许 pending 或 notified) publish_date payload.publish_date if not publish_date: publish_date datetime.now().strftime(%Y-%m-%d %H:%M:%S) with get_connection() as conn: cursor conn.execute( INSERT INTO videos (title, platform, video_url, publish_date, status) VALUES (?, ?, ?, ?, ?) , (payload.title, payload.platform, payload.video_url, publish_date, payload.status), ) row conn.execute( SELECT * FROM videos WHERE id ?, (cursor.lastrowid,) ).fetchone() return dict(row) app.get(/api/videos) def list_videos(): with get_connection() as conn: rows conn.execute(SELECT * FROM videos ORDER BY id DESC).fetchall() return [dict(row) for row in rows]创建视频接口的作用很直接接收 JSON 参数把记录写入 SQLite并返回包含自增 ID 的完整记录。业务上你可以在浏览器页面手工填写也可以后续写一个爬虫或脚本把各平台视频信息自动投递到这个新增接口。要注意cursor.lastrowid是 SQLite 插入后返回的自增主键用它再次查询可以拿到数据库自动补充的id、created_at等字段。4.2 页面渲染与生成“求关注”文案接口为了让非技术用户也可以操作我们增加一个首页路由用模板展示视频列表。# 文件路径video_manager/app.py继续追加 app FastAPI(title视频发布管理助手) templates Jinja2Templates(directorytemplates) app.get(/, response_classHTMLResponse) def index_page(request: Request): with get_connection() as conn: rows conn.execute(SELECT * FROM videos ORDER BY id DESC).fetchall() return templates.TemplateResponse( request, index.html, {videos: rows}, ) app.post(/api/videos/{video_id}/notify) def notify_video(video_id: int): with get_connection() as conn: row conn.execute(SELECT * FROM videos WHERE id ?, (video_id,)).fetchone() if row is None: raise HTTPException(status_code404, detail视频记录不存在) title row[title] platform row[platform] video_url row[video_url] copy build_follow_copy(title, platform, video_url) push_result send_serverchan(视频更新提醒, copy) conn.execute( UPDATE videos SET status notified WHERE id ?, (video_id,) ) return { id: video_id, copy: copy, push_result: push_result, }/api/videos/{video_id}/notify是两个核心动作的组合根据video_id查出视频信息。调用build_follow_copy生成关注文案。调用send_serverchan发送通知。把状态从pending改成notified防止下一次定时任务重复处理。这里并没有把文案生成逻辑塞到接口里而是放到了单独的push.py。这种拆分有利于代码复用当你以后要接入其他推送渠道时只需要改push.py而无需动接口逻辑。4.3 关注文案生成与推送模块在push.py中实现文案生成和消息推送。# 文件路径video_manager/push.py import os import urllib.parse import urllib.request SERVERCHAN_SENDKEY os.getenv(SERVERCHAN_SENDKEY, ) def build_follow_copy(title: str, platform: str, video_url: str) - str: return ( f【最新视频上线】《{title}》已经在 {platform} 发布啦。\n f观看链接{video_url}\n 如果你觉得这期内容对你有用欢迎点赞、评论、转发。\n 点击关注【烩面大王DW】后续更新不迷路最新视频第一时间推送给你 ) def send_serverchan(title: str, desp: str) - dict: if not SERVERCHAN_SENDKEY: return { success: False, message: 未配置 SERVERCHAN_SENDKEY跳过推送只返回文案, } api_url fhttps://sctapi.ftqq.com/{SERVERCHAN_SENDKEY}.send form_data urllib.parse.urlencode({title: title, desp: desp}).encode(utf-8) req urllib.request.Request(urlapi_url, dataform_data) try: with urllib.request.urlopen(req, timeout10) as resp: body resp.read().decode(utf-8) return {success: True, message: 推送成功, response: body} except Exception as exc: return {success: False, message: f推送失败{exc}}关于推送模块多说几句。这里以“Server酱”作为示例推送通道它通过一个 SendKey 将消息转发到微信。这类第三方推送服务适合个人项目不需要自己装 App、不需要维护移动端只需要在环境变量里配置SERVERCHAN_SENDKEY。如果你不方便使用 Server酱也可以替换成 Bark、PushPlus、邮件 SMTP 或企业微信机器人思路一致send_serverchan只是负责“把标题和正文发出去”对外暴露的接口参数不变。在未配置 SendKey 的情况下send_serverchan不会抛异常而是返回一个successFalse的结果。这保证你在本地测试接口时即使没有推送凭据也能正常拿到生成的“求关注”文案。5. 定时巡检与消息推送5.1 在 FastAPI 生命周期中启动调度器定时任务并不需要所有读者都用到但对一个视频账号运营者来说它的价值在于不需要每天记着“我几点发视频、几点要去互动”系统到点会自动提醒。在app.py中我们把调度器放入 FastAPI 的lifespan让 Web 服务启动时自动创建数据库并启动定时任务# 文件路径video_manager/app.py继续追加 asynccontextmanager async def lifespan(app: FastAPI): init_db() scheduler.start() yield scheduler.shutdown()然后把app的创建改成app FastAPI(title视频发布管理助手, lifespanlifespan)要注意顺序原来的代码里已经把app FastAPI(title视频发布管理助手)写在前面了请改成带lifespan的版本避免两处重复定义。5.2 定时扫描逻辑接下来定义巡检业务函数。它的任务是每隔一段时间扫描一次videos表找出已经到达发布时间、但状态还是pending的视频生成文案并推送然后把状态改成notified。# 文件路径video_manager/app.py继续追加 scheduler.scheduled_job(interval, minutes30, idscan_pending_videos_job) def scan_pending_videos_job(): now datetime.now().strftime(%Y-%m-%d %H:%M:%S) with get_connection() as conn: rows conn.execute( SELECT id, title, platform, video_url FROM videos WHERE status pending AND publish_date ? , (now,), ).fetchall() for row in rows: copy build_follow_copy( row[title], row[platform], row[video_url], ) result send_serverchan(视频更新提醒, copy) print(f[定时任务] 视频 {row[id]} 推送结果{result}) conn.execute( UPDATE videos SET status notified WHERE id ?, (row[id],), )代码整体思路比较清晰先取当前时间转成YYYY-MM-DD HH:MM:SS字符串。查询所有status pending且publish_date now的记录。对每一条记录生成文案并推送。推送完成后无论推送成功还是失败这里都先把状态置为notified避免定时任务每次启动都重复处理同一条历史记录。这里存在一个工程上的取舍把状态改成notified是在推送动作之后但如果推送失败这条视频可能就不再被巡检到你需要在页面或接口里手动重新推送。如果你希望失败自动重试可以把状态更新的条件改成“仅当推送成功时才置为 notified”并在push_result中读取success字段判断。为了演示职责边界代码保留了最简逻辑在实际项目中建议加上失败重试或告警。定时任务的触发间隔默认是 30 分钟。本地调试时不想等 30 分钟可以把scheduler.scheduled_job(interval, minutes30, ...)改成minutes1验证通过后再改回来。6. 完整页面与本地运行验证6.1 设计一个轻量的管理页面我们需要一个简单的操作页面方便在浏览器里“新增视频”和“点击生成关注文案”。在templates/index.html中写入以下代码!-- 文件路径video_manager/templates/index.html -- !DOCTYPE html html langzh-CN head meta charsetUTF-8 title视频发布管理助手/title style body { font-family: Microsoft YaHei, sans-serif; max-width: 960px; margin: 40px auto; padding: 0 16px; color: #333; } h1 { font-size: 22px; } form { background: #f7f8fa; padding: 16px 20px; border-radius: 8px; margin-bottom: 20px; } label { display: block; margin: 10px 0 4px; font-weight: 600; } input, select { box-sizing: border-box; width: 100%; padding: 8px; border: 1px solid #ccc; border-radius: 4px; } button { margin-top: 12px; padding: 8px 18px; border: 0; border-radius: 4px; background: #2d6cdf; color: #fff; cursor: pointer; } table { width: 100%; border-collapse: collapse; margin-top: 8px; } th, td { border: 1px solid #e3e3e3; padding: 8px; text-align: left; font-size: 14px; } .notice { padding: 10px 12px; margin: 16px 0; background: #eef6ff; border-left: 4px solid #2d6cdf; word-break: break-all; } /style /head body h1烩面大王DW · 视频发布管理助手/h1 form idvideoForm label fortitle视频标题/label input typetext idtitle nametitle required label forplatform发布平台/label select idplatform nameplatform option value视频号视频号/option option valueB站B站/option option value抖音抖音/option option value小红书小红书/option option value快手快手/option /select label forvideo_url视频链接/label input typeurl idvideo_url namevideo_url required label forpublish_date计划发布时间/label input typetext idpublish_date namepublish_date placeholder2025-06-01 10:00:00留空表示立即发布 button typesubmit立即登记视频/button /form div idnotice classnotice styledisplay: none;/div table thead tr thID/th th标题/th th平台/th th计划发布时间/th th状态/th th操作/th /tr /thead tbody {% for video in videos %} tr td{{ video[id] }}/td td{{ video[title] }}/td td{{ video[platform] }}/td td{{ video[publish_date] }}/td td {% if video[status] pending %} 待提醒 {% else %} 已提醒 {% endif %} /td td button onclicknotifyVideo({{ video[id] }})生成求关注文案/button /td /tr {% else %} tr td colspan6还没有视频记录先在上方表单中添加一条。/td /tr {% endfor %} /tbody /table script document.getElementById(videoForm).addEventListener(submit, async function (event) { event.preventDefault(); const formData new FormData(this); const payload { title: formData.get(title), platform: formData.get(platform), video_url: formData.get(video_url), publish_date: formData.get(publish_date) || null }; const res await fetch(/api/videos, { method: POST, headers: {Content-Type: application/json}, body: JSON.stringify(payload) }); const data await res.json(); showNotice(登记成功视频 ID data.id); setTimeout(() location.reload(), 800); }); async function notifyVideo(videoId) { const res await fetch(/api/videos/ videoId /notify, { method: POST }); const data await res.json(); showNotice(data.copy || JSON.stringify(data)); } function showNotice(text) { const box document.getElementById(notice); box.style.display block; box.innerText text; } /script /body /html页面功能不复杂核心逻辑有两点。一是表单提交时用fetch调用/api/videos接口把视频数据写入数据库二是点击表格中“生成求关注文案”按钮后调用/api/videos/{id}/notify接口页面上会显示生成的文案。为了避免小白操作时把时间格式写错页面里给出了明确的占位提示。模板中的{% for video in videos %}是 Jinja2 的循环语法它会把后端传入的视频列表逐行渲染到表格中。如果列表为空则走{% else %}分支。6.2 启动服务在video_manager目录下执行uvicorn app:app --reload --port 8000启动成功后终端会显示 Uvicorn 运行地址。打开浏览器访问http://127.0.0.1:8000你应该能看到标题为“视频发布管理助手”的页面。首次启动时会自动生成videos.db所以不用手动建库。需要特别说明的是--reload适合开发调试。它会在代码文件发生变化时自动重启服务。但在调试定时任务时如果代码频繁变更调度器会在重启后重新注册任务有可能出现重复日志建议验证定时任务时去掉--reload或者接受这种开发期小问题。真正部署时也不建议在systemd或 Docker 环境中使用--reload。6.3 用接口验证完整流程暂时不在页面操作先用curl测试接口更直观。第一步新增一条视频记录curl -X POST http://127.0.0.1:8000/api/videos \ -H Content-Type: application/json \ -d { title: 【烩面大王DW】最新视频上线求关注, platform: B站, video_url: https://www.bilibili.com/video/example, publish_date: 2025-06-01 10:00:00 }预期返回结果大致如下{ title: 【烩面大王DW】最新视频上线求关注, platform: B站, video_url: https://www.bilibili.com/video/example, publish_date: 2025-06-01 10:00:00, status: pending, id: 1, created_at: 2025-06-01 09:30:00 }第二步查询视频列表curl http://127.0.0.1:8000/api/videos第三步点击通知接口验证“生成求关注文案”逻辑curl -X POST http://127.0.0.1:8000/api/videos/1/notify接口会返回{ id: 1, copy: 【最新视频上线】《【烩面大王DW】最新视频上线求关注》已经在 B站 发布啦。\n观看链接...\n如果你觉得这期内容对你有用欢迎点赞、评论、转发。\n点击关注【烩面大王DW】后续更新不迷路最新视频第一时间推送给你, push_result: { success: false, message: 未配置 SERVERCHAN_SENDKEY跳过推送只返回文案 } }如果没有配置 SendKey你会看到push_result.successfalse但copy字段已经生成了完整文案。如果想要实际推送在启动服务前设置环境变量Windows PowerShell$env:SERVERCHAN_SENDKEY你的SendKeymacOS / Linuxexport SERVERCHAN_SENDKEY你的SendKey然后再启动服务。建议把 SendKey 放在.env文件或环境变量里不要写死在push.py源码中避免代码泄露后被人盗用推送通道。7. 常见问题与排查思路在运行这套代码时新手容易遇到下面几个问题。我把现象、可能原因和解决思路整理成一张表方便直接查阅。问题现象常见原因解决思路ModuleNotFoundError: No module named fastapi没有安装依赖或激活虚拟环境执行pip install -r requirements.txt并确认当前终端处于对应虚拟环境ModuleNotFoundError: No module named apschedulerrequirements 安装不完整单独执行pip install apscheduler或重新安装 requirements访问http://127.0.0.1:8000报 500数据库没有初始化或templates目录路径不对确认在video_manager目录下启动且lifespan中调用了init_db()sqlite3.OperationalError: no such table: videos启动时没有建表检查启动命令是否为uvicorn app:app --reload确保lifespan生效jinja2.exceptions.TemplateNotFound: index.html当前工作目录不是项目根目录在video_manager目录下执行命令不要在其他目录启动 uvicornTemplateResponse() missing ...Starlette 模板渲染写法与版本不匹配新版使用templates.TemplateResponse(request, index.html, {...})如果你的版本较旧改成templates.TemplateResponse(index.html, {request: request, ...})定时任务 30 分钟不触发间隔设置太长或者代码中还有语法错误把minutes30临时改为minutes1观察终端日志定时任务执行后数据库状态没变UPDATE语句没有提交确认用的是get_connection()上下文管理器里面包含conn.commit()Server酱返回推送失败SendKey 无效、网络不通或服务不稳定检查SERVERCHAN_SENDKEY环境变量确认 api_url 格式查看返回的response字段如果你遇到的报错不在表里最简单的方法是分三步排查先确认所有第三方依赖是否安装成功。再确认启动目录是不是video_manager根目录因为模板路径和数据库路径都用相对目录。最后看终端完整报错堆栈重点搜索File和Error两处定位是哪一行代码出错。8. 最佳实践与工程建议代码跑通只是第一步真正落到“长期使用”的时候以下几点建议会更重要。8.1 配置不要写死在代码里push.py里通过os.getenv(SERVERCHAN_SENDKEY)读取环境变量这个习惯值得保持。无论是推送 SendKey、数据库路径还是未来要接入的平台开放 API Secret都不要写死在源码里。配置文件方面可以使用python-dotenv在项目根目录创建.env文件SERVERCHAN_SENDKEY