feat: strengthen backup and mobile workflows

backend/backup/__init__.py

@@ -0,0 +1,3 @@
+from .router import router
+
+__all__ = ["router"]

backend/backup/archive.py

@@ -0,0 +1,195 @@
+from __future__ import annotations
+
+import hashlib
+import json
+import shutil
+import tempfile
+import zipfile
+from dataclasses import dataclass
+from pathlib import Path, PurePosixPath
+from uuid import UUID
+
+from fastapi import HTTPException
+
+MAX_ARCHIVE_BYTES = 256 * 1024 * 1024
+MAX_ENTRIES = 10_000
+MAX_COMPRESSION_RATIO = 200
+MAX_METADATA_ENTRY_BYTES = 16 * 1024 * 1024
+_CHUNK_SIZE = 1024 * 1024
+
+
+@dataclass(frozen=True)
+class StagedBlob:
+    archive_path: str
+    staging_path: Path
+    size: int
+    sha256: str
+
+
+@dataclass(frozen=True)
+class ParsedArchive:
+    backup_id: UUID
+    archive_sha256: str
+    entities: dict[str, list[dict]]
+    files: dict[str, StagedBlob]
+    staging_dir: Path
+
+
+
+def backup_error(code: str, message: str, status_code: int = 422) -> HTTPException:
+    return HTTPException(status_code, {"code": code, "message": message})
+
+
+
+def canonical_json(value) -> bytes:
+    return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode()
+
+
+
+def build_archive_to_path(
+    output_path: Path, backup_id: UUID, entities: dict[str, list[dict]], files: dict[str, Path]
+) -> None:
+    entity_entries = {f"data/{name}.json": canonical_json(rows) for name, rows in entities.items()}
+    checksums = {name: hashlib.sha256(data).hexdigest() for name, data in entity_entries.items()}
+    for name, path in files.items():
+        digest = hashlib.sha256()
+        with path.open("rb") as source:
+            while chunk := source.read(_CHUNK_SIZE):
+                digest.update(chunk)
+        checksums[name] = digest.hexdigest()
+    manifest = {
+        "format": "dodo-backup", "version": 2, "backup_id": str(backup_id),
+        "entities": {name: len(rows) for name, rows in entities.items()}, "checksums": checksums,
+    }
+    with zipfile.ZipFile(output_path, "w", compression=zipfile.ZIP_DEFLATED, allowZip64=True) as archive:
+        archive.writestr("manifest.json", canonical_json(manifest))
+        for name, data in entity_entries.items():
+            archive.writestr(name, data)
+        for name, path in files.items():
+            archive.write(path, name)
+
+
+
+def build_archive(backup_id: UUID, entities: dict[str, list[dict]], files: dict[str, bytes]) -> bytes:
+    with tempfile.TemporaryDirectory() as directory:
+        root = Path(directory)
+        file_paths = {}
+        for index, (name, data) in enumerate(files.items()):
+            path = root / str(index)
+            path.write_bytes(data)
+            file_paths[name] = path
+        output = root / "backup.zip"
+        build_archive_to_path(output, backup_id, entities, file_paths)
+        return output.read_bytes()
+
+
+
+def _safe_name(name: str) -> bool:
+    path = PurePosixPath(name)
+    return bool(name) and not name.startswith("/") and "\\" not in name and ".." not in path.parts
+
+
+
+def _read_metadata(archive: zipfile.ZipFile, info: zipfile.ZipInfo) -> bytes:
+    if info.file_size > MAX_METADATA_ENTRY_BYTES:
+        raise backup_error("backup_size_invalid", "备份元数据过大")
+    with archive.open(info) as source:
+        data = source.read(MAX_METADATA_ENTRY_BYTES + 1)
+    if len(data) > MAX_METADATA_ENTRY_BYTES:
+        raise backup_error("backup_size_invalid", "备份元数据过大")
+    return data
+
+
+
+def _stream_blob(archive: zipfile.ZipFile, info: zipfile.ZipInfo, destination: Path) -> tuple[int, str]:
+    digest = hashlib.sha256()
+    size = 0
+    with archive.open(info) as source, destination.open("xb") as output:
+        while chunk := source.read(_CHUNK_SIZE):
+            size += len(chunk)
+            digest.update(chunk)
+            output.write(chunk)
+    return size, digest.hexdigest()
+
+
+
+def parse_archive_path(path: Path, *, max_archive_bytes: int = MAX_ARCHIVE_BYTES) -> ParsedArchive:
+    size = path.stat().st_size
+    if size <= 0 or size > max_archive_bytes:
+        raise backup_error("backup_size_invalid", "备份文件大小无效")
+    digest = hashlib.sha256()
+    with path.open("rb") as source:
+        while chunk := source.read(_CHUNK_SIZE):
+            digest.update(chunk)
+    staging_dir = Path(tempfile.mkdtemp(prefix="dodo-backup-blobs-"))
+    try:
+        with zipfile.ZipFile(path) as archive:
+            infos = archive.infolist()
+            names = [item.filename for item in infos]
+            if len(infos) > MAX_ENTRIES:
+                raise backup_error("backup_too_many_entries", "备份条目过多")
+            if len(names) != len(set(names)):
+                raise backup_error("backup_duplicate_entry", "备份包含重复条目")
+            if any(not _safe_name(item.filename) or item.is_dir() or item.flag_bits & 1 for item in infos):
+                raise backup_error("backup_unsafe_path", "备份包含不安全路径")
+            if "manifest.json" not in names:
+                raise backup_error("backup_manifest_missing", "备份缺少 manifest")
+            if sum(item.file_size for item in infos) > max_archive_bytes:
+                raise backup_error("backup_size_invalid", "备份解压后过大")
+            if any(item.file_size and item.compress_size == 0 for item in infos):
+                raise backup_error("backup_compression_invalid", "备份压缩比异常")
+            if any(item.compress_size and item.file_size / item.compress_size > MAX_COMPRESSION_RATIO for item in infos):
+                raise backup_error("backup_compression_invalid", "备份压缩比异常")
+            by_name = {item.filename: item for item in infos}
+            manifest = json.loads(_read_metadata(archive, by_name["manifest.json"]))
+            if manifest.get("format") != "dodo-backup" or manifest.get("version") != 2:
+                raise backup_error("backup_version_unsupported", "不支持的备份版本")
+            backup_id = UUID(manifest["backup_id"])
+            checksums = manifest["checksums"]
+            declared_entities = manifest["entities"]
+            if not isinstance(checksums, dict) or not isinstance(declared_entities, dict):
+                raise TypeError
+            content_names = set(names) - {"manifest.json"}
+            if set(checksums) != content_names:
+                raise backup_error("backup_manifest_mismatch", "manifest 与 ZIP 条目不一致")
+            entities: dict[str, list[dict]] = {}
+            files: dict[str, StagedBlob] = {}
+            for index, name in enumerate(sorted(content_names)):
+                info = by_name[name]
+                if name.startswith("data/") and name.endswith(".json"):
+                    data = _read_metadata(archive, info)
+                    actual_digest = hashlib.sha256(data).hexdigest()
+                    value = json.loads(data)
+                    if not isinstance(value, list) or any(not isinstance(row, dict) for row in value):
+                        raise ValueError
+                    entities[name[5:-5]] = value
+                else:
+                    blob_path = staging_dir / str(index)
+                    blob_size, actual_digest = _stream_blob(archive, info, blob_path)
+                    files[name] = StagedBlob(name, blob_path, blob_size, actual_digest)
+                if not isinstance(checksums[name], str) or actual_digest != checksums[name]:
+                    raise backup_error("backup_checksum_mismatch", "备份校验和不匹配")
+            if set(declared_entities) != set(entities):
+                raise backup_error("backup_manifest_mismatch", "manifest 实体清单不一致")
+            if any(type(count) is not int or count < 0 or count != len(entities[name]) for name, count in declared_entities.items()):
+                raise backup_error("backup_manifest_mismatch", "manifest 实体数量不一致")
+    except HTTPException:
+        shutil.rmtree(staging_dir, ignore_errors=True)
+        raise
+    except (zipfile.BadZipFile, OSError) as exc:
+        shutil.rmtree(staging_dir, ignore_errors=True)
+        raise backup_error("backup_invalid_zip", "无效的 ZIP 备份") from exc
+    except (KeyError, TypeError, ValueError, json.JSONDecodeError) as exc:
+        shutil.rmtree(staging_dir, ignore_errors=True)
+        raise backup_error("backup_manifest_invalid", "无效的备份 manifest 或数据") from exc
+    return ParsedArchive(backup_id, digest.hexdigest(), entities, files, staging_dir)
+
+
+
+def parse_archive(content: bytes) -> ParsedArchive:
+    if not content or len(content) > MAX_ARCHIVE_BYTES:
+        raise backup_error("backup_size_invalid", "备份文件大小无效")
+    with tempfile.NamedTemporaryFile() as staged:
+        staged.write(content)
+        staged.flush()
+        return parse_archive_path(Path(staged.name))

