<commit_before>import json
import time
import logging
import itertools
import threading
import collections

import synapse.compat as s_compat
import synapse.common as s_common
import synapse.dyndeps as s_dyndeps
import synapse.reactor as s_reactor
import synapse.telepath as s_telepath
import synapse.datamodel as s_datamodel

import synapse.cores.storage as s_storage

import synapse.lib.tags as s_tags
import synapse.lib.tufo as s_tufo
import synapse.lib.cache as s_cache
import synapse.lib.queue as s_queue
import synapse.lib.ingest as s_ingest
import synapse.lib.hashset as s_hashset
import synapse.lib.threads as s_threads
import synapse.lib.modules as s_modules
import synapse.lib.hashitem as s_hashitem
import synapse.lib.interval as s_interval
import synapse.lib.userauth as s_userauth

from synapse.eventbus import EventBus, on, onfini
from synapse.lib.storm import Runtime
from synapse.datamodel import DataModel
from synapse.lib.config import Configable

logger = logging.getLogger(__name__)

def chunked(n, iterable):
    it = iter(iterable)
    while True:
        chunk = tuple(itertools.islice(it, n))
        if not chunk:
            return

        yield chunk

def reqiden(tufo):
    '''
    Raise an exception if the given tufo is ephemeral.
    '''
    if tufo[0] is None:
        raise s_common.NoSuchTufo(iden=None)
    return tufo[0]

