cnbandCursor
cf8eba082f
部署时通过 install.sh 自动编译 marknb；日常更新用 upgrade.sh 拉代码并重编译。 Co-authored-by: Cursor <cursoragent@cursor.com>
1210 lines
44 KiB
Python
1210 lines
44 KiB
Python
"""
|
||
单文件博客：FastAPI + SQLite + Jinja2，Markdown 预渲染为 HTML。
|
||
|
||
安装依赖:
|
||
bash scripts/install.sh
|
||
# 或手动: 编译 vendor/marknb/libmarknb.so 后 pip install -r requirements.txt
|
||
|
||
运行:
|
||
python blog.py
|
||
# 默认监听 127.0.0.1:8765，数据目录 ./data/（可用 BLOG_DATA_DIR 覆盖根路径）
|
||
|
||
环境变量:
|
||
BLOG_HOST 默认 127.0.0.1
|
||
BLOG_PORT 默认 8765
|
||
BLOG_DATA_DIR 默认 data（其下使用 blog.sqlite3）
|
||
BLOG_SITE_TITLE 站点标题（可选）
|
||
BLOG_ADMIN_PASSWORD 管理员密码；未设置且数据库无哈希时，启动时生成随机密码并打印一次
|
||
BLOG_SECRET_KEY 会话签名密钥；未设置则从 SQLite settings 读取或生成并持久化
|
||
BLOG_ICP_TEXT / BLOG_ICP_URL 页脚备案文案与链接（可选）
|
||
|
||
图片: 正文 Markdown 仅支持外链图床 URL，不提供本地上传。
|
||
|
||
数学公式: 使用 KaTeX auto-render，支持 $$..$$、$..$、\\(..\\)、\\[..\\]（见 base 模板内 delimiters）。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import ctypes
|
||
import hashlib
|
||
import os
|
||
import re
|
||
import secrets
|
||
import sqlite3
|
||
import sys
|
||
from contextlib import contextmanager
|
||
from datetime import datetime, timezone
|
||
from pathlib import Path
|
||
from typing import Any, Optional
|
||
from urllib.parse import quote
|
||
from fastapi import FastAPI, File, Form, HTTPException, Request, UploadFile
|
||
from fastapi.responses import HTMLResponse, RedirectResponse
|
||
from jinja2 import BaseLoader, Environment, TemplateNotFound
|
||
from starlette.middleware.sessions import SessionMiddleware
|
||
from uvicorn import Config, Server
|
||
from uvicorn.middleware.proxy_headers import ProxyHeadersMiddleware
|
||
|
||
# -----------------------------------------------------------------------------
|
||
# Config
|
||
# -----------------------------------------------------------------------------
|
||
|
||
DATA_DIR = Path(os.environ.get("BLOG_DATA_DIR", "data"))
|
||
DB_PATH = DATA_DIR / "blog.sqlite3"
|
||
BLOG_HOST = os.environ.get("BLOG_HOST", "127.0.0.1")
|
||
BLOG_PORT = int(os.environ.get("BLOG_PORT", "8765"))
|
||
BLOG_ICP_TEXT = os.environ.get("BLOG_ICP_TEXT", "").strip()
|
||
BLOG_ICP_URL = os.environ.get("BLOG_ICP_URL", "").strip()
|
||
|
||
PBKDF2_ITERATIONS = 390_000
|
||
SALT_LEN = 16
|
||
|
||
# -----------------------------------------------------------------------------
|
||
# Password (stdlib pbkdf2_hmac, no extra deps)
|
||
# -----------------------------------------------------------------------------
|
||
|
||
|
||
def hash_password(password: str) -> str:
|
||
salt = secrets.token_bytes(SALT_LEN)
|
||
dk = hashlib.pbkdf2_hmac(
|
||
"sha256", password.encode("utf-8"), salt, PBKDF2_ITERATIONS
|
||
)
|
||
return f"{PBKDF2_ITERATIONS}${salt.hex()}${dk.hex()}"
|
||
|
||
|
||
def verify_password(password: str, stored: str) -> bool:
|
||
try:
|
||
it_s, salt_hex, dk_hex = stored.split("$", 2)
|
||
iterations = int(it_s)
|
||
salt = bytes.fromhex(salt_hex)
|
||
expected = bytes.fromhex(dk_hex)
|
||
dk = hashlib.pbkdf2_hmac(
|
||
"sha256", password.encode("utf-8"), salt, iterations
|
||
)
|
||
return secrets.compare_digest(dk, expected)
|
||
except Exception:
|
||
return False
|
||
|
||
|
||
# -----------------------------------------------------------------------------
|
||
# DB
|
||
# -----------------------------------------------------------------------------
|
||
|
||
|
||
def _utc_now_iso() -> str:
|
||
return datetime.now(timezone.utc).replace(microsecond=0).isoformat()
|
||
|
||
|
||
def init_db() -> None:
|
||
DATA_DIR.mkdir(parents=True, exist_ok=True)
|
||
conn = sqlite3.connect(DB_PATH)
|
||
try:
|
||
conn.executescript(
|
||
"""
|
||
PRAGMA foreign_keys = ON;
|
||
CREATE TABLE IF NOT EXISTS settings (
|
||
key TEXT PRIMARY KEY,
|
||
value TEXT NOT NULL
|
||
);
|
||
CREATE TABLE IF NOT EXISTS groups (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
parent_id INTEGER REFERENCES groups(id) ON DELETE RESTRICT,
|
||
name TEXT NOT NULL,
|
||
created_at TEXT NOT NULL
|
||
);
|
||
CREATE TABLE IF NOT EXISTS articles (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
group_id INTEGER NOT NULL REFERENCES groups(id) ON DELETE RESTRICT,
|
||
title TEXT NOT NULL,
|
||
slug TEXT NOT NULL UNIQUE,
|
||
body_md TEXT NOT NULL,
|
||
body_html TEXT NOT NULL,
|
||
published_at TEXT NOT NULL,
|
||
display_order INTEGER,
|
||
created_at TEXT NOT NULL,
|
||
updated_at TEXT NOT NULL
|
||
);
|
||
CREATE INDEX IF NOT EXISTS idx_articles_group ON articles(group_id);
|
||
CREATE INDEX IF NOT EXISTS idx_groups_parent ON groups(parent_id);
|
||
"""
|
||
)
|
||
conn.commit()
|
||
finally:
|
||
conn.close()
|
||
|
||
|
||
@contextmanager
|
||
def get_db():
|
||
conn = sqlite3.connect(DB_PATH)
|
||
conn.row_factory = sqlite3.Row
|
||
conn.execute("PRAGMA foreign_keys = ON")
|
||
try:
|
||
yield conn
|
||
conn.commit()
|
||
except Exception:
|
||
conn.rollback()
|
||
raise
|
||
finally:
|
||
conn.close()
|
||
|
||
|
||
def get_setting(conn: sqlite3.Connection, key: str) -> Optional[str]:
|
||
row = conn.execute(
|
||
"SELECT value FROM settings WHERE key = ?", (key,)
|
||
).fetchone()
|
||
return row[0] if row else None
|
||
|
||
|
||
def set_setting(conn: sqlite3.Connection, key: str, value: str) -> None:
|
||
conn.execute(
|
||
"INSERT INTO settings(key, value) VALUES(?, ?) "
|
||
"ON CONFLICT(key) DO UPDATE SET value = excluded.value",
|
||
(key, value),
|
||
)
|
||
|
||
|
||
def bootstrap_secrets() -> str:
|
||
"""返回 SessionMiddleware 使用的 secret；副作用：初始化管理员密码与密钥。"""
|
||
with get_db() as conn:
|
||
# Session secret
|
||
env_sk = os.environ.get("BLOG_SECRET_KEY")
|
||
if env_sk:
|
||
session_secret = env_sk
|
||
else:
|
||
sk = get_setting(conn, "secret_key")
|
||
if not sk:
|
||
sk = secrets.token_hex(32)
|
||
set_setting(conn, "secret_key", sk)
|
||
session_secret = sk
|
||
|
||
# Admin password hash
|
||
env_pw = os.environ.get("BLOG_ADMIN_PASSWORD")
|
||
if env_pw:
|
||
set_setting(conn, "admin_password_hash", hash_password(env_pw))
|
||
elif not get_setting(conn, "admin_password_hash"):
|
||
pwd = secrets.token_urlsafe(14)
|
||
print("\n" + "=" * 64)
|
||
print(" 首次启动：管理员密码（请立即保存，不会再次打印）")
|
||
print(" ", pwd)
|
||
print("=" * 64 + "\n", flush=True)
|
||
set_setting(conn, "admin_password_hash", hash_password(pwd))
|
||
|
||
return session_secret
|
||
|
||
|
||
# -----------------------------------------------------------------------------
|
||
# Markdown → HTML (marknb: https://cnb.cool/whcself/marknb)
|
||
# -----------------------------------------------------------------------------
|
||
|
||
_MARKNB_DEFAULT = 65536
|
||
|
||
|
||
def _load_marknb() -> ctypes.CDLL:
|
||
lib_path = os.environ.get("MARKNB_LIB")
|
||
if not lib_path:
|
||
here = Path(__file__).resolve().parent
|
||
for candidate in (
|
||
here / "vendor" / "marknb" / "libmarknb.so",
|
||
Path("/usr/local/lib/libmarknb.so"),
|
||
):
|
||
if candidate.is_file():
|
||
lib_path = str(candidate)
|
||
break
|
||
if not lib_path:
|
||
print(
|
||
"找不到 marknb 动态库。请先运行: bash scripts/install.sh（升级: bash scripts/upgrade.sh）",
|
||
file=sys.stderr,
|
||
)
|
||
sys.exit(1)
|
||
|
||
lib = ctypes.CDLL(lib_path)
|
||
lib.marknb_parse.argtypes = [
|
||
ctypes.c_char_p,
|
||
ctypes.c_size_t,
|
||
ctypes.c_char_p,
|
||
ctypes.c_size_t,
|
||
ctypes.POINTER(ctypes.c_size_t),
|
||
]
|
||
lib.marknb_parse.restype = ctypes.c_int
|
||
return lib
|
||
|
||
|
||
_MARKNB = _load_marknb()
|
||
|
||
|
||
def render_markdown(md_text: str) -> str:
|
||
md = md_text.encode("utf-8")
|
||
cap = max(len(md) * 4 + 256, _MARKNB_DEFAULT)
|
||
limit = len(md) * 64 + 4096
|
||
while cap <= limit:
|
||
buf = ctypes.create_string_buffer(cap)
|
||
out_len = ctypes.c_size_t(0)
|
||
if _MARKNB.marknb_parse(md, len(md), buf, cap, ctypes.byref(out_len)) == 0:
|
||
return buf.raw[: out_len.value].decode("utf-8")
|
||
cap *= 2
|
||
raise RuntimeError("marknb: output buffer exhausted")
|
||
|
||
|
||
# -----------------------------------------------------------------------------
|
||
# Slug
|
||
# -----------------------------------------------------------------------------
|
||
|
||
_slug_re = re.compile(r"[^a-z0-9]+")
|
||
|
||
|
||
def slugify(title: str) -> str:
|
||
s = title.strip().lower()
|
||
s = _slug_re.sub("-", s).strip("-")
|
||
return s or "post"
|
||
|
||
|
||
def unique_slug(conn: sqlite3.Connection, base: str, exclude_id: Optional[int]) -> str:
|
||
slug = base
|
||
n = 2
|
||
while True:
|
||
row = conn.execute(
|
||
"SELECT id FROM articles WHERE slug = ?", (slug,)
|
||
).fetchone()
|
||
if not row or (exclude_id is not None and row[0] == exclude_id):
|
||
return slug
|
||
slug = f"{base}-{n}"
|
||
n += 1
|
||
|
||
|
||
# -----------------------------------------------------------------------------
|
||
# Jinja (string templates)
|
||
# -----------------------------------------------------------------------------
|
||
|
||
TEMPLATES: dict[str, str] = {}
|
||
|
||
TEMPLATES["base.html"] = r"""
|
||
<!DOCTYPE html>
|
||
<html lang="zh-CN">
|
||
<head>
|
||
<meta charset="utf-8"/>
|
||
<meta name="viewport" content="width=device-width, initial-scale=1"/>
|
||
<title>{% block title %}博客{% endblock %}</title>
|
||
<link rel="stylesheet" href="https://cdn.jsdelivr.net/npm/katex@0.16.11/dist/katex.min.css" crossorigin="anonymous"/>
|
||
<style>
|
||
:root { --fg:#1a1a1a; --muted:#666; --bd:#e5e5e5; --bg:#fafafa; --link:#0b57d0; }
|
||
* { box-sizing: border-box; }
|
||
body { font-family: system-ui, -apple-system, "Segoe UI", Roboto, "PingFang SC", "Microsoft YaHei", sans-serif;
|
||
margin:0; color:var(--fg); background:#fff; line-height:1.6; }
|
||
a { color: var(--link); text-decoration: none; }
|
||
a:hover { text-decoration: underline; }
|
||
header { border-bottom:1px solid var(--bd); padding:0.75rem 1.25rem; background:var(--bg); display:flex; align-items:center; justify-content:space-between; gap:1rem; flex-wrap:wrap;}
|
||
header h1 { margin:0; font-size:1.15rem; }
|
||
main { max-width: 52rem; margin: 0 auto; padding: 1.25rem; }
|
||
footer { border-top:1px solid var(--bd); margin-top:2rem; padding:1rem 1.25rem; color:var(--muted); font-size:.9rem; text-align:center; background:var(--bg);}
|
||
.muted { color: var(--muted); font-size: .9rem; }
|
||
article.post .body { margin-top: 1rem; }
|
||
article.post .body img { max-width: 100%; height: auto; }
|
||
article.post .body pre { overflow: auto; padding: .75rem 1rem; background: #f4f4f4; border-radius: 6px; border:1px solid var(--bd);}
|
||
article.post .body table { border-collapse: collapse; width:100%; margin:1rem 0; font-size:.95rem;}
|
||
article.post .body th, article.post .body td { border:1px solid var(--bd); padding:.35rem .5rem; }
|
||
.tree ul { list-style: none; padding-left: 1rem; margin: .25rem 0; border-left: 1px dashed var(--bd); }
|
||
.tree > ul { padding-left: 0; border-left: none; }
|
||
.tree li { margin: .35rem 0; }
|
||
.grp-name { font-weight: 600; }
|
||
.art-list { margin: .25rem 0 .5rem .5rem; padding:0; list-style: disc inside; }
|
||
.art-list li { margin: .2rem 0; }
|
||
.btn { display:inline-block; padding:.35rem .65rem; border:1px solid var(--bd); border-radius:6px; background:#fff; cursor:pointer; font-size:.9rem;}
|
||
.btn-primary { background: var(--link); color:#fff; border-color: var(--link); }
|
||
input[type=text], input[type=password], input[type=number], input[type=datetime-local], textarea, select {
|
||
width:100%; max-width:40rem; padding:.45rem .55rem; border:1px solid var(--bd); border-radius:6px; font:inherit;
|
||
}
|
||
textarea { min-height: 14rem; font-family: ui-monospace, SFMono-Regular, Menlo, Consolas, monospace; font-size:.88rem;}
|
||
label { display:block; margin:.5rem 0 .2rem; font-weight:500; }
|
||
.row { margin: .75rem 0; }
|
||
table.admin { width:100%; border-collapse:collapse; font-size:.92rem;}
|
||
table.admin th, table.admin td { border:1px solid var(--bd); padding:.4rem .5rem; text-align:left;}
|
||
.err { color:#b00020; margin:.5rem 0;}
|
||
.ok { color:#0a7; margin:.5rem 0;}
|
||
</style>
|
||
</head>
|
||
<body>
|
||
<header>
|
||
<h1><a href="/">{{ site_title }}</a></h1>
|
||
<nav class="muted">
|
||
<a href="/">首页</a>
|
||
{% if request.session.get('admin') %} · <a href="/admin/">管理</a> · <a href="/admin/logout">退出</a>{% else %} · <a href="/admin/login">登录</a>{% endif %}
|
||
</nav>
|
||
</header>
|
||
<main>
|
||
{% block content %}{% endblock %}
|
||
</main>
|
||
<footer>
|
||
{% if icp_text and icp_url %}
|
||
<a href="{{ icp_url }}" rel="noopener noreferrer">{{ icp_text }}</a>
|
||
{% elif icp_text %}
|
||
{{ icp_text }}
|
||
{% else %}
|
||
<span class="muted">Powered by blog.py</span>
|
||
{% endif %}
|
||
</footer>
|
||
<script defer src="https://cdn.jsdelivr.net/npm/katex@0.16.11/dist/katex.min.js" crossorigin="anonymous"></script>
|
||
<script defer src="https://cdn.jsdelivr.net/npm/katex@0.16.11/dist/contrib/auto-render.min.js" crossorigin="anonymous"></script>
|
||
<script>
|
||
document.addEventListener("DOMContentLoaded", function () {
|
||
if (typeof renderMathInElement === "undefined") return;
|
||
renderMathInElement(document.body, {
|
||
delimiters: [
|
||
{left: "$$", right: "$$", display: true},
|
||
{left: "$", right: "$", display: false},
|
||
{left: "\\(", right: "\\)", display: false},
|
||
{left: "\\[", right: "\\]", display: true}
|
||
],
|
||
throwOnError: false
|
||
});
|
||
});
|
||
</script>
|
||
</body>
|
||
</html>
|
||
"""
|
||
|
||
TEMPLATES["home.html"] = """
|
||
{% extends "base.html" %}
|
||
{% block title %}首页 — {{ site_title }}{% endblock %}
|
||
{% block content %}
|
||
<h2>文章目录</h2>
|
||
<p class="muted">分组按名称字典序排列；文章：有「显示顺序」时按该顺序，否则按发布时间倒序。</p>
|
||
<div class="tree">{{ tree_html|safe }}</div>
|
||
{% endblock %}
|
||
"""
|
||
|
||
TEMPLATES["post.html"] = """
|
||
{% extends "base.html" %}
|
||
{% block title %}{{ article.title }} — {{ site_title }}{% endblock %}
|
||
{% block content %}
|
||
<article class="post">
|
||
<p class="muted"><a href="/">返回</a> · {{ article.published_at }}</p>
|
||
<h2>{{ article.title }}</h2>
|
||
<div class="body">{{ article.body_html|safe }}</div>
|
||
</article>
|
||
{% endblock %}
|
||
"""
|
||
|
||
TEMPLATES["admin/login.html"] = """
|
||
{% extends "base.html" %}
|
||
{% block title %}登录 — {{ site_title }}{% endblock %}
|
||
{% block content %}
|
||
<h2>管理员登录</h2>
|
||
{% if error %}<p class="err">{{ error }}</p>{% endif %}
|
||
<form method="post" action="/admin/login">
|
||
<input type="hidden" name="csrf_token" value="{{ csrf_token }}"/>
|
||
<div class="row"><label>密码</label><input type="password" name="password" required autocomplete="current-password"/></div>
|
||
<button class="btn btn-primary" type="submit">登录</button>
|
||
</form>
|
||
{% endblock %}
|
||
"""
|
||
|
||
TEMPLATES["admin/home.html"] = """
|
||
{% extends "base.html" %}
|
||
{% block title %}管理 — {{ site_title }}{% endblock %}
|
||
{% block content %}
|
||
<h2>管理后台</h2>
|
||
<ul>
|
||
<li><a href="/admin/groups">分组（树）</a></li>
|
||
<li><a href="/admin/articles">文章</a></li>
|
||
</ul>
|
||
{% endblock %}
|
||
"""
|
||
|
||
TEMPLATES["admin/groups.html"] = """
|
||
{% extends "base.html" %}
|
||
{% block title %}分组 — {{ site_title }}{% endblock %}
|
||
{% block content %}
|
||
<h2>分组</h2>
|
||
{% if msg %}<p class="ok">{{ msg }}</p>{% endif %}
|
||
{% if err %}<p class="err">{{ err }}</p>{% endif %}
|
||
|
||
<h3>新建根分组</h3>
|
||
<form method="post" action="/admin/groups/add">
|
||
<input type="hidden" name="csrf_token" value="{{ csrf_token }}"/>
|
||
<input type="hidden" name="parent_id" value=""/>
|
||
<div class="row"><label>名称</label><input type="text" name="name" required maxlength="200"/></div>
|
||
<button class="btn btn-primary" type="submit">添加</button>
|
||
</form>
|
||
|
||
<h3>树与操作</h3>
|
||
{{ tree_admin_html|safe }}
|
||
|
||
<p class="muted" style="margin-top:1.5rem;">子分组按名称排序（不区分大小写）。删除分组要求无子分组且无文章引用。</p>
|
||
{% endblock %}
|
||
"""
|
||
|
||
TEMPLATES["admin/articles.html"] = """
|
||
{% extends "base.html" %}
|
||
{% block title %}文章 — {{ site_title }}{% endblock %}
|
||
{% block content %}
|
||
<h2>文章列表</h2>
|
||
<p><a class="btn btn-primary" href="/admin/articles/new">新建文章</a></p>
|
||
{% if msg %}<p class="ok">{{ msg }}</p>{% endif %}
|
||
{% if err %}<p class="err">{{ err }}</p>{% endif %}
|
||
<table class="admin">
|
||
<thead><tr><th>标题</th><th>Slug</th><th>分组</th><th>发布</th><th>顺序</th><th>操作</th></tr></thead>
|
||
<tbody>
|
||
{% for a in articles %}
|
||
<tr>
|
||
<td>{{ a.title }}</td>
|
||
<td><a href="/post/{{ a.slug }}" target="_blank" rel="noopener">{{ a.slug }}</a></td>
|
||
<td>{{ a.group_name }}</td>
|
||
<td>{{ a.published_at }}</td>
|
||
<td>{{ a.display_order if a.display_order is not none else '' }}</td>
|
||
<td>
|
||
<a href="/admin/articles/{{ a.id }}/edit">编辑</a>
|
||
<form style="display:inline" method="post" action="/admin/articles/{{ a.id }}/rerender" onsubmit="return confirm('重新渲染 HTML？');">
|
||
<input type="hidden" name="csrf_token" value="{{ csrf_token }}"/>
|
||
<button class="btn" type="submit">重渲染</button>
|
||
</form>
|
||
<form style="display:inline" method="post" action="/admin/articles/{{ a.id }}/delete" onsubmit="return confirm('确定删除？');">
|
||
<input type="hidden" name="csrf_token" value="{{ csrf_token }}"/>
|
||
<button class="btn" type="submit">删除</button>
|
||
</form>
|
||
</td>
|
||
</tr>
|
||
{% endfor %}
|
||
</tbody>
|
||
</table>
|
||
{% endblock %}
|
||
"""
|
||
|
||
TEMPLATES["admin/article_form.html"] = """
|
||
{% extends "base.html" %}
|
||
{% block title %}{{ form_title }} — {{ site_title }}{% endblock %}
|
||
{% block content %}
|
||
<h2>{{ form_title }}</h2>
|
||
{% if err %}<p class="err">{{ err }}</p>{% endif %}
|
||
<form method="post" enctype="multipart/form-data">
|
||
<input type="hidden" name="csrf_token" value="{{ csrf_token }}"/>
|
||
<div class="row"><label>标题</label><input type="text" name="title" value="{{ article.title or '' }}" required maxlength="300"/></div>
|
||
<div class="row"><label>Slug（可空，自动生成）</label><input type="text" name="slug" value="{{ article.slug or '' }}" maxlength="200" placeholder="留空则从标题生成"/></div>
|
||
<div class="row"><label>分组</label>
|
||
<select name="group_id" required>{{ group_options|safe }}</select>
|
||
</div>
|
||
<div class="row"><label>发布时间 (UTC 或本地 datetime-local)</label>
|
||
<input type="datetime-local" name="published_at" value="{{ article.published_at_input or '' }}" required/>
|
||
</div>
|
||
<div class="row"><label>显示顺序（可选，整数；留空则仅按发布时间排序）</label>
|
||
<input type="number" name="display_order" value="{{ article.display_order if article.display_order is not none else '' }}" step="1" placeholder="留空 = 自动"/>
|
||
</div>
|
||
<div class="row"><label>Markdown 正文</label>
|
||
<textarea name="body_md" rows="18">{{ article.body_md or '' }}</textarea>
|
||
</div>
|
||
<div class="row"><label>或上传 .md / .markdown 文件（覆盖上方正文）</label>
|
||
<input type="file" name="file" accept=".md,.markdown,text/markdown,text/plain"/>
|
||
</div>
|
||
<button class="btn btn-primary" type="submit">保存</button>
|
||
<a class="btn" href="/admin/articles">返回</a>
|
||
</form>
|
||
{% endblock %}
|
||
"""
|
||
|
||
|
||
class DictLoader(BaseLoader):
|
||
def get_source(self, environment, template):
|
||
if template in TEMPLATES:
|
||
return TEMPLATES[template], f"string:{template}", lambda: True
|
||
raise TemplateNotFound(template)
|
||
|
||
|
||
jinja_env = Environment(loader=DictLoader(), autoescape=True)
|
||
|
||
|
||
def render_template(name: str, **ctx: Any) -> str:
|
||
tpl = jinja_env.get_template(name)
|
||
return tpl.render(**ctx)
|
||
|
||
|
||
# -----------------------------------------------------------------------------
|
||
# Group tree helpers
|
||
# -----------------------------------------------------------------------------
|
||
|
||
|
||
def fetch_groups(conn: sqlite3.Connection) -> list[sqlite3.Row]:
|
||
return list(
|
||
conn.execute(
|
||
"SELECT id, parent_id, name, created_at FROM groups ORDER BY id"
|
||
).fetchall()
|
||
)
|
||
|
||
|
||
def children_by_parent(
|
||
groups: list[sqlite3.Row],
|
||
) -> dict[Optional[int], list[sqlite3.Row]]:
|
||
buckets: dict[Optional[int], list[sqlite3.Row]] = {}
|
||
for g in groups:
|
||
pid = g["parent_id"]
|
||
buckets.setdefault(pid, []).append(g)
|
||
for pid in buckets:
|
||
buckets[pid].sort(key=lambda r: (r["name"] or "").lower())
|
||
return buckets
|
||
|
||
|
||
def articles_for_group(conn: sqlite3.Connection, group_id: int) -> list[sqlite3.Row]:
|
||
return list(
|
||
conn.execute(
|
||
"""
|
||
SELECT id, title, slug, published_at, display_order
|
||
FROM articles
|
||
WHERE group_id = ?
|
||
ORDER BY CASE WHEN display_order IS NULL THEN 1 ELSE 0 END,
|
||
display_order ASC,
|
||
published_at DESC
|
||
""",
|
||
(group_id,),
|
||
).fetchall()
|
||
)
|
||
|
||
|
||
def build_public_tree_html(conn: sqlite3.Connection) -> str:
|
||
groups = fetch_groups(conn)
|
||
by_parent = children_by_parent(groups)
|
||
|
||
def node_html(gid: int) -> str:
|
||
g = next((x for x in groups if x["id"] == gid), None)
|
||
if not g:
|
||
return ""
|
||
arts = articles_for_group(conn, gid)
|
||
kids = by_parent.get(gid, [])
|
||
parts = [f'<li><span class="grp-name">{jinja_env.from_string("{{n|e}}").render(n=g["name"])}</span>']
|
||
if arts:
|
||
parts.append('<ul class="art-list">')
|
||
for a in arts:
|
||
parts.append(
|
||
f'<li><a href="/post/{jinja_env.from_string("{{s|e}}").render(s=a["slug"])}">'
|
||
f'{jinja_env.from_string("{{t|e}}").render(t=a["title"])}</a>'
|
||
f' <span class="muted">({jinja_env.from_string("{{p|e}}").render(p=a["published_at"])})</span></li>'
|
||
)
|
||
parts.append("</ul>")
|
||
if kids:
|
||
parts.append("<ul>")
|
||
for c in kids:
|
||
parts.append(node_html(c["id"]))
|
||
parts.append("</ul>")
|
||
parts.append("</li>")
|
||
return "".join(parts)
|
||
|
||
roots = by_parent.get(None, [])
|
||
if not roots and not conn.execute("SELECT 1 FROM articles LIMIT 1").fetchone():
|
||
return '<p class="muted">暂无分组与文章。请登录后台添加。</p>'
|
||
inner = "".join(node_html(r["id"]) for r in roots)
|
||
return f"<ul>{inner}</ul>"
|
||
|
||
|
||
def group_option_html(
|
||
conn: sqlite3.Connection, selected_id: Optional[int], indent: str = ""
|
||
) -> str:
|
||
groups = fetch_groups(conn)
|
||
by_parent = children_by_parent(groups)
|
||
|
||
def walk(pid: Optional[int], depth: int) -> str:
|
||
out = []
|
||
for g in by_parent.get(pid, []):
|
||
pad = " " * (depth * 4)
|
||
sel = " selected" if selected_id == g["id"] else ""
|
||
name_esc = jinja_env.from_string("{{n|e}}").render(n=g["name"])
|
||
out.append(
|
||
f'<option value="{g["id"]}"{sel}>{pad}{name_esc}</option>'
|
||
)
|
||
out.append(walk(g["id"], depth + 1))
|
||
return "".join(out)
|
||
|
||
return walk(None, 0)
|
||
|
||
|
||
def build_admin_group_tree_html(
|
||
conn: sqlite3.Connection, csrf_token: str
|
||
) -> str:
|
||
groups = fetch_groups(conn)
|
||
by_parent = children_by_parent(groups)
|
||
|
||
def subtree(pid: Optional[int]) -> str:
|
||
items = by_parent.get(pid, [])
|
||
if not items:
|
||
return ""
|
||
lis = []
|
||
for g in items:
|
||
gid = g["id"]
|
||
name_esc = jinja_env.from_string("{{n|e}}").render(n=g["name"])
|
||
block = f"""
|
||
<li>
|
||
<strong>{name_esc}</strong> <span class="muted">id={gid}</span>
|
||
<form style="display:inline" method="post" action="/admin/groups/rename">
|
||
<input type="hidden" name="csrf_token" value="{jinja_env.from_string("{{t|e}}").render(t=csrf_token)}"/>
|
||
<input type="hidden" name="id" value="{gid}"/>
|
||
<input type="text" name="name" value="{name_esc}" style="width:12rem;display:inline-block" maxlength="200"/>
|
||
<button class="btn" type="submit">改名</button>
|
||
</form>
|
||
<form style="display:inline" method="post" action="/admin/groups/add">
|
||
<input type="hidden" name="csrf_token" value="{jinja_env.from_string("{{t|e}}").render(t=csrf_token)}"/>
|
||
<input type="hidden" name="parent_id" value="{gid}"/>
|
||
<input type="text" name="name" placeholder="子分组名称" style="width:10rem;display:inline-block" maxlength="200"/>
|
||
<button class="btn" type="submit">添加子分组</button>
|
||
</form>
|
||
<form style="display:inline" method="post" action="/admin/groups/delete" onsubmit="return confirm('删除该分组？');">
|
||
<input type="hidden" name="csrf_token" value="{jinja_env.from_string("{{t|e}}").render(t=csrf_token)}"/>
|
||
<input type="hidden" name="id" value="{gid}"/>
|
||
<button class="btn" type="submit">删除</button>
|
||
</form>
|
||
<ul>{subtree(gid)}</ul>
|
||
</li>
|
||
"""
|
||
lis.append(block)
|
||
return "".join(lis)
|
||
|
||
inner = subtree(None)
|
||
return f"<ul>{inner}</ul>"
|
||
|
||
|
||
def group_is_deletable(conn: sqlite3.Connection, gid: int) -> tuple[bool, str]:
|
||
c1 = conn.execute(
|
||
"SELECT 1 FROM groups WHERE parent_id = ? LIMIT 1", (gid,)
|
||
).fetchone()
|
||
if c1:
|
||
return False, "存在子分组，无法删除"
|
||
c2 = conn.execute(
|
||
"SELECT 1 FROM articles WHERE group_id = ? LIMIT 1", (gid,)
|
||
).fetchone()
|
||
if c2:
|
||
return False, "分组下仍有文章，无法删除"
|
||
return True, ""
|
||
|
||
|
||
# -----------------------------------------------------------------------------
|
||
# CSRF + auth
|
||
# -----------------------------------------------------------------------------
|
||
|
||
|
||
def new_csrf() -> str:
|
||
return secrets.token_urlsafe(32)
|
||
|
||
|
||
def verify_csrf(request: Request, token: Optional[str]) -> None:
|
||
if not token or token != request.session.get("csrf"):
|
||
raise HTTPException(403, "CSRF 校验失败")
|
||
|
||
|
||
def require_admin(request: Request) -> None:
|
||
if not request.session.get("admin"):
|
||
raise HTTPException(302, headers={"Location": "/admin/login"})
|
||
|
||
|
||
# FastAPI doesn't support 302 via HTTPException headers well - use RedirectResponse in routes
|
||
|
||
# -----------------------------------------------------------------------------
|
||
# Datetime helpers for form
|
||
# -----------------------------------------------------------------------------
|
||
|
||
|
||
def dt_local_value(iso: str) -> str:
|
||
"""SQLite ISO → datetime-local 近似值（截断到分钟）。"""
|
||
try:
|
||
s = iso.replace("Z", "+00:00")
|
||
dt = datetime.fromisoformat(s)
|
||
if dt.tzinfo:
|
||
dt = dt.astimezone().replace(tzinfo=None)
|
||
return dt.strftime("%Y-%m-%dT%H:%M")
|
||
except Exception:
|
||
return ""
|
||
|
||
|
||
def parse_published_at(value: str) -> str:
|
||
"""接受 datetime-local 或 ISO 字符串，存 UTC ISO。"""
|
||
value = (value or "").strip()
|
||
if not value:
|
||
raise ValueError("empty published_at")
|
||
# 浏览器 datetime-local：无时区后缀
|
||
if re.match(r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}", value) and "+" not in value and not value.endswith("Z"):
|
||
dt = datetime.fromisoformat(value)
|
||
local_tz = datetime.now().astimezone().tzinfo or timezone.utc
|
||
dt = dt.replace(tzinfo=local_tz)
|
||
return dt.astimezone(timezone.utc).replace(microsecond=0).isoformat()
|
||
dt = datetime.fromisoformat(value.replace("Z", "+00:00"))
|
||
if dt.tzinfo is None:
|
||
dt = dt.replace(tzinfo=timezone.utc)
|
||
return dt.astimezone(timezone.utc).replace(microsecond=0).isoformat()
|
||
|
||
|
||
# -----------------------------------------------------------------------------
|
||
# App factory
|
||
# -----------------------------------------------------------------------------
|
||
|
||
SITE_TITLE = os.environ.get("BLOG_SITE_TITLE", "我的博客")
|
||
|
||
|
||
def create_app() -> FastAPI:
|
||
init_db()
|
||
session_secret = bootstrap_secrets()
|
||
|
||
app = FastAPI(title=SITE_TITLE, docs_url=None, redoc_url=None)
|
||
app.add_middleware(
|
||
SessionMiddleware,
|
||
secret_key=session_secret,
|
||
session_cookie="blog_session",
|
||
max_age=14 * 24 * 3600,
|
||
same_site="lax",
|
||
https_only=False,
|
||
)
|
||
app.add_middleware(ProxyHeadersMiddleware, trusted_hosts="*")
|
||
|
||
def tpl_ctx(request: Request, **extra):
|
||
return {
|
||
"request": request,
|
||
"site_title": SITE_TITLE,
|
||
"icp_text": BLOG_ICP_TEXT,
|
||
"icp_url": BLOG_ICP_URL,
|
||
**extra,
|
||
}
|
||
|
||
@app.get("/", response_class=HTMLResponse)
|
||
def home(request: Request):
|
||
with get_db() as conn:
|
||
tree_html = build_public_tree_html(conn)
|
||
return HTMLResponse(
|
||
render_template("home.html", **tpl_ctx(request, tree_html=tree_html))
|
||
)
|
||
|
||
@app.get("/post/{slug}", response_class=HTMLResponse)
|
||
def post_detail(request: Request, slug: str):
|
||
with get_db() as conn:
|
||
row = conn.execute(
|
||
"SELECT title, body_html, published_at FROM articles WHERE slug = ?",
|
||
(slug,),
|
||
).fetchone()
|
||
if not row:
|
||
raise HTTPException(404, "文章不存在")
|
||
article = dict(row)
|
||
return HTMLResponse(
|
||
render_template("post.html", **tpl_ctx(request, article=article))
|
||
)
|
||
|
||
# --- Admin ---
|
||
|
||
@app.get("/admin/login", response_class=HTMLResponse)
|
||
def admin_login_get(request: Request):
|
||
if request.session.get("admin"):
|
||
return RedirectResponse("/admin/", status_code=302)
|
||
tok = new_csrf()
|
||
request.session["csrf"] = tok
|
||
return HTMLResponse(
|
||
render_template(
|
||
"admin/login.html", **tpl_ctx(request, csrf_token=tok, error=None)
|
||
)
|
||
)
|
||
|
||
@app.post("/admin/login")
|
||
async def admin_login_post(
|
||
request: Request,
|
||
password: str = Form(...),
|
||
csrf_token: str = Form(...),
|
||
):
|
||
verify_csrf(request, csrf_token)
|
||
with get_db() as conn:
|
||
h = get_setting(conn, "admin_password_hash")
|
||
if not h or not verify_password(password, h):
|
||
tok = new_csrf()
|
||
request.session["csrf"] = tok
|
||
return HTMLResponse(
|
||
render_template(
|
||
"admin/login.html",
|
||
**tpl_ctx(
|
||
request,
|
||
csrf_token=tok,
|
||
error="密码错误",
|
||
),
|
||
),
|
||
status_code=401,
|
||
)
|
||
request.session["admin"] = True
|
||
request.session["csrf"] = new_csrf()
|
||
return RedirectResponse("/admin/", status_code=302)
|
||
|
||
@app.get("/admin/logout")
|
||
def admin_logout(request: Request):
|
||
request.session.clear()
|
||
return RedirectResponse("/", status_code=302)
|
||
|
||
@app.get("/admin/", response_class=HTMLResponse)
|
||
def admin_home(request: Request):
|
||
if not request.session.get("admin"):
|
||
return RedirectResponse("/admin/login", status_code=302)
|
||
return HTMLResponse(render_template("admin/home.html", **tpl_ctx(request)))
|
||
|
||
@app.get("/admin/groups", response_class=HTMLResponse)
|
||
def admin_groups_get(request: Request):
|
||
if not request.session.get("admin"):
|
||
return RedirectResponse("/admin/login", status_code=302)
|
||
tok = request.session.get("csrf") or new_csrf()
|
||
request.session["csrf"] = tok
|
||
with get_db() as conn:
|
||
tree_admin_html = build_admin_group_tree_html(conn, tok)
|
||
return HTMLResponse(
|
||
render_template(
|
||
"admin/groups.html",
|
||
**tpl_ctx(
|
||
request,
|
||
csrf_token=tok,
|
||
tree_admin_html=tree_admin_html,
|
||
msg=request.query_params.get("msg"),
|
||
err=request.query_params.get("err"),
|
||
),
|
||
)
|
||
)
|
||
|
||
@app.post("/admin/groups/add")
|
||
async def admin_groups_add(
|
||
request: Request,
|
||
csrf_token: str = Form(...),
|
||
name: str = Form(...),
|
||
parent_id: str = Form(default=""),
|
||
):
|
||
if not request.session.get("admin"):
|
||
return RedirectResponse("/admin/login", status_code=302)
|
||
verify_csrf(request, csrf_token)
|
||
name = name.strip()
|
||
if not name:
|
||
return RedirectResponse("/admin/groups?err=名称为空", status_code=302)
|
||
pid: Optional[int]
|
||
if parent_id is None or str(parent_id).strip() == "":
|
||
pid = None
|
||
else:
|
||
pid = int(parent_id)
|
||
with get_db() as conn:
|
||
if pid is not None:
|
||
p = conn.execute(
|
||
"SELECT id FROM groups WHERE id = ?", (pid,)
|
||
).fetchone()
|
||
if not p:
|
||
return RedirectResponse("/admin/groups?err=父分组不存在", status_code=302)
|
||
conn.execute(
|
||
"INSERT INTO groups(parent_id, name, created_at) VALUES(?,?,?)",
|
||
(pid, name, _utc_now_iso()),
|
||
)
|
||
return RedirectResponse("/admin/groups?msg=已添加", status_code=302)
|
||
|
||
@app.post("/admin/groups/rename")
|
||
async def admin_groups_rename(
|
||
request: Request,
|
||
csrf_token: str = Form(...),
|
||
id: int = Form(...),
|
||
name: str = Form(...),
|
||
):
|
||
if not request.session.get("admin"):
|
||
return RedirectResponse("/admin/login", status_code=302)
|
||
verify_csrf(request, csrf_token)
|
||
name = name.strip()
|
||
if not name:
|
||
return RedirectResponse("/admin/groups?err=名称为空", status_code=302)
|
||
with get_db() as conn:
|
||
conn.execute(
|
||
"UPDATE groups SET name = ? WHERE id = ?", (name, id)
|
||
)
|
||
return RedirectResponse("/admin/groups?msg=已更新", status_code=302)
|
||
|
||
@app.post("/admin/groups/delete")
|
||
async def admin_groups_delete(
|
||
request: Request,
|
||
csrf_token: str = Form(...),
|
||
id: int = Form(...),
|
||
):
|
||
if not request.session.get("admin"):
|
||
return RedirectResponse("/admin/login", status_code=302)
|
||
verify_csrf(request, csrf_token)
|
||
with get_db() as conn:
|
||
ok, reason = group_is_deletable(conn, id)
|
||
if not ok:
|
||
return RedirectResponse(
|
||
"/admin/groups?err=" + quote(reason), status_code=302
|
||
)
|
||
conn.execute("DELETE FROM groups WHERE id = ?", (id,))
|
||
return RedirectResponse("/admin/groups?msg=已删除", status_code=302)
|
||
|
||
@app.get("/admin/articles", response_class=HTMLResponse)
|
||
def admin_articles_list(request: Request):
|
||
if not request.session.get("admin"):
|
||
return RedirectResponse("/admin/login", status_code=302)
|
||
tok = request.session.get("csrf") or new_csrf()
|
||
request.session["csrf"] = tok
|
||
with get_db() as conn:
|
||
rows = conn.execute(
|
||
"""
|
||
SELECT a.id, a.title, a.slug, a.published_at, a.display_order,
|
||
g.name AS group_name
|
||
FROM articles a
|
||
JOIN groups g ON g.id = a.group_id
|
||
ORDER BY a.updated_at DESC
|
||
"""
|
||
).fetchall()
|
||
articles = [dict(r) for r in rows]
|
||
return HTMLResponse(
|
||
render_template(
|
||
"admin/articles.html",
|
||
**tpl_ctx(
|
||
request,
|
||
csrf_token=tok,
|
||
articles=articles,
|
||
msg=request.query_params.get("msg"),
|
||
err=request.query_params.get("err"),
|
||
),
|
||
)
|
||
)
|
||
|
||
@app.get("/admin/articles/new", response_class=HTMLResponse)
|
||
def admin_article_new_get(request: Request):
|
||
if not request.session.get("admin"):
|
||
return RedirectResponse("/admin/login", status_code=302)
|
||
tok = request.session.get("csrf") or new_csrf()
|
||
request.session["csrf"] = tok
|
||
with get_db() as conn:
|
||
opts = group_option_html(conn, None)
|
||
if not opts.strip():
|
||
return HTMLResponse(
|
||
"<p>请先<a href='/admin/groups'>创建分组</a>。</p>", status_code=400
|
||
)
|
||
article = {
|
||
"title": "",
|
||
"slug": "",
|
||
"body_md": "",
|
||
"published_at_input": dt_local_value(_utc_now_iso()),
|
||
"display_order": None,
|
||
}
|
||
return HTMLResponse(
|
||
render_template(
|
||
"admin/article_form.html",
|
||
**tpl_ctx(
|
||
request,
|
||
csrf_token=tok,
|
||
form_title="新建文章",
|
||
article=article,
|
||
group_options=opts,
|
||
err=request.query_params.get("err"),
|
||
),
|
||
)
|
||
)
|
||
|
||
@app.post("/admin/articles/new")
|
||
async def admin_article_new_post(
|
||
request: Request,
|
||
csrf_token: str = Form(...),
|
||
title: str = Form(...),
|
||
slug: str = Form(""),
|
||
group_id: int = Form(...),
|
||
published_at: str = Form(...),
|
||
display_order: str = Form(""),
|
||
body_md: str = Form(""),
|
||
file: Optional[UploadFile] = File(None),
|
||
):
|
||
if not request.session.get("admin"):
|
||
return RedirectResponse("/admin/login", status_code=302)
|
||
verify_csrf(request, csrf_token)
|
||
body = body_md or ""
|
||
if file and file.filename:
|
||
raw = await file.read()
|
||
body = raw.decode("utf-8", errors="replace")
|
||
title = title.strip()
|
||
slug = slug.strip()
|
||
disp: Optional[int]
|
||
if str(display_order).strip() == "":
|
||
disp = None
|
||
else:
|
||
try:
|
||
disp = int(str(display_order).strip())
|
||
except ValueError:
|
||
return RedirectResponse(
|
||
"/admin/articles/new?err=" + quote("显示顺序必须是整数"),
|
||
status_code=302,
|
||
)
|
||
try:
|
||
pub_iso = parse_published_at(published_at)
|
||
except Exception:
|
||
return RedirectResponse(
|
||
"/admin/articles/new?err=" + quote("发布时间无效"), status_code=302
|
||
)
|
||
with get_db() as conn:
|
||
g = conn.execute(
|
||
"SELECT id FROM groups WHERE id = ?", (group_id,)
|
||
).fetchone()
|
||
if not g:
|
||
return RedirectResponse("/admin/articles/new?err=分组不存在", status_code=302)
|
||
base = slugify(slug or title)
|
||
slug_f = unique_slug(conn, base, None)
|
||
html = render_markdown(body)
|
||
now = _utc_now_iso()
|
||
conn.execute(
|
||
"""
|
||
INSERT INTO articles(group_id, title, slug, body_md, body_html, published_at, display_order, created_at, updated_at)
|
||
VALUES(?,?,?,?,?,?,?,?,?)
|
||
""",
|
||
(
|
||
group_id,
|
||
title,
|
||
slug_f,
|
||
body,
|
||
html,
|
||
pub_iso,
|
||
disp,
|
||
now,
|
||
now,
|
||
),
|
||
)
|
||
return RedirectResponse("/admin/articles?msg=已创建", status_code=302)
|
||
|
||
@app.get("/admin/articles/{aid}/edit", response_class=HTMLResponse)
|
||
def admin_article_edit_get(request: Request, aid: int):
|
||
if not request.session.get("admin"):
|
||
return RedirectResponse("/admin/login", status_code=302)
|
||
tok = request.session.get("csrf") or new_csrf()
|
||
request.session["csrf"] = tok
|
||
with get_db() as conn:
|
||
row = conn.execute(
|
||
"SELECT id, group_id, title, slug, body_md, published_at, display_order FROM articles WHERE id = ?",
|
||
(aid,),
|
||
).fetchone()
|
||
if not row:
|
||
raise HTTPException(404)
|
||
article = dict(row)
|
||
article["published_at_input"] = dt_local_value(article["published_at"])
|
||
opts = group_option_html(conn, article["group_id"])
|
||
return HTMLResponse(
|
||
render_template(
|
||
"admin/article_form.html",
|
||
**tpl_ctx(
|
||
request,
|
||
csrf_token=tok,
|
||
form_title="编辑文章",
|
||
article=article,
|
||
group_options=opts,
|
||
err=request.query_params.get("err"),
|
||
),
|
||
)
|
||
)
|
||
|
||
@app.post("/admin/articles/{aid}/edit")
|
||
async def admin_article_edit_post(
|
||
request: Request,
|
||
aid: int,
|
||
csrf_token: str = Form(...),
|
||
title: str = Form(...),
|
||
slug: str = Form(""),
|
||
group_id: int = Form(...),
|
||
published_at: str = Form(...),
|
||
display_order: str = Form(""),
|
||
body_md: str = Form(""),
|
||
file: Optional[UploadFile] = File(None),
|
||
):
|
||
if not request.session.get("admin"):
|
||
return RedirectResponse("/admin/login", status_code=302)
|
||
verify_csrf(request, csrf_token)
|
||
body = body_md or ""
|
||
if file and file.filename:
|
||
raw = await file.read()
|
||
body = raw.decode("utf-8", errors="replace")
|
||
title = title.strip()
|
||
slug_in = slug.strip()
|
||
if str(display_order).strip() == "":
|
||
disp = None
|
||
else:
|
||
try:
|
||
disp = int(str(display_order).strip())
|
||
except ValueError:
|
||
return RedirectResponse(
|
||
f"/admin/articles/{aid}/edit?err="
|
||
+ quote("显示顺序必须是整数"),
|
||
status_code=302,
|
||
)
|
||
try:
|
||
pub_iso = parse_published_at(published_at)
|
||
except Exception:
|
||
return RedirectResponse(
|
||
f"/admin/articles/{aid}/edit?err="
|
||
+ quote("发布时间无效"),
|
||
status_code=302,
|
||
)
|
||
with get_db() as conn:
|
||
row = conn.execute("SELECT id FROM articles WHERE id = ?", (aid,)).fetchone()
|
||
if not row:
|
||
raise HTTPException(404)
|
||
g = conn.execute(
|
||
"SELECT id FROM groups WHERE id = ?", (group_id,)
|
||
).fetchone()
|
||
if not g:
|
||
return RedirectResponse(
|
||
f"/admin/articles/{aid}/edit?err=分组不存在", status_code=302
|
||
)
|
||
base = slugify(slug_in or title)
|
||
slug_f = unique_slug(conn, base, aid)
|
||
html = render_markdown(body)
|
||
now = _utc_now_iso()
|
||
conn.execute(
|
||
"""
|
||
UPDATE articles SET group_id=?, title=?, slug=?, body_md=?, body_html=?,
|
||
published_at=?, display_order=?, updated_at=?
|
||
WHERE id=?
|
||
""",
|
||
(
|
||
group_id,
|
||
title,
|
||
slug_f,
|
||
body,
|
||
html,
|
||
pub_iso,
|
||
disp,
|
||
now,
|
||
aid,
|
||
),
|
||
)
|
||
return RedirectResponse("/admin/articles?msg=已保存", status_code=302)
|
||
|
||
@app.post("/admin/articles/{aid}/delete")
|
||
async def admin_article_delete(
|
||
request: Request, aid: int, csrf_token: str = Form(...)
|
||
):
|
||
if not request.session.get("admin"):
|
||
return RedirectResponse("/admin/login", status_code=302)
|
||
verify_csrf(request, csrf_token)
|
||
with get_db() as conn:
|
||
conn.execute("DELETE FROM articles WHERE id = ?", (aid,))
|
||
return RedirectResponse("/admin/articles?msg=已删除", status_code=302)
|
||
|
||
@app.post("/admin/articles/{aid}/rerender")
|
||
async def admin_article_rerender(
|
||
request: Request, aid: int, csrf_token: str = Form(...)
|
||
):
|
||
if not request.session.get("admin"):
|
||
return RedirectResponse("/admin/login", status_code=302)
|
||
verify_csrf(request, csrf_token)
|
||
with get_db() as conn:
|
||
row = conn.execute(
|
||
"SELECT body_md FROM articles WHERE id = ?", (aid,)
|
||
).fetchone()
|
||
if not row:
|
||
raise HTTPException(404)
|
||
html = render_markdown(row["body_md"])
|
||
conn.execute(
|
||
"UPDATE articles SET body_html = ?, updated_at = ? WHERE id = ?",
|
||
(html, _utc_now_iso(), aid),
|
||
)
|
||
return RedirectResponse("/admin/articles?msg=已重新渲染", status_code=302)
|
||
|
||
return app
|
||
|
||
|
||
app = create_app()
|
||
|
||
|
||
def main():
|
||
init_db()
|
||
cfg = Config(
|
||
app,
|
||
host=BLOG_HOST,
|
||
port=BLOG_PORT,
|
||
log_level="info",
|
||
)
|
||
server = Server(cfg)
|
||
asyncio.run(server.serve())
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|