backend/backup/legacy.py

@@ -0,0 +1,5 @@
+"""Legacy JSON/CSV backup compatibility remains in backend.mvp.
+
+The v2 ZIP implementation is isolated in this package. This marker module documents
+that the old endpoints intentionally remain available during the gradual migration.
+"""

backend/backup/router.py

@@ -0,0 +1,274 @@
+from __future__ import annotations
+
+import hashlib
+import secrets
+import shutil
+from datetime import timedelta
+from pathlib import Path
+
+from fastapi import APIRouter, Depends, File, Query, UploadFile
+from fastapi.responses import FileResponse
+from sqlalchemy import func, select, update
+from sqlalchemy.ext.asyncio import AsyncSession
+from starlette.background import BackgroundTask
+
+from backend.auth import current_user, hash_token
+from backend.config import get_settings
+from backend.db import get_db
+from backend.models import BackupPreflight, Countdown, User, new_id, utcnow
+
+from .archive import MAX_ARCHIVE_BYTES as ARCHIVE_DEFAULT_LIMIT
+from .archive import backup_error, parse_archive_path
+from .schemas import RestoreRequest
+from .service import BackupRepairPending, export_v2, restore_v2, validate_archive
+
+router = APIRouter(prefix="/api/v1/backup", tags=["backup"])
+MAX_ARCHIVE_BYTES = ARCHIVE_DEFAULT_LIMIT
+MAX_PENDING_PREFLIGHTS_PER_USER = 3
+_READ_CHUNK = 1024 * 1024
+
+
+def _staging_root() -> Path:
+    root = Path(get_settings().backup_staging_dir).resolve()
+    root.mkdir(parents=True, exist_ok=True)
+    return root
+
+
+def _safe_staging_path(value: str) -> Path:
+    root = _staging_root()
+    path = Path(value).resolve()
+    if path.parent != root:
+        raise backup_error("backup_preflight_invalid", "预检暂存无效", 409)
+    return path
+
+
+def _path_size(path: str | None) -> int:
+    if not path:
+        return 0
+    root = Path(path)
+    if not root.exists():
+        return 0
+    if root.is_file():
+        return root.stat().st_size
+    return sum(item.stat().st_size for item in root.rglob("*") if item.is_file())
+
+
+async def _prune(db: AsyncSession) -> None:
+    now = utcnow()
+    rows = list((await db.scalars(select(BackupPreflight).where(
+        BackupPreflight.expires_at <= now,
+        BackupPreflight.status.in_((
+            "reserved", "pending", "failed", "consuming", "cleanup_pending",
+            "repair_pending",
+        )),
+    ).with_for_update())).all())
+    from .storage import remove_quarantine, restore_quarantine_dir
+
+    attachment_root = Path(get_settings().attachment_dir).resolve()
+    for row in rows:
+        try:
+            if row.status == "repair_pending":
+                if row.cleanup_path:
+                    restore_quarantine_dir(attachment_root, Path(row.cleanup_path))
+                    row.cleanup_path = None
+                _safe_staging_path(row.staging_path).unlink(missing_ok=True)
+                row.status = "failed"
+                continue
+            if row.cleanup_path:
+                remove_quarantine(Path(row.cleanup_path))
+                row.cleanup_path = None
+            _safe_staging_path(row.staging_path).unlink(missing_ok=True)
+        except (OSError, ValueError):
+            continue
+        if row.status == "cleanup_pending":
+            row.status = "consumed"
+            row.consumed_at = row.consumed_at or now
+        else:
+            await db.delete(row)
+    if rows:
+        await db.commit()
+
+
+@router.get("/export.zip")
+async def export_zip(user: User = Depends(current_user), db: AsyncSession = Depends(get_db)):
+    path = await export_v2(db, user)
+    await db.commit()
+    return FileResponse(
+        path,
+        media_type="application/zip",
+        filename="dodo-backup-v2.zip",
+        background=BackgroundTask(path.unlink, missing_ok=True),
+    )
+
+
+@router.post("/preflight")
+async def preflight(
+    mode: str = Query(pattern="^(merge|replace)$"),
+    file: UploadFile = File(...),
+    user: User = Depends(current_user),
+    db: AsyncSession = Depends(get_db),
+):
+    await _prune(db)
+    settings = get_settings()
+    configured_limit = settings.backup_max_archive_mb * 1024 * 1024
+    limit = min(MAX_ARCHIVE_BYTES, configured_limit)
+    max_pending = min(MAX_PENDING_PREFLIGHTS_PER_USER, settings.backup_max_pending_per_user)
+    max_staged = settings.backup_max_staged_mb_per_user * 1024 * 1024
+    token = secrets.token_urlsafe(32)
+    staging = _staging_root() / f"{new_id()}.zip"
+    reservation = BackupPreflight(
+        token_hash=hash_token(token), user_id=user.id, backup_id=new_id(),
+        archive_sha256="0" * 64, archive_size=limit, staging_path=str(staging), mode=mode,
+        status="reserved", expires_at=utcnow() + timedelta(seconds=settings.backup_preflight_ttl_seconds),
+    )
+    # A user-row write serializes quota decisions across workers on PostgreSQL;
+    # SQLite serializes writers at the database level.
+    await db.execute(update(User).where(User.id == user.id).values(username=User.username))
+    pending_count, pending_bytes = (await db.execute(select(
+        func.count(BackupPreflight.id), func.coalesce(func.sum(BackupPreflight.archive_size), 0)
+    ).where(
+        BackupPreflight.user_id == user.id,
+        BackupPreflight.status.in_((
+            "reserved", "pending", "failed", "consuming", "cleanup_pending",
+            "repair_pending",
+        )),
+    ))).one()
+    pending_rows = list((await db.scalars(select(BackupPreflight).where(
+        BackupPreflight.user_id == user.id,
+        BackupPreflight.status.in_((
+            "reserved", "pending", "failed", "consuming", "cleanup_pending",
+            "repair_pending",
+        )),
+    ))).all())
+    pending_bytes += sum(_path_size(item.cleanup_path) for item in pending_rows)
+    if pending_count >= max_pending or pending_bytes + limit > max_staged:
+        await db.rollback()
+        raise backup_error("backup_preflight_quota", "待处理预检配额已达上限", 429)
+    db.add(reservation)
+    await db.flush()
+    reservation_id = reservation.id
+    await db.commit()
+
+    size = 0
+    digest = hashlib.sha256()
+    archive = None
+    try:
+        with staging.open("xb") as output:
+            while chunk := await file.read(_READ_CHUNK):
+                size += len(chunk)
+                if size > limit:
+                    raise backup_error("backup_size_invalid", "备份文件大小无效")
+                digest.update(chunk)
+                output.write(chunk)
+        archive = parse_archive_path(staging, max_archive_bytes=limit)
+        validate_archive(archive)
+        if mode == "merge":
+            incoming_pin_ids = {
+                str(item["id"])
+                for item in archive.entities["countdowns"]
+                if item.get("pinned") is True and item.get("archived_at") is None
+            }
+            existing_pin_ids = set((await db.scalars(select(Countdown.id).where(
+                Countdown.user_id == user.id,
+                Countdown.pinned.is_(True),
+                Countdown.archived_at.is_(None),
+            ))).all())
+            if (
+                incoming_pin_ids
+                and existing_pin_ids
+                and incoming_pin_ids != {str(item) for item in existing_pin_ids}
+            ):
+                raise backup_error("backup_constraint_invalid", "合并恢复会产生多个置顶倒数日")
+        shutil.rmtree(archive.staging_dir, ignore_errors=True)
+        if archive.archive_sha256 != digest.hexdigest():
+            raise backup_error("backup_checksum_mismatch", "备份校验和不匹配")
+        reservation.backup_id = archive.backup_id
+        reservation.archive_sha256 = archive.archive_sha256
+        reservation.archive_size = size
+        reservation.status = "pending"
+        await db.commit()
+    except Exception:
+        await db.rollback()
+        if archive is not None:
+            shutil.rmtree(archive.staging_dir, ignore_errors=True)
+        failed = await db.get(BackupPreflight, reservation_id)
+        if failed is not None:
+            await db.delete(failed)
+            await db.commit()
+        staging.unlink(missing_ok=True)
+        raise
+    return {"valid": True, "preflight_token": token, "backup_id": archive.backup_id,
+            "archive_sha256": archive.archive_sha256,
+            "entities": {name: len(rows) for name, rows in archive.entities.items()}}
+
+
+@router.post("/restore")
+async def restore(
+    payload: RestoreRequest,
+    user: User = Depends(current_user),
+    db: AsyncSession = Depends(get_db),
+):
+    await _prune(db)
+    token_hash = hash_token(payload.preflight_token)
+    row = await db.scalar(select(BackupPreflight).where(
+        BackupPreflight.token_hash == token_hash, BackupPreflight.user_id == user.id
+    ).with_for_update())
+    if row is None or row.mode != payload.mode or row.expires_at <= utcnow():
+        raise backup_error("backup_preflight_invalid", "预检令牌无效或已过期", 409)
+    if row.status == "repair_pending" and row.cleanup_path:
+        from .storage import restore_quarantine_dir
+        root = Path(get_settings().attachment_dir).resolve()
+        try:
+            restore_quarantine_dir(root, Path(row.cleanup_path))
+        except OSError as exc:
+            raise backup_error("backup_repair_pending", "原附件复原尚未完成，请稍后重试", 503) from exc
+        row.status = "failed"
+        row.cleanup_path = None
+        await db.commit()
+        raise backup_error("backup_restore_retry", "附件已复原，请重新预检后重试", 409)
+    if row.status == "cleanup_pending":
+        from .storage import remove_quarantine
+        try:
+            if row.cleanup_path:
+                remove_quarantine(Path(row.cleanup_path))
+            _safe_staging_path(row.staging_path).unlink(missing_ok=True)
+        except OSError as exc:
+            raise backup_error("backup_cleanup_pending", "清理尚未完成，请稍后重试", 503) from exc
+        row.status = "consumed"
+        row.cleanup_path = None
+        await db.commit()
+        return {"restored": 0, "mode": payload.mode, "already_imported": False, "cleanup_retried": True}
+    claimed = await db.execute(update(BackupPreflight).where(
+        BackupPreflight.id == row.id, BackupPreflight.status.in_(("pending", "failed"))
+    ).values(status="consuming", consumed_at=utcnow()))
+    if claimed.rowcount != 1:
+        await db.rollback()
+        raise backup_error("backup_preflight_invalid", "预检令牌已使用", 409)
+    await db.commit()
+    staging = _safe_staging_path(row.staging_path)
+    archive = parse_archive_path(staging)
+    try:
+        if archive.archive_sha256 != row.archive_sha256 or archive.backup_id != row.backup_id:
+            row.status = "failed"
+            await db.commit()
+            raise backup_error("backup_preflight_invalid", "预检暂存已改变", 409)
+        result = await restore_v2(db, user, archive, payload.mode, operation=row)
+    except BackupRepairPending:
+        raise
+    except Exception:
+        await db.refresh(row)
+        if row.status not in {"cleanup_pending", "repair_pending"}:
+            row.status = "failed"
+            await db.commit()
+        raise
+    finally:
+        shutil.rmtree(archive.staging_dir, ignore_errors=True)
+    row.status = "cleanup_pending"
+    await db.commit()
+    try:
+        staging.unlink(missing_ok=True)
+    except OSError as exc:
+        raise backup_error("backup_cleanup_pending", "数据已恢复，但暂存清理未完成；请使用同一令牌重试", 500) from exc
+    row.status = "consumed"
+    await db.commit()
+    return result

