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, CalendarSubscription, 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, "calendar_subscriptions": CalendarSubscription, "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"), "after_completion_unit": row.get("after_completion_unit") if row.get("trigger_mode", "scheduled") == "after_completion" else None, }) 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) # Calendar subscriptions were introduced after backup v2. Treat their # absence as an empty collection so archives from older Dodo releases remain restorable. parsed.entities.setdefault("calendar_subscriptions", []) 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 if model is RecurrenceTemplate and "after_completion_unit" not in raw: values["after_completion_unit"] = "days" if raw.get("trigger_mode") == "after_completion" else None 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, CalendarSubscription, 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}