"""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 (...)"-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()