File size: 2,324 Bytes
a3c34a3 a0672e2 a3c34a3 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 | import os
import zipfile
import tempfile
import shutil
from pathlib import Path
from errors import get_logger
log = get_logger("mega_sync")
DB_PATH = "./database"
MEGA_FILENAME = "generai_db.zip"
def _zip_db(out_path: str) -> bool:
db = Path(DB_PATH)
if not db.exists():
return False
with zipfile.ZipFile(out_path, 'w', zipfile.ZIP_DEFLATED) as zf:
for f in db.rglob('*'):
if f.is_file():
zf.write(f, f.relative_to(db.parent))
return True
def _unzip_db(zip_path: str):
db = Path(DB_PATH)
if db.exists():
shutil.rmtree(db)
with zipfile.ZipFile(zip_path, 'r') as zf:
zf.extractall('.')
def download_db(email: str, password: str) -> bool:
try:
from mega import Mega
m = Mega().login(email, password)
node = m.find(MEGA_FILENAME)
if not node:
log.info("Nessun DB su MEGA — si parte da zero.")
return False
tmp_dir = tempfile.mkdtemp()
try:
m.download(node, tmp_dir)
# cerca il file scaricato nella dir temporanea
files = list(Path(tmp_dir).iterdir())
if not files:
log.warning("Download MEGA: nessun file ricevuto.")
return False
_unzip_db(str(files[0]))
log.info("DB scaricato da MEGA con successo.")
return True
finally:
shutil.rmtree(tmp_dir, ignore_errors=True)
except Exception as e:
log.warning("Download MEGA fallito: %s", e)
return False
def upload_db(email: str, password: str) -> bool:
try:
from mega import Mega
m = Mega().login(email, password)
tmp = tempfile.NamedTemporaryFile(suffix='.zip', delete=False)
tmp.close()
try:
if not _zip_db(tmp.name):
log.info("DB vuoto — niente da caricare su MEGA.")
return False
old = m.find(MEGA_FILENAME)
if old:
m.delete(old)
m.upload(tmp.name, dest_filename=MEGA_FILENAME)
log.info("DB caricato su MEGA con successo.")
return True
finally:
os.unlink(tmp.name)
except Exception as e:
log.warning("Upload MEGA fallito: %s", e)
return False
|