400 lines
14 KiB
Python
400 lines
14 KiB
Python
"""e-pub: en indsat biblioteks-reference → .epub eller .m4b til download.
|
|
|
|
Selve arbejdet laves af værktøjerne i settings["tools_dir"] (fetch.sh m.fl.); portalen
|
|
kører dem bare: frisk per-job-mappe som CWD, rent env, stderr som live-log, timeout
|
|
og en semafor fra indstillingerne. Hvert job har en job.json i sin mappe, så færdige
|
|
filer overlever en genstart; jobs der var i gang, markeres som afbrudt. Alt (fil og job)
|
|
slettes output_ttl_seconds efter jobbet sluttede, af fejer() der kører i baggrunden.
|
|
"""
|
|
import asyncio
|
|
import json
|
|
import os
|
|
import re
|
|
import secrets
|
|
import shutil
|
|
import signal
|
|
import time
|
|
from collections import deque
|
|
from pathlib import Path
|
|
|
|
import config
|
|
import store
|
|
|
|
MAX_INPUT = 2048 # tegn i det indsatte link
|
|
MAX_AKTIVE_PR_BRUGER = 3 # i kø + kørende
|
|
LOG_LINJER = 300
|
|
|
|
TILSTANDE_AKTIVE = ("koe", "koerer")
|
|
UDFALD = {
|
|
2: "Kunne ikke læse linket.",
|
|
3: "Ikke tilgængelig (udløbet?).",
|
|
4: "Serverfejl: proxy påkrævet, men ikke sat op. Sig det til en administrator.",
|
|
}
|
|
CONTENT_TYPES = {".epub": "application/epub+zip", ".m4b": "audio/mp4"}
|
|
KRAEVEDE_PROGRAMMER = ("bash", "curl", "jq", "openssl", "ffmpeg")
|
|
KRAEVEDE_FILER = ("fetch.sh", "run.sh", "run-audiobook.sh", "lib.sh")
|
|
PATH = "/usr/local/bin:/usr/bin:/bin"
|
|
|
|
_jobs: dict[str, dict] = {}
|
|
_sem: tuple[int, asyncio.Semaphore] | None = None
|
|
|
|
|
|
|
|
# ---------------- indstillinger ----------------
|
|
def get_settings() -> dict:
|
|
return {**config.EPUB_DEFAULTS, **(store.load_json(config.EPUB_SETTINGS_PATH, config.EPUB_DEFAULTS) or {})}
|
|
|
|
|
|
def save_settings(patch: dict) -> dict:
|
|
s = get_settings()
|
|
s.update(patch)
|
|
config.EPUB_SETTINGS_PATH.write_text(json.dumps(s, ensure_ascii=False, indent=2), "utf-8")
|
|
return s
|
|
|
|
|
|
def har_adgang(user, admin: bool) -> bool:
|
|
if admin:
|
|
return True
|
|
gruppe = (get_settings().get("allowed_group") or "").strip()
|
|
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()
|
|
d = Path(s["tools_dir"])
|
|
return {
|
|
"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", "")),
|
|
}
|
|
|
|
|
|
# ---------------- jobs ----------------
|
|
def _semafor(n: int) -> asyncio.Semaphore:
|
|
# Ny semafor når indstillingen ændres; jobs der allerede venter på den gamle bliver der.
|
|
global _sem
|
|
if _sem is None or _sem[0] != n:
|
|
_sem = (n, asyncio.Semaphore(n))
|
|
return _sem[1]
|
|
|
|
|
|
def valider_input(tekst: str) -> str | None:
|
|
"""Returnerer fejltekst eller None. Strengen sendes ellers ordret videre."""
|
|
if not tekst:
|
|
return "Indsæt et link."
|
|
if len(tekst) > MAX_INPUT:
|
|
return "Linket er for langt."
|
|
if any(ord(c) < 32 or ord(c) == 127 for c in tekst) or tekst.startswith("-"):
|
|
return UDFALD[2]
|
|
return None
|
|
|
|
|
|
def opret(user: str, tekst: str) -> tuple[dict | None, str | None]:
|
|
aktive = sum(1 for j in _jobs.values() if j["ejer"] == user and j["tilstand"] in TILSTANDE_AKTIVE)
|
|
if aktive >= MAX_AKTIVE_PR_BRUGER:
|
|
return None, f"Du har allerede {aktive} i gang. Vent til et af dem er færdigt."
|
|
jid = secrets.token_urlsafe(12)
|
|
_mappe({"id": jid}).mkdir(parents=True)
|
|
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, "sendt": ""}
|
|
_jobs[jid] = job
|
|
_gem(job)
|
|
return job, None
|
|
|
|
|
|
def job(jid: str) -> dict | None:
|
|
return _jobs.get(jid)
|
|
|
|
|
|
def brugerens_jobs(user: str) -> list[dict]:
|
|
return sorted((j for j in _jobs.values() if j["ejer"] == user), key=lambda j: -j["oprettet"])
|
|
|
|
|
|
def alle_jobs() -> list[dict]:
|
|
return sorted(_jobs.values(), key=lambda j: -j["oprettet"])
|
|
|
|
|
|
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"], "sendt": j["sendt"], "kan_sende": kan_sende(j)}
|
|
|
|
|
|
def _mappe(j: dict) -> Path:
|
|
return config.EPUB_SCRATCH / j["id"]
|
|
|
|
|
|
def _gem(j: dict) -> None:
|
|
d = _mappe(j)
|
|
if not d.is_dir():
|
|
return
|
|
tmp = d / ".job.json.tmp"
|
|
tmp.write_text(json.dumps({**j, "log": list(j["log"])}, ensure_ascii=False), "utf-8")
|
|
os.replace(tmp, d / "job.json")
|
|
|
|
|
|
def _indlaes() -> None:
|
|
"""Ved opstart: genindlæs jobs fra disken. Afbrudte jobs → fejl, rester uden job.json → væk."""
|
|
config.EPUB_SCRATCH.mkdir(parents=True, exist_ok=True)
|
|
for d in config.EPUB_SCRATCH.iterdir():
|
|
try:
|
|
j = json.loads((d / "job.json").read_text("utf-8"))
|
|
assert j["id"] == d.name
|
|
except Exception:
|
|
shutil.rmtree(d, ignore_errors=True)
|
|
continue
|
|
j["log"] = deque(j.get("log") or [], maxlen=LOG_LINJER)
|
|
shutil.rmtree(d / "tmp", ignore_errors=True)
|
|
if j["tilstand"] in TILSTANDE_AKTIVE:
|
|
j.update(tilstand="fejl", besked="Afbrudt, fordi serveren blev genstartet. Prøv igen.",
|
|
slut=time.time())
|
|
shutil.rmtree(d / "ud", ignore_errors=True)
|
|
elif j["tilstand"] == "faerdig" and not (j.get("fil") and Path(j["fil"]).is_file()):
|
|
j.update(tilstand="udloebet", fil=None)
|
|
_jobs[j["id"]] = j
|
|
_gem(j)
|
|
|
|
|
|
def feje(nu: float | None = None) -> None:
|
|
"""Slet jobs (mappe + fil) der sluttede for mere end output_ttl_seconds siden."""
|
|
nu = nu or time.time()
|
|
ttl = int(get_settings()["output_ttl_seconds"])
|
|
for j in list(_jobs.values()):
|
|
if j["tilstand"] not in TILSTANDE_AKTIVE and (j["slut"] or j["oprettet"]) + ttl < nu:
|
|
shutil.rmtree(_mappe(j), ignore_errors=True)
|
|
_jobs.pop(j["id"], None)
|
|
|
|
|
|
async def fejer() -> None:
|
|
while True:
|
|
try:
|
|
feje()
|
|
except Exception:
|
|
pass
|
|
await asyncio.sleep(max(1, min(60, int(get_settings()["output_ttl_seconds"]) // 4)))
|
|
|
|
|
|
def ryd(j: dict, tilstand: str = "hentet") -> None:
|
|
"""Fjern den færdige fil (fx efter download); jobbet bliver stående til TTL."""
|
|
shutil.rmtree(_mappe(j) / "ud", ignore_errors=True)
|
|
if j["tilstand"] == "faerdig":
|
|
j["tilstand"] = tilstand
|
|
j["fil"] = None
|
|
_gem(j)
|
|
|
|
|
|
def _log(j: dict, linje: str) -> None:
|
|
linje = linje.rstrip()
|
|
if not linje:
|
|
return
|
|
if config.RP_API_KEY:
|
|
linje = linje.replace(config.RP_API_KEY, "***")
|
|
j["log"].append(linje[:500])
|
|
|
|
|
|
async def _laes_stderr(stream: asyncio.StreamReader, j: dict) -> None:
|
|
# Delt på både \n og \r, så fremdriftslinjer der overskriver sig selv også kommer med.
|
|
rest = ""
|
|
while True:
|
|
bid = await stream.read(4096)
|
|
if not bid:
|
|
break
|
|
dele = re.split(r"[\r\n]", rest + bid.decode("utf-8", "replace"))
|
|
rest = dele.pop()
|
|
for d in dele:
|
|
_log(j, d)
|
|
if len(rest) > 4096:
|
|
_log(j, rest)
|
|
rest = ""
|
|
_log(j, rest)
|
|
|
|
|
|
def _find_fil(ud: Path, log: list[str]) -> Path | None:
|
|
filer = [p for p in ud.iterdir() if p.is_file() and p.suffix.lower() in CONTENT_TYPES]
|
|
if len(filer) == 1:
|
|
return filer[0]
|
|
# Flere/ingen i mappen: fald tilbage på den sidste "Wrote <sti> (...)"-linje.
|
|
for linje in reversed(log):
|
|
m = re.match(r"Wrote (/.+?) \(", linje)
|
|
if m:
|
|
p = Path(m.group(1))
|
|
if p.is_file() and p.parent == ud and p.suffix.lower() in CONTENT_TYPES:
|
|
return p
|
|
break
|
|
return None
|
|
|
|
|
|
async def koer(jid: str) -> None:
|
|
"""Køres i baggrunden: vent på plads, kør værktøjet, send evt. videre til biblioteket."""
|
|
j = _jobs[jid]
|
|
s = get_settings()
|
|
try:
|
|
async with _semafor(int(s["max_concurrent_jobs"])):
|
|
s = get_settings()
|
|
await _koer_vaerktoej(j, s)
|
|
except Exception as e: # sidste skanse: et job må aldrig hænge i "koerer"
|
|
j["tilstand"], j["besked"] = "fejl", f"Uventet fejl: {e}"
|
|
finally:
|
|
j["slut"] = j["slut"] or time.time()
|
|
shutil.rmtree(_mappe(j) / "tmp", ignore_errors=True)
|
|
if j["tilstand"] != "faerdig":
|
|
shutil.rmtree(_mappe(j) / "ud", ignore_errors=True)
|
|
_gem(j)
|
|
if s.get("auto_send") and kan_sende(j):
|
|
fejl = await asyncio.to_thread(send, j)
|
|
j["besked"] = fejl or f"Sendt automatisk til {j['sendt']}. Du kan også hente filen."
|
|
_gem(j)
|
|
|
|
|
|
async def _koer_vaerktoej(j: dict, s: dict) -> None:
|
|
fetch = Path(s["tools_dir"]) / "fetch.sh"
|
|
if not fetch.is_file():
|
|
j["tilstand"], j["besked"] = "fejl", "Serverfejl: værktøjet er ikke installeret."
|
|
return
|
|
ud, tmp = _mappe(j) / "ud", _mappe(j) / "tmp"
|
|
ud.mkdir(parents=True)
|
|
tmp.mkdir()
|
|
env = {"PATH": PATH, "TMP": str(tmp), "RP_API_KEY": config.RP_API_KEY,
|
|
"RP_BASE_URL": config.RP_BASE_URL, "RP_REQUIRED": "1" if s["rp_required"] else "0"}
|
|
j["tilstand"], j["startet"] = "koerer", time.time()
|
|
_gem(j)
|
|
proc = await asyncio.create_subprocess_exec(
|
|
"bash", str(fetch), j["input"], cwd=ud, env=env,
|
|
stdin=asyncio.subprocess.DEVNULL, stdout=asyncio.subprocess.DEVNULL,
|
|
stderr=asyncio.subprocess.PIPE, start_new_session=True)
|
|
laeser = asyncio.create_task(_laes_stderr(proc.stderr, j))
|
|
try:
|
|
rc = await asyncio.wait_for(proc.wait(), timeout=int(s["job_timeout_seconds"]))
|
|
except asyncio.TimeoutError:
|
|
_draeb(proc)
|
|
await proc.wait()
|
|
rc = None
|
|
except asyncio.CancelledError:
|
|
_draeb(proc)
|
|
raise
|
|
finally:
|
|
try:
|
|
await asyncio.wait_for(laeser, timeout=5)
|
|
except Exception:
|
|
laeser.cancel()
|
|
j["slut"] = time.time()
|
|
if rc is None:
|
|
j["tilstand"], j["besked"] = "fejl", f"Tog for lang tid (over {int(s['job_timeout_seconds']) // 60} min) og blev stoppet."
|
|
return
|
|
if rc == 0:
|
|
fil = _find_fil(ud, list(j["log"]))
|
|
if fil:
|
|
j.update(tilstand="faerdig", fil=str(fil), filnavn=fil.name, stoerrelse=fil.stat().st_size,
|
|
besked="Klar til download.")
|
|
return
|
|
rc = -1
|
|
sidste = " · ".join(list(j["log"])[-3:])
|
|
j["tilstand"] = "fejl"
|
|
j["besked"] = UDFALD.get(rc) or ("Noget gik galt." + (f" ({sidste})" if sidste else ""))
|
|
|
|
|
|
def _draeb(proc) -> None:
|
|
# Hele procesgruppen: værktøjet starter selv underprocesser.
|
|
try:
|
|
os.killpg(proc.pid, signal.SIGKILL)
|
|
except ProcessLookupError:
|
|
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
|
|
_gem(j)
|
|
return None
|
|
|
|
|
|
def content_type(filnavn: str) -> str:
|
|
return CONTENT_TYPES.get(Path(filnavn).suffix.lower(), "application/octet-stream")
|
|
|
|
|
|
def varighed(sek: int) -> str:
|
|
if sek % 86400 == 0:
|
|
d = sek // 86400
|
|
return "et døgn" if d == 1 else f"{d} døgn"
|
|
if sek >= 3600:
|
|
t = round(sek / 3600)
|
|
return "en time" if t == 1 else f"{t} timer"
|
|
return f"{max(1, sek // 60)} min"
|
|
|
|
|
|
def human(n: int) -> str:
|
|
for enhed in ("B", "KB", "MB", "GB"):
|
|
if n < 1024:
|
|
return f"{n:.0f} {enhed}" if enhed == "B" else f"{n:.1f} {enhed}"
|
|
n /= 1024
|
|
return f"{n:.1f} TB"
|
|
|
|
|
|
_indlaes()
|