ARTICLE DETAIL

资讯详情

深耕网站建设、视觉设计与SEO优化的一线实战洞察。

FastAPI+APScheduler:搭建每日自动更新的内容发布系统

FastAPI+APScheduler:搭建每日自动更新的内容发布系统 做“恐怖故事汇”这类短篇内容站点看起来只需要不断上传文章真正运营起来后会发现内容更新频率、定时发布、敏感词过滤、文件管理每一项都很占精力。尤其是像《自动更新的死亡日记网约车后座的神秘人》这种带有连续编号的短篇通常需要每天自动更新人工手动改日期、改文件、再上传效率低且容易出错。这篇文章不聊故事本身而是围绕一个可运行的示例项目讲清楚如何用 FastAPI、SQLite、APScheduler 和 Jinja2 搭建一套“短篇故事自动更新系统”。学完后你可以把它改造成自己的内容流水线让系统每天定时生成一篇 Markdown 日记、写入数据库、渲染成页面并保留发布日志和敏感内容检查能力。1. 自动更新系统要解决的核心问题以及技术选型思路1.1 内容站点真正耗时的环节不是写作运营一个短篇故事集可能每天都需要新增一则内容。以“恐怖故事汇”为例篇目通常有编号、日期、类别、简介、正文。一个人手工维护时流程大概是新建 Markdown 文件填写标题和日期把正文复制进去再打开后台发布页面确认分类和标签。这套流程第一次做还好连续做一个月就很容易出问题。常见故障包括日期写错昨天的内容今天发。编号重复后台出现两条“短篇025”。Markdown 文件存放位置不统一有些在本地有些在服务器。发布时发现正文里包含了需要过滤的敏感内容又要退回修改。定时发布需求出现后人工操作几乎无法保证每天同一时间完成。所以真正要解决的不是“能不能写一篇故事”而是“能不能让内容经过固定流水线自动生成、自动检查、自动发布同时每一步都能被追溯”。这就是自动更新系统的价值把重复劳动变成程序行为把发布过程变成可审计的数据记录。1.2 为什么用 FastAPI 而不是 Django 或 Flask选 FastAPI 的原因主要有三个异步支持好、自带 OpenAPI 文档、代码量少。短篇故事自动更新系统的核心功能不多主要是页面展示、数据库读写、定时任务。Django 自带 Admin、ORM、迁移工具功能确实强大但对这种小型内容流水线来说偏重。Flask 很轻不过需要自己拼装很多第三方扩展项目结构往往每个人都不同。FastAPI 在类型提示、请求参数解析、依赖注入上比较现代化结合 SQLAlchemy 写数据模型和接口都很直接。选择 SQLite 作为初版数据库是为了降低部署成本。一个文件就能承载数据不需要单独安装数据库服务。配合 SQLAlchemy后续切换到 PostgreSQL 时只需要改连接串业务代码基本不用动。1.3 整体工作流程模板生成、定时发布、渲染展示这套系统的完整流程可以拆成三段生成根据故事模板和当天日期用 Jinja2 渲染出一篇 Markdown 文本并写入stories目录。发布调度任务把生成的文本写入stories表状态标记为已发布同时写入一条publish_logs日志。展示FastAPI 从数据库读取故事列表和详情通过 Jinja2 渲染成 HTML 页面返回给浏览器。这里有一个关键设计定时任务只负责“生成并发布”不直接拼接 HTML。数据库是唯一的内容源页面渲染由 Web 层负责。这样分开的好处是之后如果想增加接口输出 JSON、生成 RSS、或只做静态导出都可以复用已有数据。2. 环境准备和项目结构先把依赖对齐再写代码2.1 可运行的 Python 版本和依赖清单示例项目使用 Python 3.10 及以上版本。文章里的版本号是示例落地方案时先在自己的环境里确认版本兼容性避免因为镜像源或系统 Python 版本不同导致安装失败。依赖包示例版本作用fastapi0.110.0Web 框架处理页面和接口uvicorn0.29.0ASGI 服务器负责启动应用sqlalchemy2.0.29ORM操作 SQLite 数据库apscheduler3.10.4定时任务调度jinja23.1.3模板渲染既渲染 HTML也生成 Markdownpython-multipart0.0.9FastAPI 处理表单数据时使用后续扩展后台功能需要创建requirements.txt后建议先执行安装再继续写代码。pip install -r requirements.txt这里不建议一次性安装最新版本后忽略锁定。依赖版本变化可能导致 APScheduler 的 API 调整或 SQLAlchemy 的行为差异。学习环境可以放宽生产环境必须锁定版本。2.2 建目录骨架项目目录建议按功能拆分不要把所有代码都写在一个文件里。后面的代码都基于这个目录结构。death_diary/ ├── app.py ├── config.py ├── database.py ├── models.py ├── generator.py ├── review.py ├── scheduler.py ├── run.py ├── requirements.txt ├── templates/ │ ├── index.html │ └── detail.html ├── stories/ │ └── .gitkeep └── data/ └── .gitkeepapp.py是 FastAPI 入口config.py放路径和时间配置database.py创建数据库引擎models.py定义数据表generator.py负责用模板生成 Markdown 内容review.py做敏感词检查scheduler.py管理定时任务。stories目录保存生成后的 Markdown 文件data目录保存 SQLite 数据库文件。2.3 配置文件 config.py配置项集中管理后续修改路径或发布时间时不需要翻代码。from pathlib import Path BASE_DIR Path(__file__).resolve().parent DATA_DIR BASE_DIR / data STORIES_DIR BASE_DIR / stories DATABASE_URL fsqlite:///{DATA_DIR / death_diary.db} PUBLISH_HOUR 6 PUBLISH_MINUTE 0 PUBLISH_SECOND 0 SENSITIVE_WORDS [敏感词示例1, 敏感词示例2]BASE_DIR使用Path(__file__).resolve().parent能保证无论从哪个目录启动项目路径都是相对于配置文件所在目录避免出现“找不到 data 目录”的问题。PUBLISH_HOUR、PUBLISH_MINUTE、PUBLISH_SECOND控制每天定时发布的具体时间。SENSITIVE_WORDS是敏感词列表生产环境建议从外部配置读取而不是写死在代码里。3. 数据表设计故事表、发布日志表、示例篇目3.1 stories 表字段说明stories表用于保存生成后的故事内容。设计字段时要同时考虑页面展示和后续扩展。字段类型说明idInteger主键自增titleString故事标题slugString用于生成文件名或 URL 标识categoryString分类例如“恐怖故事汇”authorString作者或来源summaryString摘要描述content_mdTextMarkdown 正文statusString状态draft 或 publishedsourceString来源manual 或 generatedpublished_atDateTime实际发布时间created_atDateTime创建时间updated_atDateTime最后更新时间这里要特别注意status和source字段。status用于区分草稿和已发布内容source用于区分人工录入和程序生成。自动更新系统上线后一定会有“这篇是程序发的那篇是人工补的”这种追溯需求少了这两个字段后期补数据很麻烦。3.2 publish_logs 表的作用publish_logs表记录每次发布任务的结果。无论发布成功还是失败都要留日志。字段类型说明idInteger主键story_idInteger关联故事 ID失败时可为空task_nameString任务名称例如 daily_publishstatusStringsuccess 或 failedmessageString错误信息或备注created_atDateTime日志时间这张表的价值在排错时体现。定时任务不触发或发布失败时先查publish_logs能快速判断问题是出在调度器、生成器还是数据库写入阶段。很多内容系统上线后出问题却查不到原因就是因为缺少这类审计日志。3.3 SQLAlchemy 模型和示例数据database.py里先定义数据库引擎和会话。from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker, declarative_base from config import DATABASE_URL engine create_engine( DATABASE_URL, connect_args{check_same_thread: False} ) SessionLocal sessionmaker(bindengine, autocommitFalse, autoflushFalse) Base declarative_base()models.py定义两张表。from datetime import datetime from sqlalchemy import Column, Integer, String, Text, DateTime from database import Base class Story(Base): __tablename__ stories id Column(Integer, primary_keyTrue, indexTrue) title Column(String(200), nullableFalse) slug Column(String(200), uniqueTrue, indexTrue) category Column(String(50), default短篇故事) author Column(String(50), default系统生成) summary Column(String(500), default) content_md Column(Text, default) status Column(String(20), defaultdraft) source Column(String(20), defaultgenerated) published_at Column(DateTime, defaultdatetime.now) created_at Column(DateTime, defaultdatetime.now) updated_at Column(DateTime, defaultdatetime.now, onupdatedatetime.now) class PublishLog(Base): __tablename__ publish_logs id Column(Integer, primary_keyTrue, indexTrue) story_id Column(Integer, nullableTrue) task_name Column(String(100), defaultdaily_publish) status Column(String(20), defaultsuccess) message Column(Text, default) created_at Column(DateTime, defaultdatetime.now)slug字段加了唯一索引这是防止重复发布的重要约束。生成内容时如果标题相同插入数据库就会触发唯一约束错误避免同一天出现两篇完全相同的文章。示例篇目可以直接放到生成器模板里也可以单独准备一份素材文件。这里用生成器内置模板来演示效果更直观。4. 实现自动更新Jinja2 模板生成 Markdown 日记4.1 为什么要用模板生成内容内容系统每天发布的文章如果结构相似只有标题编号、日期、场景名称不同非常适合模板化。例如“恐怖故事汇 短篇025”这类篇目需要把《自动更新的死亡日记网约车后座的神秘人》作为示例标题把日期自动填充进去。用 Jinja2 渲染文本而不是直接拼接字符串原因在于模板可以被维护者单独修改变量替换逻辑更清晰也不会因为忘记转义日期或标点导致格式错乱。后续如果引入大模型生成内容也可以让大模型输出结构化字段再交给模板统一排版。4.2 生成器实现日期替换和文件落盘generator.py里定义一组模板种子数据和构建方法。from datetime import datetime, timedelta from pathlib import Path from jinja2 import Template from config import STORIES_DIR STORY_SEEDS [ { title: 自动更新的死亡日记·{date}, category: 恐怖故事汇, summary: 网约车后座的神秘人留下了一本会自动更新的日记。, content_template: ( 在网约车的后座上我捡到了一本日记。\n\n 日记的封面署名是短篇025。\n\n 最奇怪的是翻开日记时里面已经写满了日期 最新一页正好是 {date}。\n\n 第二天我再次打开那本日记发现最新一页变成了 {date_plus_1}。 ), } ] def generate_next_story(publish_date: datetime | None None): if publish_date is None: publish_date datetime.now() seed STORY_SEEDS[0] template Template(seed[content_template]) content_md template.render( datepublish_date.strftime(%Y-%m-%d), date_plus_1(publish_date timedelta(days1)).strftime(%Y-%m-%d), ) title seed[title].replace({date}, publish_date.strftime(%Y-%m-%d)) summary seed[summary] slug fstory-{publish_date.strftime(%Y%m%d)} return { title: title, slug: slug, category: seed[category], summary: summary, content_md: content_md, status: published, source: generated, }代码里用{date}和{date_plus_1}两个模板变量让正文自动带上当天和明天的日期。generate_next_story返回字典字段名与Story模型保持一致之后可以直接用于数据库插入。生成 Markdown 文件是辅助功能便于人工审阅和归档。def write_story_file(story_data: dict) - Path: STORIES_DIR.mkdir(exist_okTrue) file_path STORIES_DIR / f{story_data[slug]}.md file_path.write_text(story_data[content_md], encodingutf-8) return file_path注意写入文件时指定encodingutf-8。不指定编码时Windows 环境容易写成 GBK 编码后续在页面读取时会出现乱码。4.3 敏感词过滤逻辑review.py提供两个函数检查敏感词、返回问题列表。from config import SENSITIVE_WORDS def check_sensitive(content: str) - list[str]: hits [] for word in SENSITIVE_WORDS: if word in content: hits.append(word) return hits def review_story(story_data: dict) - tuple[bool, list[str]]: content story_data[title] \n story_data[content_md] hits check_sensitive(content) if hits: return False, hits return True, []这里刻意用最简方式实现是为了让逻辑清晰。真实项目中的敏感内容检查要比这里复杂得多可能需要调用云服务、维护同义词库、处理图片和音频。但整体设计保持不变发布前先检查检查不通过就终止发布并把问题记录到发布日志。如果在自动生成任务中检查到敏感词不建议直接把内容写入数据库更合理的做法是生成一条状态为draft的记录等人工处理。5. 定时任务和 FastAPI 页面让更新真正跑起来5.1 APScheduler 每日发布任务APScheduler 是 Python 生态中常用的定时任务库。scheduler.py定义每日凌晨 6 点发布的任务。from apscheduler.schedulers.background import BackgroundScheduler from apscheduler.triggers.cron import CronTrigger from database import SessionLocal from models import Story, PublishLog from generator import generate_next_story, write_story_file from review import review_story def scheduled_publish(): db SessionLocal() try: story_data generate_next_story() ok, hits review_story(story_data) if not ok: db.add(PublishLog( story_idNone, task_namedaily_publish, statusfailed, messagefsensitive words: {hits}, )) db.commit() return story Story(**story_data) db.add(story) db.flush() write_story_file(story_data) db.add(PublishLog( story_idstory.id, task_namedaily_publish, statussuccess, )) db.commit() except Exception as exc: db.rollback() db.add(PublishLog( story_idNone, task_namedaily_publish, statusfailed, messagestr(exc), )) db.commit() finally: db.close() def start_scheduler(): scheduler BackgroundScheduler(timezoneAsia/Shanghai) scheduler.add_job( scheduled_publish, CronTrigger(hour6, minute0, second0), iddaily_story_publish, replace_existingTrue, ) scheduler.start() return schedulerBackgroundScheduler会在后台线程中执行任务不阻塞 Web 服务。CronTrigger的hour6, minute0表示每天 6 点执行。timezoneAsia/Shanghai必须显式指定否则部分服务器环境会按 UTC 时间执行发布时间就会差 8 个小时。任务内部先生成故事再检查敏感词再写数据库和文件。每一步失败都被捕获并写入publish_logs不会因为一次异常导致整个调度器退出。5.2 FastAPI 首页和详情页app.py里创建 FastAPI 应用用 lifespan 创建数据库表并启动调度器。from contextlib import asynccontextmanager from fastapi import FastAPI, Depends, HTTPException, Request from fastapi.responses import HTMLResponse from fastapi.templating import Jinja2Templates from sqlalchemy.orm import Session from database import Base, engine, SessionLocal from models import Story from scheduler import start_scheduler, scheduled_publish templates Jinja2Templates(directorytemplates) asynccontextmanager async def lifespan(app: FastAPI): Base.metadata.create_all(bindengine) scheduler start_scheduler() yield scheduler.shutdown() app FastAPI(title短篇故事自动更新系统, lifespanlifespan) def get_db(): db SessionLocal() try: yield db finally: db.close()首页查询最近 20 篇文章并渲染模板。app.get(/, response_classHTMLResponse) def index(request: Request, db: Session Depends(get_db)): stories db.query(Story).order_by(Story.published_at.desc()).limit(20).all() return templates.TemplateResponse(index.html, { request: request, stories: stories, }) app.get(/story/{story_id}, response_classHTMLResponse) def detail(request: Request, story_id: int, db: Session Depends(get_db)): story db.query(Story).filter(Story.id story_id).first() if not story: raise HTTPException(status_code404, detailstory not found) return templates.TemplateResponse(detail.html, { request: request, story: story, })数据库查询没有分页少量数据没问题。生产环境故事数量增长后要改成offset或order_by加limit、offset的分页方式避免一次查出上万条记录。5.3 手动触发接口和模板文件新增一个手动触发接口方便联调时立刻执行发布任务。app.post(/admin/trigger) def trigger_publish(db: Session Depends(get_db)): scheduled_publish() return {message: triggered}这个接口只适合在本地或内网环境使用。如果部署到公网必须增加鉴权不能允许任何匿名请求触发写库操作。模板文件index.html核心代码如下。!DOCTYPE html html langzh-CN head meta charsetUTF-8 title短篇故事自动更新系统/title /head body h1自动更新的日记/h1 ul {% for story in stories %} li a href/story/{{ story.id }}{{ story.title }}/a span{{ story.summary }}/span /li {% endfor %} /ul /body /htmldetail.html展示正文。!DOCTYPE html html langzh-CN head meta charsetUTF-8 title{{ story.title }}/title /head body h1{{ story.title }}/h1 p{{ story.summary }}/p pre{{ story.content_md }}/pre /body /html模板中直接使用{% if %}、{% for %}等 Jinja2 语法能安全渲染数据库中的文本。需要注意这里用pre展示 Markdown 原文方便确认生成结果。生产环境建议引入 Markdown 转 HTML 的渲染库例如markdown再配合安全过滤不能直接把用户提供的内容当作 HTML 输出。6. 运行验证从启动服务到看到新日记6.1 安装依赖和启动命令在项目根目录执行以下命令pip install -r requirements.txt python run.pyrun.py的内容是import uvicorn if __name__ __main__: uvicorn.run(app:app, host0.0.0.0, port8000, reloadTrue)使用reloadTrue可以在开发时自动加载代码改动生产环境不要开。启动后看到类似日志说明服务正常运行。INFO: Uvicorn running on http://0.0.0.0:8000 INFO: Application startup complete.浏览器访问http://127.0.0.1:8000/正常情况下能看到空列表或已生成的故事。6.2 手动触发、数据库和文件检查因为定时任务默认设置在凌晨 6 点联调时直接等不太现实。用手动接口触发一次curl -X POST http://127.0.0.1:8000/admin/trigger成功响应{message:triggered}然后验证三处结果。第一访问首页刷新后能看到新故事标题。第二检查 SQLite 数据库sqlite3 data/death_diary.db SELECT id, title, status, source, published_at FROM stories;1|自动更新的死亡日记·2026-05-14|published|generated|2026-05-14 06:00:00第三检查stories目录生成的文件ls stories/story-20260514.md文件内容应该是日期替换后的 Markdown 正文。6.3 验证代码参考如果不想频繁手动触发可以在启动后运行一段简单脚本直接调用generate_next_story和review_story验证生成链路。from generator import generate_next_story from review import review_story data generate_next_story() print(data[title]) print(data[content_md]) print(review_story(data))输出应包含标题、替换了日期的正文以及敏感词检查结果。这一步能确认问题到底出在“生成”还是“发布”。7. 常见问题排查按现象倒推根因自动更新系统运行一段时间后最常遇到的不是复杂逻辑错误而是环境、时间、并发这几种问题。问题现象常见原因检查方式处理建议每天 6 点没有新文章服务器时区不是 Asia/Shanghai或调度器未启动查看启动日志是否出现 Application startup complete检查系统时区显式设置 CronTrigger 的 timezone确认服务未因异常退出发布时间比预期早或晚 8 小时APScheduler 使用 UTC 时间打印当前时间和调度器 next_run_time在调度器初始化时指定 timezoneAsia/Shanghai手动触发后接口报 500SQLite 文件锁冲突或被其他进程占用查看 publish_logs 中 failed 记录检查 message避免多进程同时启动任务或改用 PostgreSQL页面中文乱码文件写入或模板读取编码不一致检查模板文件头的 meta charset检查代码是否指定 utf-8统一使用 utf-8 编码文件读取写入都显式指定模板变量渲染后为空Jinja2 变量名拼写不一致打印渲染前后内容使用字典传参避免多层模板作用域变量名不一致同一天生成两篇相同文章定时任务重复启动检查是否多次执行 start_scheduler增加唯一约束和 replace_existingTrue7.1 定时任务不触发这是最常见的现象服务正常启动页面能访问但每天到了设定时间就是没有新文章。先确认调度器有没有启动。开发环境运行python run.py时如果代码里scheduled_publish有异常但被except吞掉了日志表里应该有failed记录。如果publish_logs表为空说明任务根本没有执行。检查顺序查看启动日志确认Application startup complete出现。在start_scheduler中打印scheduler.get_job(daily_story_publish)确认 job 存在。检查系统时区。Linux 服务器可能默认 UTC。查看next_run_time是否正确。scheduler start_scheduler() job scheduler.get_job(daily_story_publish) print(job.next_run_time)7.2 SQLite 写锁冲突SQLite 适合单机、低并发场景。如果开发时开了 uvicorn 的reloadTrueFastAPI 会启动两个进程再加上后台调度线程就可能出现数据库被锁的现象。常见报错是database is locked。学习环境可以接受生产环境要注意调度任务和 Web 服务尽量部署在同一个入口但不要多个进程同时执行同一个定时任务。多实例部署时建议把定时任务单独部署成任务进程或使用 PostgreSQL 之类支持并发写入的数据库。7.3 中文乱码和模板变量为空中文乱码主要出现在 Windows。写文件时Path.write_text没有指定编码读取模板时 FastAPI 默认按 UTF-8 读取两边不一致就会乱码。解决方案是统一使用encodingutf-8。模板变量为空往往是变量名问题。比如模板中写了{{ date_plus_1 }}传参时却是date_plus_oneJinja2 不会报错只会渲染空字符串。排查时直接 print 渲染后的结果不要只看页面输出。8. 学习环境与生产环境的差别以及发布检查清单8.1 调度任务不要和 Web 服务捆在一起学习环境里用BackgroundScheduler配合 FastAPI 启动足够跑通整套逻辑。但生产环境要考虑几个问题应用进程重启时后台定时任务可能重复注册。多 Worker 部署时每个 Worker 都会启动一个调度器。定时任务执行时间较长时会与请求处理争抢线程资源。更稳妥的做法是拆分任务进程Web 服务只处理页面渲染和接口请求调度服务单独运行只负责定时生成和发布。两个服务共用同一个数据库职责清晰。8.2 安全、日志、备份和内容审核上线前要额外处理四件事。安全方面/admin/trigger这种接口必须加鉴权至少使用 API Key 或内部网络访问控制。如果后续加入编辑后台还要考虑登录态、CSRF 防护和操作权限。日志方面不能只依赖print。生产环境建议使用logging模块按任务名称记录生成、审核、发布三个阶段的日志并保留一定时间周期。备份方面SQLite 数据库是一个文件每天跑一个sqlite3 .backup或文件复制任务即可。如果换成 PostgreSQL再设计自动备份策略。内容审核方面示例里的敏感词列表只是占位。真实运营需要根据目标平台规则维护词库同时建立“机器初审 人工抽查”的流程。自动更新系统如果完全无人值守遇到内容违规会带来比人工运营更大的风险。8.3 扩展方向更大规模的内容流水线这套系统可以朝三个方向扩展。第一内容源接口化。把generate_next_story改成从内部接口获取结构化内容让编辑可以在后台填写模板字段系统再定时发布。第二渲染升级。当前页面用pre显示 Markdown 原文正式站点应引入markdown库转为 HTML并启用代码高亮、目录生成等能力。第三发布渠道扩展。数据库里保存的是结构化内容后续可以增加 RSS、公众号同步、静态站点导出等渠道。核心思路不变内容生成与渠道发布分开数据统一管理。项目学习环境做法生产环境建议数据库SQLite 单文件PostgreSQL支持并发调度BackgroundScheduler 内嵌独立任务服务日志print 输出logging 文件或日志平台手动触发无鉴权接口API Key、内网访问内容审核内置敏感词列表外部审核服务 人工抽查缓存无列表页可加 Redis 或静态化9. 落地时最需要记住的三个实践判断在真实项目中落地“自动更新的日记”这类内容系统最值得记住的判断有三条。第一不要把发布逻辑散落在接口里。无论页面、手动触发还是定时任务最终都调用同一个发布函数并写入发布日志。这样执行入口不同行为完全一致排查问题时只需要盯住一条链路。第二敏感词检查和文件落盘要放在数据库写入之前。数据一旦进入正式表清理成本会成倍增加。先检查、先备份文件再写正式数据失败时能直接回滚。第三定时任务必须显式指定时区。国内服务器通常使用北京时间云服务器却可能默认 UTC。没有指定时区的调度任务往往要等上线第二天发现发布时间和预期不一致时才能发现。如果你是从零开始做内容站建议先按本文结构跑通最小闭环生成、审核、发布、展示、日志。把这条链路稳定下来再考虑接入大模型生成、静态站点导出或更多发布渠道。系统最核心的竞争力不是某一个功能有多强而是每一次自动发布都可以被解释、被追踪、被回滚。
返回列表