Allow changing username and password in settings; admin can delete duplicate accounts.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
dekun
2026-10-17 19:05:41 +08:00
parent 972d5a1e8c
commit d25a4c6862
6 changed files with 137 additions and 20 deletions
+62 -5
View File
@@ -387,20 +387,77 @@ def verify_login(username: str, password: str) -> Optional[dict]:
def change_user_password(user_id: int, old_password: str, new_password: str) -> None:
change_user_credentials(user_id, old_password, new_username=None, new_password=new_password)
def change_user_credentials(
user_id: int,
old_password: str,
new_username: Optional[str] = None,
new_password: Optional[str] = None,
) -> dict:
"""校验当前密码后,可改用户名和/或密码。返回更新后的用户信息。"""
from werkzeug.security import check_password_hash, generate_password_hash
if len(new_password) < 6:
new_username = (new_username or "").strip() or None
new_password = new_password if new_password else None
if new_password is not None and len(new_password) < 6:
raise ValueError("新密码至少 6 位")
if new_username is not None and len(new_username) < 2:
raise ValueError("用户名至少 2 位")
with get_db() as conn:
user = get_user(conn, user_id)
if not user:
raise ValueError("用户不存在")
if not check_password_hash(user["password_hash"], old_password):
raise ValueError("当前密码不正确")
conn.execute(
"UPDATE users SET password_hash=? WHERE id=?",
(generate_password_hash(new_password), user_id),
)
username_changed = bool(new_username and new_username != user["username"])
password_changed = new_password is not None
if not username_changed and not password_changed:
raise ValueError("请修改用户名或填写新密码")
if username_changed:
taken = conn.execute(
"SELECT id FROM users WHERE username=? AND id!=?",
(new_username, user_id),
).fetchone()
if taken:
raise ValueError("用户名已被占用")
conn.execute("UPDATE users SET username=? WHERE id=?", (new_username, user_id))
if password_changed:
conn.execute(
"UPDATE users SET password_hash=? WHERE id=?",
(generate_password_hash(new_password), user_id),
)
# 同步旧 settings 键(兼容)
updated = get_user(conn, user_id)
if updated and updated.get("role") == "admin":
set_setting(conn, "admin_user", updated["username"])
if password_changed:
set_setting(conn, "admin_password_hash", updated["password_hash"])
return updated or user
def delete_account(conn: sqlite3.Connection, account_id: int) -> None:
"""删除交易账户;若仍被用户绑定则拒绝。"""
bound = conn.execute(
"SELECT id, username FROM users WHERE account_id=?", (account_id,)
).fetchone()
if bound:
raise ValueError(f"账户已绑定用户 {bound['username']},请先调整用户绑定")
pending = conn.execute(
"SELECT COUNT(*) AS c FROM activations WHERE account_id=? AND status='pending'",
(account_id,),
).fetchone()["c"]
if pending:
raise ValueError("该账户仍有未使用的激活码,请先作废或用完后再删")
if not conn.execute("SELECT id FROM accounts WHERE id=?", (account_id,)).fetchone():
raise ValueError("账户不存在")
conn.execute("DELETE FROM accounts WHERE id=?", (account_id,))
def create_user(