<commit_before>import os
from os.path import join as pjoin
import tarfile
import json
import logging
from shutil import copyfile

from .utils import sha1_for_path, canonical_path, makedirs, mygzip
from .defaults import defaults

# TODO: real ignore system
IGNORE = ['.DS_Store']

### Errors ###################################################################

class ZincError(object):
    def __init__(self, code, message):
        self.code = code
        self.message = message

# TODO: might be worst python ever
class ZincErrors(object):
    OK = ZincError(0, "OK")
    INCORRECT_SHA = ZincError(1, "SHA did not match")
    DOES_NOT_EXIST = ZincError(2, "Does not exist")

### Utils

def bundle_name_from_bundle_descriptor(bundle_descriptor):
    return bundle_descriptor[:bundle_descriptor.rfind('-')]

def bundle_version_from_bundle_descritor(bundle_descriptor):
    return int(bundle_descriptor[bundle_descriptor.rfind('-') + 1:])

### ZincOperation ############################################################

class ZincOperation(object):

    def commit():
        pass

### ZincIndex ################################################################

class ZincIndex(object):

    def __init__(self, id=None):
        self.format = defaults['zinc_format']
        self.id = id
        self.bundle_info_by_name = dict()

    def to_json(self):
       return {
				'id' : self.id,
                'bundles' : self.bundle_info_by_name,
                'format' : self.format,
                }

    def write(self, path, gzip=False):
        if self.id is None:
            raise ValueError("catalog id is None") # TODO: better exception?
        index_file = open(path, 'w')
        dict = self.to_json()
        index_file.write(json.dumps(dict))
        index_file.close()
        if gzip:
            gzpath = path + '.gz'
            mygzip(path, gzpath)

    def _get_or_create_bundle_info(self, bundle_name):
        if self.bundle_info_by_name.get(bundle_name) is None:
            self.bundle_info_by_name[bundle_name] = {
                    'versions':[],
                    'distributions':{},
                    }
        return self.bundle_info_by_name.get(bundle_name)

    def add_version_for_bundle(self, bundle_name, version):
        bundle_info = self._get_or_create_bundle_info(bundle_name)
        if version not in bundle_info['versions']:
            bundle_info['versions'].append(version)
            bundle_info['versions'] = sorted(bundle_info['versions'])

    def versions_for_bundle(self, bundle_name):
        return self._get_or_create_bundle_info(bundle_name).get('versions')

    def delete_bundle_version(self, bundle_name, bundle_version):
        bundle_info = self.bundle_info_by_name.get(bundle_name)
        if bundle_info is None:
            raise Exception("Unknown bundle %s" % (bundle_name))
        for distro_name, distro_version in bundle_info['distributions'].iteritems():
            if distro_version == bundle_version:
                raise Exception("bundle '%s' v%d is referenced by the distribution '%s'" 
                        % (bundle_name, bundle_version, distro_name))
        versions = bundle_info['versions']
        if bundle_version in versions:
            versions.remove(bundle_version)
        if len(versions) == 0: # remove info if no more versions
            del self.bundle_info_by_name[bundle_name]
        else:
            bundle_info['versions'] = versions
        
    def distributions_for_bundle(self, bundle_name):
        bundle_info = self.bundle_info_by_name.get(bundle_name)
        if bundle_name is None:
            raise ValueError("Unknown bundle %s" % (bundle_name))
        return bundle_info['distributions']

    def version_for_bundle(self, bundle_name, distro):
        return self.distributions_for_bundle(bundle_name)[distro]

    def update_distribution(self, distribution_name, bundle_name, bundle_version):
        if bundle_version == 'latest':
            bundle_version = self.versions_for_bundle(bundle_name)[-1]
        elif int(bundle_version) not in self.versions_for_bundle(bundle_name):
            raise ValueError("Invalid bundle version")
        bundle_info = self._get_or_create_bundle_info(bundle_name)
        bundle_info['distributions'][distribution_name] = bundle_version

    def delete_distribution(self, distribution_name, bundle_name):
        bundle_info = self.bundle_info_by_name.get(bundle_name)
        if bundle_name is None:
            raise ValueError("Unknown bundle %s" % (bundle_name))
        del bundle_info['distributions'][distribution_name]

def load_index(path):
    index_file = open(path, 'r')
    dict = json.load(index_file)
    index_file.close()
    index = ZincIndex()
    index.id = dict['id']
    index.format = dict['format']
    index.bundle_info_by_name = dict['bundles']
    return index


### ZincConfig ###############################################################

class ZincConfig(object):

    def __init__(self):
        self.gzip_threshhold = 0.85

    def to_json(self):
        d = {}
        if self.gzip_threshhold is not None:
            d['gzip_threshhold'] = self.gzip_threshhold
        return d
   
    def write(self, path):
        config_file = open(path, 'w')
        dict = self.to_json()
        config_file.write(json.dumps(dict))
        config_file.close()

def load_config(path):
    config_file = open(path, 'r')
    dict = json.load(config_file)
    config_file.close()
    config = ZincConfig()
    if dict.get('gzip_threshhold'):<commit_msg>Class method to load a ZincIndex from JSON dict<commit_after>import os
from os.path import join as pjoin
import tarfile
import json
import logging
from shutil import copyfile

from .utils import sha1_for_path, canonical_path, makedirs, mygzip
from .defaults import defaults

