Coverage for chalicelib/utils/helper.py: 33%
240 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-10 12:56 +0000
1import math
2import random
3import re
4import string
5from typing import Union
6from urllib.parse import urlparse
7import hashlib
8from decouple import config
10import schemas
11from chalicelib.utils.TimeUTC import TimeUTC
14def random_string(length=36):
15 return "".join(random.choices(string.hexdigits, k=length))
18def list_to_camel_case(items: list[dict], flatten: bool = False, ignore_keys=[]) -> list[dict]:
19 for i in range(len(items)):
20 if flatten:
21 items[i] = flatten_nested_dicts(items[i])
22 items[i] = dict_to_camel_case(items[i], ignore_keys=[])
24 return items
27def dict_to_camel_case(variable, delimiter='_', ignore_keys=[]):
28 if variable is None:
29 return None
30 if isinstance(variable, str):
31 return variable
32 elif isinstance(variable, dict): 32 ↛ 45line 32 didn't jump to line 45 because the condition on line 32 was always true
33 aux = {}
34 for key in variable.keys():
35 if key in ignore_keys: 35 ↛ 36line 35 didn't jump to line 36 because the condition on line 35 was never true
36 aux[key] = variable[key]
37 elif isinstance(variable[key], dict):
38 aux[key_to_camel_case(key, delimiter)] = dict_to_camel_case(variable[key])
39 elif isinstance(variable[key], list):
40 aux[key_to_camel_case(key, delimiter)] = list_to_camel_case(variable[key])
41 else:
42 aux[key_to_camel_case(key, delimiter)] = variable[key]
43 return aux
44 else:
45 return variable
48def dict_to_CAPITAL_keys(variable):
49 if variable is None:
50 return None
51 if isinstance(variable, str):
52 return variable.upper()
53 elif isinstance(variable, dict):
54 aux = {}
55 for key in variable.keys():
56 if isinstance(variable[key], dict):
57 aux[key.upper()] = dict_to_CAPITAL_keys(variable[key])
58 else:
59 aux[key.upper()] = variable[key]
60 return aux
61 else:
62 return variable
65def variable_to_snake_case(variable, delimiter='_', split_number=False):
66 if isinstance(variable, str):
67 return key_to_snake_case(variable, delimiter, split_number)
68 elif isinstance(variable, dict):
69 aux = {}
70 for key in variable.keys():
71 if isinstance(variable[key], dict):
72 aux[key_to_snake_case(key, delimiter, split_number)] = variable_to_snake_case(variable[key], delimiter,
73 split_number)
74 else:
75 aux[key_to_snake_case(key, delimiter, split_number)] = variable[key]
76 return aux
77 else:
78 return variable
81def key_to_camel_case(snake_str, delimiter='_'):
82 if snake_str.startswith(delimiter): 82 ↛ 83line 82 didn't jump to line 83 because the condition on line 82 was never true
83 snake_str = snake_str[1:]
84 components = snake_str.split(delimiter)
85 return components[0] + ''.join(x.title() for x in components[1:])
88def key_to_snake_case(name, delimiter='_', split_number=False):
89 s1 = re.sub('(.)([A-Z][a-z]+)', fr'\1{delimiter}\2', name)
90 return re.sub('([a-z])([A-Z0-9])' if split_number else '([a-z0-9])([A-Z])', fr'\1{delimiter}\2', s1).lower()
93def key_to_title_case(name, delimiter='_'):
94 snake = key_to_snake_case(name, delimiter)
95 return ' '.join(word.capitalize() for word in snake.split(delimiter) if word)
98TRACK_TIME = True
101def allow_captcha():
102 return config("captcha_server", default=None) is not None and config("captcha_key", default=None) is not None \
103 and len(config("captcha_server")) > 0 and len(config("captcha_key")) > 0
106def string_to_sql_like(value):
107 if value is None:
108 return None
109 value = re.sub(' +', ' ', value)
110 value = value.replace("*", "%")
111 if value.startswith("^"):
112 value = value[1:]
113 elif not value.startswith("%"):
114 value = '%' + value
116 if value.endswith("$"):
117 value = value[:-1]
118 elif not value.endswith("%"):
119 value = value + '%'
120 # value = value.replace(" ", "%")
121 return value
124def string_to_sql_like_with_op(value, op):
125 if isinstance(value, list):
126 r = []
127 for v in value:
128 r.append(string_to_sql_like_with_op(v, op))
129 return r
130 else:
131 _value = value
132 if _value is None:
133 return _value
134 if op.upper() != 'ILIKE':
135 return _value.replace("%", "%%")
136 _value = _value.replace("*", "%")
137 if _value.startswith("^"):
138 _value = _value[1:]
139 elif not _value.startswith("%"):
140 _value = '%' + _value
142 if _value.endswith("$"):
143 _value = _value[:-1]
144 elif not _value.endswith("%"):
145 _value = _value + '%'
146 return _value.replace("%", "%%")
149likable_operators = [schemas.SearchEventOperator.STARTS_WITH, schemas.SearchEventOperator.ENDS_WITH,
150 schemas.SearchEventOperator.CONTAINS, schemas.SearchEventOperator.NOT_CONTAINS,
151 schemas.ClickEventExtraOperator.STARTS_WITH, schemas.ClickEventExtraOperator.ENDS_WITH,
152 schemas.ClickEventExtraOperator.CONTAINS, schemas.ClickEventExtraOperator.NOT_CONTAINS]
155def is_likable(op: Union[schemas.SearchEventOperator, schemas.ClickEventExtraOperator]):
156 return op in likable_operators
159def values_for_operator(value: Union[str, list],
160 op: Union[schemas.SearchEventOperator, schemas.ClickEventExtraOperator]):
161 if not is_likable(op):
162 return value
163 if isinstance(value, list):
164 r = []
165 for v in value:
166 r.append(values_for_operator(v, op))
167 return r
168 else:
169 if value is None:
170 return value
171 if op in (schemas.SearchEventOperator.STARTS_WITH, schemas.ClickEventExtraOperator.STARTS_WITH):
172 return f"{value}%"
173 elif op in (schemas.SearchEventOperator.ENDS_WITH, schemas.ClickEventExtraOperator.ENDS_WITH):
174 return f"%{value}"
175 elif op in (schemas.SearchEventOperator.CONTAINS, schemas.SearchEventOperator.NOT_CONTAINS,
176 schemas.ClickEventExtraOperator.CONTAINS, schemas.ClickEventExtraOperator.NOT_CONTAINS):
177 return f"%{value}%"
178 return value
181def is_alphabet_space_dash(word):
182 r = re.compile("^[a-zA-Z -]*$")
183 return r.match(word) is not None
186def merge_lists_by_key(l1, l2, key):
187 merged = {}
188 for item in l1 + l2:
189 if item[key] in merged:
190 merged[item[key]].update(item)
191 else:
192 merged[item[key]] = item
193 return [val for (_, val) in merged.items()]
196def flatten_nested_dicts(obj):
197 if obj is None:
198 return None
199 result = {}
200 for key in obj.keys():
201 if isinstance(obj[key], dict):
202 result = {**result, **flatten_nested_dicts(obj[key])}
203 else:
204 result[key] = obj[key]
205 return result
208def delete_keys_from_dict(d, to_delete):
209 if isinstance(to_delete, str):
210 to_delete = [to_delete]
211 if isinstance(d, dict):
212 for single_to_delete in set(to_delete):
213 if single_to_delete in d:
214 del d[single_to_delete]
215 for k, v in d.items():
216 delete_keys_from_dict(v, to_delete)
217 elif isinstance(d, list):
218 for i in d:
219 delete_keys_from_dict(i, to_delete)
220 return d
223def explode_widget(data, key=None):
224 result = []
225 for k in data.keys():
226 if k.endswith("Progress") or k == "chart":
227 continue
228 result.append({"key": key_to_snake_case(k) if key is None else key, "data": {"value": data[k]}})
229 if k + "Progress" in data:
230 result[-1]["data"]["progress"] = data[k + "Progress"]
231 if "chart" in data:
232 result[-1]["data"]["chart"] = []
233 for c in data["chart"]:
234 result[-1]["data"]["chart"].append({"timestamp": c["timestamp"], "value": c[k]})
235 return result
238def get_issue_title(issue_type):
239 return {'click_rage': "Click Rage",
240 'dead_click': "Dead Click",
241 'excessive_scrolling': "Excessive Scrolling",
242 'bad_request': "Bad Request",
243 'missing_resource': "Missing Image",
244 'memory': "High Memory Usage",
245 'cpu': "High CPU",
246 'slow_resource': "Slow Resource",
247 'slow_page_load': "Slow Page",
248 'crash': "Crash",
249 'ml_cpu': "High CPU",
250 'ml_memory': "High Memory Usage",
251 'ml_dead_click': "Dead Click",
252 'ml_click_rage': "Click Rage",
253 'ml_mouse_thrashing': "Mouse Thrashing",
254 'ml_excessive_scrolling': "Excessive Scrolling",
255 'ml_slow_resources': "Slow Resource",
256 'custom': "Custom Event",
257 'js_exception': "Error",
258 'custom_event_error': "Custom Error",
259 'js_error': "Error",
260 "mouse_thrashing": "Mouse Thrashing"}.get(issue_type, issue_type)
263def __progress(old_val, new_val):
264 return ((old_val - new_val) / new_val) * 100 if new_val > 0 else 0 if old_val == 0 else 100
267def __decimal_limit(value, limit):
268 factor = pow(10, limit)
269 value = math.floor(value * factor)
270 if value % factor == 0:
271 return value // factor
272 return value / factor
275def old_search_payload_to_flat(values):
276 # in case the old search body was passed
277 if values.get("events") is not None:
278 for v in values["events"]:
279 v["isEvent"] = True
280 for v in values.get("filters", []):
281 v["isEvent"] = False
282 values["filters"] = values.pop("events") + values.get("filters", [])
283 return values
286def custom_alert_to_front(values):
287 # to support frontend format for payload
288 if values.get("seriesId") is not None and values["query"]["left"] == schemas.AlertColumn.CUSTOM: 288 ↛ 289line 288 didn't jump to line 289 because the condition on line 288 was never true
289 values["query"]["left"] = values["seriesId"]
290 values["seriesId"] = None
291 return values
294def __time_value(row):
295 row["unit"] = schemas.TemplatePredefinedUnits.MILLISECOND
296 factor = 1
297 if row["value"] > TimeUTC.MS_MINUTE:
298 row["value"] = row["value"] / TimeUTC.MS_MINUTE
299 row["unit"] = schemas.TemplatePredefinedUnits.MINUTE
300 factor = TimeUTC.MS_MINUTE
301 elif row["value"] > 1 * 1000:
302 row["value"] = row["value"] / 1000
303 row["unit"] = schemas.TemplatePredefinedUnits.SECOND
304 factor = 1000
306 if "chart" in row and factor > 1:
307 for r in row["chart"]:
308 r["value"] /= factor
311def is_saml2_available():
312 return config("hastSAML2", default=False, cast=bool)
315def get_domain():
316 _url = config("SITE_URL")
317 if not _url.startswith("http"):
318 _url = "http://" + _url
319 return '.'.join(urlparse(_url).netloc.split(".")[-2:])
322def obfuscate(text, keep_last: int = 4):
323 if text is None or not isinstance(text, str): 323 ↛ 324line 323 didn't jump to line 324 because the condition on line 323 was never true
324 return text
325 if len(text) <= keep_last:
326 return "*" * len(text)
327 return "*" * (len(text) - keep_last) + text[-keep_last:]
330def cast_session_id_to_string(data):
331 if not isinstance(data, dict) and not isinstance(data, list):
332 return data
333 if isinstance(data, list):
334 for i, item in enumerate(data):
335 data[i] = cast_session_id_to_string(item)
336 elif isinstance(data, dict): 336 ↛ 343line 336 didn't jump to line 343 because the condition on line 336 was always true
337 keys = data.keys()
338 if "sessionId" in keys: 338 ↛ 339line 338 didn't jump to line 339 because the condition on line 338 was never true
339 data["sessionId"] = str(data["sessionId"])
340 else:
341 for key in keys:
342 data[key] = cast_session_id_to_string(data[key])
343 return data
346def string_to_id(s: str):
347 if s is None:
348 return None
349 return str(int(hashlib.md5(s.encode()).hexdigest(), 16))