""" storage.py ---------- Abstrakce úložiště pro faktury. Dva backendy se stejným rozhraním: - LocalStorage — lokální filesystem (Windows, Dropbox mount přes najdi_dropbox). - DropboxStorage — Dropbox HTTP API (unraid/server, kde není Dropbox mount). Výběr přes proměnnou STORAGE: "local" (default) | "dropbox". Rozhraní (oba backendy): load_hashes() -> set[str] # otisky souborů už v cíli (dedup baseline) hash_bytes(data) -> str # otisk vstupních bajtů (stejný algoritmus) save(name, data) -> str # ulož, vrať finální název (řeší kolizi názvu) describe() -> str # popis cíle do logu Pozn.: každý backend je vnitřně konzistentní (load_hashes i hash_bytes používají týž algoritmus). Local = sha256 obsahu. Dropbox = Dropbox "content_hash" (blokový sha256, viz dokumentace Dropboxu) — bere se přímo z metadat souboru, takže není potřeba nic stahovat. """ import hashlib import os from pathlib import Path _DROPBOX_BLOCK = 4 * 1024 * 1024 # 4 MiB def _dropbox_content_hash(data: bytes) -> str: """Dropbox content_hash: sha256 z konkatenace sha256 jednotlivých 4MiB bloků.""" h = hashlib.sha256() for i in range(0, len(data), _DROPBOX_BLOCK): h.update(hashlib.sha256(data[i:i + _DROPBOX_BLOCK]).digest()) return h.hexdigest() # ========================= # LOKÁLNÍ FILESYSTEM # ========================= class LocalStorage: def __init__(self, target_dir: Path): self.dir = Path(target_dir) self.dir.mkdir(parents=True, exist_ok=True) def hash_bytes(self, data: bytes) -> str: return hashlib.sha256(data).hexdigest() def load_hashes(self) -> set: return {self.hash_bytes(p.read_bytes()) for p in self.dir.glob("*.pdf")} def save(self, name: str, data: bytes) -> str: out = self.dir / name stem, suffix = Path(name).stem, Path(name).suffix i = 2 while out.exists(): out = self.dir / f"{stem} ({i}){suffix}" i += 1 out.write_bytes(data) return out.name def describe(self) -> str: return str(self.dir) # ========================= # DROPBOX HTTP API # ========================= class DropboxStorage: def __init__(self, folder: str, app_key: str, app_secret: str, refresh_token: str): import dropbox # lazy import — potřeba jen pro tento backend self._dbx_mod = dropbox self.dbx = dropbox.Dropbox( app_key=app_key, app_secret=app_secret, oauth2_refresh_token=refresh_token, ) self.folder = "/" + folder.strip("/") def hash_bytes(self, data: bytes) -> str: return _dropbox_content_hash(data) def load_hashes(self) -> set: hashes = set() try: res = self.dbx.files_list_folder(self.folder) except Exception: return hashes # složka ještě nemusí existovat while True: for e in res.entries: ch = getattr(e, "content_hash", None) if ch: hashes.add(ch) if not res.has_more: break res = self.dbx.files_list_folder_continue(res.cursor) return hashes def save(self, name: str, data: bytes) -> str: # autorename=True -> při kolizi názvu (jiný obsah) Dropbox přidá " (1)". md = self.dbx.files_upload( data, f"{self.folder}/{name}", mode=self._dbx_mod.files.WriteMode.add, autorename=True, ) return md.name def describe(self) -> str: return f"Dropbox:{self.folder}" # ========================= # VÝBĚR BACKENDU # ========================= def get_storage(local_dir: Path, dropbox_path: str): """ STORAGE=dropbox -> DropboxStorage (klíče DROPBOX_APP_KEY/SECRET/REFRESH_TOKEN z prostředí), jinak LocalStorage(local_dir). """ if os.getenv("STORAGE", "local").lower() == "dropbox": return DropboxStorage( dropbox_path, os.environ["DROPBOX_APP_KEY"], os.environ["DROPBOX_APP_SECRET"], os.environ["DROPBOX_APP_REFRESH_TOKEN"], ) return LocalStorage(local_dir)