85 lines
2.5 KiB
Python
85 lines
2.5 KiB
Python
import hashlib
|
|
import logging
|
|
import shutil
|
|
from pathlib import Path
|
|
|
|
import requests
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
|
|
def compute_sha256(path: Path) -> str:
|
|
h = hashlib.sha256()
|
|
with path.open("rb") as f:
|
|
for chunk in iter(lambda: f.read(65536), b""):
|
|
h.update(chunk)
|
|
return h.hexdigest()
|
|
|
|
|
|
def _is_in_grimmory(filename: str, url: str, user: str, password: str) -> bool:
|
|
"""Search Grimmory Komga-compatible API by filename stem. Non-fatal on error."""
|
|
stem = Path(filename).stem
|
|
try:
|
|
r = requests.get(
|
|
url.rstrip("/") + "/api/v1/books",
|
|
params={"search": stem},
|
|
auth=(user, password),
|
|
timeout=10,
|
|
)
|
|
if r.status_code == 200:
|
|
return r.json().get("totalElements", 0) > 0
|
|
log.debug("Grimmory search returned HTTP %s for '%s' — skipping API check", r.status_code, stem)
|
|
except Exception as e:
|
|
log.debug("Grimmory duplicate check unavailable for '%s': %s", filename, e)
|
|
return False
|
|
|
|
|
|
class PlacementResult:
|
|
__slots__ = ("status", "sha256", "error_msg")
|
|
|
|
def __init__(self, status: str, sha256: str = "", error_msg: str = ""):
|
|
self.status = status
|
|
self.sha256 = sha256
|
|
self.error_msg = error_msg
|
|
|
|
|
|
def place_book(
|
|
book_path: Path,
|
|
bookdrop_path: str,
|
|
url: str,
|
|
user: str,
|
|
password: str,
|
|
sha256: str | None = None,
|
|
) -> PlacementResult:
|
|
if sha256 is None:
|
|
sha256 = compute_sha256(book_path)
|
|
|
|
if _is_in_grimmory(book_path.name, url, user, password):
|
|
log.info("Skipping '%s' — already in Grimmory", book_path.name)
|
|
return PlacementResult("skipped", sha256)
|
|
|
|
dest_dir = Path(bookdrop_path)
|
|
dest_dir.mkdir(parents=True, exist_ok=True)
|
|
dest = dest_dir / book_path.name
|
|
counter = 1
|
|
while dest.exists():
|
|
dest = dest_dir / f"{book_path.stem}_{counter}{book_path.suffix}"
|
|
counter += 1
|
|
|
|
shutil.copy2(book_path, dest)
|
|
log.info("Placed '%s' → %s", book_path.name, dest)
|
|
return PlacementResult("success", sha256)
|
|
|
|
|
|
def test_connection(url: str, user: str, password: str) -> tuple[bool, str]:
|
|
base = url.rstrip("/")
|
|
try:
|
|
r = requests.get(base + "/api/v1/healthcheck", timeout=10)
|
|
if r.status_code == 200:
|
|
return True, "Connected to Grimmory successfully"
|
|
return False, f"Grimmory not reachable (HTTP {r.status_code})"
|
|
except requests.exceptions.ConnectionError:
|
|
return False, "Could not connect — check the URL"
|
|
except Exception as e:
|
|
return False, str(e)
|