<commit_before>import os.path
import shutil
import json
from pathlib import Path

from glob import glob
from zipfile import ZipFile
from urllib.parse import urljoin

import requests

from fac.files import JSONFile
from fac.utils import JSONDict
from fac.api import AuthError, ModNotFoundError


class Mod:
    location = None

    def __init__(self, manager, location):
        self.manager = manager
        self.location = location

    def get_enabled(self):
        return self.manager.is_mod_enabled(self.name)

    def set_enabled(self, val):
        self.manager.set_mod_enabled(self.name, val)

    enabled = property(get_enabled, set_enabled)

    @property
    def name(self):
        return self.info.name

    @property
    def version(self):
        return self.info.version

    def _check_valid(self):
        expected_basename = "%s_%s" % (self.name, self.version)

        assert self.basename == expected_basename, \
            "Invalid file name %s, expected %s" % (
                    self.basename,
                    expected_basename
            )

    @classmethod
    def _find(cls, pattern, manager, name, version):
        name = name or '*'
        version = version or '*'

        files = glob(
            os.path.join(
                manager.config.mods_path,
                pattern % (name, version)
            )
        )
        for file in files:
            try:
                mod = cls(manager, file)
                yield mod
            except Exception as ex:
                print('Warning: invalid mod %s: %s' % (file, ex))


class ZippedMod(Mod):
    packed = True

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.basename = os.path.splitext(
                os.path.basename(
                    self.location
                )
        )[0]
        self._check_valid()

    def remove(self):
        print('Removing file: %s' % self.location)
        os.remove(self.location)

    @property
    def info(self):
        with ZipFile(self.location) as f:
            info = json.loads(
                    f.read(
                        '%s/info.json' % self.basename,
                        ).decode('utf-8'),
                    )
            return JSONDict(info)

    def unpack(self):
        mod_directory = self.manager.config.mods_path
        unpacked_location = os.path.join(mod_directory, self.basename)

        print('Unpacking: %s' % self.location)

        with ZipFile(self.location) as f:
            os.makedirs(unpacked_location)

            for info in f.infolist():
                if not info.filename.startswith(self.basename + '/'):
                    print("Warning: out-of-directory file %s ignored" % (
                        info.filename))
                    continue
                f.extract(info, mod_directory)

        self.remove()
        return UnpackedMod(self.manager, unpacked_location)

    @classmethod
    def find(cls, *args, **kwargs):
        return cls._find("%s_%s.zip", *args, **kwargs)


class UnpackedMod(Mod):
    packed = False

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.basename = os.path.basename(
            os.path.realpath(
                self.location
            )
        )
        self._check_valid()

    def remove(self):
        print('Removing directory: %s' % self.location)
        shutil.rmtree(self.location)

    @property
    def info(self):
        path = os.path.join(self.location, 'info.json')
        return JSONFile(path)

    def pack(self):
        packed_location = os.path.join(
            self.manager.config.mods_path,
            self.basename + '.zip'
        )

        print('Packing: %s' % self.location)

        if os.path.exists(packed_location):
            raise Exception("File already exists: %s" % packed_location)

        with ZipFile(packed_location, "w") as f:
            for root, dirs, files in os.walk(self.location):
                zip_root = Path(root).relative_to(
                    self.manager.config.mods_path).as_posix()

                for file_name in files:
                    f.write(
                        '%s/%s' % (root, file_name),
                        '%s/%s' % (zip_root, file_name),
                    )

        self.remove()

        return ZippedMod(self.manager, packed_location)

    @classmethod
    def find(cls, *args, **kwargs):<commit_msg>Use the first zip entry to determine the toplevel zip directory<commit_after>import os.path
import shutil
import json
from pathlib import Path

from glob import glob
from zipfile import ZipFile
from urllib.parse import urljoin

import requests

from fac.files import JSONFile
from fac.utils import JSONDict
from fac.api import AuthError, ModNotFoundError


class Mod:
    location = None

    def __init__(self, manager, location):
        self.manager = manager
        self.location = location

    def get_enabled(self):
        return self.manager.is_mod_enabled(self.name)

    def set_enabled(self, val):
        self.manager.set_mod_enabled(self.name, val)

    enabled = property(get_enabled, set_enabled)

    @property
    def name(self):
        return self.info.name

    @property
    def version(self):
        return self.info.version

    def _check_valid(self):
        expected_basename = "%s_%s" % (self.name, self.version)

        assert self.basename == expected_basename, \
            "Invalid file name %s, expected %s" % (
                    self.basename,
                    expected_basename
            )

    @classmethod
    def _find(cls, pattern, manager, name, version):
        name = name or '*'
        version = version or '*'

        files = glob(
            os.path.join(
                manager.config.mods_path,
                pattern % (name, version)
            )
        )
        for file in files:
            try:
                mod = cls(manager, file)
                yield mod
            except Exception as ex:
                print('Warning: invalid mod %s: %s' % (file, ex))


