feat(e-pub): Tilføj til CWA (.epub) og Send til Audiobookshelf (.m4b)
CD / deploy (push) Successful in 26s
CI / check (push) Successful in 40s

Claude-Session: https://claude.ai/code/session_01AUj1LSMmrowcomkn1ciAgj
This commit is contained in:
root
2026-10-10 11:49:57 +02:00
parent 9a57e178b5
commit 873500f5d4
6 changed files with 127 additions and 10 deletions
+72 -2
View File
@@ -61,6 +61,10 @@ def har_adgang(user, admin: bool) -> bool:
return bool(user and gruppe and gruppe in (user.get("groups") or []))
def _skrivbar(d: str) -> bool:
return bool(d) and os.path.isdir(d) and os.access(d, os.W_OK | os.X_OK)
def tjek_opsaetning(s: dict | None = None) -> dict:
"""Til admin-siden: mangler der værktøjer eller programmer hvor portalen kører?"""
s = s or get_settings()
@@ -69,6 +73,8 @@ def tjek_opsaetning(s: dict | None = None) -> dict:
"mangler_filer": [f for f in KRAEVEDE_FILER if not (d / f).is_file()],
"mangler_programmer": [p for p in KRAEVEDE_PROGRAMMER if not shutil.which(p, path=PATH)],
"rp_noegle": bool(config.RP_API_KEY),
"cwa_ok": _skrivbar(s.get("cwa_ingest_dir", "")),
"abs_ok": _skrivbar(s.get("abs_dir", "")),
}
@@ -99,7 +105,7 @@ def opret(user: str, tekst: str) -> tuple[dict | None, str | None]:
jid = secrets.token_urlsafe(12)
job = {"id": jid, "ejer": user, "input": tekst, "tilstand": "koe", "log": deque(maxlen=LOG_LINJER),
"besked": "", "fil": None, "filnavn": "", "stoerrelse": 0,
"oprettet": time.time(), "startet": None, "slut": None, "hentet": False}
"oprettet": time.time(), "startet": None, "slut": None, "hentet": False, "sendt": ""}
_jobs[jid] = job
return job, None
@@ -119,7 +125,7 @@ def alle_jobs() -> list[dict]:
def status(j: dict) -> dict:
return {"id": j["id"], "tilstand": j["tilstand"], "besked": j["besked"], "log": list(j["log"])[-80:],
"filnavn": j["filnavn"], "stoerrelse": j["stoerrelse"], "oprettet": j["oprettet"],
"startet": j["startet"], "slut": j["slut"]}
"startet": j["startet"], "slut": j["slut"], "sendt": j["sendt"], "kan_sende": kan_sende(j)}
def _mappe(j: dict) -> Path:
@@ -250,6 +256,70 @@ def _draeb(proc) -> None:
pass
# ---------------- videre til biblioteket ----------------
MAAL = {".epub": ("cwa", "cwa_ingest_dir", "CWA"), ".m4b": ("abs", "abs_dir", "Audiobookshelf")}
def kan_sende(j: dict) -> str:
""""cwa"/"abs" hvis den færdige fil kan sendes videre (mappen findes og kan skrives), ellers ""."""
if j["tilstand"] != "faerdig" or j["sendt"] or not j["filnavn"]:
return ""
maal = MAAL.get(Path(j["filnavn"]).suffix.lower())
return maal[0] if maal and _skrivbar(get_settings().get(maal[1], "")) else ""
def _rent_navn(t: str) -> str:
t = re.sub(r'[/\\\x00-\x1f]', " ", t).strip(" .")
return re.sub(r"\s+", " ", t)[:150] or "Ukendt"
def _ledig(p: Path) -> Path:
n = 2
ud = p
while ud.exists():
ud = p.with_name(f"{p.stem} ({n}){p.suffix}")
n += 1
return ud
def _kopier(kilde: Path, maal: Path) -> None:
# Via et skjult .part-navn og rename, så CWA/Audiobookshelf aldrig ser en halv fil.
tmp = maal.with_name("." + maal.name + ".part")
try:
shutil.copyfile(kilde, tmp)
os.chmod(tmp, 0o666)
os.replace(tmp, maal)
finally:
tmp.unlink(missing_ok=True)
def send(j: dict) -> str | None:
"""Kopiér den færdige fil til CWA's indbakke (.epub) eller Audiobookshelf (.m4b). Fejltekst eller None.
Synkron (kan være flere hundrede MB): kaldes via asyncio.to_thread."""
hvor = kan_sende(j)
if not hvor:
return "Filen kan ikke sendes videre (allerede sendt, udløbet eller ikke sat op)."
kilde = Path(j["fil"])
ext = kilde.suffix.lower()
_, noegle, navn = MAAL[ext]
rod = Path(get_settings()[noegle])
try:
if hvor == "cwa":
# CWA importerer kun små filendelser.
maal = _ledig(rod / (_rent_navn(kilde.stem) + ext))
else:
# Audiobookshelf: Forfatter/Titel/fil når navnet er "Forfatter - Titel", ellers Titel/fil.
forfatter, sep, titel = kilde.stem.partition(" - ")
mappe = rod / _rent_navn(forfatter) / _rent_navn(titel) if sep else rod / _rent_navn(kilde.stem)
mappe.mkdir(parents=True, exist_ok=True)
maal = _ledig(mappe / (_rent_navn(kilde.stem) + ext))
_kopier(kilde, maal)
except OSError as e:
return f"Kunne ikke sende til {navn}: {e.strerror or e}"
j["sendt"] = navn
return None
def content_type(filnavn: str) -> str:
return CONTENT_TYPES.get(Path(filnavn).suffix.lower(), "application/octet-stream")