#!/usr/bin/env python3 """ ftp_sync.py — Incremental FTP/FTPS mirror for the ThorStorage Scheduler. Downloads ONLY what is missing or changed on the remote server. ftp_sync.py — Oglindă incrementală FTP/FTPS pentru Scheduler-ul ThorStorage. Descarcă DOAR ce lipsește sau s-a schimbat pe serverul remote. HOW IT WORKS IN THE SCHEDULER (important): - Downloaded files are written to OUTPUT_DIR. After each run, ThorStorage PUBLISHES the contents of OUTPUT_DIR into the job's output folder (adds/replaces, never deletes), so the full mirror builds up over time. - OUTPUT_DIR is EMPTY on every run (you don't see last run's files), so the "already downloaded?" decision CANNOT rely on local files — it relies on an INDEX (remote size + date) kept in STATE_DIR. - STATE_DIR is a small folder that PERSISTS between runs. It holds .ftp_index.json, so the second run downloads only the delta, not everything. CUM FUNCȚIONEAZĂ ÎN SCHEDULER (important): - Fișierele descărcate se scriu în OUTPUT_DIR. După fiecare rulare, ThorStorage PUBLICĂ conținutul lui OUTPUT_DIR în folderul de output al jobului (adaugă/înlocuiește, nu șterge), așa se adună oglinda completă în timp. - OUTPUT_DIR e GOL la fiecare rulare (nu vezi fișierele descărcate data trecută), deci decizia „am descărcat deja?" NU se poate baza pe fișierele locale — se bazează pe un INDEX (dimensiune + dată remote) ținut în STATE_DIR. - STATE_DIR e un folder mic care PERSISTĂ între rulări. Aici se ține .ftp_index.json, așa că la a doua rulare descărcăm doar delta, nu tot. Edit the CONFIG block. Password: put it in CONFIG or in the FTP_PASSWORD env var. Editează blocul CONFIG. Parola: pune-o în CONFIG sau în variabila de mediu FTP_PASSWORD. """ import ftplib import hashlib import json import logging import os import sys from datetime import datetime, timezone # ================== CONFIG ================== HOST = "ftp.example.com" # EN: server IP or hostname | RO: IP sau hostname server PORT = 21 USER = "username" # EN: FTP account name | RO: numele contului FTP PASSWORD = "" # EN: leave "" and use FTP_PASSWORD at runtime # RO: lasă "" și folosește FTP_PASSWORD la rulare REMOTE_DIR = "/pub/data" # EN: the folder on the server | RO: directorul de pe server RECURSIVE = True # EN: walk subfolders too | RO: parcurge subdirectoare USE_TLS = True # EN: True = FTPS (FTP over TLS), recommended # RO: True = FTPS (FTP peste TLS), recomandat PASSIVE = True # EN: False = active mode | RO: False = mod activ TIMEOUT = 30 DRY_RUN = False # EN: True = report only, don't download # RO: True = doar raportează, nu descarcă VERBOSE = False # ============================================ # EN: In the Scheduler these come from the environment; outside it, they fall back below. # RO: În Scheduler acestea vin din mediu; în afara lui, cad pe căile de mai jos. OUTPUT_DIR = os.environ.get("OUTPUT_DIR") or "./out" # EN: downloaded files (published) | RO: fișierele descărcate (se publică) STATE_DIR = os.environ.get("STATE_DIR") or OUTPUT_DIR # EN: the index (persists) | RO: indexul (persistă între rulări) INDEX_PATH = os.path.join(STATE_DIR, ".ftp_index.json") log = logging.getLogger("ftp_sync") # ---------------------------------------------------------------- index def load_index(): if not os.path.exists(INDEX_PATH): return {"files": {}, "last_sync": None} try: with open(INDEX_PATH, "r", encoding="utf-8") as f: data = json.load(f) data.setdefault("files", {}) return data except (json.JSONDecodeError, OSError) as e: # EN: index is corrupt — rebuild from scratch (everything is re-downloaded once). # RO: index corupt — se reconstruiește de la zero (se redescarcă tot o dată). log.warning("Index corrupt (%s), rebuilding from scratch (everything re-downloads once).", e) return {"files": {}, "last_sync": None} def save_index(index): os.makedirs(STATE_DIR, exist_ok=True) # EN: write to a temp file then rename, so the index is never left half-written. # RO: scrie într-un fișier temporar apoi redenumește, ca indexul să nu rămână scris pe jumătate. tmp = INDEX_PATH + ".tmp" with open(tmp, "w", encoding="utf-8") as f: json.dump(index, f, indent=2, ensure_ascii=False) os.replace(tmp, INDEX_PATH) def sha256_file(path, chunk=1 << 20): # EN: hash the file in chunks, so a big file costs no memory. # RO: calculează hash-ul pe bucăți, ca un fișier mare să nu consume memorie. h = hashlib.sha256() with open(path, "rb") as f: for block in iter(lambda: f.read(chunk), b""): h.update(block) return h.hexdigest() # ---------------------------------------------------------------- ftp def connect(): cls = ftplib.FTP_TLS if USE_TLS else ftplib.FTP ftp = cls(timeout=TIMEOUT) ftp.connect(HOST, PORT) ftp.login(USER, PASSWORD) if USE_TLS: ftp.prot_p() # EN: encrypt the data channel too, not just the login # RO: criptează și canalul de date, nu doar autentificarea ftp.set_pasv(PASSIVE) return ftp def list_remote(ftp, remote_dir, _prefix=""): """EN: {relative_path: {"size": int, "mtime": str|None}} — MLSD if available, else SIZE+MDTM. RO: {cale_relativă: {"size": int, "mtime": str|None}} — MLSD dacă există, altfel SIZE+MDTM.""" out = {} ftp.cwd(remote_dir) try: for name, facts in list(ftp.mlsd()): if name in (".", ".."): continue typ = facts.get("type", "") rel = _prefix + name if typ == "file": out[rel] = {"size": int(facts.get("size", -1)), "mtime": facts.get("modify")} elif typ == "dir" and RECURSIVE: out.update(list_remote(ftp, name, rel + "/")) ftp.cwd("..") return out except (ftplib.error_perm, ftplib.error_proto): pass # EN: MLSD unavailable — fall back below | RO: MLSD indisponibil — fallback mai jos for name in ftp.nlst(): if name in (".", ".."): continue size = mtime = None try: size = ftp.size(name) except (ftplib.error_perm, ftplib.error_reply): pass if size is None: continue # EN: probably a directory (no recursion on the fallback path) # RO: probabil director (fără recursivitate pe fallback) try: resp = ftp.sendcmd("MDTM " + name) if resp.startswith("213"): mtime = resp[4:].strip()[:14] except (ftplib.error_perm, ftplib.error_reply): pass out[_prefix + name] = {"size": size, "mtime": mtime} return out def download(ftp, rel_path, local_path): os.makedirs(os.path.dirname(local_path) or ".", exist_ok=True) # EN: download to a .part file, then rename — a half-file never looks complete. # RO: descarcă într-un fișier .part, apoi redenumește — un fișier pe jumătate nu pare complet. tmp = local_path + ".part" ftp.cwd(REMOTE_DIR) with open(tmp, "wb") as f: ftp.retrbinary("RETR " + rel_path, f.write, blocksize=1 << 16) os.replace(tmp, local_path) # ---------------------------------------------------------------- logic / logica def needs_download(meta, entry): """EN: The decision is made FROM THE INDEX (not from local files — OUTPUT_DIR is empty each run). RO: Decizia se ia DUPĂ INDEX (nu după fișierele locale — OUTPUT_DIR e gol la fiecare rulare).""" if entry is None: return "new (not in the index)" if entry.get("remote_size") != meta["size"]: return "size changed on remote" if meta.get("mtime") and entry.get("remote_mtime") != meta["mtime"]: return "date changed on remote" return None def sync(): os.makedirs(OUTPUT_DIR, exist_ok=True) os.makedirs(STATE_DIR, exist_ok=True) index = load_index() files = index["files"] log.info("Connecting to %s:%s (%s) ...", HOST, PORT, "FTPS" if USE_TLS else "FTP") ftp = connect() try: remote = list_remote(ftp, REMOTE_DIR) log.info("Remote: %d files. Index: %d known.", len(remote), len(files)) new = updated = skipped = failed = 0 for rel in sorted(remote): meta = remote[rel] entry = files.get(rel) reason = needs_download(meta, entry) if reason is None: skipped += 1 continue local_path = os.path.join(OUTPUT_DIR, rel.replace("/", os.sep)) log.info("%s -> %s", rel, reason) if DRY_RUN: continue try: download(ftp, rel, local_path) except (OSError, *ftplib.all_errors) as e: log.error("Failed to download %s: %s", rel, e) failed += 1 continue files[rel] = { "remote_size": meta["size"], "remote_mtime": meta.get("mtime"), "downloaded_size": os.path.getsize(local_path), "downloaded_at": datetime.now(timezone.utc).isoformat(), "sha256": sha256_file(local_path), } if entry is None: new += 1 else: updated += 1 # EN: files that no longer exist on the remote — we only flag them in the index. # (We can't delete them from the published mirror — the real folder isn't mounted here.) # RO: fișiere care nu mai există pe remote — doar le însemnăm în index. # (Nu le putem șterge din oglinda publicată — folderul real nu e montat în sandbox.) orphans = [r for r in list(files) if r not in remote] for rel in orphans: files[rel]["missing_on_remote"] = True if orphans: log.warning("%d file(s) no longer exist on the remote (kept in the mirror).", len(orphans)) index["last_sync"] = datetime.now(timezone.utc).isoformat() if not DRY_RUN: save_index(index) log.info("Done. New: %d | Updated: %d | Unchanged: %d | Failed: %d | Orphaned: %d", new, updated, skipped, failed, len(orphans)) if failed: sys.exit(1) # EN: mark the run as failed if any download failed # RO: marchează rularea ca eșuată dacă a picat vreo descărcare finally: try: ftp.quit() except ftplib.all_errors: ftp.close() def main(): global PASSWORD logging.basicConfig( level=logging.DEBUG if VERBOSE else logging.INFO, format="%(asctime)s %(levelname)s %(message)s", datefmt="%H:%M:%S", stream=sys.stderr, ) # EN: the password comes from the FTP_PASSWORD env var (preferred) or from CONFIG above. # RO: parola vine din variabila de mediu FTP_PASSWORD (de preferat) sau din CONFIG de sus. PASSWORD = os.environ.get("FTP_PASSWORD", PASSWORD) if not PASSWORD: log.error("No password. Set PASSWORD in CONFIG or the FTP_PASSWORD env var. / " "Nicio parolă. Pune PASSWORD în CONFIG sau variabila de mediu FTP_PASSWORD.") sys.exit(2) log.info("Syncing %s:%s%s -> OUTPUT_DIR=%s (index in STATE_DIR=%s)", HOST, PORT, REMOTE_DIR, OUTPUT_DIR, STATE_DIR) try: sync() except ftplib.all_errors as e: log.error("FTP error: %s", e) sys.exit(1) if __name__ == "__main__": main()