bvandeusen
5bb25245a5
Patreon attachment downloads land with sanitized URL-blob filenames
(01_https___www.patreon.com_media-u_v3_<id>) whose Path.suffix is junk, never
.zip — so the extension-only is_archive() filed them as opaque PostAttachments
and never extracted them ("No images attached to this post").
Add archive_extractor.detect_archive_format() — extension first, then magic-byte
sniff (zipfile.is_zipfile + RAR/7z signatures). is_archive(), extract_archive(),
and safe_probe._inspect_archive() (the bomb-guard) all route through it, so a
mis-named/extension-less archive is now detected, bomb-guarded, integrity-tested,
AND extracted regardless of filename. Stops new ones; part 2 re-extracts the
already-imported backlog.
Tests: mis-named zip detected + extracted; non-archive dotted name not
misdetected; _inspect_archive on a mis-named zip; signature updated.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
94 lines
3.1 KiB
Python
94 lines
3.1 KiB
Python
"""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"}
|
|
|
|
# Magic-byte signatures, so an archive with a mangled / extension-less filename
|
|
# is still recognised. Patreon attachment download URLs sanitize to names like
|
|
# `01_https___www.patreon.com_media-u_v3_131083093`, whose `Path.suffix` is junk
|
|
# (`.com_media-u_v3_131083093`), never `.zip` — an extension-only gate filed
|
|
# those as opaque PostAttachments and NEVER extracted them (operator-flagged
|
|
# 2026-06-06). Detection is by extension first (cheap), then header sniff.
|
|
_RAR_MAGIC = b"Rar!\x1a\x07"
|
|
_7Z_MAGIC = b"7z\xbc\xaf\x27\x1c"
|
|
|
|
|
|
def detect_archive_format(path: Path) -> str | None:
|
|
"""Return "zip" | "rar" | "7z" for an archive, else None.
|
|
|
|
Trusts a known extension first, then falls back to magic-byte sniffing so a
|
|
mis-named or extension-less archive is still handled. (zip covers .cbz too.)
|
|
"""
|
|
ext = Path(path).suffix.lower()
|
|
if ext in (".zip", ".cbz"):
|
|
return "zip"
|
|
if ext == ".rar":
|
|
return "rar"
|
|
if ext == ".7z":
|
|
return "7z"
|
|
try:
|
|
if zipfile.is_zipfile(path):
|
|
return "zip"
|
|
with open(path, "rb") as fh:
|
|
head = fh.read(8)
|
|
except OSError:
|
|
return None
|
|
if head.startswith(_RAR_MAGIC):
|
|
return "rar"
|
|
if head.startswith(_7Z_MAGIC):
|
|
return "7z"
|
|
return None
|
|
|
|
|
|
def is_archive(path: Path) -> bool:
|
|
return detect_archive_format(path) is not None
|
|
|
|
|
|
@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:
|
|
fmt = detect_archive_format(path)
|
|
if fmt == "zip":
|
|
with zipfile.ZipFile(path) as zf:
|
|
zf.extractall(base)
|
|
elif fmt == "rar":
|
|
import rarfile
|
|
|
|
with rarfile.RarFile(path) as rf:
|
|
rf.extractall(base)
|
|
elif fmt == "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()
|