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 []