<commit_before>import sys, re, math
import os, glob
import time
import copy
import importlib
import pickle
from datetime import datetime
from pprint import pformat
import asyncio
from functools import partial
import glob, random
import aiocron

from .mapper import TransparentMapper
from ..dataload.uploader import ResourceNotReady
from .differ import set_pending_to_diff
from ..databuild.backend import SourceDocMongoBackend, TargetDocMongoBackend, \
                                LinkTargetDocMongoBackend
from biothings.utils.common import timesofar, iter_n, get_timestamp, dotdict, \
                                   dump, rmdashfr, loadobj, open_compressed_file, \
                                   get_class_from_classpath, find_classes_subclassing
from biothings.utils.mongo import doc_feeder, id_feeder
from biothings.utils.loggers import get_logger
from biothings.utils.manager import BaseManager, ManagerError
from biothings.utils.dataload import update_dict_recur, merge_struct
import biothings.utils.mongo as mongo
from biothings.utils.hub_db import get_source_fullname, get_src_build_config, \
                                   get_src_build, get_src_dump, get_src_master
from biothings import config as btconfig
from biothings.hub import UPLOADER_CATEGORY, BUILDER_CATEGORY
from .backend import create_backend

logging = btconfig.logger

class BuilderException(Exception):
    pass
class ResumeException(Exception):
    pass


