智能摘要 DeepSeek
本文介绍用 FastAPI + SQLite 自建一言 API：单文件约150行代码，兼容 hitokoto 返回字段，支持随机句子、分类过滤、增删管理及统计，实测可运行。部署采用 uvicorn + systemd + Nginx 反代，管理接口以环境变量令牌鉴权。作者主张以自建替代依赖第三方免费接口，保证可用性与语料自主，并可作练手项目复用。

一句话结论： 与其依赖第三方一言接口的稳定性，不如花 20 分钟用 FastAPI + SQLite 自己写一个——单文件 150 行代码，部署到任意一台 VPS 上，随机句子、分类过滤、管理增删全部可控，本文所有代码实测可运行。

给博客配一个「一言」API：用 FastAPI + SQLite 从零自建-缘梦の镇

为什么自己写一个？

博客侧边栏挂「每日一句」是老传统了。市面上的公共一言 API（比如 hitokoto 社区版）确实好用，但把博客上一个显眼的动态区块押在免费服务的可用性上，心里总不踏实：限流、改版、挂掉，你说了不算。之前分享过 nuoxian's API 这类免费接口服务，思路一致——但对开发者来说，这种轻量服务最好的用法是：别人的拿来过渡，自己的拿来生产。

自建成本极低，收益明确：

  • 可用性自主：跟你的博客同一台服务器，博客活着它就活着；
  • 语料自主：想收什么句子、想加什么分类，一个 POST 的事；
  • 响应极快：SQLite 单机随机查询，内网 <5ms；
  • 顺便练手：FastAPI 是目前 Python Web 框架里类型提示做得最舒服的一个，拿来当入门项目刚好。

技术选型一句话说明：FastAPI（自带 OpenAPI 文档、Pydantic 校验）+ SQLite（零运维，WAL 模式支持并发读）+ uvicorn（ASGI 服务器）。如果日请求量到几十万，再考虑换 PostgreSQL，个人博客场景 SQLite 就是正解。

项目结构

hitokoto-api/
├── app.py            # 全部逻辑，单文件
├── requirements.txt  # fastapi + uvicorn
└── hitokoto.db       # 首次启动自动生成
# requirements.txt
fastapi>=0.115
uvicorn>=0.30

核心代码（实测通过）

下面是 app.py 完整代码，接口设计兼容 hitokoto 社区的返回字段（hitokoto 字段名不变），前端迁移零成本：

"""一言 API —— FastAPI + SQLite 极简实现。

启动：uvicorn app:app --host 127.0.0.1 --port 8100
接口：
  GET  /hitokoto          随机返回一句话，可用 c 参数按分类过滤（可多值）
  GET  /hitokoto/{qid}    按 id 精确获取
  GET  /stats             语料统计
  POST /admin/quotes      新增一句话（请求头 X-Admin-Token 鉴权）
  DELETE /admin/quotes/{qid}  删除一句话（同上鉴权）
"""

import os
import sqlite3
from pathlib import Path
from typing import List

from fastapi import Depends, FastAPI, Header, HTTPException, Query
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel, Field

BASE_DIR = Path(__file__).resolve().parent
DB_PATH = Path(os.getenv("HITOKOTO_DB", str(BASE_DIR / "hitokoto.db")))
ADMIN_TOKEN = os.getenv("HITOKOTO_ADMIN_TOKEN", "change-me-in-production")

# 分类代码沿用一言(hitokoto)社区惯例，方便前端无缝迁移
CATEGORIES = {
    "a": "动画", "b": "漫画", "c": "文学", "d": "哲学", "e": "网络",
    "f": "其他", "g": "诗词", "h": "网易云", "i": "哲学", "j": "随笔",
}

app = FastAPI(title="hitokoto-api", version="1.0.0")

# 允许任意来源跨域：博客前端直接 fetch 调用
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_methods=["GET", "POST", "DELETE"],
    allow_headers=["*"],
)


def get_db():
    conn = sqlite3.connect(DB_PATH, timeout=5)
    conn.row_factory = sqlite3.Row
    try:
        yield conn
    finally:
        conn.close()