class Cortex(EventBus, DataModel, Runtime, Configable, s_ingest.IngestApi):
    '''
    Top level Cortex key/valu storage object.
    '''
    def __init__(self, link, store, **conf):

        Runtime.__init__(self)
        EventBus.__init__(self)
        Configable.__init__(self)

        # a cortex may have a ref to an axon
        self.axon = None
        self.seedctors = {}

        self.modules = [(ctor, conf) for ctor, smod, conf in s_modules.ctorlist]
        self.modsdone = False

        self.noauto = {'syn:form', 'syn:type', 'syn:prop'}

        self.addConfDef('autoadd', type='bool', asloc='autoadd', defval=1,
                        doc='Automatically add forms for props where type is form')
        self.addConfDef('enforce', type='bool', asloc='enforce', defval=0, doc='Enables data model enforcement')
        self.addConfDef('caching', type='bool', asloc='caching', defval=0, doc='Enables caching layer in the cortex')
        self.addConfDef('cache:maxsize', type='int', asloc='cache_maxsize', defval=1000,
                        doc='Enables caching layer in the cortex')

        self.addConfDef('rev:model', type='bool', defval=1, doc='Set to 0 to disallow model version updates')

        self.addConfDef('axon:url', type='str', doc='Allows cortex to be aware of an axon blob store')

        self.addConfDef('log:save', type='bool', asloc='logsave', defval=0,
                        doc='Enables saving exceptions to the cortex as syn:log nodes')
        self.addConfDef('log:level', type='int', asloc='loglevel', defval=0, doc='Filters log events to >= level')

        self.addConfDef('modules', defval=(),
                        doc='An optional list of (pypath,conf) tuples for synapse modules to load')

        self.onConfOptSet('modules', self._onSetMods)
        self.onConfOptSet('caching', self._onSetCaching)
        self.onConfOptSet('axon:url', self._onSetAxonUrl)

        self.setConfOpts(conf)

        self._link = link

        self.lock = threading.Lock()

        self.coremods = {}  # name:module ( CoreModule() )
        self.statfuncs = {}

        self.auth = None
        self.snaps = s_cache.Cache(maxtime=60)

        self.seqs = s_cache.KeyCache(self.getSeqNode)

        self.formed = collections.defaultdict(int)      # count tufos formed since startup

        self._core_tags = s_cache.FixedCache(maxsize=10000, onmiss=self._getFormFunc('syn:tag'))
        self._core_tagforms = s_cache.FixedCache(maxsize=10000, onmiss=self._getFormFunc('syn:tagform'))

        self.tufosbymeths = {}

        self.cache_fifo = collections.deque()               # [ ((prop,valu,limt), {
        self.cache_bykey = {}                               # (prop,valu,limt):( (prop,valu,limt), {iden:tufo,...} )
        self.cache_byiden = s_cache.RefDict()
        self.cache_byprop = collections.defaultdict(dict)   # (<prop>,<valu>):[ ((prop, valu, limt),  answ), ... ]

        #############################################################
        # Handlers for each splice event action
        self.spliceact = s_reactor.Reactor()
        self.spliceact.act('node:add', self._actNodeAdd)
        self.spliceact.act('node:del', self._actNodeDel)
        self.spliceact.act('node:prop:set', self._actNodePropSet)
        self.spliceact.act('node:prop:del', self._actNodePropDel)
        self.spliceact.act('node:tag:add', self._actNodeTagAdd)
        self.spliceact.act('node:tag:del', self._actNodeTagDel)
        self.spliceact.act('node:ival:set', self._actNodeIvalSet)
        self.spliceact.act('node:ival:del', self._actNodeIvalDel)

        #############################################################

        self.addStatFunc('any', self._calcStatAny)
        self.addStatFunc('all', self._calcStatAll)
        self.addStatFunc('min', self._calcStatMin)
        self.addStatFunc('max', self._calcStatMax)
        self.addStatFunc('sum', self._calcStatSum)
        self.addStatFunc('mean', self._calcStatMean)
        self.addStatFunc('count', self._calcStatCount)
        self.addStatFunc('histo', self._calcStatHisto)

        # Strap in default initTufosBy functions - Storage layer may override these.
        self.initTufosBy('eq', self._tufosByEq)
        self.initTufosBy('in', self._tufosByIn)
        self.initTufosBy('has', self._tufosByHas)
        self.initTufosBy('tag', self._tufosByTag)
        self.initTufosBy('type', self._tufosByType)
        self.initTufosBy('inet:cidr', self._tufosByInetCidr)
        self.initTufosBy('dark', self._tufosByDark)

        # Initialize the storage layer
        self.store = store  # type: s_storage.Storage()
        self.store.register_cortex(self)
        self.onfini(self.store.fini)

        self.isok = True

        DataModel.__init__(self, load=False)

        self.myfo = self.formTufoByProp('syn:core', 'self')
        self.isnew = self.myfo[1].get('.new', False)

        self.modelrevlist = []
        with self.getCoreXact() as xact:
            self._initCoreMods()

        # and finally, strap in our event handlers...
        self.on('node:add', self._onTufoAddSynType, form='syn:type')
        self.on('node:add', self._onTufoAddSynForm, form='syn:form')
        self.on('node:add', self._onTufoAddSynProp, form='syn:prop')

        self.addTufoForm('syn:log', ptype='guid', local=1)
        self.addTufoProp('syn:log', 'subsys', defval='??', help='Named subsystem which originated the log event')
        self.addTufoProp('syn:log', 'level', ptype='int', defval=logging.WARNING)
        self.addTufoProp('syn:log', 'time', ptype='time', doc='When did the log event occur')
        self.addTufoProp('syn:log', 'exc', ptype='str', help='Exception class name if caused by an exception')
        self.addTufoProp('syn:log', 'info:*', glob=1)

        self.addTufoForm('syn:ingest', ptype='str:lwr', local=1)
        self.addTufoProp('syn:ingest', 'time', ptype='time')
        self.addTufoProp('syn:ingest', 'text', ptype='json')

        # storm operators specific to the cortex
        self.setOperFunc('stat', self._stormOperStat)
        self.setOperFunc('dset', self._stormOperDset)

        # allow modules a shot at hooking cortex events for model ctors
        for name, ret, exc in s_modules.call('addCoreOns', self):
            if exc is not None:
                logger.warning('%s.addCoreOns: %s' % (name, exc))

        self.onfini(self._finiCoreMods)

        s_ingest.IngestApi.__init__(self, self)

    def getModlVers(self, name):
        '''
        Retrieve the model version for the given model name.

        Args:
            name (str): The name of the model

        Returns:
            (int):  The model version ( linear version number ) or -1
        '''
        prop = '.:modl:vers:' + name
        rows = self.getRowsByProp(prop)
        if not rows:
            return -1<commit_msg>Use the confdef() decorator for loading configable options into Cortex class.<commit_after>import json
import time
import logging
import itertools
import threading
import collections

import synapse.compat as s_compat
import synapse.common as s_common
import synapse.dyndeps as s_dyndeps
import synapse.reactor as s_reactor
import synapse.telepath as s_telepath
import synapse.datamodel as s_datamodel

import synapse.cores.storage as s_storage

