开场:我在项目中遇到的痛点
最近在项目中遇到一个问题:内容营销团队需要每周产出 20 条社交媒体帖文、10 篇博客和 5 条邮件文案,手工排期不仅耗时,还容易出现主题重复或发布时间冲突。于是我决定用 ChatGPT 来自动生成主题、撰写草稿并把结果写进 Google Sheet,配合 APScheduler 自动推送到内容管理系统(CMS),实现真正的内容日历自动化。
关键在于:把「创意生成」交给模型,把「排期执行」交给调度脚本,两者之间用结构化数据桥接。
1. 原理分析:从 Prompt 到结构化输出
1.1 为什么要让模型输出 JSON
ChatGPT 天生擅长自然语言,但在实际业务中我们需要的是机器可直接读取的字段。通过在 Prompt 中明确要求返回 JSON(如 {"title":"...","summary":"...","publish_date":"..."}),可以让后续脚本省去正则解析的麻烦,也更易于调试。
1.2 内容主题生成的核心 Prompt
你是一名内容策划专家,请基于以下关键词生成 5 条适合在社交媒体发布的主题,每条返回 JSON,字段包括:title、angle(切入角度)、target(目标受众)和publish_date(本周三、周五任选)。关键词:ChatGPT, 自动化, 内容营销。这里有个坑要注意:如果不在 Prompt 里限制 publish_date 的取值范围,模型有时会返回不符合业务规则的日期,需要在后置代码里做二次校验。
2. 实践应用:完整技术栈与实现步骤
2.1 技术选型概览
| 组件 | 作用 |
|---|---|
| OpenAI API | 文本生成、主题提炼 |
| Python 3.11 | 主脚本语言 |
| APScheduler | 定时任务调度 |
| Google Sheets API | 结构化存储、协作编辑 |
| Flask (可选) | 提供 Webhook 接口给 CMS |
2.2 项目结构
content-calendar/
├─ config.py # 配置文件,存放 API Key、Sheet ID 等
├─ generator.py # 与 OpenAI 交互、返回 JSON
├─ scheduler.py # APScheduler 任务定义
├─ sheets_client.py # Google Sheets 封装
└─ main.py # 启动入口2.3 关键代码实现
2.3.1 与 OpenAI 对话的封装(generator.py)
import os, json
import openai
from config import OPENAI_API_KEY, MODEL_NAME
openai.api_key = OPENAI_API_KEY
def generate_topics(keywords: list, count: int = 5) -> list:
prompt = (
"你是一名内容策划专家,请基于以下关键词生成 %d 条适合在社交媒体发布的主题,"
"每条返回 JSON,字段包括 title、angle、target、publish_date。"
"publish_date 只能是本周三或本周五的日期(YYYY-MM-DD),关键词:%s。"
) % (count, ", ".join(keywords))
response = openai.ChatCompletion.create(
model=MODEL_NAME,
messages=[{"role": "user", "content": prompt}],
temperature=0.7,
max_tokens=800,
)
# 直接让模型输出 JSON 列表,省去逐行解析
raw = response.choices[0].message.content.strip()
try:
topics = json.loads(raw)
except json.JSONDecodeError:
# 若模型返回了多段 JSON,做一次容错合并
topics = json.loads('[' + raw.replace('}{', '},{') + ']')
return topics2.3.2 写入 Google Sheet(sheets_client.py)
from google.oauth2.service_account import Credentials
from googleapiclient.discovery import build
from config import GOOGLE_SHEET_ID, GOOGLE_CREDS_JSON
SCOPES = ['https://www.googleapis.com/auth/spreadsheets']
creds = Credentials.from_service_account_file(GOOGLE_CREDS_JSON, scopes=SCOPES)
service = build('sheets', 'v4', credentials=creds)
def append_rows(rows: list):
body = {'values': rows}
result = service.spreadsheets().values().append(
spreadsheetId=GOOGLE_SHEET_ID,
range='Content!A:D',
valueInputOption='RAW',
body=body
).execute()
return result2.3.3 调度任务(scheduler.py)
from apscheduler.schedulers.background import BackgroundScheduler
from generator import generate_topics
from sheets_client import append_rows
from datetime import datetime, timedelta
scheduler = BackgroundScheduler()
def job_generate_and_save():
# 业务里常用的关键词集合,可放在配置中
keywords = ['ChatGPT', '自动化', '内容营销']
topics = generate_topics(keywords)
rows = []
for t in topics:
rows.append([
t.get('title'),
t.get('angle'),
t.get('target'),
t.get('publish_date')
])
append_rows(rows)
print(f"[{datetime.now()}] 已写入 {len(rows)} 条内容计划")
# 设定每周一 09:00 触发一次
scheduler.add_job(job_generate_and_save, 'cron', day_of_week='mon', hour=9, minute=0)
if __name__ == '__main__':
scheduler.start()
try:
# 让主线程保持运行
while True:
pass
except (KeyboardInterrupt, SystemExit):
scheduler.shutdown()2.3.4 可选:提供 Flask Webhook 给 CMS(main.py)
from flask import Flask, request, jsonify
from generator import generate_topics
app = Flask(__name__)
@app.route('/api/content', methods=['POST'])
def create_content():
data = request.json
keywords = data.get('keywords', [])
count = data.get('count', 5)
topics = generate_topics(keywords, count)
return jsonify(topics)
if __name__ == '__main__':
app.run(host='0.0.0.0', port=5000)2.4 部署注意事项
- 环境变量安全:API Key、Google Service Account JSON 建议放在 Docker secret 或 k8s secret 中,切勿硬编码。
- 限流防护:OpenAI 对每分钟请求数有限制,使用
time.sleep或者在 APScheduler 中加上misfire_grace_time防止突发重试导致超额。 - 错误监控:建议接入 Sentry 或自行写日志,尤其是 JSON 解析失败时要记录原始返回内容,便于快速定位模型输出变化。
3. 经验总结:常见坑与最佳实践
3.1 坑点回顾
- Prompt 不够明确:最初的 Prompt 只要求「生成主题」而没有限制日期格式,导致返回的
publish_date有时是「下周一」这种相对表达,需要额外正则处理。 - Google Sheet 并发写入冲突:在高频率触发时,Sheets API 会返回
429 Too Many Requests,通过在append_rows前加retry(指数退避)解决。 - 模型漂移:模型升级后,同一 Prompt 可能返回不同字段顺序或多余换行,使用
json.loads包裹容错逻辑是必须的。
3.2 最佳实践
- Prompt 采用结构化模板,每次调用保持一致,便于后期复用。
- 统一时间库:所有日期均使用
datetime的strftime('%Y-%m-%d'),避免时区混乱。 - 分层日志:DEBUG 级别记录原始 Prompt 与模型返回,INFO 级别记录成功写入行数,ERROR 级别捕获异常并发送告警。
- 可视化审阅:在 Google Sheet 中加入「审核状态」列,内容团队可手动标记「已审」或「待修」,后续脚本可以根据该列决定是否推送到 CMS。
- 持续迭代 Prompt:每月通过 A/B 测试对比不同 Prompt 生成的点击率,选出表现最好的版本。
4. 实战演练:从零部署到产出
准备工作
- 在 OpenAI 平台申请 API Key 并记录在
.env中。 - 在 Google Cloud 控制台创建 Service Account,下载 JSON 并放在项目根目录。
- 创建 Google Sheet,命名为「Content」并在第一行写入标题
Title, Angle, Target, Publish Date。
- 在 OpenAI 平台申请 API Key 并记录在
本地测试
pip install openai google-auth google-api-python-client apscheduler flask- 运行
python scheduler.py,观察控制台打印的「已写入」日志。
容器化部署(示例 Dockerfile)
FROM python:3.11-slim WORKDIR /app COPY . /app RUN pip install --no-cache-dir -r requirements.txt CMD ["python", "scheduler.py"]- 构建镜像
docker build -t content-calendar .,在服务器上使用docker run -d --env-file .env content-calendar启动。
- 构建镜像
上线后监控
- 设置 Grafana + Prometheus 采集脚本日志。
- 每周检查 Sheet 中「审核状态」列,确保内容质量符合品牌调性。
5. 结语:让内容生产真正解放
通过上述方案,团队把「每周主题策划」的 2 小时工作压缩到几分钟,且每次输出都有统一结构、可追溯的来源。关键在于:把创意交给语言模型,把流程交给调度脚本,剩下的只需要人审校和业务系统对接。
如果你正面临内容排期混乱、创意枯竭或人力成本高的痛点,不妨尝试本文的实现思路。后续可以进一步引入 向量数据库(如 Pinecone)做主题相似度去重,或使用 LangChain 编排更复杂的多步骤工作流。
行动建议:先在本地跑通「生成+写入」两步,再逐步添加审校列和 CMS 推送,循序渐进更容易定位问题。
祝你的内容日历跑得更快、更稳!