def init_db() -> None:
    """建表并写入种子数据。WAL 模式下多 worker 并发读写互不阻塞。"""
    conn = sqlite3.connect(DB_PATH)
    try:
        conn.execute("PRAGMA journal_mode=WAL")
        conn.execute(
            """
            CREATE TABLE IF NOT EXISTS quotes (
                id       INTEGER PRIMARY KEY AUTOINCREMENT,
                content  TEXT    NOT NULL,
                author   TEXT    NOT NULL DEFAULT '',
                source   TEXT    NOT NULL DEFAULT '',
                category TEXT    NOT NULL DEFAULT 'f',
                created_at INTEGER NOT NULL DEFAULT (unixepoch())
            )
            """
        )
        seed = [
            ("越是试着忘记，越记得深刻。", "", "慕星", "a"),
            ("不要停止奔跑，不要回顾来路，来路无可眷恋，值得期待的只有前方。", "", "《马男波杰克》", "a"),
            ("我们听过无数的道理，却仍旧过不好这一生。", "", "《后会无期》", "f"),
            ("世界上只有一种真正的英雄主义，那就是认清生活的真相后依然热爱生活。", "罗曼·罗兰", "《米开朗琪罗传》", "c"),
            ("山不过来，我就过去。", "", "", "j"),
        ]
        count = conn.execute("SELECT COUNT(*) FROM quotes").fetchone()[0]
        if count == 0:
            conn.executemany(
                "INSERT INTO quotes (content, author, source, category) VALUES (?,?,?,?)",
                seed,
            )
        conn.commit()
    finally:
        conn.close()


init_db()


def require_admin(x_admin_token: str = Header(default="")) -> None:
    if ADMIN_TOKEN == "change-me-in-production":
        raise HTTPException(500, "请先通过环境变量 HITOKOTO_ADMIN_TOKEN 设置管理令牌")
    if x_admin_token != ADMIN_TOKEN:
        raise HTTPException(401, "无效的管理令牌")


class QuoteIn(BaseModel):
    content: str = Field(min_length=1, max_length=500)
    author: str = Field(default="", max_length=100)
    source: str = Field(default="", max_length=200)
    category: str = Field(default="f", max_length=1)


@app.get("/hitokoto")
def random_hitokoto(
    c: List[str] = Query(default=[], description="分类代码，可重复传多次，如 ?c=a&c=b"),
    db: sqlite3.Connection = Depends(get_db),
):
    codes = [x for x in c if x in CATEGORIES]
    if codes:
        marks = ",".join("?" * len(codes))
        row = db.execute(
            f"SELECT * FROM quotes WHERE category IN ({marks}) ORDER BY RANDOM() LIMIT 1",
            codes,
        ).fetchone()
    else:
        row = db.execute("SELECT * FROM quotes ORDER BY RANDOM() LIMIT 1").fetchone()
    if row is None:
        raise HTTPException(404, "语料库为空")
    return {
        "id": row["id"],
        "hitokoto": row["content"],
        "author": row["author"] or None,
        "source": row["source"] or None,
        "category": row["category"],
    }


@app.get("/hitokoto/{qid}")
def get_hitokoto(qid: int, db: sqlite3.Connection = Depends(get_db)):
    row = db.execute("SELECT * FROM quotes WHERE id=?", (qid,)).fetchone()
    if row is None:
        raise HTTPException(404, "未找到该条目")
    return {
        "id": row["id"],
        "hitokoto": row["content"],
        "author": row["author"] or None,
        "source": row["source"] or None,
        "category": row["category"],
    }


@app.get("/stats")
def stats(db: sqlite3.Connection = Depends(get_db)):
    total = db.execute("SELECT COUNT(*) FROM quotes").fetchone()[0]
    by_cat = db.execute(
        "SELECT category, COUNT(*) AS n FROM quotes GROUP BY category ORDER BY n DESC"
    ).fetchall()
    return {"total": total, "categories": {r["category"]: r["n"] for r in by_cat}}


@app.post("/admin/quotes", status_code=201)
def add_quote(q: QuoteIn, _: None = Depends(require_admin),
              db: sqlite3.Connection = Depends(get_db)):
    if q.category not in CATEGORIES:
        raise HTTPException(422, f"category 必须是 {sorted(CATEGORIES)} 之一")
    cur = db.execute(
        "INSERT INTO quotes (content, author, source, category) VALUES (?,?,?,?)",
        (q.content, q.author, q.source, q.category),
    )
    db.commit()
    return {"id": cur.lastrowid}


@app.delete("/admin/quotes/{qid}")
def delete_quote(qid: int, _: None = Depends(require_admin),
                 db: sqlite3.Connection = Depends(get_db)):
    cur = db.execute("DELETE FROM quotes WHERE id=?", (qid,))
    db.commit()
    if cur.rowcount == 0:
        raise HTTPException(404, "未找到该条目")
    return {"deleted": qid}

几个设计决策交代一下：

  • ORDER BY RANDOM() 够用吗？ 1 万条以内毫秒级返回，个人语料库完全没压力。真到十万条量级再改成「随机 id 区间两次查询」的方案，不要过早优化；
  • 每个请求开一个连接：SQLite 连接开销极低（微秒级），比维护连接池省心，配 WAL 模式不会出现读写锁等待；
  • 管理令牌走环境变量：代码里默认值 change-me-in-production 会直接拒绝所有写操作，防止带默认口令上线。

本地实测记录

装依赖并启动：

python3 -m venv .venv && source .venv/bin/activate
pip install -r requirements.txt

