diff --git a/backend/app/services/archive_extractor.py b/backend/app/services/archive_extractor.py new file mode 100644 index 0000000..45babef --- /dev/null +++ b/backend/app/services/archive_extractor.py @@ -0,0 +1,57 @@ +"""Best-effort archive member extraction (filesystem-import aid). + +zip/cbz via stdlib zipfile; rar via rarfile (needs an unrar/bsdtar +binary); 7z via py7zr (pure-Python). NEVER raises: on any failure the +context yields an empty list, and the caller still preserves the archive +file itself as a PostAttachment, so nothing is lost. +""" + +import logging +import zipfile +from contextlib import contextmanager +from pathlib import Path +from tempfile import TemporaryDirectory + +log = logging.getLogger(__name__) + +ARCHIVE_EXTS = {".zip", ".cbz", ".rar", ".7z"} + + +def is_archive(path: Path) -> bool: + return Path(path).suffix.lower() in ARCHIVE_EXTS + + +@contextmanager +def extract_archive(path: Path): + """Context manager yielding a list of (member_name, extracted_path) + for every regular-file member. Temp dir is removed on exit. Members + are only ever under the temp dir (libs' safe extractall + rglob from + base). Fail-soft: any error → yields [].""" + tmp = TemporaryDirectory(prefix="fc_arch_") + base = Path(tmp.name) + members: list[tuple[str, Path]] = [] + try: + try: + ext = Path(path).suffix.lower() + if ext in (".zip", ".cbz"): + with zipfile.ZipFile(path) as zf: + zf.extractall(base) + elif ext == ".rar": + import rarfile + + with rarfile.RarFile(path) as rf: + rf.extractall(base) + elif ext == ".7z": + import py7zr + + with py7zr.SevenZipFile(path, "r") as zf: + zf.extractall(base) + for p in sorted(base.rglob("*")): + if p.is_file(): + members.append((str(p.relative_to(base)), p)) + except Exception as exc: + log.warning("archive extract failed for %s: %s", path, exc) + members = [] + yield members + finally: + tmp.cleanup() diff --git a/requirements.txt b/requirements.txt index 72edeb3..a6e009b 100644 --- a/requirements.txt +++ b/requirements.txt @@ -29,3 +29,7 @@ structlog>=25.5,<26.0 # HTML sanitization for scraped post descriptions (FC-2d provenance) nh3>=0.2,<0.3 + +# Archive extraction (FC-2d-iii filesystem-import aid) +rarfile>=4.2,<5 +py7zr>=1,<2 diff --git a/tests/test_archive_extractor.py b/tests/test_archive_extractor.py new file mode 100644 index 0000000..24e3a95 --- /dev/null +++ b/tests/test_archive_extractor.py @@ -0,0 +1,64 @@ +import zipfile +from pathlib import Path + +import pytest + +from backend.app.services.archive_extractor import extract_archive, is_archive + + +def test_is_archive(): + assert is_archive(Path("a.zip")) + assert is_archive(Path("a.CBZ")) + assert is_archive(Path("a.rar")) + assert is_archive(Path("a.7z")) + assert not is_archive(Path("a.jpg")) + + +def _zip(path, entries): + with zipfile.ZipFile(path, "w") as zf: + for name, data in entries.items(): + zf.writestr(name, data) + + +def test_zip_members_enumerated(tmp_path): + z = tmp_path / "set.zip" + _zip(z, {"a.jpg": b"img", "sub/b.png": b"img2", "notes.txt": b"hi"}) + with extract_archive(z) as members: + names = sorted(n for n, _ in members) + assert names == ["a.jpg", "notes.txt", "sub/b.png"] + for _, p in members: + assert p.is_file() + for _, p in members: + assert not p.exists() # temp cleaned on exit + + +def test_cbz_alias(tmp_path): + z = tmp_path / "set.cbz" + _zip(z, {"p1.jpg": b"i"}) + with extract_archive(z) as members: + assert [n for n, _ in members] == ["p1.jpg"] + + +def test_corrupt_archive_yields_nothing(tmp_path): + bad = tmp_path / "broken.zip" + bad.write_bytes(b"not a zip at all") + with extract_archive(bad) as members: + assert members == [] + + +def test_7z_roundtrip(tmp_path): + py7zr = pytest.importorskip("py7zr") + a = tmp_path / "s.7z" + (tmp_path / "x.jpg").write_bytes(b"img") + with py7zr.SevenZipFile(a, "w") as zf: + zf.write(tmp_path / "x.jpg", "x.jpg") + with extract_archive(a) as members: + assert [n for n, _ in members] == ["x.jpg"] + + +def test_rar_corrupt_is_failsoft(tmp_path): + pytest.importorskip("rarfile") + bad = tmp_path / "b.rar" + bad.write_bytes(b"Rar! not really") + with extract_archive(bad) as members: + assert members == [] # invalid / no unrar → fail-soft, nothing lost