switch to grimmory as book web app and a more modern dashboard
This commit is contained in:
+90
@@ -0,0 +1,90 @@
|
||||
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.warning("Grimmory search returned HTTP %s for '%s'", r.status_code, stem)
|
||||
except Exception as e:
|
||||
log.warning("Grimmory duplicate check failed 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]:
|
||||
try:
|
||||
r = requests.get(
|
||||
url.rstrip("/") + "/api/v1/books",
|
||||
params={"size": 1},
|
||||
auth=(user, password),
|
||||
timeout=10,
|
||||
)
|
||||
if r.status_code == 200:
|
||||
return True, "Connected to Grimmory successfully"
|
||||
if r.status_code == 401:
|
||||
return False, "Authentication failed — check username and password"
|
||||
return False, f"HTTP {r.status_code}"
|
||||
except requests.exceptions.ConnectionError:
|
||||
return False, "Could not connect — check the URL"
|
||||
except Exception as e:
|
||||
return False, str(e)
|
||||
Reference in New Issue
Block a user