export HITOKOTO_ADMIN_TOKEN="你的管理令牌"
uvicorn app:app --host 127.0.0.1 --port 8100

实测输出（与本文写作时运行结果一致）：

$ curl -s http://127.0.0.1:8100/hitokoto
{"id":2,"hitokoto":"不要停止奔跑，不要回顾来路，来路无可眷恋，值得期待的只有前方。","author":null,"source":"《马男波杰克》","category":"a"}

$ curl -s "http://127.0.0.1:8100/hitokoto?c=a"
{"id":1,"hitokoto":"越是试着忘记，越记得深刻。","author":null,"source":"慕星","category":"a"}

$ curl -s http://127.0.0.1:8100/stats
{"total":5,"categories":{"a":2,"j":1,"f":1,"c":1}}

管理接口实测——无令牌请求被正确拒绝，带令牌可增删：

$ curl -s -X POST http://127.0.0.1:8100/admin/quotes \
    -H "Content-Type: application/json" -d '{"content":"测试"}'
{"detail":"无效的管理令牌"}          # 401

$ curl -s -X POST http://127.0.0.1:8100/admin/quotes \
    -H "Content-Type: application/json" -H "X-Admin-Token: 你的管理令牌" \
    -d '{"content":"测试一句话","author":"测试者","category":"j"}'
{"id":6}                            # 201

$ curl -s -X DELETE http://127.0.0.1:8100/admin/quotes/6 -H "X-Admin-Token: 你的管理令牌"
{"deleted":6}

另外 FastAPI 自动生成了交互式文档，浏览器打开 http://127.0.0.1:8100/docs 就是，接口联调零成本。

部署：systemd + Nginx

1. 先把代码放到固定位置（/opt/hitokoto-api），venv 建在项目里：

sudo mkdir -p /opt/hitokoto-api
sudo chown $USER /opt/hitokoto-api
# 把 app.py、requirements.txt 传上去，然后：
cd /opt/hitokoto-api
python3 -m venv .venv && .venv/bin/pip install -r requirements.txt

2. systemd 守护，写 /etc/systemd/system/hitokoto.service：

[Unit]
Description=Hitokoto API (FastAPI + uvicorn)
After=network.target

[Service]
Type=simple
User=www-data
WorkingDirectory=/opt/hitokoto-api
Environment="HITOKOTO_ADMIN_TOKEN=换成强随机令牌"
ExecStart=/opt/hitokoto-api/.venv/bin/uvicorn app:app --host 127.0.0.1 --port 8100 --workers 2
Restart=always
RestartSec=3

[Install]
WantedBy=multi-user.target
sudo chown -R www-data:www-data /opt/hitokoto-api
sudo systemctl daemon-reload
sudo systemctl enable --now hitokoto
systemctl status hitokoto --no-pager   # active (running) 即成功

关于 --workers 2：多 worker 意味着多进程共享 SQLite 文件。WAL 模式下读不阻塞写，实测无锁冲突；但要注意 init_db() 里 CREATE TABLE IF NOT EXISTS 是幂等的，多进程同时启动也不会建重表。

3. Nginx 反代，在博客站点的 server 块里加一段：

location /api/hitokoto/ {
    proxy_pass http://127.0.0.1:8100/;
    proxy_set_header Host $host;
    proxy_set_header X-Real-IP $remote_addr;
}
sudo nginx -t && sudo systemctl reload nginx
curl -s https://blog.loyur.cn/api/hitokoto/    # 走域名验证

4. 管理令牌建议固定下来，以后加句子一条 curl 的事：

curl -X POST https://blog.loyur.cn/api/hitokoto/admin/quotes \
  -H "Content-Type: application/json" -H "X-Admin-Token: 你的管理令牌" \
  -d '{"content":"代码是写给人看的，只是顺便能在机器上运行。","author":"Harold Abelson","source":"《SICP》","category":"c"}'

前端接入：给博客侧边栏挂一句

WordPress 的主题自定义 HTML 小工具里贴这段，每次进页面随机换一句：

<p id="hitokoto" style="text-align:center;color:#666;">加载中…</p>
<script>
fetch('/api/hitokoto/')
  .then(r => r.json())
  .then(d => {
    const who = d.author ? ` —— ${d.author}` : '';
    document.getElementById('hitokoto').textContent = `${d.hitokoto}${who}`;
  })
  .catch(() => document.getElementById('hitokoto').textContent = '一言服务打盹了');
</script>

小结

150 行 Python、一张 SQLite 表、一个 systemd 单元，一个完全属于你的一言 API 就跑起来了。这种小服务的意义不止于「能用」：语料库是你的品味，接口是你的作品，部署链路是你的肌肉记忆。下次想给博客加「历史上的今天」或者「随机壁纸」接口，照着这个模子改一改就能交付——欢迎把你的成品链接发到评论区来串门。