backend/backup/schemas.py

@@ -0,0 +1,6 @@
+from pydantic import BaseModel, Field
+
+
+class RestoreRequest(BaseModel):
+    preflight_token: str = Field(min_length=32, max_length=128)
+    mode: str = Field(pattern="^(merge|replace)$")

backend/backup/service.py

@@ -0,0 +1,771 @@
+from __future__ import annotations
+
+import hashlib
+import math
+import tempfile
+from datetime import date, datetime
+from itertools import pairwise
+from pathlib import Path
+from uuid import UUID, uuid5
+
+from fastapi import HTTPException
+from pydantic import ValidationError
+from sqlalchemy import Boolean, Date, DateTime, Float, Integer, String, Text, Uuid, delete, select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from backend.config import get_settings
+from backend.models import (
+    Attachment,
+    BackupImport,
+    BackupImportEntity,
+    BackupPreflight,
+    Countdown,
+    Folder,
+    Habit,
+    HabitLog,
+    HabitPause,
+    Memo,
+    RecurrenceException,
+    RecurrenceTemplate,
+    Task,
+    TaskList,
+    User,
+    new_id,
+)
+from backend.mvp import (
+    CountdownInput,
+    HabitCreate,
+    MemoCreate,
+    PauseInput,
+    RecurrenceCreate,
+    is_occurrence,
+    parse_rrule,
+)
+from backend.schemas import FolderCreate, ListCreate, TaskCreate
+
+from .archive import ParsedArchive, backup_error, build_archive_to_path, canonical_json
+from .storage import contained_file, quarantine_files, remove_quarantine, restore_quarantine
+
+
+class BackupRepairPending(HTTPException):
+    def __init__(self) -> None:
+        super().__init__(500, {
+            "code": "backup_repair_pending",
+            "message": "恢复失败且原附件尚未完全复原；请使用同一令牌重试",
+        })
+
+ENTITY_MODELS = {
+    "folders": Folder,
+    "lists": TaskList,
+    "tasks": Task,
+    "recurrences": RecurrenceTemplate,
+    "recurrence_exceptions": RecurrenceException,
+    "habits": Habit,
+    "habit_logs": HabitLog,
+    "habit_pauses": HabitPause,
+    "countdowns": Countdown,
+    "memos": Memo,
+    "attachments": Attachment,
+}
+RELATIONS = {
+    "lists": {"folder_id": "folders"},
+    "tasks": {"list_id": "lists", "parent_id": "tasks"},
+    "recurrences": {"task_id": "tasks"},
+    "recurrence_exceptions": {"template_id": "recurrences"},
+    "habit_logs": {"habit_id": "habits"},
+    "habit_pauses": {"habit_id": "habits"},
+    "attachments": {"task_id": "tasks"},
+}
+ORDER = tuple(ENTITY_MODELS)
+
+
+def _value(value):
+    if isinstance(value, (UUID, date, datetime)):
+        return value.isoformat() if not isinstance(value, UUID) else str(value)
+    return value
+
+
+def _row(row, *, attachment_path: str | None = None) -> dict:
+    result = {
+        column.name: _value(getattr(row, column.name))
+        for column in row.__table__.columns
+        if column.name != "user_id" and column.name != "storage_name"
+    }
+    if attachment_path is not None:
+        result["archive_path"] = attachment_path
+    return result
+
+
+async def export_v2(db: AsyncSession, user: User) -> Path:
+    # Serialize exports with restores for this user. PostgreSQL keeps the row
+    # lock until this function's transaction is committed by the route.
+    await db.scalar(select(User).where(User.id == user.id).with_for_update())
+    entities: dict[str, list[dict]] = {}
+    files: dict[str, Path] = {}
+    root = Path(get_settings().attachment_dir).resolve()
+    total_size = 0
+    limit = get_settings().backup_max_archive_mb * 1024 * 1024
+    for name, model in ENTITY_MODELS.items():
+        if name in {"recurrence_exceptions", "habit_logs", "habit_pauses"}:
+            if name == "recurrence_exceptions":
+                ids = select(RecurrenceTemplate.id).where(RecurrenceTemplate.user_id == user.id)
+                rows = list((await db.scalars(select(model).where(model.template_id.in_(ids)))).all())
+            else:
+                ids = select(Habit.id).where(Habit.user_id == user.id)
+                rows = list((await db.scalars(select(model).where(model.habit_id.in_(ids)))).all())
+        else:
+            rows = list((await db.scalars(select(model).where(model.user_id == user.id))).all())
+        output = []
+        for item in rows:
+            if isinstance(item, Attachment):
+                archive_path = f"attachments/{item.id}/content"
+                try:
+                    path = contained_file(root, item.storage_name)
+                except ValueError as exc:
+                    raise backup_error("backup_attachment_path_invalid", "附件存储路径无效") from exc
+                if not path.is_file():
+                    raise backup_error("backup_attachment_missing", "附件文件缺失")
+                actual_size = path.stat().st_size
+                if actual_size != item.size:
+                    raise backup_error("backup_attachment_size_mismatch", "附件大小不匹配")
+                total_size += actual_size
+                if total_size > limit:
+                    raise backup_error("backup_size_invalid", "备份文件大小超出导入合同")
+                files[archive_path] = path
+                output.append(_row(item, attachment_path=archive_path))
+            else:
+                output.append(_row(item))
+        entities[name] = output
+    handle, output_name = tempfile.mkstemp(prefix="dodo-backup-", suffix=".zip")
+    import os
+    os.close(handle)
+    output_path = Path(output_name)
+    try:
+        build_archive_to_path(output_path, new_id(), entities, files)
+        if output_path.stat().st_size > limit:
+            raise backup_error("backup_size_invalid", "备份文件大小超出导入合同")
+        return output_path
+    except Exception:
+        output_path.unlink(missing_ok=True)
+        raise
+
+
+def _validate_scalar(entity: str, column, row: dict) -> None:
+    name = column.name
+    if name in {"user_id", "storage_name"}:
+        return
+    if name not in row:
+        if not column.nullable and column.default is None and not column.primary_key:
+            raise backup_error("backup_entity_invalid", f"{entity} 缺少必填字段")
+        return
+    value = row[name]
+    if value is None:
+        if not column.nullable and not column.primary_key:
+            raise backup_error("backup_entity_invalid", f"{entity}.{name} 不可为空")
+        return
+    try:
+        effective_type = getattr(column.type, "impl", column.type)
+        if isinstance(effective_type, Uuid):
+            UUID(str(value))
+        elif isinstance(effective_type, DateTime):
+            if not isinstance(value, str):
+                raise TypeError
+            datetime.fromisoformat(value)
+        elif isinstance(effective_type, Date):
+            if not isinstance(value, str):
+                raise TypeError
+            date.fromisoformat(value)
+        elif isinstance(effective_type, Boolean):
+            if type(value) is not bool:
+                raise TypeError
+        elif isinstance(effective_type, Integer):
+            if type(value) is not int:
+                raise TypeError
+        elif isinstance(effective_type, Float):
+            if isinstance(value, bool) or not isinstance(value, (int, float)) or not math.isfinite(value):
+                raise TypeError
+        elif isinstance(effective_type, (String, Text)):
+            if not isinstance(value, str):
+                raise TypeError
+            if isinstance(effective_type, String) and effective_type.length is not None and len(value) > effective_type.length:
+                raise ValueError
+    except (TypeError, ValueError, OverflowError) as exc:
+        raise backup_error("backup_entity_invalid", f"{entity}.{name} 类型无效") from exc
+
+
+def _parse_habit_days(value: object, *, minimum: int, maximum: int) -> list[int] | None:
+    if value is None:
+        return None
+    if not isinstance(value, str) or not value or any(part.strip() != part for part in value.split(",")):
+        raise ValueError
+    parts = value.split(",")
+    if any(not part.isascii() or not part.isdecimal() for part in parts):
+        raise ValueError
+    days = [int(part) for part in parts]
+    if value != ",".join(map(str, days)):
+        raise ValueError
+    if len(days) != len(set(days)) or any(day < minimum or day > maximum for day in days):
+        raise ValueError
+    return days
+
+
+def _validate_habit_graph(parsed: ParsedArchive) -> None:
+    habits: dict[str, HabitCreate] = {}
+    try:
+        for row in parsed.entities["habits"]:
+            weekdays = _parse_habit_days(row.get("weekdays"), minimum=0, maximum=6)
+            month_days = _parse_habit_days(row.get("month_days"), minimum=1, maximum=31)
+            payload = {
+                field: row.get(field)
+                for field in HabitCreate.model_fields
+                if field in row
+            }
+            payload["weekdays"] = weekdays
+            payload["month_days"] = month_days
+            habit = HabitCreate.model_validate(payload)
+            if habit.kind == "boolean" and (row.get("target") != 1 or row.get("max_value") != 1):
+                raise ValueError
+            if habit.weekdays != weekdays or habit.month_days != month_days:
+                raise ValueError
+            if habit.interval_days != row.get("interval_days"):
+                raise ValueError
+            habits[str(row["id"])] = habit
+
+        for row in parsed.entities["habit_logs"]:
+            habit = habits[str(row["habit_id"])]
+            value = row["value"]
+            if not math.isfinite(value) or value < 0:
+                raise ValueError
+            if habit.kind == "boolean":
+                if value not in (0, 1):
+                    raise ValueError
+            elif habit.max_value is not None and value > habit.max_value:
+                raise ValueError
+
+        pauses_by_habit: dict[str, list[tuple[date, date]]] = {}
+        for row in parsed.entities["habit_pauses"]:
+            pause = PauseInput.model_validate({
+                "start_date": row.get("start_date"),
+                "end_date": row.get("end_date"),
+            })
+            pauses_by_habit.setdefault(str(row["habit_id"]), []).append(
+                (pause.start_date, pause.end_date)
+            )
+        for pauses in pauses_by_habit.values():
+            pauses.sort()
+            if any(current[0] <= previous[1] for previous, current in pairwise(pauses)):
+                raise ValueError
+    except (KeyError, TypeError, ValueError, ValidationError) as exc:
+        raise backup_error("backup_habit_invalid", "习惯数据不符合业务规则") from exc
+
+
+def _required_trimmed(value: object, *, maximum: int | None = None) -> str:
+    if not isinstance(value, str) or value != value.strip() or not value:
+        raise ValueError
+    if maximum is not None and len(value) > maximum:
+        raise ValueError
+    return value
+
+
+def _ordered_datetimes(row: dict, earlier: str, later: str) -> None:
+    if row.get(earlier) and row.get(later) and datetime.fromisoformat(row[later]) < datetime.fromisoformat(row[earlier]):
+        raise ValueError
+
+
+def _validate_entity_contracts(parsed: ParsedArchive) -> None:
+    try:
+        for row in parsed.entities["folders"]:
+            FolderCreate.model_validate({"name": row.get("name")})
+            _required_trimmed(row.get("name"), maximum=120)
+            if row.get("position", 0) < 0:
+                raise ValueError
+            _ordered_datetimes(row, "created_at", "deleted_at")
+        for row in parsed.entities["lists"]:
+            ListCreate.model_validate({"name": row.get("name"), "folder_id": row.get("folder_id")})
+            _required_trimmed(row.get("name"), maximum=120)
+            if row.get("position", 0) < 0:
+                raise ValueError
+            _ordered_datetimes(row, "created_at", "deleted_at")
+        for row in parsed.entities["tasks"]:
+            task = TaskCreate.model_validate({
+                "title": row.get("title"),
+                "list_id": row.get("list_id"),
+                "description": row.get("description", ""),
+                "priority": row.get("priority", 0),
+                "due_at": row.get("due_at"),
+                "due_has_time": row.get("due_has_time", False),
+                "parent_id": row.get("parent_id"),
+            })
+            _required_trimmed(row.get("title"), maximum=500)
+            if row.get("version", 1) < 1 or row.get("position", 0) < 0:
+                raise ValueError
+            if task.due_at is None and task.due_has_time:
+                raise ValueError
+            completed_at = row.get("completed_at")
+            if row.get("completed") is True and completed_at is None:
+                raise ValueError
+            if row.get("completed") is False and completed_at is not None:
+                raise ValueError
+            _ordered_datetimes(row, "created_at", "updated_at")
+            _ordered_datetimes(row, "created_at", "deleted_at")
+        for row in parsed.entities["countdowns"]:
+            CountdownInput.model_validate({
+                field: row[field] for field in CountdownInput.model_fields if field in row
+            })
+            _required_trimmed(row.get("title"), maximum=200)
+            _required_trimmed(row.get("icon"), maximum=32)
+            _ordered_datetimes(row, "created_at", "updated_at")
+            _ordered_datetimes(row, "created_at", "archived_at")
+        for row in parsed.entities["memos"]:
+            MemoCreate.model_validate({"title": row.get("title"), "content": row.get("content", "")})
+            _required_trimmed(row.get("title"), maximum=200)
+            if row.get("version", 1) < 1:
+                raise ValueError
+            _ordered_datetimes(row, "created_at", "updated_at")
+            _ordered_datetimes(row, "created_at", "deleted_at")
+        allowed_mime = {
+            "text/plain", "text/csv", "application/pdf", "image/jpeg", "image/png",
+            "image/gif", "application/json", "application/zip",
+        }
+        for row in parsed.entities["attachments"]:
+            filename = row.get("filename")
+            mime_type = row.get("mime_type")
+            if (
+                not isinstance(filename, str)
+                or not filename.strip()
+                or Path(filename).name != filename
+                or mime_type not in allowed_mime
+                or row.get("size", -1) < 0
+            ):
+                raise ValueError
+    except (KeyError, TypeError, ValueError, ValidationError) as exc:
+        raise backup_error("backup_entity_invalid", "实体数据不符合业务规则") from exc
+
+
+def _validate_recurrence_graph(parsed: ParsedArchive) -> None:
+    tasks = {str(row["id"]): row for row in parsed.entities["tasks"]}
+    try:
+        for row in parsed.entities["recurrences"]:
+            task = tasks[str(row["task_id"])]
+            recurrence = RecurrenceCreate.model_validate({
+                "task_id": row["task_id"],
+                "rrule": row.get("rrule"),
+                "trigger_mode": row.get("trigger_mode", "scheduled"),
+                "after_completion_days": row.get("after_completion_days"),
+            })
+            starts_at = datetime.fromisoformat(row["starts_at"])
+            ends_at = datetime.fromisoformat(row["ends_at"]) if row.get("ends_at") else None
+            last_completed_at = (
+                datetime.fromisoformat(row["last_completed_at"])
+                if row.get("last_completed_at")
+                else None
+            )
+            if task.get("parent_id") is not None or task.get("due_at") is None:
+                raise ValueError
+            due_at = datetime.fromisoformat(task["due_at"])
+            if recurrence.trigger_mode == "scheduled":
+                if recurrence.rrule is None:
+                    raise ValueError
+                parse_rrule(recurrence.rrule)
+                if starts_at != due_at:
+                    raise ValueError
+            if ends_at is not None and ends_at < starts_at:
+                raise ValueError
+            if last_completed_at is not None and last_completed_at > starts_at:
+                raise ValueError
+
+        recurrences = {str(row["id"]): row for row in parsed.entities["recurrences"]}
+        for row in parsed.entities["recurrence_exceptions"]:
+            recurrence = recurrences[str(row["template_id"])]
+            if recurrence.get("trigger_mode", "scheduled") != "scheduled":
+                raise ValueError
+            occurrence_at = datetime.fromisoformat(row["occurrence_at"])
+            if not is_occurrence(
+                recurrence["rrule"], datetime.fromisoformat(recurrence["starts_at"]), occurrence_at
+            ):
+                raise ValueError
+            if recurrence.get("ends_at") and occurrence_at > datetime.fromisoformat(recurrence["ends_at"]):
+                raise ValueError
+    except (HTTPException, KeyError, TypeError, ValueError, ValidationError) as exc:
+        raise backup_error("backup_recurrence_invalid", "重复规则不符合业务规则") from exc
+
+
+def validate_archive(parsed: ParsedArchive) -> None:
+    unknown = set(parsed.entities) - set(ENTITY_MODELS)
+    missing = set(ENTITY_MODELS) - set(parsed.entities)
+    if unknown:
+        raise backup_error("backup_entity_unknown", "备份包含未知实体")
+    if missing:
+        raise backup_error("backup_entity_missing", "备份缺少必需实体")
+    ids: dict[str, set[str]] = {}
+    for entity in ORDER:
+        rows = parsed.entities.get(entity, [])
+        model = ENTITY_MODELS[entity]
+        allowed = {column.name for column in model.__table__.columns} | ({"archive_path"} if entity == "attachments" else set())
+        entity_ids = []
+        for row in rows:
+            if set(row) - allowed or "user_id" in row or "storage_name" in row:
+                raise backup_error("backup_entity_invalid", f"{entity} 包含未知或受保护字段")
+            for column in model.__table__.columns:
+                _validate_scalar(entity, column, row)
+            try:
+                entity_ids.append(str(UUID(str(row["id"]))))
+            except (KeyError, TypeError, ValueError) as exc:
+                raise backup_error("backup_entity_invalid", f"{entity} 包含无效 ID") from exc
+        if len(entity_ids) != len(set(entity_ids)):
+            raise backup_error("backup_duplicate_id", f"{entity} 包含重复 ID")
+        ids[entity] = set(entity_ids)
+    for entity, fields in RELATIONS.items():
+        for row in parsed.entities.get(entity, []):
+            for field, target in fields.items():
+                value = row.get(field)
+                if value is not None and str(value) not in ids[target]:
+                    raise backup_error("backup_reference_invalid", f"{entity}.{field} 引用不存在")
+    attachment_paths = set()
+    for row in parsed.entities.get("attachments", []):
+        path = row.get("archive_path")
+        if not isinstance(path, str) or path not in parsed.files or not path.startswith("attachments/"):
+            raise backup_error("backup_attachment_missing", "附件内容缺失")
+        if path in attachment_paths:
+            raise backup_error("backup_attachment_duplicate", "附件内容被重复引用")
+        attachment_paths.add(path)
+        blob = parsed.files[path]
+        if row.get("size") != blob.size:
+            raise backup_error("backup_attachment_size_mismatch", "附件大小不匹配")
+    if set(parsed.files) != attachment_paths:
+        raise backup_error("backup_unreferenced_file", "备份包含未引用文件")
+
+    def unique(entity: str, fields: tuple[str, ...], *, ignore_null: bool = False) -> None:
+        seen = set()
+        for row in parsed.entities[entity]:
+            key = tuple(row.get(field) for field in fields)
+            if ignore_null and any(value is None for value in key):
+                continue
+            if key in seen:
+                raise backup_error("backup_constraint_invalid", f"{entity} 唯一约束冲突")
+            seen.add(key)
+
+    unique("recurrences", ("task_id",))
+    unique("recurrence_exceptions", ("template_id", "occurrence_at"))
+    unique("habit_logs", ("habit_id", "day"))
+    unique("tasks", ("external_id",), ignore_null=True)
+    active_pinned = sum(
+        row.get("pinned") is True and row.get("archived_at") is None
+        for row in parsed.entities["countdowns"]
+    )
+    if active_pinned > 1:
+        raise backup_error("backup_constraint_invalid", "最多只能有一个置顶倒数日")
+    _validate_entity_contracts(parsed)
+    _validate_habit_graph(parsed)
+    _validate_recurrence_graph(parsed)
+    inboxes = [row for row in parsed.entities["lists"] if row.get("is_inbox") is True]
+    if len(inboxes) != 1:
+        raise backup_error("backup_constraint_invalid", "备份必须包含且仅包含一个收集箱")
+    tasks = {str(row["id"]): row for row in parsed.entities["tasks"]}
+    for task_id, row in tasks.items():
+        parent_id = row.get("parent_id")
+        if parent_id is None:
+            continue
+        parent_id = str(parent_id)
+        if parent_id == task_id:
+            raise backup_error("backup_constraint_invalid", "任务不能以自身为父任务")
+        parent = tasks[parent_id]
+        if parent.get("parent_id") is not None:
+            raise backup_error("backup_constraint_invalid", "任务父子关系仅支持一层")
+        if parent.get("list_id") != row.get("list_id"):
+            raise backup_error("backup_constraint_invalid", "父子任务必须属于同一清单")
+
+
+def _ordered_rows(entity: str, rows: list[dict]) -> list[dict]:
+    if entity != "tasks":
+        return rows
+    pending = {str(row["id"]): row for row in rows}
+    ordered: list[dict] = []
+    while pending:
+        ready = [row for row in pending.values() if row.get("parent_id") is None or str(row["parent_id"]) not in pending]
+        if not ready:
+            raise backup_error("backup_constraint_invalid", "任务父子关系存在环")
+        for row in ready:
+            ordered.append(row)
+            pending.pop(str(row["id"]))
+    return ordered
+
+
+def _normalized_values(values: dict) -> dict:
+    return {key: _value(value) for key, value in values.items() if key not in {"created_at", "updated_at"}}
+
+
+def _coerce(model, raw: dict, mapping: dict[str, dict[str, UUID]], user_id: UUID) -> dict:
+    values = {}
+    relations = RELATIONS.get(next(name for name, item in ENTITY_MODELS.items() if item is model), {})
+    for column in model.__table__.columns:
+        name = column.name
+        if name == "user_id":
+            values[name] = user_id
+        elif name == "storage_name":
+            continue
+        elif name in raw:
+            value = raw[name]
+            if name in relations and value is not None:
+                value = mapping[relations[name]][str(value)]
+            elif value is not None:
+                column_type = column.type
+                effective_type = getattr(column_type, "impl", column_type)
+                if isinstance(effective_type, Uuid):
+                    value = UUID(str(value))
+                elif isinstance(effective_type, DateTime) and isinstance(value, str):
+                    value = datetime.fromisoformat(value)
+                elif isinstance(effective_type, Date) and isinstance(value, str):
+                    value = date.fromisoformat(value)
+            values[name] = value
+    return values
+
+
+def _content_digest(entity: str, raw: dict) -> str:
+    # Source-form relations are deliberate: digest identity remains stable even
+    # when another user needs different target UUIDs.
+    payload = {key: value for key, value in raw.items() if key not in {"created_at", "updated_at", "storage_name"}}
+    return hashlib.sha256(canonical_json(payload)).hexdigest()
+
+
+async def _all_uuid_primary_keys(db: AsyncSession) -> set[UUID]:
+    occupied: set[UUID] = set()
+    for table in User.metadata.sorted_tables:
+        primary_keys = list(table.primary_key.columns)
+        if len(primary_keys) != 1:
+            continue
+        column = primary_keys[0]
+        effective_type = getattr(column.type, "impl", column.type)
+        if isinstance(effective_type, Uuid):
+            occupied.update((await db.scalars(select(column))).all())
+    return occupied
+
+
+async def _mapped_id(
+    db: AsyncSession, model, source: UUID, user_id: UUID, backup_id: UUID, entity: str,
+    reserved: set[UUID], incoming: set[UUID],
+) -> UUID:
+    ledger = await db.scalar(select(BackupImportEntity).where(
+        BackupImportEntity.user_id == user_id,
+        BackupImportEntity.backup_id == backup_id,
+        BackupImportEntity.entity_type == entity,
+        BackupImportEntity.source_id == source,
+    ))
+    if ledger is not None:
+        reserved.add(ledger.target_id)
+        return ledger.target_id
+    existing = await db.get(model, source)
+    owner_id = getattr(existing, "user_id", None)
+    if owner_id is None:
+        if isinstance(existing, RecurrenceException):
+            owner_id = await db.scalar(select(RecurrenceTemplate.user_id).where(RecurrenceTemplate.id == existing.template_id))
+        elif isinstance(existing, (HabitLog, HabitPause)):
+            owner_id = await db.scalar(select(Habit.user_id).where(Habit.id == existing.habit_id))
+    if existing is not None and owner_id == user_id:
+        reserved.add(source)
+        return source
+    if source not in reserved:
+        reserved.add(source)
+        return source
+    candidate = uuid5(user_id, str(source))
+    while candidate in reserved or candidate in incoming:
+        candidate = uuid5(user_id, str(candidate))
+    reserved.add(candidate)
+    return candidate
+
+
+async def restore_v2(
+    db: AsyncSession, user: User, parsed: ParsedArchive, mode: str,
+    *, operation: BackupPreflight | None = None,
+) -> dict:
+    validate_archive(parsed)
+    # The user row is the cross-worker restore/export mutex on PostgreSQL.
+    await db.scalar(select(User).where(User.id == user.id).with_for_update())
+    existing_import = await db.scalar(select(BackupImport).where(
+        BackupImport.user_id == user.id, BackupImport.backup_id == parsed.backup_id
+    ))
+    already_imported = existing_import is not None
+    if mode == "merge" and existing_import and existing_import.archive_sha256 != parsed.archive_sha256:
+        raise HTTPException(409, {"code": "backup_id_conflict", "message": "备份标识与内容不一致"})
+
+    root = Path(get_settings().attachment_dir).resolve()
+    root.mkdir(parents=True, exist_ok=True)
+    quarantine = root.parent / f".backup-quarantine-{new_id()}"
+    moved = []
+    written: list[Path] = []
+    try:
+        if mode == "replace":
+            old_names = list((await db.scalars(
+                select(Attachment.storage_name).where(Attachment.user_id == user.id)
+            )).all())
+            moved = quarantine_files(root, old_names, quarantine)
+            await db.execute(delete(RecurrenceException).where(
+                RecurrenceException.template_id.in_(
+                    select(RecurrenceTemplate.id).where(RecurrenceTemplate.user_id == user.id)
+                )
+            ))
+            await db.execute(delete(RecurrenceTemplate).where(RecurrenceTemplate.user_id == user.id))
+            await db.execute(delete(HabitLog).where(
+                HabitLog.habit_id.in_(select(Habit.id).where(Habit.user_id == user.id))
+            ))
+            await db.execute(delete(HabitPause).where(
+                HabitPause.habit_id.in_(select(Habit.id).where(Habit.user_id == user.id))
+            ))
+            for model in (Attachment, Memo, Countdown, Task, Habit, TaskList, Folder):
+                await db.execute(delete(model).where(model.user_id == user.id))
+            await db.execute(delete(BackupImportEntity).where(BackupImportEntity.user_id == user.id))
+            await db.execute(delete(BackupImport).where(BackupImport.user_id == user.id))
+
+        mapping: dict[str, dict[str, UUID]] = {name: {} for name in ORDER}
+        incoming_ids: set[UUID] = {
+            UUID(str(raw["id"]))
+            for rows in parsed.entities.values()
+            for raw in rows
+        }
+        reserved_ids = await _all_uuid_primary_keys(db)
+        for entity in ORDER:
+            model = ENTITY_MODELS[entity]
+            for raw in _ordered_rows(entity, parsed.entities[entity]):
+                source = UUID(str(raw["id"]))
+                mapping[entity][str(source)] = await _mapped_id(
+                    db, model, source, user.id, parsed.backup_id, entity,
+                    reserved_ids, incoming_ids - {source},
+                )
+
+        restored = 0
+        for entity in ORDER:
+            model = ENTITY_MODELS[entity]
+            for raw in _ordered_rows(entity, parsed.entities[entity]):
+                source = UUID(str(raw["id"]))
+                target = mapping[entity][str(source)]
+                digest_payload = dict(raw)
+                if model is Attachment:
+                    digest_payload["archive_sha256"] = parsed.files[raw["archive_path"]].sha256
+                content_digest = _content_digest(entity, digest_payload)
+                ledger = await db.scalar(select(BackupImportEntity).where(
+                    BackupImportEntity.user_id == user.id,
+                    BackupImportEntity.backup_id == parsed.backup_id,
+                    BackupImportEntity.entity_type == entity,
+                    BackupImportEntity.source_id == source,
+                ))
+                existing = await db.get(model, target)
+                values = _coerce(model, raw, mapping, user.id)
+                values["id"] = target
+                if ledger is not None:
+                    if ledger.target_id != target or ledger.content_digest != content_digest:
+                        raise HTTPException(409, {"code": "backup_entity_conflict", "message": f"{entity} 映射账本冲突"})
+                    if existing is None:
+                        raise HTTPException(409, {"code": "backup_entity_missing", "message": f"{entity} 映射目标不存在"})
+                    current = _row(existing, attachment_path=raw.get("archive_path") if model is Attachment else None)
+                    expected = {key: _value(value) for key, value in values.items() if key not in {"user_id", "storage_name"}}
+                    if model is Attachment:
+                        expected["archive_path"] = raw["archive_path"]
+                        stored = contained_file(root, existing.storage_name)
+                        file_digest = hashlib.sha256()
+                        actual_size = 0
+                        try:
+                            with stored.open("rb") as source_file:
+                                while chunk := source_file.read(1024 * 1024):
+                                    actual_size += len(chunk)
+                                    file_digest.update(chunk)
+                        except OSError as exc:
+                            raise HTTPException(409, {"code": "backup_entity_missing", "message": "attachments 映射目标文件不存在"}) from exc
+                        blob = parsed.files[raw["archive_path"]]
+                        if actual_size != blob.size or file_digest.hexdigest() != blob.sha256:
+                            raise HTTPException(409, {"code": "backup_entity_conflict", "message": "attachments 映射目标内容冲突"})
+                    if _normalized_values(current) != _normalized_values(expected):
+                        raise HTTPException(409, {"code": "backup_entity_conflict", "message": f"{entity} 映射目标内容冲突"})
+                    continue
+                if existing is not None and mode == "merge":
+                    current = _row(existing, attachment_path=raw.get("archive_path") if model is Attachment else None)
+                    expected = {key: _value(value) for key, value in values.items() if key not in {"user_id", "storage_name"}}
+                    if model is Attachment:
+                        expected["archive_path"] = raw["archive_path"]
+                        stored = contained_file(root, existing.storage_name)
+                        blob = parsed.files[raw["archive_path"]]
+                        file_digest = hashlib.sha256()
+                        actual_size = 0
+                        try:
+                            with stored.open("rb") as source_file:
+                                while chunk := source_file.read(1024 * 1024):
+                                    actual_size += len(chunk)
+                                    file_digest.update(chunk)
+                        except OSError as exc:
+                            raise HTTPException(409, {
+                                "code": "backup_entity_missing",
+                                "message": "attachments 已存在但文件不存在",
+                            }) from exc
+                        if actual_size != blob.size or file_digest.hexdigest() != blob.sha256:
+                            raise HTTPException(409, {
+                                "code": "backup_entity_conflict",
+                                "message": "attachments 已存在不同内容",
+                            })
+                    if _normalized_values(current) != _normalized_values(expected):
+                        raise HTTPException(409, {"code": "backup_entity_conflict", "message": f"{entity} 已存在不同内容"})
+                else:
+                    if model is Attachment:
+                        archive_path = raw["archive_path"]
+                        storage_name = str(new_id())
+                        destination = contained_file(root, storage_name)
+                        digest = hashlib.sha256()
+                        copied_size = 0
+                        blob = parsed.files[archive_path]
+                        with blob.staging_path.open("rb") as source_file, destination.open("xb") as output:
+                            while chunk := source_file.read(1024 * 1024):
+                                copied_size += len(chunk)
+                                digest.update(chunk)
+                                output.write(chunk)
+                        if copied_size != blob.size or digest.hexdigest() != blob.sha256:
+                            destination.unlink(missing_ok=True)
+                            raise backup_error("backup_checksum_mismatch", "附件暂存校验失败", 409)
+                        written.append(destination)
+                        values["storage_name"] = storage_name
+                    db.add(model(**values))
+                    await db.flush()
+                    restored += 1
+                db.add(BackupImportEntity(
+                    user_id=user.id, backup_id=parsed.backup_id, entity_type=entity,
+                    source_id=source, target_id=target, content_digest=content_digest,
+                ))
+                await db.flush()
+        if mode == "replace" or existing_import is None:
+            db.add(BackupImport(
+                user_id=user.id, backup_id=parsed.backup_id,
+                archive_sha256=parsed.archive_sha256, mode=mode,
+            ))
+        await db.commit()
+    except HTTPException:
+        await db.rollback()
+        for path in written:
+            path.unlink(missing_ok=True)
+        try:
+            restore_quarantine(moved)
+        except (OSError, ValueError):
+            if operation is not None:
+                operation.status = "repair_pending"
+                operation.cleanup_path = str(quarantine)
+                await db.commit()
+            raise BackupRepairPending()
+        raise
+    except Exception:
+        await db.rollback()
+        for path in written:
+            path.unlink(missing_ok=True)
+        try:
+            restore_quarantine(moved)
+        except (OSError, ValueError):
+            if operation is not None:
+                operation.status = "repair_pending"
+                operation.cleanup_path = str(quarantine)
+                await db.commit()
+            raise BackupRepairPending()
+        raise
+    try:
+        remove_quarantine(quarantine)
+    except OSError as exc:
+        if operation is not None:
+            operation.status = "cleanup_pending"
+            operation.cleanup_path = str(quarantine)
+            await db.commit()
+        raise HTTPException(500, {"code": "backup_cleanup_pending", "message": "数据已恢复，但旧附件清理未完成；请使用同一令牌重试"}) from exc
+    return {"restored": restored, "mode": mode, "already_imported": already_imported, "cleanup_retried": False}

