243 lines
8.3 KiB
Python
243 lines
8.3 KiB
Python
import os
|
|
import sys
|
|
|
|
from flask import Flask, jsonify, request, send_from_directory, session
|
|
|
|
ROOT_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
|
if ROOT_DIR not in sys.path:
|
|
sys.path.insert(0, ROOT_DIR)
|
|
|
|
from server.auth import admin_required, authenticate, create_user, login_required, set_password
|
|
from server.db import BASE_DIR, get_db, init_db
|
|
from server.entries_api import entries_bp
|
|
from server.settings_api import settings_bp
|
|
from server.tts_api import tts_bp
|
|
from server.typing_api import typing_bp
|
|
from server.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.register_blueprint(tts_bp)
|
|
app.register_blueprint(typing_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.post("/api/change-password")
|
|
@login_required
|
|
def change_password(user):
|
|
data = request.get_json(silent=True) or {}
|
|
old_password = data.get("oldPassword") or ""
|
|
new_password = data.get("newPassword") or ""
|
|
if len(new_password) < 6:
|
|
return jsonify({"error": "新密码至少 6 位"}), 400
|
|
if authenticate(user["username"], old_password) is None:
|
|
return jsonify({"error": "当前密码不正确"}), 400
|
|
set_password(user["id"], new_password)
|
|
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.put("/api/admin/users/<int:user_id>/password")
|
|
@admin_required
|
|
def admin_reset_password(user, user_id):
|
|
data = request.get_json(silent=True) or {}
|
|
new_password = data.get("password") or data.get("newPassword") or ""
|
|
if len(new_password) < 6:
|
|
return jsonify({"error": "新密码至少 6 位"}), 400
|
|
conn = get_db()
|
|
try:
|
|
target = conn.execute(
|
|
"SELECT id FROM users WHERE id = ?", (user_id,)
|
|
).fetchone()
|
|
finally:
|
|
conn.close()
|
|
if target is None:
|
|
return jsonify({"error": "用户不存在"}), 404
|
|
set_password(user_id, new_password)
|
|
return jsonify({"ok": True})
|
|
|
|
@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)
|