""" 单文件博客：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""" {% block title %}博客{% endblock %}

{{ site_title }}

{% block content %}{% endblock %}
""" TEMPLATES["home.html"] = """ {% extends "base.html" %} {% block title %}首页 — {{ site_title }}{% endblock %} {% block content %}

文章目录

分组按名称字典序排列；文章：有「显示顺序」时按该顺序，否则按发布时间倒序。

{{ tree_html|safe }}
{% endblock %} """ TEMPLATES["post.html"] = """ {% extends "base.html" %} {% block title %}{{ article.title }} — {{ site_title }}{% endblock %} {% block content %}

返回 · {{ article.published_at }}

{{ article.title }}

{{ article.body_html|safe }}
{% endblock %} """ TEMPLATES["admin/login.html"] = """ {% extends "base.html" %} {% block title %}登录 — {{ site_title }}{% endblock %} {% block content %}

管理员登录

{% if error %}

{{ error }}

{% endif %}
{% endblock %} """ TEMPLATES["admin/home.html"] = """ {% extends "base.html" %} {% block title %}管理 — {{ site_title }}{% endblock %} {% block content %}

管理后台

{% endblock %} """ TEMPLATES["admin/groups.html"] = """ {% extends "base.html" %} {% block title %}分组 — {{ site_title }}{% endblock %} {% block content %}

分组

{% if msg %}

{{ msg }}

{% endif %} {% if err %}

{{ err }}

{% endif %}

新建根分组

树与操作

{{ tree_admin_html|safe }}

子分组按名称排序（不区分大小写）。删除分组要求无子分组且无文章引用。

{% endblock %} """ TEMPLATES["admin/articles.html"] = """ {% extends "base.html" %} {% block title %}文章 — {{ site_title }}{% endblock %} {% block content %}

文章列表

新建文章

{% if msg %}

{{ msg }}

{% endif %} {% if err %}

{{ err }}

{% endif %} {% for a in articles %} {% endfor %}
标题Slug分组发布顺序操作
{{ a.title }} {{ a.slug }} {{ a.group_name }} {{ a.published_at }} {{ a.display_order if a.display_order is not none else '' }} 编辑
{% endblock %} """ TEMPLATES["admin/article_form.html"] = """ {% extends "base.html" %} {% block title %}{{ form_title }} — {{ site_title }}{% endblock %} {% block content %}

{{ form_title }}

{% if err %}

{{ err }}

{% endif %}
返回
{% 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'
  • {jinja_env.from_string("{{n|e}}").render(n=g["name"])}'] if arts: parts.append('") if kids: parts.append("") parts.append("
    • ") return "".join(parts) roots = by_parent.get(None, []) if not roots and not conn.execute("SELECT 1 FROM articles LIMIT 1").fetchone(): return '

    暂无分组与文章。请登录后台添加。

    ' inner = "".join(node_html(r["id"]) for r in roots) return f"" 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'' ) 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"""
  • {name_esc} id={gid}
    • """ lis.append(block) return "".join(lis) inner = subtree(None) return f"" 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( "

    请先创建分组。

    ", 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()