weircon-front/epub.py

"""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. Jobs lever i hukommelsen; en genstart dræber
igangværende jobs, og scratch-roden ryddes ved opstart.
"""
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

# Rester fra før en genstart kan ingen længere hente: væk med dem.
shutil.rmtree(config.EPUB_SCRATCH, ignore_errors=True)
config.EPUB_SCRATCH.mkdir(parents=True, exist_ok=True)


# ---------------- 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 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),
    }


# ---------------- 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)
    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}
    _jobs[jid] = 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"]}


def _mappe(j: dict) -> Path:
    return config.EPUB_SCRATCH / j["id"]


def ryd(j: dict, tilstand: str = "udloebet") -> None:
    shutil.rmtree(_mappe(j), ignore_errors=True)
    if j["tilstand"] == "faerdig":
        j["tilstand"] = tilstand
        j["fil"] = None


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, ryd op efter TTL."""
    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), ignore_errors=True)
    await asyncio.sleep(int(s["output_ttl_seconds"]))
    ryd(j)
    # Jobbet glemmes efter endnu en TTL, så "udløbet" kan ses et stykke tid.
    await asyncio.sleep(int(s["output_ttl_seconds"]))
    _jobs.pop(jid, None)


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


def content_type(filnavn: str) -> str:
    return CONTENT_TYPES.get(Path(filnavn).suffix.lower(), "application/octet-stream")


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"