feat(attachments): fail-soft archive_extractor (zip/cbz/rar/7z)
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -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()
|
||||||
@@ -29,3 +29,7 @@ structlog>=25.5,<26.0
|
|||||||
|
|
||||||
# HTML sanitization for scraped post descriptions (FC-2d provenance)
|
# HTML sanitization for scraped post descriptions (FC-2d provenance)
|
||||||
nh3>=0.2,<0.3
|
nh3>=0.2,<0.3
|
||||||
|
|
||||||
|
# Archive extraction (FC-2d-iii filesystem-import aid)
|
||||||
|
rarfile>=4.2,<5
|
||||||
|
py7zr>=1,<2
|
||||||
|
|||||||
@@ -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
|
||||||
Reference in New Issue
Block a user