# TODO: real ignore system
IGNORE = ['.DS_Store']

### Errors ###################################################################

class ZincError(object):
    def __init__(self, code, message):
        self.code = code
        self.message = message

# TODO: might be worst python ever
class ZincErrors(object):
    OK = ZincError(0, "OK")
    INCORRECT_SHA = ZincError(1, "SHA did not match")
    DOES_NOT_EXIST = ZincError(2, "Does not exist")

### Utils

def bundle_name_from_bundle_descriptor(bundle_descriptor):
    return bundle_descriptor[:bundle_descriptor.rfind('-')]

def bundle_version_from_bundle_descritor(bundle_descriptor):
    return int(bundle_descriptor[bundle_descriptor.rfind('-') + 1:])

### ZincOperation ############################################################

class ZincOperation(object):

    def commit():
        pass

### ZincIndex ################################################################

class ZincIndex(object):

    def __init__(self, id=None):
        self.format = defaults['zinc_format']
        self.id = id
        self.bundle_info_by_name = dict()

    def to_json(self):
       return {
				'id' : self.id,
                'bundles' : self.bundle_info_by_name,
                'format' : self.format,
                }

    @classmethod
    def from_dict(cls, json_dict):
        index = cls()
        index.id = json_dict['id']
        index.format = json_dict['format']
        index.bundle_info_by_name = json_dict['bundles']
        return index

    def write(self, path, gzip=False):
        if self.id is None:
            raise ValueError("catalog id is None") # TODO: better exception?
        index_file = open(path, 'w')
        dict = self.to_json()
        index_file.write(json.dumps(dict))
        index_file.close()
        if gzip:
            gzpath = path + '.gz'
            mygzip(path, gzpath)

    def _get_or_create_bundle_info(self, bundle_name):
        if self.bundle_info_by_name.get(bundle_name) is None:
            self.bundle_info_by_name[bundle_name] = {
                    'versions':[],
                    'distributions':{},
                    }
        return self.bundle_info_by_name.get(bundle_name)

    def add_version_for_bundle(self, bundle_name, version):
        bundle_info = self._get_or_create_bundle_info(bundle_name)
        if version not in bundle_info['versions']:
            bundle_info['versions'].append(version)
            bundle_info['versions'] = sorted(bundle_info['versions'])

    def versions_for_bundle(self, bundle_name):
        return self._get_or_create_bundle_info(bundle_name).get('versions')

    def delete_bundle_version(self, bundle_name, bundle_version):
        bundle_info = self.bundle_info_by_name.get(bundle_name)
        if bundle_info is None:
            raise Exception("Unknown bundle %s" % (bundle_name))
        for distro_name, distro_version in bundle_info['distributions'].iteritems():
            if distro_version == bundle_version:
                raise Exception("bundle '%s' v%d is referenced by the distribution '%s'" 
                        % (bundle_name, bundle_version, distro_name))
        versions = bundle_info['versions']
        if bundle_version in versions:
            versions.remove(bundle_version)
        if len(versions) == 0: # remove info if no more versions
            del self.bundle_info_by_name[bundle_name]
        else:
            bundle_info['versions'] = versions
        
    def distributions_for_bundle(self, bundle_name):
        bundle_info = self.bundle_info_by_name.get(bundle_name)
        if bundle_name is None:
            raise ValueError("Unknown bundle %s" % (bundle_name))
        return bundle_info['distributions']

    def version_for_bundle(self, bundle_name, distro):
        return self.distributions_for_bundle(bundle_name)[distro]

    def update_distribution(self, distribution_name, bundle_name, bundle_version):
        if bundle_version == 'latest':
            bundle_version = self.versions_for_bundle(bundle_name)[-1]
        elif int(bundle_version) not in self.versions_for_bundle(bundle_name):
            raise ValueError("Invalid bundle version")
        bundle_info = self._get_or_create_bundle_info(bundle_name)
        bundle_info['distributions'][distribution_name] = bundle_version

    def delete_distribution(self, distribution_name, bundle_name):
        bundle_info = self.bundle_info_by_name.get(bundle_name)
        if bundle_name is None:
            raise ValueError("Unknown bundle %s" % (bundle_name))
        del bundle_info['distributions'][distribution_name]

def load_index(path):
    with open(path, 'r') as index_file:
        json_dict = json.load(index_file)
    return ZincIndex.from_dict(json_dict)


### ZincConfig ###############################################################

class ZincConfig(object):

    def __init__(self):
        self.gzip_threshhold = 0.85

    def to_json(self):
        d = {}
        if self.gzip_threshhold is not None:
            d['gzip_threshhold'] = self.gzip_threshhold
        return d
   
    def write(self, path):
        config_file = open(path, 'w')
        dict = self.to_json()
        config_file.write(json.dumps(dict))
        config_file.close()

def load_config(path):
    config_file = open(path, 'r')
    dict = json.load(config_file)
    config_file.close()
    config = ZincConfig()
    if dict.get('gzip_threshhold'):
{'fraction_non_alphanumeric': 0.1405429461504228, 'fraction_numerical': 0.0017801513128615932, 'mean_word_length': 3.100729927007299, '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': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3580245', 'n_tokens_mistral': 3311, 'n_tokens_neox': 3240, 'n_words': 702}