<commit_before>
import getpass
import os
import re
import time
import types

from itertools import cycle
from concurrent.futures import ThreadPoolExecutor, as_completed

from libdlt.util import util
from libdlt.depot import Depot
from libdlt.protocol import factory
from libdlt.settings import DEPOT_TYPES, THREADS, COPIES, BLOCKSIZE, TIMEOUT
from unis.models import Exnode, Service
from unis.runtime import Runtime

class Session(object):
    def __init__(self, url, depots, bs=BLOCKSIZE, timeout=TIMEOUT, **kwargs):
        self._validate_url(url)
        self._runtime = Runtime(url, defer_update=True, **kwargs)
        self._do_flush = True
        self._blocksize = int(util.human2bytes(bs))
        self._timeout = timeout
        self._plan = cycle
        self._depots = {}
        
        if not depots:
            for depot in self._runtime.services.where(lambda x: x.serviceType in DEPOT_TYPES):
                self._depots[depot.selfRef] = depot
        elif isinstance(depots, str):
            self._validate_url(depots)
            with Runtime(depots) as rt:
                for depot in rt.services.where(lambda x: x.serviceType in DEPOT_TYPES):
                    self._depots[depot.selfRef] = depot
        elif isinstance(depots, dict):
            for name, depot in depots.items():
                if isinstance(depot, dict):
                    self._depots[name] = Service(depot)
        else:
            raise ValueError("depots argument must contain a list of depot description objects or a valid unis url")

        if not len(self._depots):
            raise ValueError("No depots found for session, unable to continue")
            
    def upload(self, filepath, folder=None, copies=COPIES, duration=None):
        def _chunked(fh, bs, size):
            offset = 0
            while True:
                bs = min(bs, size - offset)
                data = fh.read(bs)
                if not data:
                    return
                yield (offset, bs, data)
                offset += bs
            
        if isinstance(folder, str):
            do_flush = self._do_flush
            self._do_flush = False
            folder = self.mkdir(folder)
            self._do_flush = do_flush
        
        stat = os.stat(filepath)
        ex = Exnode({ "parent": folder, "created": int(time.time()), "mode": "file", "size": stat.st_size,
                      "permission": format(stat.st_mode & 0o0777, 'o'), "owner": getpass.getuser(),
                      "name": os.path.basename(filepath) })
        ex.group = ex.owner
        ex.updated = ex.created
        self._runtime.insert(ex, commit=True)
        
        executor = ThreadPoolExecutor(max_workers=THREADS)
        futures = []
        time_s = time.time()
        
        depots = self._get_plan(self._depots)
        with open(filepath, "rb") as fh:
            for offset, size, data in _chunked(fh, self._blocksize, ex.size):
                for n in range(copies):
                    d = Depot(next(depots))
                    futures.append(executor.submit(factory.makeAllocation, data, offset, d,
                                                   **self._depots[d.endpoint].to_JSON()))
                    
        for future in as_completed(futures):
            ext = future.result().GetMetadata()
            self._runtime.insert(ext, commit=True)
            ext.parent = ex
            ex.extents.append(ext)

        time_e = time.time()
            
        if self._do_flush:
            self._runtime.flush()
        return (time.time() - time_e, ex)
    
    def download(self, href, filepath, length=0, offset=0):
        def _download_chunk(ls):
            for ext in ls:
                try:
                    alloc = factory.buildAllocation(ext)
                    d = Depot(ext.location)
                    return alloc.Read(**self._depots[d.endpoint].to_JSON())
                except Exception as exp:
                    print ("READ Error: {}".format(exp))
                    pass
            return None

        self._validate_url(href)
        time_s = time.time()
        ex = self._runtime.find(href)
        exts = ex.extents
        chunks = {}
        
        # bin extents
        for ext in exts:
            if ext.offset in chunks:
                chunks[ext.offset].append(ext)
            else:
                chunks[ext.offset] = [ext]

        if not filepath:
            filepath = ex.name        
                
        with open(filepath, "wb") as fh:
            with ThreadPoolExecutor(max_workers=THREADS) as executor:
                for ext, data in zip(exts, executor.map(_download_chunk, chunks.values())):
                    if data:
                        fh.seek(ext.offset)
                        fh.write(data)
        
        return (time.time() - time_s, ex)
        
    def mkdir(self, path):
        def _traverse(ls, obj):
            if not ls:
                return ([], obj)
            for child in obj.children:
                if child.name == ls[0]:
                    return _traverse(ls[1:], child)
            return (ls, obj)
        
        path = list(filter(None, path.split('/')))
        if not path:
            return
        
        folder_ls = list(self._runtime.exnodes.where({"name": path[0], "mode": "directory", "parent": None}))
        if folder_ls:
            path, root = _traverse(path, folder_ls[0])
        else:
            root = None
        
        for folder in path:
            new_folder = Exnode({"name": folder, "parent": root, "mode": "directory"})
            self._runtime.insert(new_folder, commit=True)
            if root:
                if not hasattr(root, "children"):
                    root.children = [new_folder]
                    root.commit("children")
                else:
                    root.children.append(new_folder)
                    
            root = new_folder
        
        if self._do_flush:
            self._runtime.flush()
        
        return root
    
    def setDistributionPlan(self, plan):<commit_msg>Trying to make mkdir work.<commit_after>
import getpass
import os
import re
import time
import types

from itertools import cycle
from concurrent.futures import ThreadPoolExecutor, as_completed

