这篇能做出什么
跑完这篇，你会得到一条能在本地或服务器上定时运行的流水线，输入是「一批公开热点标题」，输出是「可直接开拍的口播稿 + 分镜时间轴字幕文件 + 候选标题和话题标签」，按日期归档到文件夹里。
举个具体例子：早上八点定时任务跑一次，从若干资讯源抓到 30 条标题，打分筛出 5 条，模型把它们改写成 60 秒口播稿。其中一条是「某款新耳机发布，续航翻倍、价格没动」，稿子的钩子写成「续航翻倍、价格没动，这次是升级还是清库存」，正文给两个信息点，结尾一句互动引导。你打开文件夹，复制、微调、开拍。
有一件事必须先说清楚：AI 负责把「话题」变成「能念出口的句子」，事实核对和观点表达仍然由人把关。稿子里出现的数字、时间、结论，发布前逐条对一遍。
前置条件清单
- Python 3.10 以上，以官方文档当前版本为准
- 一个可用的对话模型 API Key（兼容 OpenAI 风格接口的服务都能套用下面的代码）
- 若干公开资讯源，优先用官方 RSS，其次用开放 API
- 能读环境变量、会写简单函数
- 一台能跑定时任务的机器：Linux/macOS 用 cron，Windows 用任务计划程序
- 一个确定好的账号定位（例如「给职场新人的效率工具测评」），越具体，生成质量越高
步骤一：搭目录与依赖
```bash
mkdir -p hot2script/{data,out,logs}
cd hot2script
python3 -m venv .venv
source .venv/bin/activate # Windows 用 .venv\Scripts\activate
pip install feedparser requests
```
API Key 放进环境变量，不要写进代码：
```bash
export LLM_API_KEY="你的密钥"
export LLM_BASE_URL="https://你的服务地址/v1" # 以服务商官方文档为准
export LLM_MODEL="你账号下可用的对话模型"
```
如果服务商提供官方 SDK，按官方文档安装即可；下面的示例走 HTTP 请求，避免绑定具体客户端版本。
步骤二：抓热点标题
只抓标题、链接和摘要，不整段搬运正文，既省 token 也避开版权风险。
```python
fetch_hot.py
import feedparser, hashlib, json, time
from datetime import datetime, timezone
FEEDS = [
"https://换成你关注的公开RSS地址",
多加几个源，覆盖面更广
]
def fetch_all():
items = []
for url in FEEDS:
feed = feedparser.parse(url)
for entry in feed.entries:
title = (getattr(entry, "title", "") or "").strip()
link = getattr(entry, "link", "")
summary = (getattr(entry, "summary", "") or "")[:300]
if not title:
continue
uid = hashlib.md5((title + link).encode("utf-8")).hexdigest()[:12]
items.append({
"id": uid,
"title": title,
"link": link,
"summary": summary,
"fetched_at": datetime.now(timezone.utc).isoformat(),
})
time.sleep(1) # 温和抓取，别打满对方服务器
return items
if __name__ == "__main__":
data = fetch_all()
json.dump(data, open("data/raw.json", "w", encoding="utf-8"),
ensure_ascii=False, indent=2)
print(f"抓到 {len(data)} 条")
```
抓取前看一眼目标站点的 robots.txt 和使用条款，允许再用。频率放低，一天几次足够。
步骤三：给热点打分，先筛后写
模型调用是要花钱的，别把 30 条全丢进去。用规则先打分，只把前几名送去写稿。
```python
rank.py
import json
POSITIVE = ["教程", "避坑", "对比", "清单", "实测", "怎么选", "新手", "变化", "涨价", "降价"]
NEGATIVE = ["抽奖", "招聘", "广告"]
def score(item, niche_words):
text = item["title"] + " " + item["summary"]
s = 0
for w in niche_words: # 与账号定位相关的词，权重高
if w in text:
s += 2
for w in POSITIVE:
if w in text:
s += 1
for w in NEGATIVE:
if w in text:
s -= 3
return s
def pick(items, niche_words, top_n=5):
ranked = sorted(items, key=lambda i: score(i, niche_words), reverse=True)
return ranked[:top_n]
```
niche_words 按你的账号填，比如做职场效率就填 ["效率", "会议", "汇报", "表格", "工具"]。这套打分是启发式的，后面你可以用「历史发布数据」反过来调权重。
步骤四：把热点写成口播稿
关键是提示词要给出可执行的约束：时长、字数、结构、禁止项、输出格式。
中文口播语速大致每分钟 240 到 280 字，取 250 字估算比较稳。60 秒视频，正文字数控制在 230 到 260 字之间，留出停顿。这个值按你自己的实际语速测一遍再定。
```python
write_script.py
import json, os, re, requests
PROMPT = """你是短视频口播稿作者。把下面这条热点改写成一条 {duration} 秒的竖屏口播稿。
账号定位：{niche}
目标观众：{audience}
正文字数控制在 {word_count} 字左右（按每分钟 250 字估算）。
要求：
1. 前 3 秒必须是钩子：抛问题、给结论或制造反差，不要“大家好”式开场。
2. 正文只讲 2 到 3 个信息点，每点一句话，能落到具体动作或例子。
3. 结尾给一句互动引导，不超过 15 字。
4. 不编造数据、时间、人物言论。热点里没写的信息不要补，必要时用“据公开报道”这类表述。
5. 全部用短句，只用逗号和句号，不要括号、破折号、书名号。
6. 只输出 JSON，不要代码块，不要解释。字段：
{{"title":"标题候选","hook":"钩子台词","body":["信息点1","信息点2"],"cta":"互动引导","tags":["话题1","话题2"]}}
热点标题：{title}
热点摘要：{summary}
"""
def call_llm(prompt):
url = os.environ["LLM_BASE_URL"].rstrip("/") + "/chat/completions"
payload = {
"model": os.environ["LLM_MODEL"],
"messages": [{"role": "user", "content": prompt}],
"temperature": 0.7,
}
headers = {
"Authorization": "Bearer " + os.environ["LLM_API_KEY"],
"Content-Type": "application/json",
}
r = requests.post(url, headers=headers, json=payload, timeout=120)
r.raise_for_status()
return r.json()["choices"][0]["message"]["content"]
def parse_json(text):
text = text.strip()
text = re.sub(r"^``(?:json)?|``$", "", text, flags=re.M).strip()
start, end = text.find("{"), text.rfind("}")
return json.loads(text[start:end + 1])
def make_script(item, niche, audience, duration=60, word_count=250):
prompt = PROMPT.format(duration=duration, niche=niche, audience=audience,
word_count=word_count, title=item["title"],
summary=item["summary"])
for attempt in range(3): # 解析失败就重试，温度略降
try:
return parse_json(call_llm(prompt))
except Exception as e:
print("第", attempt + 1, "次失败：", e)
return None
```
拿到 JSON 之后再校验一次字数：
```python
def count_chars(script):
return len(script["hook"]) + sum(len(x) for x in script["body"]) + len(script["cta"])
```
如果比目标字数差 30 字以上，就把「实际字数 X，请压缩/扩写到 Y 字」补进提示词再跑一次。
步骤五：生成分镜时间轴字幕
口播稿有了，还要能直接用。把每句按字数估算时长，切成 SRT 字幕，剪辑软件和提词器都能读。
```python
to_srt.py
def fmt(t):
h, t = divmod(t, 3600)
m, s = divmod(t, 60)
ms = int((s - int(s)) * 1000)
return f"{int(h):02d}:{int(m):02d}:{int(s):02d},{ms:03d}"
def build_srt(lines, cps=4.2, gap=0.15):
"""cps: 每秒口播字数，250字/分钟约等于 4.17"""
out, clock = [], 0.0
for i, line in enumerate(lines, 1):
dur = max(len(line) / cps, 1.0)
out.append(f"{i}\n{fmt(clock)} --> {fmt(clock + dur)}\n{line}\n")
clock += dur + gap
return "\n".join(out)
def script_to_lines(script):
return [script["hook"]] + list(script["body"]) + [script["cta"]]
```
渲染成一份可直接看的脚本卡：
```python
def render_md(script, item):
lines = script_to_lines(script)
body = "\n".join(f"- {x}" for x in script["body"])
srt = build_srt(lines)
return f"""# {script['title']}
来源热点：{item['title']}
参考链接：{item['link']}
标签：{'、'.join(script.get('tags', []))}
预估字数：{sum(len(x) for x in lines)}
口播稿
钩子：{script['hook']}
{body}
结尾：{script['cta']}
字幕
```srt
{srt}```
"""
```
步骤六：接上定时任务与去重
不记录已用过的热点，第二天会重复推同一条。用一个 SQLite 表兜住：
```python
store.py
import sqlite3
def init_db(path="data/seen.db"):
conn = sqlite3.connect(path)
conn.execute("CREATE TABLE IF NOT EXISTS seen (id TEXT PRIMARY KEY, used_at TEXT)")
conn.commit()
return conn
def is_new(conn, uid):
return conn.execute("SELECT 1 FROM seen WHERE id=?", (uid,)).fetchone() is None
def mark(conn, uid, ts):
conn.execute("INSERT OR IGNORE INTO seen VALUES (?, ?)", (uid, ts))
conn.commit()
```
串起整条流水线：
```python
pipeline.py
import json, os
from datetime import datetime
from fetch_hot import fetch_all
from rank import pick
from write_script import make_script
from to_srt import render_md
from store import init_db, is_new, mark
NICHE = "给职场新人的效率工具测评"
AUDIENCE = "工作一到三年、想少加班的上班族"
NICHE_WORDS = ["效率", "会议", "汇报", "表格", "工具"]
def main():
conn = init_db()
items = [i for i in fetch_all() if is_new(conn, i["id"])]
print("新热点：", len(items))
today = datetime.now().strftime("%Y-%m-%d")
folder = f"out/{today}"
os.makedirs(folder, exist_ok=True)
for item in pick(items, NICHE_WORDS, top_n=5):
script = make_script(item, NICHE, AUDIENCE)
if not script:
continue
path = f"{folder}/{item['id']}.md"
open(path, "w", encoding="utf-8").write(render_md(script, item))
mark(conn, item["id"], datetime.now().isoformat())
print("已生成：", path)
if __name__ == "__main__":
main()
```
定时执行（Linux/macOS）：
```bash
crontab -e
```
```cron
0 8 * * * cd /绝对路径/hot2script && /绝对路径/hot2script/.venv/bin/python pipeline.py >> logs/run.log 2>&1
```
Windows 用任务计划程序，操作填 python.exe 的绝对路径，起始位置填项目目录。
常见坑与排错
JSON 解析失败。 模型经常在 JSON 外面加 ``json 围栏，或者前面加一句「好的，以下是」。代码里的 parse_json` 已经做了去围栏和截取首尾大括号，但更稳的做法是重试两三次，并把失败原文写进日志，方便回头调提示词。
生成的稿子读起来拗口。 多半是书面语太重。在提示词里加两条限制：每句不超过 25 字；不要出现「综上所述」「值得注意的是」这类连接词。生成后可以写个简单检查，把超过 30 字的句子打出来人工看。
字数老是超标。 模型的字数和你的直觉不一致，别指望一次到位。把「目标字数」和「实际字数」都算出来，差得多了就带反馈重跑一次，通常一次就够。
定时任务里 API Key 读不到。 cron 不加载你的 shell 配置文件。把环境变量写进 crontab 顶部，或者在脚本里显式读取一个 .env 文件，别依赖 ~/.bashrc。
抓到的热点是旧的。 RSS 有缓存，检查条目里的发布时间字段，超过设定时效（比如 24 小时）的直接丢掉。别拿三天前的新闻当热点。
模型补了不存在的事实。 这是最需要防的一类问题。提示词里明确禁止编造，同时把「来源链接」原样写进生成的 Markdown 文件，发布前点开对一遍。
接口限流或超时。 加指数退避重试，批量处理时每条之间 sleep 一秒。并发别开太高，多数服务的免费档扛不住。
抓取踩线。 只取标题和短摘要，不存全文，不绕过登录墙，遵守 robots.txt 和目标站点的使用条款。
下一步建议
一是接上语音合成，把口播稿直接转成配音文件，和 SRT 一起丢进剪辑软件，出片时间还能再压一截。二是给每条热点生成三个候选标题，发布后用播放完成率回测，把表现好的表达方式沉淀回提示词。三是把发布数据（完播率、点赞、评论关键词）回流到步骤三的打分函数里，用真实反馈调关键词权重，比拍脑袋准确。四是加一道人工审核状态位，脚本生成后默认是「待审」，你确认过事实再改成「可用」，这样自动化跑得再快也不会把没核对的内容发出去。