From c153703b575ee037878c6e58d3d5fb2de4863995 Mon Sep 17 00:00:00 2001 From: Frank Vitrano <90512559+WHC2006@users.noreply.github.com> Date: Sun, 10 May 2026 11:04:30 +0800 Subject: [PATCH] first upload --- blog.py | 1172 ++++++++++++++++++++++++++++++++++++++++++++++ requirements.txt | 6 + 2 files changed, 1178 insertions(+) create mode 100644 blog.py create mode 100644 requirements.txt diff --git a/blog.py b/blog.py new file mode 100644 index 0000000..f4ec945 --- /dev/null +++ b/blog.py @@ -0,0 +1,1172 @@ +""" +单文件博客：FastAPI + SQLite + Jinja2，Markdown 预渲染为 HTML。 + +安装依赖: + pip install -r requirements.txt + # 或: pip install fastapi "uvicorn[standard]" jinja2 python-multipart markdown + +运行: + 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 hashlib +import os +import re +import secrets +import sqlite3 +from contextlib import contextmanager +from datetime import datetime, timezone +from pathlib import Path +from typing import Any, Optional +from urllib.parse import quote + +import markdown +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 +# ----------------------------------------------------------------------------- + +_MD_EXTENSIONS = [ + "markdown.extensions.fenced_code", + "markdown.extensions.tables", + "markdown.extensions.nl2br", + "markdown.extensions.sane_lists", +] + + +def render_markdown(md_text: str) -> str: + return markdown.markdown(md_text, extensions=_MD_EXTENSIONS) + + +# ----------------------------------------------------------------------------- +# 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() diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..0e100f6 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,6 @@ +fastapi>=0.110,<1 +uvicorn[standard]>=0.27,<1 +jinja2>=3.1,<4 +python-multipart>=0.0.9,<1 +markdown>=3.5,<4 +isdangerous \ No newline at end of file