from libdlt.util import util
from libdlt.depot import Depot
from libdlt.protocol import factory
from libdlt.settings import DEPOT_TYPES, THREADS, COPIES, BLOCKSIZE, TIMEOUT
from unis.models import Exnode, Service
from unis.runtime import Runtime

class Session(object):
    def __init__(self, url, depots, bs=BLOCKSIZE, timeout=TIMEOUT, **kwargs):
        self._validate_url(url)
        self._runtime = Runtime(url, defer_update=True, **kwargs)
        self._do_flush = True
        self._blocksize = int(util.human2bytes(bs))
        self._timeout = timeout
        self._plan = cycle
        self._depots = {}
        
        if not depots:
            for depot in self._runtime.services.where(lambda x: x.serviceType in DEPOT_TYPES):
                self._depots[depot.selfRef] = depot
        elif isinstance(depots, str):
            self._validate_url(depots)
            with Runtime(depots) as rt:
                for depot in rt.services.where(lambda x: x.serviceType in DEPOT_TYPES):
                    self._depots[depot.selfRef] = depot
        elif isinstance(depots, dict):
            for name, depot in depots.items():
                if isinstance(depot, dict):
                    self._depots[name] = Service(depot)
        else:
            raise ValueError("depots argument must contain a list of depot description objects or a valid unis url")

        if not len(self._depots):
            raise ValueError("No depots found for session, unable to continue")
            
    def upload(self, filepath, folder=None, copies=COPIES, duration=None):
        def _chunked(fh, bs, size):
            offset = 0
            while True:
                bs = min(bs, size - offset)
                data = fh.read(bs)
                if not data:
                    return
                yield (offset, bs, data)
                offset += bs
            
        if isinstance(folder, str):
            do_flush = self._do_flush
            self._do_flush = False
            folder = self.mkdir(folder)
            self._do_flush = do_flush
        
        stat = os.stat(filepath)
        ex = Exnode({ "parent": folder, "created": int(time.time()), "mode": "file", "size": stat.st_size,
                      "permission": format(stat.st_mode & 0o0777, 'o'), "owner": getpass.getuser(),
                      "name": os.path.basename(filepath) })
        ex.group = ex.owner
        ex.updated = ex.created
        self._runtime.insert(ex, commit=True)
        
        executor = ThreadPoolExecutor(max_workers=THREADS)
        futures = []
        time_s = time.time()
        
        depots = self._get_plan(self._depots)
        with open(filepath, "rb") as fh:
            for offset, size, data in _chunked(fh, self._blocksize, ex.size):
                for n in range(copies):
                    d = Depot(next(depots))
                    futures.append(executor.submit(factory.makeAllocation, data, offset, d,
                                                   **self._depots[d.endpoint].to_JSON()))
                    
        for future in as_completed(futures):
            ext = future.result().GetMetadata()
            self._runtime.insert(ext, commit=True)
            ext.parent = ex
            ex.extents.append(ext)

        time_e = time.time()
            
        if self._do_flush:
            self._runtime.flush()
        return (time.time() - time_e, ex)
    
    def download(self, href, filepath, length=0, offset=0):
        def _download_chunk(ls):
            for ext in ls:
                try:
                    alloc = factory.buildAllocation(ext)
                    d = Depot(ext.location)
                    return alloc.Read(**self._depots[d.endpoint].to_JSON())
                except Exception as exp:
                    print ("READ Error: {}".format(exp))
                    pass
            return None

        self._validate_url(href)
        time_s = time.time()
        ex = self._runtime.find(href)
        exts = ex.extents
        chunks = {}
        
        # bin extents
        for ext in exts:
            if ext.offset in chunks:
                chunks[ext.offset].append(ext)
            else:
                chunks[ext.offset] = [ext]

        if not filepath:
            filepath = ex.name        
                
        with open(filepath, "wb") as fh:
            with ThreadPoolExecutor(max_workers=THREADS) as executor:
                for ext, data in zip(exts, executor.map(_download_chunk, chunks.values())):
                    if data:
                        fh.seek(ext.offset)
                        fh.write(data)
        
        return (time.time() - time_s, ex)
        
    def mkdir(self, path):
        def _traverse(ls, obj):
            if not ls:
                return ([], obj)
            for child in obj.children:
                if child.name == ls[0]:
                    return _traverse(ls[1:], child)
            return (ls, obj)
        
        path = list(filter(None, path.split('/')))
        if not path:
            return
        
        folder_ls = list(self._runtime.exnodes.where({"name": path[0], "mode": "directory", "parent": None}))
        if folder_ls:
            path, root = _traverse(path, folder_ls[0])
        else:
            root = None
        
        for folder in path:
            new_folder = Exnode({"name": folder, "parent": root, "mode": "directory",
                                 "created": int(time.time()),
                                 "permission": format(0o0755, 'o'),
                                 "owner": getpass.getuser()})
            new_folder.group = new_folder.owner
            new_folder.modified = new_folder.created
            self._runtime.insert(new_folder, commit=True)
            if root:
                if not hasattr(root, "children"):
                    root.children = [new_folder]
                    root.commit("children")
                else:
                    root.children.append(new_folder)
                new_folder.parent = root
                    
            root = new_folder
        
        if self._do_flush:
            self._runtime.flush()
        
        return root
    
    def setDistributionPlan(self, plan):
{'fraction_non_alphanumeric': 0.09835542072489223, 'fraction_numerical': 0.002474852307201022, 'mean_word_length': 1.5440698619008937, 'pattern_counts': {'":': 29, '<': 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': 13, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4074934', 'n_tokens_mistral': 3513, 'n_tokens_neox': 3398, 'n_words': 866}