nomadweb/backend/app/services/auth.py

"""Auth + favorites + synced user plans. File-backed so demo restarts keep data."""
from __future__ import annotations

import hashlib
import json
import secrets
import time
from pathlib import Path
from typing import Any

STORE_PATH = Path(__file__).resolve().parents[1] / "data" / "user_store.json"

_users: dict[str, dict] = {}
_sessions: dict[str, str] = {}  # token -> user_id
_favorites: dict[str, list[str]] = {}  # user_id -> [slugs]
_plans: dict[str, dict] = {}  # user_id -> { items, meta, updated_at }


def _hash_password(password: str) -> str:
  return hashlib.sha256(password.encode()).hexdigest()


def _persist() -> None:
  STORE_PATH.parent.mkdir(parents=True, exist_ok=True)
  payload = {
    "users": _users,
    "sessions": _sessions,
    "favorites": _favorites,
    "plans": _plans,
  }
  STORE_PATH.write_text(json.dumps(payload, ensure_ascii=False), encoding="utf-8")


def _restore() -> bool:
  if not STORE_PATH.exists():
    return False
  try:
    data = json.loads(STORE_PATH.read_text(encoding="utf-8"))
  except (OSError, json.JSONDecodeError):
    return False
  global _users, _sessions, _favorites, _plans
  _users = data.get("users") or {}
  _sessions = data.get("sessions") or {}
  _favorites = data.get("favorites") or {}
  _plans = data.get("plans") or {}
  return True


def _user_profile(user: dict) -> dict:
  return {
    "id": user["id"],
    "email": user["email"],
    "name": user["name"],
    "avatar": user.get("avatar", "🧑‍💻"),
  }


def _empty_plan() -> dict[str, Any]:
  return {
    "items": [],
    "meta": {
      "title": "我的旅居计划",
      "startMonth": "",
      "monthlyBudget": 0,
      "checklist": {},
    },
    "updated_at": 0,
  }


def register_user(email: str, password: str, name: str) -> dict[str, Any] | None:
  if email in _users:
    return None
  uid = secrets.token_hex(8)
  _users[email] = {
    "id": uid,
    "email": email,
    "password_hash": _hash_password(password),
    "name": name,
    "avatar": "🧑‍💻",
  }
  _favorites[uid] = []
  _plans[uid] = _empty_plan()
  token = secrets.token_urlsafe(32)
  _sessions[token] = uid
  _persist()
  return {"token": token, "user": _user_profile(_users[email])}


def login_user(email: str, password: str) -> dict[str, Any] | None:
  user = _users.get(email)
  if not user or user["password_hash"] != _hash_password(password):
    return None
  token = secrets.token_urlsafe(32)
  _sessions[token] = user["id"]
  _persist()
  return {"token": token, "user": _user_profile(user)}


def login_google(email: str, name: str) -> dict[str, Any] | None:
  if email not in _users:
    uid = secrets.token_hex(8)
    _users[email] = {
      "id": uid,
      "email": email,
      "password_hash": _hash_password(secrets.token_hex(16)),
      "name": name or email.split("@")[0],
      "avatar": "🌐",
      "oauth": "google",
    }
    _favorites[uid] = []
    _plans[uid] = _empty_plan()
  user = _users[email]
  token = secrets.token_urlsafe(32)
  _sessions[token] = user["id"]
  _persist()
  return {"token": token, "user": _user_profile(user)}


def login_demo() -> dict[str, Any] | None:
  email = "demo@nomadro.com"
  if email not in _users:
    created = register_user(email, "demo123", "演示用户")
    return created
  user = _users[email]
  token = secrets.token_urlsafe(32)
  _sessions[token] = user["id"]
  _persist()
  return {"token": token, "user": _user_profile(user)}


def _reload_sessions() -> None:
  if not STORE_PATH.exists():
    return
  try:
    data = json.loads(STORE_PATH.read_text(encoding="utf-8"))
    global _sessions
    _sessions.update(data.get("sessions") or {})
  except (OSError, json.JSONDecodeError):
    return


def get_user_by_token(token: str) -> dict | None:
  uid = _sessions.get(token)
  if not uid:
    _reload_sessions()
    uid = _sessions.get(token)
  if not uid:
    return None
  for user in _users.values():
    if user["id"] == uid:
      return _user_profile(user)
  return None


def get_favorites(token: str) -> list[str]:
  uid = _sessions.get(token)
  if not uid:
    return []
  return list(_favorites.get(uid, []))


def toggle_favorite(token: str, slug: str) -> list[str]:
  uid = _sessions.get(token)
  if not uid:
    return []
  favs = _favorites.setdefault(uid, [])
  if slug in favs:
    favs.remove(slug)
  else:
    favs.append(slug)
  _persist()
  return list(favs)


def get_plan(token: str) -> dict[str, Any] | None:
  uid = _sessions.get(token)
  if not uid:
    return None
  plan = _plans.get(uid) or _empty_plan()
  return {
    "items": plan.get("items") or [],
    "meta": plan.get("meta") or _empty_plan()["meta"],
    "updated_at": int(plan.get("updated_at") or 0),
  }


def save_plan(token: str, items: list, meta: dict, updated_at: int | None = None) -> dict[str, Any] | None:
  uid = _sessions.get(token)
  if not uid:
    return None
  ts = int(updated_at or time.time() * 1000)
  _plans[uid] = {
    "items": items,
    "meta": meta,
    "updated_at": ts,
  }
  _persist()
  return get_plan(token)


# Boot: restore disk store or seed demo
if not _restore():
  login_demo()
elif "demo@nomadro.com" not in _users:
  login_demo()