121 lines
3.3 KiB
Python
121 lines
3.3 KiB
Python
import json
|
|
import os
|
|
import time
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
|
|
DB_PATH = Path(__file__).parent / "database.json"
|
|
SETTINGS_PATH = Path(__file__).parent / "settings.json"
|
|
|
|
def get_settings():
|
|
if SETTINGS_PATH.exists():
|
|
try:
|
|
with open(SETTINGS_PATH, "r", encoding="utf-8") as f:
|
|
return json.load(f)
|
|
except Exception:
|
|
pass
|
|
return {"smtp_enabled": False, "smtp_server": "", "smtp_port": 465, "smtp_user": "", "smtp_pass": ""}
|
|
|
|
def save_settings(settings: dict):
|
|
with open(SETTINGS_PATH, "w", encoding="utf-8") as f:
|
|
json.dump(settings, f, indent=2)
|
|
|
|
def load_db():
|
|
if DB_PATH.exists():
|
|
try:
|
|
with open(DB_PATH, "r", encoding="utf-8") as f:
|
|
return json.load(f)
|
|
except Exception:
|
|
return []
|
|
return []
|
|
|
|
def save_db(users):
|
|
with open(DB_PATH, "w", encoding="utf-8") as f:
|
|
json.dump(users, f, indent=2)
|
|
|
|
def get_user_by_username(username: str):
|
|
for u in load_db():
|
|
if u.get("username") == username:
|
|
return u
|
|
return None
|
|
|
|
def update_user_login(user_id: int):
|
|
users = load_db()
|
|
for u in users:
|
|
if u.get("id") == user_id:
|
|
u["last_login"] = datetime.now().isoformat()
|
|
save_db(users)
|
|
return u
|
|
return None
|
|
|
|
def get_all_users():
|
|
users = load_db()
|
|
# 移除敏感的历史记录数据以便返回概览
|
|
result = []
|
|
for u in users:
|
|
u_copy = {k: v for k, v in u.items() if k != "history"}
|
|
result.append(u_copy)
|
|
return result
|
|
|
|
def get_user_by_id(user_id: int):
|
|
for u in load_db():
|
|
if u.get("id") == user_id:
|
|
return u
|
|
return None
|
|
|
|
def create_user(user_data: dict, ip_address: str = "Unknown"):
|
|
users = load_db()
|
|
new_user = {
|
|
**user_data,
|
|
"id": int(time.time() * 1000),
|
|
"createdAt": datetime.now().isoformat(),
|
|
"register_ip": ip_address,
|
|
"last_login": datetime.now().isoformat(),
|
|
"history": []
|
|
}
|
|
users.append(new_user)
|
|
save_db(users)
|
|
return new_user
|
|
|
|
def update_user(user_id: int, updates: dict):
|
|
users = load_db()
|
|
for u in users:
|
|
if u.get("id") == user_id:
|
|
u.update(updates)
|
|
save_db(users)
|
|
return u
|
|
return None
|
|
|
|
def add_history_record(user_id: int, markdown_text: str):
|
|
users = load_db()
|
|
for u in users:
|
|
if u.get("id") == user_id:
|
|
if "history" not in u:
|
|
u["history"] = []
|
|
|
|
# 取前50个字符作为摘要
|
|
summary = markdown_text[:50].replace('\n', ' ') + ('...' if len(markdown_text) > 50 else '')
|
|
|
|
record = {
|
|
"id": int(time.time() * 1000),
|
|
"timestamp": datetime.now().isoformat(),
|
|
"summary": summary if summary.strip() else "空文档",
|
|
"markdown": markdown_text
|
|
}
|
|
# 插入到最前面
|
|
u["history"].insert(0, record)
|
|
|
|
# 最多保存 100 条
|
|
if len(u["history"]) > 100:
|
|
u["history"] = u["history"][:100]
|
|
|
|
save_db(users)
|
|
return record
|
|
return None
|
|
|
|
def get_history_records(user_id: int):
|
|
user = get_user_by_id(user_id)
|
|
if user:
|
|
return user.get("history", [])
|
|
return []
|