feat: add wallet types, image upload, backup tabs, and manage.sh

Add mnemonic/key credentials with image upload, tabbed UI for add/query
and settings (auth/types/backup), daily backup to /root, and interactive
deploy script that updates via git pull only.

Co-authored-by: Cursor <cursoragent@cursor.com>

backup.py

@@ -0,0 +1,189 @@
+"""备份与恢复：打包 .env / data.json / settings.json / uploads"""
+import io
+import json
+import os
+import shutil
+import tarfile
+import threading
+import time
+from datetime import datetime, timedelta
+from pathlib import Path
+from typing import Optional, Tuple
+
+BASE_DIR = Path(__file__).resolve().parent
+UPLOAD_DIR = BASE_DIR / "uploads"
+BACKUP_META = BASE_DIR / "backup_meta.json"
+
+# Linux 生产环境写到 /root；其它环境回退到项目 backups/
+ROOT_BACKUP_DIR = Path("/root/crypto_key_backups")
+LOCAL_BACKUP_DIR = BASE_DIR / "backups"
+
+BACKUP_FILES = (".env", "data.json", "settings.json")
+
+_scheduler_started = False
+_scheduler_lock = threading.Lock()
+
+
+def backup_dir() -> Path:
+    if Path("/root").exists() and os.access("/root", os.W_OK):
+        try:
+            ROOT_BACKUP_DIR.mkdir(parents=True, exist_ok=True)
+            return ROOT_BACKUP_DIR
+        except OSError:
+            pass
+    LOCAL_BACKUP_DIR.mkdir(parents=True, exist_ok=True)
+    return LOCAL_BACKUP_DIR
+
+
+def _load_meta() -> dict:
+    if not BACKUP_META.exists():
+        return {}
+    try:
+        return json.loads(BACKUP_META.read_text(encoding="utf-8"))
+    except (json.JSONDecodeError, OSError):
+        return {}
+
+
+def _save_meta(data: dict):
+    BACKUP_META.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8")
+
+
+def create_backup_archive(dest: Optional[Path] = None) -> Path:
+    """创建 tar.gz 备份，返回文件路径。"""
+    ts = datetime.now().strftime("%Y%m%d_%H%M%S")
+    out_dir = dest.parent if dest else backup_dir()
+    out_dir.mkdir(parents=True, exist_ok=True)
+    out_path = dest or (out_dir / f"crypto_key_backup_{ts}.tar.gz")
+
+    with tarfile.open(out_path, "w:gz") as tar:
+        for name in BACKUP_FILES:
+            p = BASE_DIR / name
+            if p.exists():
+                tar.add(p, arcname=name)
+        if UPLOAD_DIR.exists():
+            for f in UPLOAD_DIR.rglob("*"):
+                if f.is_file():
+                    tar.add(f, arcname=str(Path("uploads") / f.relative_to(UPLOAD_DIR)))
+
+    meta = _load_meta()
+    meta["last_backup"] = datetime.now().isoformat(timespec="seconds")
+    meta["last_backup_path"] = str(out_path)
+    meta["last_backup_size"] = out_path.stat().st_size
+    _save_meta(meta)
+    return out_path
+
+
+def create_backup_bytes() -> Tuple[bytes, str]:
+    """生成内存中的备份包，供前端下载。"""
+    buf = io.BytesIO()
+    with tarfile.open(fileobj=buf, mode="w:gz") as tar:
+        for name in BACKUP_FILES:
+            p = BASE_DIR / name
+            if p.exists():
+                tar.add(p, arcname=name)
+        if UPLOAD_DIR.exists():
+            for f in UPLOAD_DIR.rglob("*"):
+                if f.is_file():
+                    tar.add(f, arcname=str(Path("uploads") / f.relative_to(UPLOAD_DIR)))
+    buf.seek(0)
+    filename = f"crypto_key_backup_{datetime.now().strftime('%Y%m%d_%H%M%S')}.tar.gz"
+    return buf.read(), filename
+
+
+def restore_from_archive(fileobj) -> None:
+    """从 tar.gz 恢复到项目目录。"""
+    # 先解压到临时目录再覆盖，避免半成品
+    tmp = BASE_DIR / f".restore_tmp_{int(time.time())}"
+    if tmp.exists():
+        shutil.rmtree(tmp)
+    tmp.mkdir(parents=True)
+
+    try:
+        with tarfile.open(fileobj=fileobj, mode="r:gz") as tar:
+            for member in tar.getmembers():
+                name = member.name.replace("\\", "/").lstrip("./")
+                if not name or name.startswith("/") or ".." in name.split("/"):
+                    raise ValueError(f"非法备份路径: {member.name}")
+                allowed = (
+                    name in BACKUP_FILES
+                    or name == "uploads"
+                    or name.startswith("uploads/")
+                )
+                if not allowed:
+                    continue
+                member.name = name
+                tar.extract(member, path=tmp)
+
+        for name in BACKUP_FILES:
+            src = tmp / name
+            if src.exists():
+                shutil.copy2(src, BASE_DIR / name)
+
+        uploads_src = tmp / "uploads"
+        if uploads_src.exists():
+            UPLOAD_DIR.mkdir(parents=True, exist_ok=True)
+            for f in uploads_src.rglob("*"):
+                if f.is_file():
+                    rel = f.relative_to(uploads_src)
+                    dest = UPLOAD_DIR / rel
+                    dest.parent.mkdir(parents=True, exist_ok=True)
+                    shutil.copy2(f, dest)
+
+        meta = _load_meta()
+        meta["last_restore"] = datetime.now().isoformat(timespec="seconds")
+        _save_meta(meta)
+    finally:
+        shutil.rmtree(tmp, ignore_errors=True)
+
+
+def get_backup_status() -> dict:
+    meta = _load_meta()
+    return {
+        "last_backup": meta.get("last_backup"),
+        "last_backup_path": meta.get("last_backup_path"),
+        "last_backup_size": meta.get("last_backup_size"),
+        "last_restore": meta.get("last_restore"),
+        "auto_backup_dir": str(backup_dir()),
+        "auto_backup_enabled": True,
+    }
+
+
+def _seconds_until_hour(hour: int = 3) -> float:
+    now = datetime.now()
+    target = now.replace(hour=hour, minute=0, second=0, microsecond=0)
+    if target <= now:
+        target += timedelta(days=1)
+    return (target - now).total_seconds()
+
+
+def _auto_backup_loop():
+    while True:
+        try:
+            time.sleep(max(60, _seconds_until_hour(3)))
+            create_backup_archive()
+            # 清理超过 30 天的自动备份
+            _cleanup_old_backups(keep_days=30)
+        except Exception:
+            time.sleep(3600)
+
+
+def _cleanup_old_backups(keep_days: int = 30):
+    cutoff = time.time() - keep_days * 86400
+    d = backup_dir()
+    for f in d.glob("crypto_key_backup_*.tar.gz"):
+        try:
+            if f.stat().st_mtime < cutoff:
+                f.unlink()
+        except OSError:
+            pass
+
+
+def start_auto_backup_scheduler():
+    """启动每日自动备份后台线程（仅启动一次）。"""
+    global _scheduler_started
+    with _scheduler_lock:
+        if _scheduler_started:
+            return
+        _scheduler_started = True
+        t = threading.Thread(target=_auto_backup_loop, name="crypto-key-backup", daemon=True)
+        t.start()