feat: 阶段2 Python后端、用户管理、统计、单词收藏与响应式适配

2026-08-01 17:46:04 +08:00
parent a2e705aa30
commit 55705683ff
26 changed files with 1929 additions and 42 deletions
@@ -0,0 +1 @@
+"""Spanish Diary backend package."""
@@ -0,0 +1,201 @@
+import os
+
+from flask import Flask, jsonify, request, send_from_directory, session
+
+from .auth import admin_required, authenticate, create_user, login_required
+from .db import BASE_DIR, get_db, init_db
+from .entries_api import entries_bp
+from .settings_api import settings_bp
+from .vocab_api import vocab_bp
+
+
+def create_app():
+    app = Flask(__name__, static_folder=None)
+    app.secret_key = os.environ.get("SECRET_KEY", "spanish-diary-dev-secret-change-me")
+    app.config.update(
+        SESSION_COOKIE_HTTPONLY=True,
+        SESSION_COOKIE_SAMESITE="Lax",
+        SESSION_COOKIE_SECURE=os.environ.get("COOKIE_SECURE", "0") == "1",
+        MAX_CONTENT_LENGTH=2 * 1024 * 1024,
+    )
+
+    init_db()
+    app.register_blueprint(entries_bp)
+    app.register_blueprint(vocab_bp)
+    app.register_blueprint(settings_bp)
+
+    @app.post("/api/register")
+    def register():
+        data = request.get_json(silent=True) or {}
+        username = (data.get("username") or "").strip()
+        password = data.get("password") or ""
+        display_name = (data.get("displayName") or username or "用户").strip()
+        if len(username) < 3 or len(username) > 50:
+            return jsonify({"error": "用户名长度需为 3-50 个字符"}), 400
+        if len(password) < 6:
+            return jsonify({"error": "密码至少 6 位"}), 400
+        conn = get_db()
+        try:
+            exists = conn.execute(
+                "SELECT id FROM users WHERE username = ?", (username,)
+            ).fetchone()
+        finally:
+            conn.close()
+        if exists:
+            return jsonify({"error": "用户名已存在"}), 409
+        user = create_user(username, password, display_name)
+        session["user_id"] = user["id"]
+        session.permanent = True
+        return jsonify(dict(user)), 201
+
+    @app.post("/api/login")
+    def login():
+        data = request.get_json(silent=True) or {}
+        username = (data.get("username") or "").strip()
+        password = data.get("password") or ""
+        user = authenticate(username, password)
+        if user is None:
+            return jsonify({"error": "用户名或密码错误"}), 401
+        if user.get("disabled"):
+            return jsonify({"error": "账号已被禁用"}), 403
+        session["user_id"] = user["id"]
+        session.permanent = True
+        return jsonify(
+            {
+                "id": user["id"],
+                "username": user["username"],
+                "displayName": user["display_name"],
+                "isAdmin": bool(user["is_admin"]),
+                "isActive": bool(user["is_active"]),
+            }
+        )
+
+    @app.post("/api/logout")
+    def logout():
+        session.clear()
+        return jsonify({"ok": True})
+
+    @app.get("/api/me")
+    @login_required
+    def me(user):
+        return jsonify(
+            {
+                "id": user["id"],
+                "username": user["username"],
+                "displayName": user["display_name"],
+                "isAdmin": bool(user["is_admin"]),
+                "isActive": bool(user["is_active"]),
+            }
+        )
+
+    @app.get("/api/admin/users")
+    @admin_required
+    def admin_list_users(user):
+        conn = get_db()
+        try:
+            rows = conn.execute(
+                """
+                SELECT id, username, display_name, is_admin, is_active, created_at
+                FROM users
+                ORDER BY id
+                """
+            ).fetchall()
+            return jsonify([dict(r) for r in rows])
+        finally:
+            conn.close()
+
+    @app.post("/api/admin/users")
+    @admin_required
+    def admin_create_user(user):
+        data = request.get_json(silent=True) or {}
+        username = (data.get("username") or "").strip()
+        password = data.get("password") or ""
+        display_name = (data.get("displayName") or username or "用户").strip()
+        if len(username) < 3 or len(username) > 50:
+            return jsonify({"error": "用户名长度需为 3-50 个字符"}), 400
+        if len(password) < 6:
+            return jsonify({"error": "密码至少 6 位"}), 400
+        conn = get_db()
+        try:
+            exists = conn.execute(
+                "SELECT id FROM users WHERE username = ?", (username,)
+            ).fetchone()
+        finally:
+            conn.close()
+        if exists:
+            return jsonify({"error": "用户名已存在"}), 409
+        created = create_user(
+            username,
+            password,
+            display_name,
+            is_admin=bool(data.get("isAdmin")),
+            is_active=bool(data.get("isActive", True)),
+        )
+        return jsonify(dict(created)), 201
+
+    @app.put("/api/admin/users/<int:user_id>/status")
+    @admin_required
+    def admin_update_status(user, user_id):
+        data = request.get_json(silent=True) or {}
+        if "isActive" not in data:
+            return jsonify({"error": "缺少 isActive 参数"}), 400
+        conn = get_db()
+        try:
+            target = conn.execute(
+                "SELECT id FROM users WHERE id = ?", (user_id,)
+            ).fetchone()
+            if target is None:
+                return jsonify({"error": "用户不存在"}), 404
+            conn.execute(
+                "UPDATE users SET is_active = ? WHERE id = ?",
+                (1 if data["isActive"] else 0, user_id),
+            )
+            conn.commit()
+            row = conn.execute(
+                """
+                SELECT id, username, display_name, is_admin, is_active, created_at
+                FROM users WHERE id = ?
+                """,
+                (user_id,),
+            ).fetchone()
+            return jsonify(dict(row))
+        finally:
+            conn.close()
+
+    @app.delete("/api/admin/users/<int:user_id>")
+    @admin_required
+    def admin_delete_user(user, user_id):
+        if user_id == user["id"]:
+            return jsonify({"error": "不能删除当前登录账号"}), 400
+        conn = get_db()
+        try:
+            target = conn.execute(
+                "SELECT id FROM users WHERE id = ?", (user_id,)
+            ).fetchone()
+            if target is None:
+                return jsonify({"error": "用户不存在"}), 404
+            conn.execute("DELETE FROM users WHERE id = ?", (user_id,))
+            conn.commit()
+            return jsonify({"ok": True})
+        finally:
+            conn.close()
+
+    @app.get("/")
+    def index():
+        return send_from_directory(BASE_DIR, "index.html")
+
+    @app.get("/<path:file_path>")
+    def static_files(file_path):
+        full_path = os.path.join(BASE_DIR, file_path)
+        if os.path.isfile(full_path):
+            return send_from_directory(BASE_DIR, file_path)
+        return jsonify({"error": "页面不存在"}), 404
+
+    return app
+
+
+app = create_app()
+
+
+if __name__ == "__main__":
+    app.run(host="0.0.0.0", port=5000, debug=False)
@@ -0,0 +1,84 @@
+from functools import wraps
+
+from flask import jsonify, session
+from werkzeug.security import check_password_hash, generate_password_hash
+
+from .db import get_db, now_iso
+
+
+def get_current_user():
+    user_id = session.get("user_id")
+    if not user_id:
+        return None
+    conn = get_db()
+    try:
+        row = conn.execute("SELECT * FROM users WHERE id = ?", (user_id,)).fetchone()
+        if row is None or not row["is_active"]:
+            session.clear()
+            return None
+        return dict(row)
+    finally:
+        conn.close()
+
+
+def login_required(view):
+    @wraps(view)
+    def wrapped(*args, **kwargs):
+        user = get_current_user()
+        if user is None:
+            return jsonify({"error": "请先登录"}), 401
+        return view(user=user, *args, **kwargs)
+
+    return wrapped
+
+
+def admin_required(view):
+    @wraps(view)
+    def wrapped(*args, **kwargs):
+        user = get_current_user()
+        if user is None:
+            return jsonify({"error": "请先登录"}), 401
+        if not user["is_admin"]:
+            return jsonify({"error": "无管理员权限"}), 403
+        return view(user=user, *args, **kwargs)
+
+    return wrapped
+
+
+def create_user(username, password, display_name, is_admin=False, is_active=True):
+    conn = get_db()
+    try:
+        conn.execute(
+            """
+            INSERT INTO users (username, password_hash, display_name, is_admin, is_active, created_at)
+            VALUES (?, ?, ?, ?, ?, ?)
+            """,
+            (
+                username,
+                generate_password_hash(password),
+                display_name,
+                1 if is_admin else 0,
+                1 if is_active else 0,
+                now_iso(),
+            ),
+        )
+        conn.commit()
+        return conn.execute(
+            "SELECT id, username, display_name, is_admin, is_active, created_at FROM users WHERE username = ?",
+            (username,),
+        ).fetchone()
+    finally:
+        conn.close()
+
+
+def authenticate(username, password):
+    conn = get_db()
+    try:
+        row = conn.execute("SELECT * FROM users WHERE username = ?", (username,)).fetchone()
+        if row is None or not check_password_hash(row["password_hash"], password):
+            return None
+        if not row["is_active"]:
+            return {"disabled": True}
+        return dict(row)
+    finally:
+        conn.close()
@@ -0,0 +1,93 @@
+import os
+import sqlite3
+from datetime import datetime, timezone
+
+from werkzeug.security import generate_password_hash
+
+BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+DATA_DIR = os.path.join(BASE_DIR, "server", "data")
+DB_PATH = os.path.join(DATA_DIR, "spanish_diary.db")
+
+SCHEMA = """
+CREATE TABLE IF NOT EXISTS users (
+  id INTEGER PRIMARY KEY AUTOINCREMENT,
+  username TEXT UNIQUE NOT NULL,
+  password_hash TEXT NOT NULL,
+  display_name TEXT NOT NULL,
+  is_admin INTEGER NOT NULL DEFAULT 0,
+  is_active INTEGER NOT NULL DEFAULT 1,
+  created_at TEXT NOT NULL
+);
+
+CREATE TABLE IF NOT EXISTS entries (
+  id INTEGER PRIMARY KEY AUTOINCREMENT,
+  user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE,
+  date TEXT NOT NULL,
+  spanish TEXT NOT NULL,
+  english TEXT NOT NULL DEFAULT '',
+  chinese TEXT NOT NULL DEFAULT '',
+  created_at TEXT NOT NULL,
+  updated_at TEXT NOT NULL
+);
+
+CREATE INDEX IF NOT EXISTS idx_entries_user_date ON entries(user_id, date);
+
+CREATE TABLE IF NOT EXISTS vocabulary (
+  id INTEGER PRIMARY KEY AUTOINCREMENT,
+  user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE,
+  word TEXT NOT NULL,
+  english TEXT NOT NULL DEFAULT '',
+  chinese TEXT NOT NULL DEFAULT '',
+  category TEXT NOT NULL DEFAULT '',
+  created_at TEXT NOT NULL,
+  UNIQUE(user_id, word)
+);
+
+CREATE TABLE IF NOT EXISTS settings (
+  user_id INTEGER PRIMARY KEY REFERENCES users(id) ON DELETE CASCADE,
+  translator TEXT NOT NULL DEFAULT 'local',
+  api_url TEXT NOT NULL DEFAULT '',
+  api_key TEXT NOT NULL DEFAULT '',
+  extra_config TEXT NOT NULL DEFAULT '{}',
+  updated_at TEXT NOT NULL
+);
+"""
+
+
+def now_iso():
+    return datetime.now(timezone.utc).isoformat()
+
+
+def get_db():
+    conn = sqlite3.connect(DB_PATH)
+    conn.row_factory = sqlite3.Row
+    conn.execute("PRAGMA foreign_keys = ON")
+    return conn
+
+
+def init_db():
+    os.makedirs(DATA_DIR, exist_ok=True)
+    conn = get_db()
+    try:
+        conn.executescript(SCHEMA)
+        conn.commit()
+
+        admin = conn.execute(
+            "SELECT id FROM users WHERE username = ?", ("admin",)
+        ).fetchone()
+        if admin is None:
+            conn.execute(
+                """
+                INSERT INTO users (username, password_hash, display_name, is_admin, is_active, created_at)
+                VALUES (?, ?, ?, 1, 1, ?)
+                """,
+                (
+                    "admin",
+                    generate_password_hash("admin123"),
+                    "管理员",
+                    now_iso(),
+                ),
+            )
+            conn.commit()
+    finally:
+        conn.close()
@@ -0,0 +1,173 @@
+from datetime import date, timedelta
+
+from flask import Blueprint, jsonify, request
+
+from .auth import login_required
+from .db import get_db, now_iso
+
+entries_bp = Blueprint("entries", __name__)
+
+
+def row_to_dict(row):
+    return dict(row)
+
+
+@entries_bp.get("/api/entries")
+@login_required
+def list_entries(user):
+    conn = get_db()
+    try:
+        rows = conn.execute(
+            """
+            SELECT id, date, spanish, english, chinese, created_at, updated_at
+            FROM entries
+            WHERE user_id = ?
+            ORDER BY date DESC, id DESC
+            """,
+            (user["id"],),
+        ).fetchall()
+        return jsonify([row_to_dict(r) for r in rows])
+    finally:
+        conn.close()
+
+
+@entries_bp.post("/api/entries")
+@login_required
+def create_entry(user):
+    data = request.get_json(silent=True) or {}
+    spanish = (data.get("spanish") or "").strip()
+    if not spanish:
+        return jsonify({"error": "日记内容不能为空"}), 400
+    entry_date = (data.get("date") or date.today().isoformat()).strip()
+    conn = get_db()
+    try:
+        cur = conn.execute(
+            """
+            INSERT INTO entries (user_id, date, spanish, english, chinese, created_at, updated_at)
+            VALUES (?, ?, ?, ?, ?, ?, ?)
+            """,
+            (
+                user["id"],
+                entry_date,
+                spanish,
+                (data.get("english") or "").strip(),
+                (data.get("chinese") or "").strip(),
+                now_iso(),
+                now_iso(),
+            ),
+        )
+        conn.commit()
+        row = conn.execute(
+            "SELECT id, date, spanish, english, chinese, created_at, updated_at FROM entries WHERE id = ?",
+            (cur.lastrowid,),
+        ).fetchone()
+        return jsonify(row_to_dict(row)), 201
+    finally:
+        conn.close()
+
+
+@entries_bp.put("/api/entries/<int:entry_id>")
+@login_required
+def update_entry(user, entry_id):
+    data = request.get_json(silent=True) or {}
+    spanish = (data.get("spanish") or "").strip()
+    if not spanish:
+        return jsonify({"error": "日记内容不能为空"}), 400
+    conn = get_db()
+    try:
+        row = conn.execute(
+            "SELECT id FROM entries WHERE id = ? AND user_id = ?",
+            (entry_id, user["id"]),
+        ).fetchone()
+        if row is None:
+            return jsonify({"error": "日记不存在或无权访问"}), 404
+        conn.execute(
+            """
+            UPDATE entries
+            SET date = ?, spanish = ?, english = ?, chinese = ?, updated_at = ?
+            WHERE id = ?
+            """,
+            (
+                (data.get("date") or row["date"]).strip(),
+                spanish,
+                (data.get("english") or "").strip(),
+                (data.get("chinese") or "").strip(),
+                now_iso(),
+                entry_id,
+            ),
+        )
+        conn.commit()
+        updated = conn.execute(
+            "SELECT id, date, spanish, english, chinese, created_at, updated_at FROM entries WHERE id = ?",
+            (entry_id,),
+        ).fetchone()
+        return jsonify(row_to_dict(updated))
+    finally:
+        conn.close()
+
+
+@entries_bp.delete("/api/entries/<int:entry_id>")
+@login_required
+def delete_entry(user, entry_id):
+    conn = get_db()
+    try:
+        row = conn.execute(
+            "SELECT id FROM entries WHERE id = ? AND user_id = ?",
+            (entry_id, user["id"]),
+        ).fetchone()
+        if row is None:
+            return jsonify({"error": "日记不存在或无权访问"}), 404
+        conn.execute("DELETE FROM entries WHERE id = ?", (entry_id,))
+        conn.commit()
+        return jsonify({"ok": True})
+    finally:
+        conn.close()
+
+
+@entries_bp.get("/api/entries/stats")
+@login_required
+def entry_stats(user):
+    conn = get_db()
+    try:
+        rows = conn.execute(
+            """
+            SELECT date, spanish FROM entries
+            WHERE user_id = ?
+            """,
+            (user["id"],),
+        ).fetchall()
+        total = len(rows)
+        total_chars = sum(len(r["spanish"]) for r in rows)
+        dates = [r["date"] for r in rows]
+
+        today = date.today()
+        month_start = today.replace(day=1)
+        week_start = today - timedelta(days=today.weekday())
+        month_count = sum(1 for d in dates if d >= month_start.isoformat())
+        week_count = sum(1 for d in dates if d >= week_start.isoformat())
+
+        streak = 0
+        cursor = today
+        if cursor.isoformat() not in dates:
+            cursor -= timedelta(days=1)
+        while cursor.isoformat() in dates:
+            streak += 1
+            cursor -= timedelta(days=1)
+
+        daily = []
+        for offset in range(6, -1, -1):
+            day = today - timedelta(days=offset)
+            daily.append({"date": day.isoformat(), "count": sum(1 for d in dates if d == day.isoformat())})
+
+        return jsonify(
+            {
+                "total": total,
+                "month": month_count,
+                "week": week_count,
+                "streak": streak,
+                "totalChars": total_chars,
+                "daily": daily,
+            }
+        )
+    finally:
+        conn.close()
@@ -0,0 +1 @@
+Flask==3.1.3
@@ -0,0 +1,94 @@
+import json
+
+from flask import Blueprint, jsonify, request
+
+from .auth import login_required
+from .db import get_db, now_iso
+from .translate_service import perform_translate
+
+settings_bp = Blueprint("settings", __name__)
+
+
+def get_user_settings(user_id):
+    conn = get_db()
+    try:
+        row = conn.execute(
+            "SELECT translator, api_url, api_key, extra_config, updated_at FROM settings WHERE user_id = ?",
+            (user_id,),
+        ).fetchone()
+        if row is None:
+            return {
+                "translator": "local",
+                "api_url": "",
+                "api_key": "",
+                "extra_config": {},
+                "updated_at": "",
+            }
+        result = dict(row)
+        try:
+            result["extra_config"] = json.loads(result["extra_config"] or "{}")
+        except Exception:
+            result["extra_config"] = {}
+        return result
+    finally:
+        conn.close()
+
+
+@settings_bp.get("/api/settings/translation")
+@login_required
+def get_translation_settings(user):
+    return jsonify(get_user_settings(user["id"]))
+
+
+@settings_bp.put("/api/settings/translation")
+@login_required
+def update_translation_settings(user):
+    data = request.get_json(silent=True) or {}
+    translator = (data.get("translator") or "local").strip().lower()
+    if translator not in ("local", "mymemory", "custom"):
+        return jsonify({"error": "不支持的翻译方式"}), 400
+    extra_config = data.get("extra_config") or {}
+    if not isinstance(extra_config, dict):
+        return jsonify({"error": "extra_config 必须是对象"}), 400
+    conn = get_db()
+    try:
+        conn.execute(
+            """
+            INSERT INTO settings (user_id, translator, api_url, api_key, extra_config, updated_at)
+            VALUES (?, ?, ?, ?, ?, ?)
+            ON CONFLICT(user_id) DO UPDATE SET
+              translator = excluded.translator,
+              api_url = excluded.api_url,
+              api_key = excluded.api_key,
+              extra_config = excluded.extra_config,
+              updated_at = excluded.updated_at
+            """,
+            (
+                user["id"],
+                translator,
+                (data.get("api_url") or "").strip(),
+                (data.get("api_key") or "").strip(),
+                json.dumps(extra_config, ensure_ascii=False),
+                now_iso(),
+            ),
+        )
+        conn.commit()
+        return jsonify(get_user_settings(user["id"]))
+    finally:
+        conn.close()
+
+
+@settings_bp.post("/api/translate")
+@login_required
+def translate_text(user):
+    data = request.get_json(silent=True) or {}
+    text = (data.get("text") or "").strip()
+    if not text:
+        return jsonify({"error": "翻译内容不能为空"}), 400
+    source = (data.get("source") or "zh").strip()
+    target = (data.get("target") or "es").strip()
+    settings = get_user_settings(user["id"])
+    result = perform_translate(text, settings, source, target)
+    if not result.get("ok"):
+        return jsonify({"error": result.get("error", "翻译失败")}), 502
+    return jsonify(result)
@@ -0,0 +1,83 @@
+import json
+import urllib.parse
+import urllib.request
+
+
+def translate_mymemory(text, source="zh-CN", target="es"):
+    url = (
+        "https://api.mymemory.translated.net/get?"
+        + urllib.parse.urlencode({"q": text, "langpair": source + "|" + target})
+    )
+    try:
+        with urllib.request.urlopen(url, timeout=15) as resp:
+            data = json.loads(resp.read().decode("utf-8"))
+        translated = data.get("responseData", {}).get("translatedText", "")
+        if not translated or data.get("quotaFinished"):
+            return {"ok": False, "error": "MyMemory 配额已用完或未返回结果"}
+        return {"ok": True, "translatedText": translated, "provider": "mymemory"}
+    except Exception as exc:
+        return {"ok": False, "error": str(exc)}
+
+
+def translate_custom(text, settings, source="zh", target="es"):
+    api_url = (settings.get("api_url") or "").strip()
+    if not api_url:
+        return {"ok": False, "error": "未配置自定义翻译接口地址"}
+
+    extra = {}
+    try:
+        extra = json.loads(settings.get("extra_config") or "{}")
+    except Exception:
+        extra = {}
+
+    payload = extra.get("payload", {})
+    payload["q"] = text
+    payload["source"] = source
+    payload["target"] = target
+
+    headers = {"Content-Type": "application/json"}
+    api_key = settings.get("api_key") or ""
+    header_key = extra.get("api_key_header", "Authorization")
+    if api_key:
+        headers[header_key] = api_key
+    headers.update(extra.get("headers", {}) or {})
+
+    request = urllib.request.Request(
+        api_url,
+        data=json.dumps(payload).encode("utf-8"),
+        headers=headers,
+        method="POST",
+    )
+    try:
+        with urllib.request.urlopen(request, timeout=20) as resp:
+            data = json.loads(resp.read().decode("utf-8"))
+        text_key = extra.get("result_key", "translatedText")
+        translated = data
+        for part in text_key.split("."):
+            if isinstance(translated, dict):
+                translated = translated.get(part, "")
+            else:
+                translated = ""
+        if not translated:
+            return {"ok": False, "error": "自定义接口未返回可识别的翻译字段"}
+        return {"ok": True, "translatedText": translated, "provider": "custom"}
+    except Exception as exc:
+        return {"ok": False, "error": str(exc)}
+
+
+def perform_translate(text, settings, source="zh", target="es"):
+    provider = (settings.get("translator") or "local").lower()
+    if provider == "mymemory":
+        lang_map = {
+            ("zh", "es"): ("zh-CN", "es"),
+            ("es", "zh"): ("es", "zh-CN"),
+            ("zh", "en"): ("zh-CN", "en"),
+            ("en", "zh"): ("en", "zh-CN"),
+            ("es", "en"): ("es", "en"),
+            ("en", "es"): ("en", "es"),
+        }
+        src, dst = lang_map.get((source, target), (source, target))
+        return translate_mymemory(text, src, dst)
+    if provider == "custom":
+        return translate_custom(text, settings, source, target)
+    return {"ok": False, "error": "本地词典翻译请在前端完成"}
@@ -0,0 +1,90 @@
+from flask import Blueprint, jsonify, request
+
+from .auth import login_required
+from .db import get_db, now_iso
+
+vocab_bp = Blueprint("vocabulary", __name__)
+
+
+def row_to_dict(row):
+    return dict(row)
+
+
+@vocab_bp.get("/api/vocabulary")
+@login_required
+def list_vocabulary(user):
+    conn = get_db()
+    try:
+        rows = conn.execute(
+            """
+            SELECT id, word, english, chinese, category, created_at
+            FROM vocabulary
+            WHERE user_id = ?
+            ORDER BY created_at DESC
+            """,
+            (user["id"],),
+        ).fetchall()
+        return jsonify([row_to_dict(r) for r in rows])
+    finally:
+        conn.close()
+
+
+@vocab_bp.post("/api/vocabulary")
+@login_required
+def add_vocabulary(user):
+    data = request.get_json(silent=True) or {}
+    word = (data.get("word") or "").strip()
+    if not word:
+        return jsonify({"error": "单词不能为空"}), 400
+    conn = get_db()
+    try:
+        existing = conn.execute(
+            "SELECT id FROM vocabulary WHERE user_id = ? AND word = ?",
+            (user["id"], word),
+        ).fetchone()
+        if existing:
+            row = conn.execute(
+                "SELECT id, word, english, chinese, category, created_at FROM vocabulary WHERE id = ?",
+                (existing["id"],),
+            ).fetchone()
+            return jsonify(row_to_dict(row))
+        cur = conn.execute(
+            """
+            INSERT INTO vocabulary (user_id, word, english, chinese, category, created_at)
+            VALUES (?, ?, ?, ?, ?, ?)
+            """,
+            (
+                user["id"],
+                word,
+                (data.get("english") or "").strip(),
+                (data.get("chinese") or "").strip(),
+                (data.get("category") or "").strip(),
+                now_iso(),
+            ),
+        )
+        conn.commit()
+        row = conn.execute(
+            "SELECT id, word, english, chinese, category, created_at FROM vocabulary WHERE id = ?",
+            (cur.lastrowid,),
+        ).fetchone()
+        return jsonify(row_to_dict(row)), 201
+    finally:
+        conn.close()
+
+
+@vocab_bp.delete("/api/vocabulary/<int:item_id>")
+@login_required
+def delete_vocabulary(user, item_id):
+    conn = get_db()
+    try:
+        row = conn.execute(
+            "SELECT id FROM vocabulary WHERE id = ? AND user_id = ?",
+            (item_id, user["id"]),
+        ).fetchone()
+        if row is None:
+            return jsonify({"error": "收藏不存在或无权访问"}), 404
+        conn.execute("DELETE FROM vocabulary WHERE id = ?", (item_id,))
+        conn.commit()
+        return jsonify({"ok": True})
+    finally:
+        conn.close()