#!/usr/bin/env python3 """极简博客 —— 单文件 Flask 应用，文章以 JSON 存在 PVC 上。""" import os import json import re import html import datetime import markdown from flask import Flask, request, redirect, abort, render_template_string DATA_DIR = os.environ.get("DATA_DIR", "/data") DB_FILE = os.path.join(DATA_DIR, "posts.json") ADMIN_TOKEN = os.environ.get("ADMIN_TOKEN", "change-me") SITE_TITLE = os.environ.get("SITE_TITLE", "我的博客") app = Flask(__name__) def load_posts(): if not os.path.exists(DB_FILE): return [] try: with open(DB_FILE, encoding="utf-8") as f: posts = json.load(f) return sorted(posts, key=lambda p: p.get("created", ""), reverse=True) except (json.JSONDecodeError, OSError): return [] def save_posts(posts): os.makedirs(DATA_DIR, exist_ok=True) tmp = DB_FILE + ".tmp" with open(tmp, "w", encoding="utf-8") as f: json.dump(posts, f, ensure_ascii=False, indent=2) os.replace(tmp, DB_FILE) def slugify(text): slug = re.sub(r"[^\w\u4e00-\u9fff-]+", "-", text.strip().lower()).strip("-") return slug[:60] or "post" BASE_CSS = """ """ LIST_TPL = """ {{ site }}""" + BASE_CSS + """

{{ site }}

写文章 →
{% for p in posts %}

{{ p.title }}

{{ p.created }} · {{ p.views }} 次阅读

{{ p.excerpt }}

{% else %}

还没有文章。点右上角"写文章"开始。

{% endfor %}
""" POST_TPL = """ {{ p.title }} · {{ site }}""" + BASE_CSS + """

{{ site }}

← 返回列表

{{ p.title }}

{{ p.created }}
{{ body|safe }}
""" NEW_TPL = """ 写文章 · {{ site }}""" + BASE_CSS + """

{{ site }}

← 返回列表
{% if error %}

{{ error }}

{% endif %}
""" @app.get("/healthz") def healthz(): return "ok", 200 @app.get("/") def index(): posts = [] for p in load_posts(): posts.append({ "slug": p["slug"], "title": p["title"], "created": p["created"], "views": p.get("views", 0), "excerpt": p.get("excerpt", ""), }) return render_template_string(LIST_TPL, site=SITE_TITLE, posts=posts) @app.get("/post/") def post_detail(slug): posts = load_posts() for p in posts: if p["slug"] == slug: p["views"] = p.get("views", 0) + 1 save_posts(posts) body = markdown.markdown(p["body"], extensions=["fenced_code", "tables"]) return render_template_string(POST_TPL, site=SITE_TITLE, p=p, body=body) abort(404) @app.route("/new", methods=["GET", "POST"]) def new_post(): if request.method == "GET": return render_template_string(NEW_TPL, site=SITE_TITLE, error=None, title="", body="") title = (request.form.get("title") or "").strip() body = (request.form.get("body") or "").strip() token = request.form.get("token") or "" if token != ADMIN_TOKEN: return render_template_string(NEW_TPL, site=SITE_TITLE, error="管理口令不正确", title=title, body=body), 403 if not title or not body: return render_template_string(NEW_TPL, site=SITE_TITLE, error="标题和正文都不能为空", title=title, body=body), 400 posts = load_posts() base = slugify(title) slug = base n = 1 existing = {p["slug"] for p in posts} while slug in existing: n += 1 slug = f"{base}-{n}" plain = re.sub(r"[#>*_`\[\]()!-]", "", body) posts.append({ "slug": slug, "title": title, "body": body, "excerpt": html.escape(plain[:120].replace("\n", " ")), "created": datetime.datetime.now().strftime("%Y-%m-%d %H:%M"), "views": 0, }) save_posts(posts) return redirect(f"/post/{slug}") @app.post("/post//delete") def delete_post(slug): token = request.form.get("token") or "" if token != ADMIN_TOKEN: abort(403) posts = [p for p in load_posts() if p["slug"] != slug] save_posts(posts) return redirect("/") if __name__ == "__main__": app.run(host="0.0.0.0", port=int(os.environ.get("PORT", "5000")))