<commit_before>import json
from couchdbkit.exceptions import ResourceNotFound
from dimagi.ext.jsonobject import JsonObject, StringProperty, ListProperty, DictProperty
from jsonobject.base_properties import DefaultProperty
from corehq.apps.userreports.expressions.getters import (
    DictGetter,
    NestedDictGetter,
    TransformedGetter,
    transform_from_datatype)
from corehq.apps.userreports.indicators.specs import DataTypeProperty
from corehq.apps.userreports.specs import TypeProperty, EvaluationContext
from corehq.util.quickcache import quickcache
from dimagi.utils.couch.database import get_db


class ConstantGetterSpec(JsonObject):
    type = TypeProperty('constant')
    constant = DefaultProperty(required=True)

    def __call__(self, item, context=None):
        return self.constant


class PropertyNameGetterSpec(JsonObject):
    type = TypeProperty('property_name')
    property_name = StringProperty(required=True)
    datatype = DataTypeProperty(required=False)

    @property
    def expression(self):
        transform = transform_from_datatype(self.datatype)
        getter = DictGetter(self.property_name)
        return TransformedGetter(getter, transform)

    def __call__(self, item, context=None):
        return self.expression(item, context)


class PropertyPathGetterSpec(JsonObject):
    type = TypeProperty('property_path')
    property_path = ListProperty(unicode, required=True)
    datatype = DataTypeProperty(required=False)

    @property
    def expression(self):
        transform = transform_from_datatype(self.datatype)
        getter = NestedDictGetter(self.property_path)
        return TransformedGetter(getter, transform)

    def __call__(self, item, context=None):
        return self.expression(item, context)


class ConditionalExpressionSpec(JsonObject):
    type = TypeProperty('conditional')
    test = DictProperty(required=True)
    expression_if_true = DictProperty(required=True)
    expression_if_false = DictProperty(required=True)

    def configure(self, test_function, true_expression, false_expression):
        self._test_function = test_function
        self._true_expression = true_expression
        self._false_expression = false_expression

    def __call__(self, item, context=None):
        if self._test_function(item, context):
            return self._true_expression(item, context)
        else:
            return self._false_expression(item, context)


class RootDocExpressionSpec(JsonObject):
    type = TypeProperty('root_doc')
    expression = DictProperty(required=True)

    def configure(self, expression):
        self._expression_fn = expression

    def __call__(self, item, context=None):
        if context is None:
            return None
        return self._expression_fn(context.root_doc, context)


class RelatedDocExpressionSpec(JsonObject):
    type = TypeProperty('related_doc')
    related_doc_type = StringProperty()
    doc_id_expression = DictProperty(required=True)
    value_expression = DictProperty(required=True)

    db_lookup = lambda self, type: get_db()

    def configure(self, related_doc_type, doc_id_expression, value_expression):
        self._related_doc_type = related_doc_type
        self._doc_id_expression = doc_id_expression
        self._value_expression = value_expression

        # used in caching
        self._vary_on = json.dumps(self.value_expression, sort_keys=True)

    def __call__(self, item, context=None):
        doc_id = self._doc_id_expression(item, context)
        if doc_id:
            return self.get_value(doc_id, context)

    @quickcache(['self._vary_on', 'doc_id'])
    def get_value(self, doc_id, context):
        try:
            doc = self.db_lookup(self.related_doc_type).get(doc_id)
            # ensure no cross-domain lookups of different documents
            assert context.root_doc['domain']
            if context.root_doc['domain'] != doc.get('domain'):
                return None
            # explicitly use a new evaluation context since this is a new document
            return self._value_expression(doc, EvaluationContext(doc, 0))
        except ResourceNotFound:
            return None


class AbtSupervisorExpressionSpec(JsonObject):
    type = TypeProperty('abt_supervisor')

    def __call__(self, item, context=None):
        # TODO: What does context do (I think I can ignore it)?
        # TODO: Instead, take a list of tuples which are paths and flag answers.
        docs = []
        if item['form']['breakfast'] == 'no':
            docs.append({'flag': 'breakfast'})
        if item['form']['equipment'] == 'yes':
            docs.append({'flag': 'equipment'})
        return docs<commit_msg>moar temp<commit_after>import json
