88 lines
3.0 KiB
Python
88 lines
3.0 KiB
Python
"""Persistencia del estado de migración (migration-state.json).
|
|
|
|
El estado permite reanudar una migración interrumpida y evitar volver a
|
|
copiar mensajes ya migrados. Nunca se almacenan contraseñas aquí.
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
import tempfile
|
|
import threading
|
|
from datetime import datetime, timezone
|
|
|
|
STATE_VERSION = 1
|
|
|
|
|
|
class MigrationState:
|
|
def __init__(self, path: str, account: str):
|
|
self.path = path
|
|
self.account = account
|
|
self._lock = threading.Lock()
|
|
self._data = self._load()
|
|
|
|
def _load(self) -> dict:
|
|
if os.path.exists(self.path):
|
|
with open(self.path, "r", encoding="utf-8") as f:
|
|
data = json.load(f)
|
|
data.setdefault("folders", {})
|
|
return data
|
|
return {
|
|
"version": STATE_VERSION,
|
|
"account": self.account,
|
|
"created_at": datetime.now(timezone.utc).isoformat(),
|
|
"folders": {},
|
|
}
|
|
|
|
def save(self) -> None:
|
|
with self._lock:
|
|
self._data["updated_at"] = datetime.now(timezone.utc).isoformat()
|
|
target_dir = os.path.dirname(os.path.abspath(self.path)) or "."
|
|
os.makedirs(target_dir, exist_ok=True)
|
|
fd, tmp_path = tempfile.mkstemp(prefix=".migration-state-", dir=target_dir)
|
|
try:
|
|
with os.fdopen(fd, "w", encoding="utf-8") as f:
|
|
json.dump(self._data, f, indent=2, ensure_ascii=False)
|
|
os.replace(tmp_path, self.path)
|
|
finally:
|
|
if os.path.exists(tmp_path):
|
|
try:
|
|
os.remove(tmp_path)
|
|
except OSError:
|
|
pass
|
|
|
|
def _folder(self, folder: str) -> dict:
|
|
return self._data["folders"].setdefault(
|
|
folder,
|
|
{"fingerprints": {}, "last_scanned_a_uid": 0, "a_uidvalidity": None},
|
|
)
|
|
|
|
def get_uidvalidity(self, folder: str):
|
|
return self._folder(folder).get("a_uidvalidity")
|
|
|
|
def set_uidvalidity(self, folder: str, value) -> None:
|
|
f = self._folder(folder)
|
|
if f.get("a_uidvalidity") != value:
|
|
f["a_uidvalidity"] = value
|
|
f["last_scanned_a_uid"] = 0
|
|
|
|
def get_last_scanned_uid(self, folder: str) -> int:
|
|
return self._folder(folder).get("last_scanned_a_uid", 0)
|
|
|
|
def set_last_scanned_uid(self, folder: str, uid: int) -> None:
|
|
f = self._folder(folder)
|
|
f["last_scanned_a_uid"] = max(f.get("last_scanned_a_uid", 0), uid)
|
|
|
|
def has_fingerprint(self, folder: str, fingerprint: str) -> bool:
|
|
return fingerprint in self._folder(folder)["fingerprints"]
|
|
|
|
def mark_migrated(self, folder: str, fingerprint: str, a_uid: int, b_uid, size: int) -> None:
|
|
self._folder(folder)["fingerprints"][fingerprint] = {
|
|
"a_uid": a_uid,
|
|
"b_uid": b_uid,
|
|
"size": size,
|
|
"migrated_at": datetime.now(timezone.utc).isoformat(),
|
|
}
|
|
|
|
def folders_seen(self) -> list[str]:
|
|
return list(self._data["folders"].keys())
|