class DataBuilder(object):
    """
    Generic data builder.
    """

    keep_archive = 10 # number of archived collection to keep. Oldest get dropped first.

    def __init__(self, build_name, source_backend, target_backend, log_folder,
                 doc_root_key="root", mappers=[], default_mapper_class=TransparentMapper,
                 sources=None, target_name=None,**kwargs):
        self.init_state()
        self.build_name = build_name
        self.sources = sources
        self.target_name = target_name
        self._partial_source_backend = None
        self._partial_target_backend = None
        if type(source_backend) == partial:
            self._partial_source_backend = source_backend
        else:
            self._state["source_backend"] = source_backend
        if type(target_backend) == partial:
            self._partial_target_backend = target_backend
        else:
            self._state["target_backend"] = target_backend
        # doc_root_key is a key name within src_build_config doc.
        # it's a list of datasources that are able to create a document
        # even it doesn't exist. It root documents list is not empty, then
        # any other datasources not listed there won't be able to create
        # a document, only will they able to update it.
        # If no root documets, any datasources can create/update a doc
        # and thus there's no priority nor limitations
        # note: negations can be used, like "!source1". meaning source1 is not
        # root document datasource.
        # Usefull to express; "all resources except source1"
        self.doc_root_key = doc_root_key
        # overall merge start time
        self.t0 = time.time()
        # step merge start time
        self.ti = time.time()
        self.logfile = None
        self.log_folder = log_folder
        self.mappers = {}
        self.timestamp = datetime.now()
        self.merge_stats = {} # keep track of cnt per source, etc...
        self.src_meta = {} # sources involved in this build (includes versions)
        self.stats = {} # can be customized
        self.mapping = {} # ES mapping (merged from src_master's docs)

        for mapper in mappers + [default_mapper_class()]:
            self.mappers[mapper.name] = mapper

        self.step = kwargs.get("step",10000)
        self.prepared = False

    def init_state(self):
        self._state = {
                "logger" : None,
                "source_backend" : None,
                "target_backend" : None,
                "build_config" : None,
        }
    @property
    def logger(self):
        if not self._state["logger"]:
            self.prepare()
        return self._state["logger"]
    @property
    def source_backend(self):
        if self._state["source_backend"] is None:
            self.prepare()
            self._state["build_config"] = self._state["source_backend"].get_build_configuration(self.build_name)
            self._state["source_backend"].validate_sources(self.sources)
        return self._state["source_backend"]
    @property
    def target_backend(self):
        if self._state["target_backend"] is None:
            self.prepare()
        return self._state["target_backend"]
    @property
    def build_config(self):
        self._state["build_config"] = self.source_backend.get_build_configuration(self.build_name)
        return self._state["build_config"]
    @logger.setter
    def logger(self, value):
        self._state["logger"] = value
    @build_config.setter
    def build_config(self, value):
        self._state["build_config"] = value

    def prepare(self,state={}):
        if self.prepared:
            return
        if state:
            # let's be explicit, _state takes what it wants
            for k in self._state:
                self._state[k] = state[k]
            return
        if self._partial_source_backend:
            self._state["source_backend"] = self._partial_source_backend()
        if self._partial_target_backend:
            self._state["target_backend"] = self._partial_target_backend()
        self.setup()
        self.setup_log()
        self.prepared = True

    def unprepare(self):
        """
        reset anything that's not pickable (so self can be pickled)
        return what's been reset as a dict, so self can be restored
        once pickled
        """
        # TODO: use copy ?
        state = {
            "logger" : self._state["logger"],
            "source_backend" : self._state["source_backend"],
            "target_backend" : self._state["target_backend"],
            "build_config" : self._state["build_config"],
        }
        for k in state:
            self._state[k] = None
        self.prepared = False
        return state

    def get_predicates(self):
        def no_uploader_running(job_manager):
            """Uploaders could change the data to be merged..."""
            return len([j for j in job_manager.jobs.values() if j["category"] == UPLOADER_CATEGORY]) == 0
        #def no_merger_running():
        #    """
        #    Mergers use cache files, if more than one running and caches need to be built
        #    both would try to write on the same cache file
        #    """
        #    return len([j for j in job_manager.jobs.values() if j["category"] == BUILDER_CATEGORY]) == 0
        return [no_uploader_running]

    def get_pinfo(self):
        """
        Return dict containing information about the current process
        (used to report in the hub)
        """
        pinfo = {"category" : BUILDER_CATEGORY,
                "source" : "%s:%s" % (self.build_name,self.target_backend.target_name),
                "step" : "",
                "description" : ""}
        preds = self.get_predicates()
        if preds:
            pinfo["__predicates__"] = preds
        return pinfo

    def setup_log(self):
        self.logger,_ = get_logger('build_%s' % self.build_name)

    def check_ready(self,force=False):
        if force:
            # don't even bother
            return
        src_build_config = self.source_backend.build_config
        src_dump = self.source_backend.dump
        _cfg = src_build_config.find_one({'_id': self.build_config['name']})
        # check if all resources are uploaded
        for src_name in _cfg["sources"]:
            fullname = get_source_fullname(src_name)
            if not fullname:
                raise ResourceNotReady("Can't find source '%s'" % src_name)
            main_name = fullname.split(".")[0]
            src_doc = src_dump.find_one({"_id":main_name})
            if not src_doc:
                raise ResourceNotReady("Missing information for source '%s' to start merging" % src_name)
            if not src_doc.get("upload",{}).get("jobs",{}).get(src_name,{}).get("status") == "success":
                raise ResourceNotReady("No successful upload found for resource '%s'" % src_name)

    def get_build_version(self):
        """
        Generate an arbitrary major build version. Default is using a timestamp (YYMMDD)
        '.' char isn't allowed in build version as it's reserved for minor versions
        """
        d = datetime.fromtimestamp(self.t0)
        return "%d%02d%02d" % (d.year,d.month,d.day)

    def register_status(self,status,transient=False,init=False,**extra):
        """
        Register current build status. A build status is a record in src_build
        The key used in this dict the target_name. Then, any operation
        acting on this target_name is registered in a "jobs" list.
        """
        assert self.build_config, "build_config needs to be specified first"
        # get it from source_backend, kind of weird...
        src_build = self.source_backend.build
        src_build_config = self.source_backend.build_config
        all_sources = self.build_config.get("sources",[])
        target_name = "%s" % self.target_backend.target_name
        backend_url = self.target_backend.get_backend_url()
        build_info = {
                '_id' : target_name,
                # TODO: deprecate target_backend & target_name, use backed_url instead
                'target_backend': self.target_backend.name,
                'target_name': target_name,
                'backend_url' : backend_url,
                'build_config': self.build_config,
                # these are all the sources required to build target
                # (not just the ones being processed, those are registered in jobs
                'sources' : all_sources, 
                }
        job_info = {
                'status': status,
                'step_started_at': datetime.now(),
                'logfile': self.logfile,
                }
        if transient:
            # record some "in-progress" information
            job_info['pid'] = os.getpid()
        else:
            # only register time when it's a final state
            job_info["time"] = timesofar(self.ti)
            t1 = round(time.time() - self.ti, 0)
            job_info["time_in_s"] = t1
        if "build" in extra:
            build_info.update(extra["build"])
        if "job" in extra:
            job_info.update(extra["job"])
        # create a new build entry in "build" dict if none exists
        build = src_build.find_one({'_id': target_name})
        if not build:
            # first record for target_name, keep a timestamp
            build_info["started_at"] = datetime.fromtimestamp(self.t0)
            build_info["jobs"] = []
            src_build.insert_one(build_info)
        if init:
            # init timer for this step
            self.ti = time.time()
            src_build.update({'_id': target_name}, {"$push": {'jobs': job_info}})
            # now refresh/sync
            build = src_build.find_one({'_id': target_name})
        else:
            # merge extra at root level
            # (to keep building data...) and update the last one
            # (it's been properly created before when init=True)
            build["jobs"] and build["jobs"][-1].update(job_info)
            # build_info is common to all jobs, so we want to keep
            # any existing data (well... except if it's explicitely specified)
            def merge_build_info(target,d):
                if "__REPLACE__" in d.keys():
                    d.pop("__REPLACE__")
                    target = d
                else:
                    for k,v in d.items():
                        if type(v) == dict:
                            if k in target:
                                target[k] = merge_build_info(target[k],v) 
                            else:
                                v.pop("__REPLACE__",None)
                                # merge v with "nothing" just to make sure to remove any "__REPLACE__"
                                v = merge_build_info({},v)
                                target[k] = v
                        else:
                            target[k] = v
                return target
            build = merge_build_info(build,build_info)
            src_build.replace_one({"_id" : build["_id"]}, build)

    def clean_old_collections(self):
        # use target_name is given, otherwise build name will be used 
        # as collection name prefix, so they should start like that
        prefix = "%s_" % (self.target_name or self.build_name)
        db = mongo.get_target_db()
        cols = [c for c in db.collection_names() if c.startswith(prefix)]
        # timestamp is what's after _archive_, YYYYMMDD, so we can sort it safely
        cols = sorted(cols,reverse=True)
        to_drop = cols[self.keep_archive:]
        for colname in to_drop:
            self.logger.info("Cleaning old archive collection '%s'" % colname)
            db[colname].drop()

    def init_mapper(self,mapper_name):
        if self.mappers[mapper_name].need_load():
            if mapper_name is None:
                self.logger.info("Initializing default mapper")
            else:
                self.logger.info("Initializing mapper name '%s'" % mapper_name)
            self.mappers[mapper_name].load()

    def generate_document_query(self, src_name):
        return None

    def get_root_document_sources(self):
        root_srcs = self.build_config.get(self.doc_root_key,[]) or []
        # check for "not this resource" and adjust the list
        none_root_srcs = [src.replace("!","") for src in root_srcs if src.startswith("!")]
        if none_root_srcs:
            if len(none_root_srcs) != len(root_srcs):
                raise BuilderException("If using '!' operator, all datasources must use it (cannot mix), got: %s" % \
                        (repr(root_srcs)))
            # ok, grab sources for this build, 
            srcs = self.build_config.get("sources",[])
            root_srcs = list(set(srcs).difference(set(none_root_srcs)))
            #self.logger.info("'except root' sources %s resolves to root source = %s" % (repr(none_root_srcs),root_srcs))

        # resolve possible regex based source name (split-collections sources)
        root_srcs = self.resolve_sources(root_srcs)
        return root_srcs

    def setup(self,sources=None, target_name=None):
        sources = sources or self.sources
        target_name = target_name or self.target_name
        self.target_backend.set_target_name(self.target_name, self.build_name)
        # root key is optional but if set, it must exist in build config
        if self.doc_root_key and not self.doc_root_key in self.build_config:
            raise BuilderException("Root document key '%s' can't be found in build configuration" % self.doc_root_key)

    def get_stats(self,sources,job_manager):
        """
        Return a dictionnary of metadata for this build. It's usually app-specific 
        and this method may be overridden as needed.

        Return dictionary will be merged with any existing metadata in
        src_build collection. This behavior can be changed by setting a special
        key within metadata dict: {"__REPLACE__" : True} will... replace existing 
        metadata with the one returned here.

        "job_manager" is passed in case parallelization is needed. Be aware
        that this method is already running in a dedicated thread, in order to
        use job_manager, the following code must be used at the very beginning
        of its implementation:
        asyncio.set_event_loop(job_manager.loop)
        """
        return {}

    def get_custom_metadata(self, sources, job_manager):
        """
        If more metadata is required, this method can be overridden and should
        return a dict. Existing metadata dict will be update with that one
        before storage.
        """
        return {}


    def get_mapping(self,sources):
        """
        Merge mappings from src_master
        """
        mapping = {}
        src_master = self.source_backend.master
        for collection in self.build_config['sources']:
            meta = src_master.find_one({"_id" : collection})
            if 'mapping' in meta and meta["mapping"]:
                mapping = merge_struct(mapping,meta['mapping'])
            else:
                raise BuilderException('"%s" has no mapping data' % collection)
        return mapping

    def store_metadata(self,res,sources,job_manager):
        self.target_backend.post_merge()
        self.src_meta = self.source_backend.get_src_metadata()
        # now that we have merge stats (count/srcs) + all src involved
        # we can propagate stats
        self.update_src_meta_stats()
        self.mapping = self.get_mapping(sources)
        self.stats = self.get_stats(sources,job_manager)
        self.custom_metadata = self.get_custom_metadata(sources,job_manager)
        self.build_metadata = self.get_build_metadata()
        # also search for _meta in build_config
        bmeta = self.build_config.get("_meta")
        if bmeta:
            self.logger.info("Found _meta in build_config, merging: %s" % pformat(bmeta))
            self.build_metadata.update(self.build_config.get("_meta",{}))

    def update_src_meta_stats(self):
        for src,count in self.merge_stats.items():
            mainsrc = get_source_fullname(src).split(".")[0]
            self.src_meta.setdefault(mainsrc,{}).setdefault("stats",{})
            self.src_meta[mainsrc]["stats"].update({src:count})

    def resolve_sources(self,sources):
        """
        Source can be a string that may contain regex chars. It's usefull
        when you have plenty of sub-collections prefixed with a source name.
        For instance, given a source named 'blah' stored in as many collections<commit_msg>hook for custom metadata (fix)<commit_after>import sys, re, math
import os, glob
import time
import copy
import importlib
import pickle
from datetime import datetime
from pprint import pformat
import asyncio
from functools import partial
import glob, random
import aiocron

from .mapper import TransparentMapper
from ..dataload.uploader import ResourceNotReady
from .differ import set_pending_to_diff
from ..databuild.backend import SourceDocMongoBackend, TargetDocMongoBackend, \
                                LinkTargetDocMongoBackend
from biothings.utils.common import timesofar, iter_n, get_timestamp, dotdict, \
                                   dump, rmdashfr, loadobj, open_compressed_file, \
                                   get_class_from_classpath, find_classes_subclassing
from biothings.utils.mongo import doc_feeder, id_feeder
from biothings.utils.loggers import get_logger
from biothings.utils.manager import BaseManager, ManagerError
from biothings.utils.dataload import update_dict_recur, merge_struct
import biothings.utils.mongo as mongo
from biothings.utils.hub_db import get_source_fullname, get_src_build_config, \
                                   get_src_build, get_src_dump, get_src_master
from biothings import config as btconfig
from biothings.hub import UPLOADER_CATEGORY, BUILDER_CATEGORY
from .backend import create_backend

logging = btconfig.logger

class BuilderException(Exception):
    pass
class ResumeException(Exception):
    pass


class DataBuilder(object):
    """
    Generic data builder.
    """

    keep_archive = 10 # number of archived collection to keep. Oldest get dropped first.

    def __init__(self, build_name, source_backend, target_backend, log_folder,
                 doc_root_key="root", mappers=[], default_mapper_class=TransparentMapper,
                 sources=None, target_name=None,**kwargs):
        self.init_state()
        self.build_name = build_name
        self.sources = sources
        self.target_name = target_name
        self._partial_source_backend = None
        self._partial_target_backend = None
        if type(source_backend) == partial:
            self._partial_source_backend = source_backend
        else:
            self._state["source_backend"] = source_backend
        if type(target_backend) == partial:
            self._partial_target_backend = target_backend
        else:
            self._state["target_backend"] = target_backend
        # doc_root_key is a key name within src_build_config doc.
        # it's a list of datasources that are able to create a document
        # even it doesn't exist. It root documents list is not empty, then
        # any other datasources not listed there won't be able to create
        # a document, only will they able to update it.
        # If no root documets, any datasources can create/update a doc
        # and thus there's no priority nor limitations
        # note: negations can be used, like "!source1". meaning source1 is not
        # root document datasource.
        # Usefull to express; "all resources except source1"
        self.doc_root_key = doc_root_key
        # overall merge start time
        self.t0 = time.time()
        # step merge start time
        self.ti = time.time()
        self.logfile = None
        self.log_folder = log_folder
        self.mappers = {}
        self.timestamp = datetime.now()
        self.merge_stats = {} # keep track of cnt per source, etc...
        self.src_meta = {} # sources involved in this build (includes versions)
        self.stats = {} # can be customized
        self.mapping = {} # ES mapping (merged from src_master's docs)

        for mapper in mappers + [default_mapper_class()]:
            self.mappers[mapper.name] = mapper

        self.step = kwargs.get("step",10000)
        self.prepared = False

    def init_state(self):
        self._state = {
                "logger" : None,
                "source_backend" : None,
                "target_backend" : None,
                "build_config" : None,
        }
    @property
    def logger(self):
        if not self._state["logger"]:
            self.prepare()
        return self._state["logger"]
    @property
    def source_backend(self):
        if self._state["source_backend"] is None:
            self.prepare()
            self._state["build_config"] = self._state["source_backend"].get_build_configuration(self.build_name)
            self._state["source_backend"].validate_sources(self.sources)
        return self._state["source_backend"]
    @property
    def target_backend(self):
        if self._state["target_backend"] is None:
            self.prepare()
        return self._state["target_backend"]
    @property
    def build_config(self):
        self._state["build_config"] = self.source_backend.get_build_configuration(self.build_name)
        return self._state["build_config"]
    @logger.setter
    def logger(self, value):
        self._state["logger"] = value
    @build_config.setter
    def build_config(self, value):
        self._state["build_config"] = value

    def prepare(self,state={}):
        if self.prepared:
            return
        if state:
            # let's be explicit, _state takes what it wants
            for k in self._state:
                self._state[k] = state[k]
            return
        if self._partial_source_backend:
            self._state["source_backend"] = self._partial_source_backend()
        if self._partial_target_backend:
            self._state["target_backend"] = self._partial_target_backend()
        self.setup()
        self.setup_log()
        self.prepared = True

    def unprepare(self):
        """
        reset anything that's not pickable (so self can be pickled)
        return what's been reset as a dict, so self can be restored
        once pickled
        """
        # TODO: use copy ?
        state = {
            "logger" : self._state["logger"],
            "source_backend" : self._state["source_backend"],
            "target_backend" : self._state["target_backend"],
            "build_config" : self._state["build_config"],
        }
        for k in state:
            self._state[k] = None
        self.prepared = False
        return state

    def get_predicates(self):
        def no_uploader_running(job_manager):
            """Uploaders could change the data to be merged..."""
            return len([j for j in job_manager.jobs.values() if j["category"] == UPLOADER_CATEGORY]) == 0
        #def no_merger_running():
        #    """
        #    Mergers use cache files, if more than one running and caches need to be built
        #    both would try to write on the same cache file
        #    """
        #    return len([j for j in job_manager.jobs.values() if j["category"] == BUILDER_CATEGORY]) == 0
        return [no_uploader_running]

    def get_pinfo(self):
        """
        Return dict containing information about the current process
        (used to report in the hub)
        """
        pinfo = {"category" : BUILDER_CATEGORY,
                "source" : "%s:%s" % (self.build_name,self.target_backend.target_name),
                "step" : "",
                "description" : ""}
        preds = self.get_predicates()
        if preds:
            pinfo["__predicates__"] = preds
        return pinfo

    def setup_log(self):
        self.logger,_ = get_logger('build_%s' % self.build_name)

    def check_ready(self,force=False):
        if force:
            # don't even bother
            return
        src_build_config = self.source_backend.build_config
        src_dump = self.source_backend.dump
        _cfg = src_build_config.find_one({'_id': self.build_config['name']})
        # check if all resources are uploaded
        for src_name in _cfg["sources"]:
            fullname = get_source_fullname(src_name)
            if not fullname:
                raise ResourceNotReady("Can't find source '%s'" % src_name)
            main_name = fullname.split(".")[0]
            src_doc = src_dump.find_one({"_id":main_name})
            if not src_doc:
                raise ResourceNotReady("Missing information for source '%s' to start merging" % src_name)
            if not src_doc.get("upload",{}).get("jobs",{}).get(src_name,{}).get("status") == "success":
                raise ResourceNotReady("No successful upload found for resource '%s'" % src_name)

    def get_build_version(self):
        """
        Generate an arbitrary major build version. Default is using a timestamp (YYMMDD)
        '.' char isn't allowed in build version as it's reserved for minor versions
        """
        d = datetime.fromtimestamp(self.t0)
        return "%d%02d%02d" % (d.year,d.month,d.day)

    def register_status(self,status,transient=False,init=False,**extra):
        """
        Register current build status. A build status is a record in src_build
        The key used in this dict the target_name. Then, any operation
        acting on this target_name is registered in a "jobs" list.
        """
        assert self.build_config, "build_config needs to be specified first"
        # get it from source_backend, kind of weird...
        src_build = self.source_backend.build
        src_build_config = self.source_backend.build_config
        all_sources = self.build_config.get("sources",[])
        target_name = "%s" % self.target_backend.target_name
        backend_url = self.target_backend.get_backend_url()
        build_info = {
                '_id' : target_name,
                # TODO: deprecate target_backend & target_name, use backed_url instead
                'target_backend': self.target_backend.name,
                'target_name': target_name,
                'backend_url' : backend_url,
                'build_config': self.build_config,
                # these are all the sources required to build target
                # (not just the ones being processed, those are registered in jobs
                'sources' : all_sources, 
                }
        job_info = {
                'status': status,
                'step_started_at': datetime.now(),
                'logfile': self.logfile,
                }
        if transient:
            # record some "in-progress" information
            job_info['pid'] = os.getpid()
        else:
            # only register time when it's a final state
            job_info["time"] = timesofar(self.ti)
            t1 = round(time.time() - self.ti, 0)
            job_info["time_in_s"] = t1
        if "build" in extra:
            build_info.update(extra["build"])
        if "job" in extra:
            job_info.update(extra["job"])
        # create a new build entry in "build" dict if none exists
        build = src_build.find_one({'_id': target_name})
        if not build:
            # first record for target_name, keep a timestamp
            build_info["started_at"] = datetime.fromtimestamp(self.t0)
            build_info["jobs"] = []
            src_build.insert_one(build_info)
        if init:
            # init timer for this step
            self.ti = time.time()
            src_build.update({'_id': target_name}, {"$push": {'jobs': job_info}})
            # now refresh/sync
            build = src_build.find_one({'_id': target_name})
        else:
            # merge extra at root level
            # (to keep building data...) and update the last one
            # (it's been properly created before when init=True)
            build["jobs"] and build["jobs"][-1].update(job_info)
            # build_info is common to all jobs, so we want to keep
            # any existing data (well... except if it's explicitely specified)
            def merge_build_info(target,d):
                if "__REPLACE__" in d.keys():
                    d.pop("__REPLACE__")
                    target = d
                else:
                    for k,v in d.items():
                        if type(v) == dict:
                            if k in target:
                                target[k] = merge_build_info(target[k],v) 
                            else:
                                v.pop("__REPLACE__",None)
                                # merge v with "nothing" just to make sure to remove any "__REPLACE__"
                                v = merge_build_info({},v)
                                target[k] = v
                        else:
                            target[k] = v
                return target
            build = merge_build_info(build,build_info)
            src_build.replace_one({"_id" : build["_id"]}, build)

    def clean_old_collections(self):
        # use target_name is given, otherwise build name will be used 
        # as collection name prefix, so they should start like that
        prefix = "%s_" % (self.target_name or self.build_name)
        db = mongo.get_target_db()
        cols = [c for c in db.collection_names() if c.startswith(prefix)]
        # timestamp is what's after _archive_, YYYYMMDD, so we can sort it safely
        cols = sorted(cols,reverse=True)
        to_drop = cols[self.keep_archive:]
        for colname in to_drop:
            self.logger.info("Cleaning old archive collection '%s'" % colname)
            db[colname].drop()

    def init_mapper(self,mapper_name):
        if self.mappers[mapper_name].need_load():
            if mapper_name is None:
                self.logger.info("Initializing default mapper")
            else:
                self.logger.info("Initializing mapper name '%s'" % mapper_name)
            self.mappers[mapper_name].load()

    def generate_document_query(self, src_name):
        return None

    def get_root_document_sources(self):
        root_srcs = self.build_config.get(self.doc_root_key,[]) or []
        # check for "not this resource" and adjust the list
        none_root_srcs = [src.replace("!","") for src in root_srcs if src.startswith("!")]
        if none_root_srcs:
            if len(none_root_srcs) != len(root_srcs):
                raise BuilderException("If using '!' operator, all datasources must use it (cannot mix), got: %s" % \
                        (repr(root_srcs)))
            # ok, grab sources for this build, 
            srcs = self.build_config.get("sources",[])
            root_srcs = list(set(srcs).difference(set(none_root_srcs)))
            #self.logger.info("'except root' sources %s resolves to root source = %s" % (repr(none_root_srcs),root_srcs))

        # resolve possible regex based source name (split-collections sources)
        root_srcs = self.resolve_sources(root_srcs)
        return root_srcs

    def setup(self,sources=None, target_name=None):
        sources = sources or self.sources
        target_name = target_name or self.target_name
        self.target_backend.set_target_name(self.target_name, self.build_name)
        # root key is optional but if set, it must exist in build config
        if self.doc_root_key and not self.doc_root_key in self.build_config:
            raise BuilderException("Root document key '%s' can't be found in build configuration" % self.doc_root_key)

    def get_stats(self,sources,job_manager):
        """
        Return a dictionnary of metadata for this build. It's usually app-specific 
        and this method may be overridden as needed.

        Return dictionary will be merged with any existing metadata in
        src_build collection. This behavior can be changed by setting a special
        key within metadata dict: {"__REPLACE__" : True} will... replace existing 
        metadata with the one returned here.

        "job_manager" is passed in case parallelization is needed. Be aware
        that this method is already running in a dedicated thread, in order to
        use job_manager, the following code must be used at the very beginning
        of its implementation:
        asyncio.set_event_loop(job_manager.loop)
        """
        return {}

    def get_custom_metadata(self, sources, job_manager):
        """
        If more metadata is required, this method can be overridden and should
        return a dict. Existing metadata dict will be update with that one
        before storage.
        """
        return {}


    def get_mapping(self,sources):
        """
        Merge mappings from src_master
        """
        mapping = {}
        src_master = self.source_backend.master
        for collection in self.build_config['sources']:
            meta = src_master.find_one({"_id" : collection})
            if 'mapping' in meta and meta["mapping"]:
                mapping = merge_struct(mapping,meta['mapping'])
            else:
                raise BuilderException('"%s" has no mapping data' % collection)
        return mapping

    def store_metadata(self,res,sources,job_manager):
        self.target_backend.post_merge()
        self.src_meta = self.source_backend.get_src_metadata()
        # now that we have merge stats (count/srcs) + all src involved
        # we can propagate stats
        self.update_src_meta_stats()
        self.mapping = self.get_mapping(sources)
        self.stats = self.get_stats(sources,job_manager)
        self.custom_metadata = self.get_custom_metadata(sources,job_manager)
        # also search for _meta in build_config
        bmeta = self.build_config.get("_meta")
        if bmeta:
            self.logger.info("Found _meta in build_config, merging: %s" % pformat(bmeta))
            self.custom_metadata.update(self.build_config.get("_meta",{}))

    def update_src_meta_stats(self):
        for src,count in self.merge_stats.items():
            mainsrc = get_source_fullname(src).split(".")[0]
            self.src_meta.setdefault(mainsrc,{}).setdefault("stats",{})
            self.src_meta[mainsrc]["stats"].update({src:count})

    def resolve_sources(self,sources):
        """
        Source can be a string that may contain regex chars. It's usefull
        when you have plenty of sub-collections prefixed with a source name.
        For instance, given a source named 'blah' stored in as many collections
{'fraction_non_alphanumeric': 0.09110803324099723, 'fraction_numerical': 0.0013850415512465374, 'mean_word_length': 2.2106901458555672, 'pattern_counts': {'":': 6, '<': 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': 10, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '953452', 'n_tokens_mistral': 10079, 'n_tokens_neox': 9795, 'n_words': 2819}