from couchdbkit.exceptions import ResourceNotFound
from dimagi.ext.jsonobject import JsonObject, StringProperty, ListProperty, DictProperty
from jsonobject.base_properties import DefaultProperty
from corehq.apps.userreports.expressions.getters import (
    DictGetter,
    NestedDictGetter,
    TransformedGetter,
    transform_from_datatype)
from corehq.apps.userreports.indicators.specs import DataTypeProperty
from corehq.apps.userreports.specs import TypeProperty, EvaluationContext
from corehq.util.quickcache import quickcache
from dimagi.utils.couch.database import get_db


class ConstantGetterSpec(JsonObject):
    type = TypeProperty('constant')
    constant = DefaultProperty(required=True)

    def __call__(self, item, context=None):
        return self.constant


class PropertyNameGetterSpec(JsonObject):
    type = TypeProperty('property_name')
    property_name = StringProperty(required=True)
    datatype = DataTypeProperty(required=False)

    @property
    def expression(self):
        transform = transform_from_datatype(self.datatype)
        getter = DictGetter(self.property_name)
        return TransformedGetter(getter, transform)

    def __call__(self, item, context=None):
        return self.expression(item, context)


class PropertyPathGetterSpec(JsonObject):
    type = TypeProperty('property_path')
    property_path = ListProperty(unicode, required=True)
    datatype = DataTypeProperty(required=False)

    @property
    def expression(self):
        transform = transform_from_datatype(self.datatype)
        getter = NestedDictGetter(self.property_path)
        return TransformedGetter(getter, transform)

    def __call__(self, item, context=None):
        return self.expression(item, context)


class ConditionalExpressionSpec(JsonObject):
    type = TypeProperty('conditional')
    test = DictProperty(required=True)
    expression_if_true = DictProperty(required=True)
    expression_if_false = DictProperty(required=True)

    def configure(self, test_function, true_expression, false_expression):
        self._test_function = test_function
        self._true_expression = true_expression
        self._false_expression = false_expression

    def __call__(self, item, context=None):
        if self._test_function(item, context):
            return self._true_expression(item, context)
        else:
            return self._false_expression(item, context)


class RootDocExpressionSpec(JsonObject):
    type = TypeProperty('root_doc')
    expression = DictProperty(required=True)

    def configure(self, expression):
        self._expression_fn = expression

    def __call__(self, item, context=None):
        if context is None:
            return None
        return self._expression_fn(context.root_doc, context)


class RelatedDocExpressionSpec(JsonObject):
    type = TypeProperty('related_doc')
    related_doc_type = StringProperty()
    doc_id_expression = DictProperty(required=True)
    value_expression = DictProperty(required=True)

    db_lookup = lambda self, type: get_db()

    def configure(self, related_doc_type, doc_id_expression, value_expression):
        self._related_doc_type = related_doc_type
        self._doc_id_expression = doc_id_expression
        self._value_expression = value_expression

        # used in caching
        self._vary_on = json.dumps(self.value_expression, sort_keys=True)

    def __call__(self, item, context=None):
        doc_id = self._doc_id_expression(item, context)
        if doc_id:
            return self.get_value(doc_id, context)

    @quickcache(['self._vary_on', 'doc_id'])
    def get_value(self, doc_id, context):
        try:
            doc = self.db_lookup(self.related_doc_type).get(doc_id)
            # ensure no cross-domain lookups of different documents
            assert context.root_doc['domain']
            if context.root_doc['domain'] != doc.get('domain'):
                return None
            # explicitly use a new evaluation context since this is a new document
            return self._value_expression(doc, EvaluationContext(doc, 0))
        except ResourceNotFound:
            return None


class AbtSupervisorExpressionSpec(JsonObject):
    type = TypeProperty('abt_supervisor')

    def __call__(self, item, context=None):
        """
        Given a document (item), return a list of documents representing each
        of the flagged questions.
        """
        # TODO: Instead, take a list of tuples which are paths and flag answers.
        flag_specs = [
            (['breakfast'], 'no'),
            (['equipment'], 'yes'),
        ]
        docs = []
        for path, danger_value in flag_specs:
            v = item['form']
            for key in path:
                v = v[key]
            if v == danger_value:
                docs.append({'flag': path[-1]})
        return docs
{'fraction_non_alphanumeric': 0.08507743825868565, 'fraction_numerical': 0.00031393888656341566, 'mean_word_length': 3.4471847370870172, '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': '360154', 'n_tokens_mistral': 2650, 'n_tokens_neox': 2585, 'n_words': 607}