fix(import): survive file names that are not valid utf-8

README.md

@@ -32,6 +32,10 @@ Audible führt Reihen manchmal anders, als man sie ablegen möchte (*Harry Potte
 - **Vor dem Anlegen**: Im Import-Dialog übernimmt ✎ an einem Treffer dessen Felder (Titel, Autor, Serie, Band) in die Eingabemaske — Sprecher, Cover, Jahr und ASIN bleiben erhalten. Erst *Anlegen & verbinden* erzeugt die Anfrage.
 - **Danach**: Im Tab *Missing* bzw. *Importiert* öffnet ein Klick auf die Karte alle Felder. *Speichern & neu taggen* schreibt die ID3-Tags neu, **Speichern & neu ablegen** wendet das Namensschema der Library auf die bereits verschobenen Dateien an — der Ordner wird also umbenannt bzw. verschoben, Cover ziehen mit um und leer gewordene Ordner verschwinden. Ein belegter Zielordner bricht den Vorgang ab, statt etwas zu überschreiben — dasselbe gilt beim Import selbst: Rendern zwei Titel nach dem Namensschema denselben Pfad (etwa zwei Bände mit identischem Titel bei einem Schema ohne `{Volume}`), scheitert der zweite mit einer Meldung, statt den ersten zu überschreiben.
 
+## Dateinamen mit kaputter Kodierung
+
+Linux erlaubt in Dateinamen beliebige Bytes, ältere Rips tragen ihre Umlaute deshalb manchmal als Latin-1 (`Die Fu\xdfball-Falle` statt UTF-8). Python liest solche Namen mit Ersatzzeichen ein, die sich nicht als JSON ausliefern lassen — früher scheiterte daran der **gesamte** Scan mit einem 500er. wordarr zeigt betroffene Einträge jetzt mit `�` an und überträgt den Pfad verlustfrei im Hintergrund, sodass Scan und Import normal funktionieren. Beim Import bekommt die Datei ohnehin den Namen aus dem Schema — danach ist der Name sauber.
+
 ## Sprachen
 
 Jede Library kann eine **Sprache** tragen (`Deutsch`/`Englisch`, Default: egal). Audible liefert zu jedem Titel die Sprache mit, deshalb zeigt die Suche für eine englische Library nur englische Ausgaben und für eine deutsche nur deutsche — praktisch, wenn dieselbe Reihe in beiden Sprachen in getrennten Libraries liegt (*A Song of Ice and Fire* vs. *Das Lied von Eis und Feuer*). Gesucht wird dann auch auf dem passenden Marktplatz (`audible.com` bzw. `audible.de`), was die Trefferqualität deutlich hebt. Im Tab *Suche* lässt sich die Sprache zusätzlich frei filtern; im Import-Dialog kommt sie automatisch aus der gewählten Ziel-Library.

tests/test_fsnames.py

@@ -0,0 +1,37 @@
+"""Names that are not valid UTF-8 must not break the API."""
+import json
+import os
+from pathlib import Path
+
+from wordarr.fsnames import decode_path, encode_path, is_broken, readable
+
+# a Latin-1 "ß" in an otherwise normal name, as it comes from older rips
+BROKEN = os.fsdecode(b"Die drei ??? - Die Fu\xdfball-Falle")
+
+
+def test_detects_and_cleans_broken_names():
+    assert is_broken(BROKEN)
+    assert not is_broken("Die drei ??? - Die Fußball-Falle")
+    shown = readable(BROKEN)
+    assert shown.startswith("Die drei ??? - Die Fu") and "�" in shown
+    # the cleaned name survives the same encoding that used to raise
+    json.dumps({"name": shown}, ensure_ascii=False).encode("utf-8")
+
+
+def test_encoded_paths_round_trip(tmp_path):
+    folder = tmp_path / BROKEN
+    folder.mkdir()
+    (folder / "01.mp3").write_bytes(b"x")
+
+    encoded = encode_path(str(folder))
+    json.dumps({"path": encoded}, ensure_ascii=False).encode("utf-8")  # transportable
+    assert not is_broken(encoded)
+    assert Path(decode_path(encoded)) == folder
+    assert (Path(decode_path(encoded)) / "01.mp3").read_bytes() == b"x"
+
+
+def test_plain_names_are_left_alone():
+    plain = "/downloads/Die drei ??? - Die Fußball-Falle"
+    assert encode_path(plain) == plain
+    assert decode_path(plain) == plain
+    assert readable(plain) == plain

tests/test_import_flow.py

@@ -1,4 +1,5 @@
 import importlib
+import os
 import shutil
 from pathlib import Path
 
@@ -928,3 +929,39 @@ def test_two_titles_rendering_to_the_same_name_do_not_overwrite(client):
     assert b"Teil 1 Datei 1" in content and b"Teil 2" not in content
     assert sorted(p.name for p in (client.downloads / "teil2").iterdir()) == \
         ["01.mp3", "02.mp3"]
+
+
+def test_scan_and_import_survive_a_name_that_is_not_utf8(client):
+    """A single Latin-1 byte in a folder name used to fail the whole scan with a
+    500 (UnicodeEncodeError: surrogates not allowed)."""
+    broken = os.fsdecode(b"Die drei ??? - Die Fu\xdfball-Falle")
+    folder = client.downloads / broken
+    folder.mkdir()
+    for i in (1, 2):
+        (folder / f"{i:02d}.mp3").write_bytes(b"")
+
+    root = client.tmp_path / "library" / "kids"
+    lib_id = client.post("/api/libraries", json={
+        "name": "kids", "media_type": "audiobook", "root_path": str(root),
+        "folder_template": "{Title}", "file_template": "{Title}",
+    }).json()["id"]
+    req_id = client.post("/api/requests", json={
+        "library_id": lib_id, "title": "Die drei ??? und die Fußball-Falle",
+    }).json()["id"]
+
+    resp = client.get("/api/import/scan")
+    assert resp.status_code == 200
+    item = resp.json()["items"][0]
+    assert "�" in item["name"]          # shown with a replacement character
+    assert item["path"].startswith("b64:")   # but transported losslessly
+
+    res = client.post("/api/import", json={"items": [{
+        "path": item["path"], "is_dir": True, "files": item["files"],
+        "request_id": req_id,
+    }]}).json()["results"][0]
+    assert res["ok"], res
+    assert sorted(p.name for p in Path(res["dest"]).iterdir()) == [
+        "Die drei ??? und die Fußball-Falle - Part 01.mp3",
+        "Die drei ??? und die Fußball-Falle - Part 02.mp3",
+    ]
+    assert not folder.exists()

wordarr/api/imports.py

@@ -7,6 +7,7 @@ from sqlalchemy.orm import Session
 
 from .. import config
 from ..db import BookRequest, get_session
+from ..fsnames import decode_path, readable
 from ..importer import matcher, mover, scanner
 
 router = APIRouter(prefix="/api/import", tags=["import"])
@@ -57,9 +58,13 @@ def do_import(data: ImportIn, session: Session = Depends(get_session)):
     results = []
     download_root = Path(config.DOWNLOAD_DIR).resolve()
     for item in data.items:
+        # the client hands back whatever the scan produced, base64 included
+        item.path = decode_path(item.path)
+        item.files = [decode_path(f) for f in item.files]
+        shown = readable(item.path)
         req = session.get(BookRequest, item.request_id)
         if not req:
-            results.append({"path": item.path, "ok": False,
+            results.append({"path": shown, "ok": False,
                             "error": f"Anfrage #{item.request_id} existiert nicht mehr"})
             continue
         path_gone = not req.imported_path or not Path(req.imported_path).exists()
@@ -70,16 +75,16 @@ def do_import(data: ImportIn, session: Session = Depends(get_session)):
             try:
                 dest = mover.append_to_import(item.files, req, req.library)
             except Exception as exc:
-                results.append({"path": item.path, "ok": False, "error": str(exc)})
+                results.append({"path": shown, "ok": False, "error": str(exc)})
                 continue
             req.imported_path = dest
             session.commit()
-            results.append({"path": item.path, "ok": True, "dest": dest, "appended": True})
+            results.append({"path": shown, "ok": True, "dest": dest, "appended": True})
             continue
         elif req.status != "missing":
             # usually several entries pointing at the same request: the first one
             # imported and flipped it, the rest land here
-            results.append({"path": item.path, "ok": False, "error": (
+            results.append({"path": shown, "ok": False, "error": (
                 f"Anfrage „{req.title}“ (#{req.id}) ist bereits importiert"
                 f"{' nach ' + req.imported_path if req.imported_path else ''}"
                 " — mehrere Einträge auf dieselbe Anfrage? Dann vorher zusammenfassen."
@@ -89,17 +94,17 @@ def do_import(data: ImportIn, session: Session = Depends(get_session)):
         # carry files from several folders, so check them all
         paths = [Path(p).resolve() for p in [item.path, *item.files]]
         if any(download_root not in p.parents and p != download_root for p in paths):
-            results.append({"path": item.path, "ok": False, "error": "path outside download dir"})
+            results.append({"path": shown, "ok": False, "error": "path outside download dir"})
             continue
         try:
             dest = mover.import_item(item.path, item.files, item.is_dir, req, req.library)
         except Exception as exc:
-            results.append({"path": item.path, "ok": False, "error": str(exc)})
+            results.append({"path": shown, "ok": False, "error": str(exc)})
             continue
         req.status = "imported"
         req.imported_path = dest
         session.commit()
-        results.append({"path": item.path, "ok": True, "dest": dest})
+        results.append({"path": shown, "ok": True, "dest": dest})
     if not results:
         raise HTTPException(400, "nothing to import")
     return {"results": results}

wordarr/fsnames.py

@@ -0,0 +1,39 @@
+"""Carrying file names that are not valid UTF-8 through the JSON API.
+
+Linux allows any byte in a name, so an old rip can hold a Latin-1 "ü". Python
+decodes such names with surrogateescape ("\\udcfc"), and those characters cannot
+be encoded as UTF-8 - a single one of them used to make the whole scan fail with
+a 500. Display names are cleaned up, while paths keep a lossless representation
+so the import still finds the file.
+"""
+
+import base64
+import os
+
+B64_PREFIX = "b64:"
+
+
+def is_broken(text: str) -> bool:
+    """True when the name contains bytes that are not valid UTF-8."""
+    return any("\udc80" <= ch <= "\udcff" for ch in text)
+
+
+def readable(text: str) -> str:
+    """Name for the UI: undecodable bytes become the replacement character."""
+    if not is_broken(text):
+        return text
+    return os.fsencode(text).decode("utf-8", "replace")
+
+
+def encode_path(path: str) -> str:
+    """Path for the API: base64 of the raw bytes when it cannot travel as text."""
+    if not is_broken(path):
+        return path
+    return B64_PREFIX + base64.urlsafe_b64encode(os.fsencode(path)).decode("ascii")
+
+
+def decode_path(path: str) -> str:
+    """Turn an encoded path back into something the filesystem understands."""
+    if not path.startswith(B64_PREFIX):
+        return path
+    return os.fsdecode(base64.urlsafe_b64decode(path[len(B64_PREFIX):]))

wordarr/importer/scanner.py

@@ -2,6 +2,7 @@ import re
 from pathlib import Path
 
 from .. import config
+from ..fsnames import encode_path, readable
 
 # "CD", "CD1", "CD 2", "Disc_03", "Teil 1", "Teil B" as folder name. The index is
 # optional (single-disc rips just use "CD") and may be a letter. A bare number is
@@ -130,23 +131,23 @@ def scan(root: Path, split_dirs: bool = False) -> list[dict]:
             discs = disc_audio(d) if d != root else None
             if discs:
                 items.append({
-                    "path": str(d),
+                    "path": encode_path(str(d)),
-                    "name": d.name,
+                    "name": readable(d.name),
-                    "rel_dir": rel_dir(d.parent),
+                    "rel_dir": readable(rel_dir(d.parent)),
                     "media_type": "audiobook",
                     "is_dir": True,
-                    "files": [str(f) for f in audio + discs],
+                    "files": [encode_path(str(f)) for f in audio + discs],
                     "maybe_separate": False,  # disc folders are one book by definition
                 })
                 return
             if len(audio) > 1:
                 items.append({
-                    "path": str(d),
+                    "path": encode_path(str(d)),
-                    "name": d.name,
+                    "name": readable(d.name),
-                    "rel_dir": rel_dir(d.parent),
+                    "rel_dir": readable(rel_dir(d.parent)),
                     "media_type": "audiobook",
                     "is_dir": True,
-                    "files": [str(f) for f in audio],
+                    "files": [encode_path(str(f)) for f in audio],
                     "maybe_separate": separate_titles(audio),
                 })
                 return
@@ -159,12 +160,12 @@ def scan(root: Path, split_dirs: bool = False) -> list[dict]:
                 mt = media_type_for(entry.suffix)
                 if mt:
                     items.append({
-                        "path": str(entry),
+                        "path": encode_path(str(entry)),
-                        "name": entry.stem,
+                        "name": readable(entry.stem),
-                        "rel_dir": rel_dir(d),
+                        "rel_dir": readable(rel_dir(d)),
                         "media_type": mt,
                         "is_dir": False,
-                        "files": [str(entry)],
+                        "files": [encode_path(str(entry))],
                     })
 
     walk(root)