class ZippedMod(Mod):
    """
    A zipped mod consists of a strictly named name_version.zip file.

    The top-level directory of the first zip entry must
    contain the info.json file and other mod files.

    Any subsequent top-level directories will be ignored.
    """

    packed = True

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.basename = os.path.splitext(
                os.path.basename(
                    self.location
                )
        )[0]
        self._read_info()
        self._check_valid()

    def remove(self):
        print('Removing file: %s' % self.location)
        os.remove(self.location)

    def _read_info(self):
        with ZipFile(self.location) as f:
            first_entry = f.namelist()[0]
            self.toplevel = first_entry.split('/')[0]

            if not self.toplevel:
                raise Exception('Could not find a top-level directory')

            info = json.loads(
                f.read(
                    '%s/info.json' % self.toplevel,
                ).decode('utf-8'),
            )

            self.info = JSONDict(info)

    def unpack(self):
        mod_directory = self.manager.config.mods_path
        unpacked_location = os.path.join(mod_directory, self.basename)

        print('Unpacking: %s' % self.location)

        with ZipFile(self.location) as f:
            os.makedirs(unpacked_location)

            for arcname in f.namelist():
                if not arcname.startswith(self.toplevel + '/'):
                    print("Warning: out-of-place file %s ignored" % (
                        arcname))
                    continue

                dest = arcname[len(self.toplevel) + 1:]
                dest = self._sanitize_arcname(dest)
                dest = os.path.join(unpacked_location, dest)
                self._extract_member(f, arcname, dest)

        self.remove()
        return UnpackedMod(self.manager, unpacked_location)

    def _sanitize_arcname(self, arcname):
        arcname = arcname.replace('/', os.path.sep)

        if os.path.altsep:
            arcname = arcname.replace(os.path.altsep, os.path.sep)
        # interpret absolute pathname as relative, remove drive letter or
        # UNC path, redundant separators, "." and ".." components.
        arcname = os.path.splitdrive(arcname)[1]
        invalid_path_parts = ('', os.path.curdir, os.path.pardir)
        arcname = os.path.sep.join(x for x in arcname.split(os.path.sep)
                                   if x not in invalid_path_parts)
        if os.path.sep == '\\':
            # filter illegal characters on Windows
            arcname = ZipFile._sanitize_windows_name(arcname, os.path.sep)

        return arcname

    def _extract_member(self, zipfile, arcname, dest):
        # Create all upper directories if necessary.
        upperdirs = os.path.dirname(dest)
        if upperdirs and not os.path.exists(upperdirs):
            os.makedirs(upperdirs)

        if arcname[-1] == '/':
            if not os.path.isdir(dest):
                os.mkdir(dest)
            return

        with zipfile.open(arcname) as source, \
                open(dest, "wb") as target:
            shutil.copyfileobj(source, target)

    @classmethod
    def find(cls, *args, **kwargs):
        return cls._find("%s_%s.zip", *args, **kwargs)


class UnpackedMod(Mod):
    packed = False

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.basename = os.path.basename(
            os.path.realpath(
                self.location
            )
        )
        self._read_info()
        self._check_valid()

    def remove(self):
        print('Removing directory: %s' % self.location)
        shutil.rmtree(self.location)

    def _read_info(self):
        path = os.path.join(self.location, 'info.json')
        self.info = JSONFile(path)

    def pack(self):
        packed_location = os.path.join(
            self.manager.config.mods_path,
            self.basename + '.zip'
        )

        print('Packing: %s' % self.location)

        if os.path.exists(packed_location):
            raise Exception("File already exists: %s" % packed_location)

        with ZipFile(packed_location, "w") as f:
            for root, dirs, files in os.walk(self.location):
                zip_root = Path(root).relative_to(
                    self.manager.config.mods_path).as_posix()

                for file_name in files:
                    f.write(
                        '%s/%s' % (root, file_name),
                        '%s/%s' % (zip_root, file_name),
                    )

        self.remove()

        return ZippedMod(self.manager, packed_location)

    @classmethod
    def find(cls, *args, **kwargs):
{'fraction_non_alphanumeric': 0.10526800515748756, 'fraction_numerical': 0.0008288819303739178, 'mean_word_length': 1.7701530612244898, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 1, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '285139', 'n_tokens_mistral': 3224, 'n_tokens_neox': 3071, 'n_words': 742}