backend/backup/storage.py

@@ -0,0 +1,52 @@
+import shutil
+from pathlib import Path
+
+
+def contained_file(root: Path, storage_name: str) -> Path:
+    root = root.resolve()
+    candidate = (root / storage_name).resolve()
+    if candidate.parent != root:
+        raise ValueError("attachment path escapes storage root")
+    return candidate
+
+
+def quarantine_files(root: Path, storage_names: list[str], quarantine: Path) -> list[tuple[Path, Path]]:
+    moved = []
+    quarantine.mkdir(parents=True, exist_ok=True)
+    try:
+        for name in storage_names:
+            source = contained_file(root, name)
+            if source.exists():
+                target = quarantine / name
+                target.parent.mkdir(parents=True, exist_ok=True)
+                source.replace(target)
+                moved.append((source, target))
+        return moved
+    except Exception:
+        restore_quarantine(moved)
+        raise
+
+
+def restore_quarantine(moved: list[tuple[Path, Path]]) -> None:
+    for original, quarantined in reversed(moved):
+        if quarantined.exists():
+            original.parent.mkdir(parents=True, exist_ok=True)
+            quarantined.replace(original)
+
+
+def restore_quarantine_dir(root: Path, quarantine: Path) -> None:
+    if not quarantine.exists():
+        return
+    for quarantined in sorted(quarantine.rglob("*")):
+        if not quarantined.is_file():
+            continue
+        relative = quarantined.relative_to(quarantine)
+        original = contained_file(root, relative.as_posix())
+        original.parent.mkdir(parents=True, exist_ok=True)
+        quarantined.replace(original)
+    shutil.rmtree(quarantine)
+
+
+def remove_quarantine(path: Path) -> None:
+    if path.exists():
+        shutil.rmtree(path)