Vladimir Buzalka
129 lines
4.2 KiB
Python
129 lines
4.2 KiB
Python
"""
|
|
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)
|