kitconcept.solr/backend/src/kitconcept/solr/services/solr_utils_extra.py at eb21206d3c3fd8ae97ca60380a0f1af219955fee · kitconcept/kitconcept.solr · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
from kitconcept.solr.services.solr_utils import escape
from kitconcept.solr.services.solr_utils import replace_reserved
from zExceptions import BadRequest

import base64
import binascii
import json
import logging


logger = logging.getLogger("kitconcept.solr")
logger.setLevel(logging.DEBUG)


def escape_fieldname(fieldname):
    return replace_reserved(fieldname)


def escape_value(value):
    return '"' + (escape(value)) + '"'


class SolrExtraConditions:
    config: object

    def __init__(self, config: dict):
        self.config = config

    @classmethod
    def from_encoded(cls, raw: str):
        # An empty string (e.g. a bare extra_conditions= URL
        # parameter from the results page) means no conditions, like an
        # absent parameter - not invalid input worth a log warning.
        if raw:
            try:
                config = json.loads(base64.b64decode(raw))
            except (
                UnicodeDecodeError,
                json.decoder.JSONDecodeError,
                binascii.Error,
            ):
                logger.warning("Ignoring invalid base64 encoded string", exc_info=True)
                config = []
        else:
            config = []
        return cls(config)

    def query_list(self):
        results = []
        for row in self.config:
            try:
                [fieldname, kind, condition] = row
            except (TypeError, ValueError) as err:
                raise BadRequest(
                    f"Invalid extra condition row [{row}], needs: [fieldname, kind, condition]"  # noqa: E501
                ) from err
            fieldname = escape_fieldname(fieldname)
            if kind == "date-range":
                keys = set(condition.keys())
                if (
                    not keys.issubset({"ge", "le", "gr", "ls"})
                    or {"ge", "gr"}.issubset(keys)
                    or {"le", "ls"}.issubset(keys)
                ):
                    raise BadRequest(
                        f"invalid keys in options for condition 'date-range' [{keys}]"
                    )
                result = f"{fieldname}:"
                if "ge" in condition:
                    value = replace_reserved(condition["ge"])
                    result += f"[{value} TO "
                elif "gr" in condition:
                    value = replace_reserved(condition["gr"])
                    result += f"{{{value} TO "
                else:
                    result += "[* TO "
                if "le" in condition:
                    value = replace_reserved(condition["le"])
                    result += f"{value}]"
                elif "ls" in condition:
                    value = replace_reserved(condition["ls"])
                    result += f"{value}}}"
                else:
                    result += "*]"
            elif kind == "string":
                keys = set(condition.keys())
                if not keys.issubset({"in"}):
                    raise BadRequest(
                        "invalid keys in options for condition 'string', "
                        f"supported: 'in' [{keys}]"
                    )
                if "in" in condition:
                    if type(condition["in"]) is not list:
                        raise BadRequest(
                            "invalid type for condition 'string' "
                            f"[{type(condition['in'])}]"
                        )
                    if len(condition["in"]) == 0:
                        # Empty list, ignore
                        continue
                    # Terms are exact values (e.g. a portal_type like
                    # "News Item"): quote them, otherwise a space would
                    # split the term and leak half of it out of the
                    # field query. escape() keeps a quote inside a term
                    # from breaking out of the quoting.
                    result = f"{fieldname}:"
                    result += (
                        "("
                        + " OR ".join([
                            f'"{escape(replace_reserved(term))}"'
                            for term in condition["in"]
                        ])
                        + ")"
                    )
            else:
                raise BadRequest(f"Wrong condition type [{kind}]")
            results.append(result)
        return results