FabledCurator/backend/app/services/archive_extractor.py

"""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


# Files that only a program ships: executables and libraries, and the packed
# data and compiled scripts of the common game engines (Ren'Py, Unity, Godot,
# RPG Maker, Python builds). An archive holding any of them is a game or app
# build. Its images are sprites, UI and CGs bundled for the program, not art
# for the library, so the archive is kept whole as an attachment.
SOFTWARE_EXTS = frozenset({
    ".exe", ".msi", ".dll", ".so", ".dylib", ".apk",
    ".rpa", ".rpyc", ".rpy", ".pyc", ".pyo",
    ".pak", ".pck", ".unity3d", ".assets", ".ress",
    ".rgssad", ".rgss2a", ".rgss3a", ".rxdata", ".rvdata", ".rvdata2",
})


def member_names(path: Path) -> list[str]:
    """Every member name in the archive, read from its directory without
    decompressing anything. Empty when the archive can't be listed."""
    try:
        fmt = detect_archive_format(path)
        if fmt == "zip":
            with zipfile.ZipFile(path) as zf:
                return zf.namelist()
        if fmt == "rar":
            import rarfile

            with rarfile.RarFile(path) as rf:
                return rf.namelist()
        if fmt == "7z":
            import py7zr

            with py7zr.SevenZipFile(path, "r") as zf:
                return zf.getnames()
    except Exception as exc:
        log.warning("archive listing failed for %s: %s", path, exc)
    return []


def software_marker(path: Path) -> str | None:
    """The first member that marks the archive as a software build, or None."""
    for name in member_names(path):
        if Path(name).suffix.lower() in SOFTWARE_EXTS:
            return name
    return 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()