import synapse.lib.tags as s_tags
import synapse.lib.tufo as s_tufo
import synapse.lib.cache as s_cache
import synapse.lib.queue as s_queue
import synapse.lib.ingest as s_ingest
import synapse.lib.hashset as s_hashset
import synapse.lib.threads as s_threads
import synapse.lib.modules as s_modules
import synapse.lib.hashitem as s_hashitem
import synapse.lib.interval as s_interval
import synapse.lib.userauth as s_userauth

from synapse.eventbus import EventBus, on, onfini
from synapse.lib.storm import Runtime
from synapse.datamodel import DataModel
from synapse.lib.config import Configable, confdef

logger = logging.getLogger(__name__)

def chunked(n, iterable):
    it = iter(iterable)
    while True:
        chunk = tuple(itertools.islice(it, n))
        if not chunk:
            return

        yield chunk

def reqiden(tufo):
    '''
    Raise an exception if the given tufo is ephemeral.
    '''
    if tufo[0] is None:
        raise s_common.NoSuchTufo(iden=None)
    return tufo[0]

class Cortex(EventBus, DataModel, Runtime, Configable, s_ingest.IngestApi):
    '''
    Top level Cortex key/valu storage object.
    '''
    def __init__(self, link, store, **conf):

        Runtime.__init__(self)
        EventBus.__init__(self)
        Configable.__init__(self)

        # a cortex may have a ref to an axon
        self.axon = None
        self.seedctors = {}

        self.modules = [(ctor, conf) for ctor, smod, conf in s_modules.ctorlist]
        self.modsdone = False

        self.noauto = {'syn:form', 'syn:type', 'syn:prop'}

        self.onConfOptSet('modules', self._onSetMods)
        self.onConfOptSet('caching', self._onSetCaching)
        self.onConfOptSet('axon:url', self._onSetAxonUrl)

        self.setConfOpts(conf)

        self._link = link

        self.lock = threading.Lock()

        self.coremods = {}  # name:module ( CoreModule() )
        self.statfuncs = {}

        self.auth = None
        self.snaps = s_cache.Cache(maxtime=60)

        self.seqs = s_cache.KeyCache(self.getSeqNode)

        self.formed = collections.defaultdict(int)      # count tufos formed since startup

        self._core_tags = s_cache.FixedCache(maxsize=10000, onmiss=self._getFormFunc('syn:tag'))
        self._core_tagforms = s_cache.FixedCache(maxsize=10000, onmiss=self._getFormFunc('syn:tagform'))

        self.tufosbymeths = {}

        self.cache_fifo = collections.deque()               # [ ((prop,valu,limt), {
        self.cache_bykey = {}                               # (prop,valu,limt):( (prop,valu,limt), {iden:tufo,...} )
        self.cache_byiden = s_cache.RefDict()
        self.cache_byprop = collections.defaultdict(dict)   # (<prop>,<valu>):[ ((prop, valu, limt),  answ), ... ]

        #############################################################
        # Handlers for each splice event action
        self.spliceact = s_reactor.Reactor()
        self.spliceact.act('node:add', self._actNodeAdd)
        self.spliceact.act('node:del', self._actNodeDel)
        self.spliceact.act('node:prop:set', self._actNodePropSet)
        self.spliceact.act('node:prop:del', self._actNodePropDel)
        self.spliceact.act('node:tag:add', self._actNodeTagAdd)
        self.spliceact.act('node:tag:del', self._actNodeTagDel)
        self.spliceact.act('node:ival:set', self._actNodeIvalSet)
        self.spliceact.act('node:ival:del', self._actNodeIvalDel)

        #############################################################

        self.addStatFunc('any', self._calcStatAny)
        self.addStatFunc('all', self._calcStatAll)
        self.addStatFunc('min', self._calcStatMin)
        self.addStatFunc('max', self._calcStatMax)
        self.addStatFunc('sum', self._calcStatSum)
        self.addStatFunc('mean', self._calcStatMean)
        self.addStatFunc('count', self._calcStatCount)
        self.addStatFunc('histo', self._calcStatHisto)

        # Strap in default initTufosBy functions - Storage layer may override these.
        self.initTufosBy('eq', self._tufosByEq)
        self.initTufosBy('in', self._tufosByIn)
        self.initTufosBy('has', self._tufosByHas)
        self.initTufosBy('tag', self._tufosByTag)
        self.initTufosBy('type', self._tufosByType)
        self.initTufosBy('inet:cidr', self._tufosByInetCidr)
        self.initTufosBy('dark', self._tufosByDark)

        # Initialize the storage layer
        self.store = store  # type: s_storage.Storage()
        self.store.register_cortex(self)
        self.onfini(self.store.fini)

        self.isok = True

        DataModel.__init__(self, load=False)

        self.myfo = self.formTufoByProp('syn:core', 'self')
        self.isnew = self.myfo[1].get('.new', False)

        self.modelrevlist = []
        with self.getCoreXact() as xact:
            self._initCoreMods()

        # and finally, strap in our event handlers...
        self.on('node:add', self._onTufoAddSynType, form='syn:type')
        self.on('node:add', self._onTufoAddSynForm, form='syn:form')
        self.on('node:add', self._onTufoAddSynProp, form='syn:prop')

        self.addTufoForm('syn:log', ptype='guid', local=1)
        self.addTufoProp('syn:log', 'subsys', defval='??', help='Named subsystem which originated the log event')
        self.addTufoProp('syn:log', 'level', ptype='int', defval=logging.WARNING)
        self.addTufoProp('syn:log', 'time', ptype='time', doc='When did the log event occur')
        self.addTufoProp('syn:log', 'exc', ptype='str', help='Exception class name if caused by an exception')
        self.addTufoProp('syn:log', 'info:*', glob=1)

        self.addTufoForm('syn:ingest', ptype='str:lwr', local=1)
        self.addTufoProp('syn:ingest', 'time', ptype='time')
        self.addTufoProp('syn:ingest', 'text', ptype='json')

        # storm operators specific to the cortex
        self.setOperFunc('stat', self._stormOperStat)
        self.setOperFunc('dset', self._stormOperDset)

        # allow modules a shot at hooking cortex events for model ctors
        for name, ret, exc in s_modules.call('addCoreOns', self):
            if exc is not None:
                logger.warning('%s.addCoreOns: %s' % (name, exc))

        self.onfini(self._finiCoreMods)

        s_ingest.IngestApi.__init__(self, self)

    @staticmethod
    @confdef()
    def _cortex_condefs():
        confdefs = (
            ('autoadd', {'type': 'bool', 'asloc': 'autoadd', 'defval': 1,
                         'doc': 'Automatically add forms for props where type is form'}),
            ('enforce', {'type': 'bool', 'asloc': 'enforce', 'defval': 0, 'doc': 'Enables data model enforcement'}),
            ('caching', {'type': 'bool', 'asloc': 'caching', 'defval': 0,
                         'doc': 'Enables caching layer in the cortex'}),
            ('cache:maxsize', {'type': 'int', 'asloc': 'cache_maxsize', 'defval': 1000,
                               'doc': 'Enables caching layer in the cortex'}),
            ('rev:model', {'type': 'bool', 'defval': 1, 'doc': 'Set to 0 to disallow model version updates'}),
            ('rev:storage', {'type': 'bool', 'defval': 1, 'doc': 'Set to 0 to disallow storage version updates'}),
            ('axon:url', {'type': 'str', 'doc': 'Allows cortex to be aware of an axon blob store'}),
            ('log:save', {'type': 'bool', 'asloc': 'logsave', 'defval': 0,
                          'doc': 'Enables saving exceptions to the cortex as syn:log nodes'}),
            ('log:level', {'type': 'int', 'asloc': 'loglevel', 'defval': 0, 'doc': 'Filters log events to >= level'}),
            ('modules', {'defval': (), 'doc': 'An optional list of (pypath,conf) tuples for synapse modules to load'})
        )
        return confdefs

    def getModlVers(self, name):
        '''
        Retrieve the model version for the given model name.

        Args:
            name (str): The name of the model

        Returns:
            (int):  The model version ( linear version number ) or -1
        '''
        prop = '.:modl:vers:' + name
        rows = self.getRowsByProp(prop)
        if not rows:
            return -1
{'fraction_non_alphanumeric': 0.14035402463820118, 'fraction_numerical': 0.0038272933859586173, 'mean_word_length': 2.9459650778669184, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 9, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5968011', 'n_tokens_mistral': 5458, 'n_tokens_neox': 5290, 'n_words': 1203}