from functools import reduce from typing import List, Optional from service.tasks.analytics import get_group_map from service.utils.aws_connectors import run_query from service.utils.data_model_utils import ( get_collection_attribute_names, get_collection_type, get_parent, get_related_collections, update_collection, ) from .filters import ( add_input_to_filter_conf, get_filter, get_global_filter_configurations, get_global_filter_map_id, ) from .sets import Set def apply_filters( schema: str, user_id: str, source_id: int = None, set_id: int = None, filter_input: List = None, ): """Apply filters, save collection_fan""" filter_input = filter_input or [] if set_id is not None: # existing set? set_id = int(set_id) set_type = get_collection_type(schema, set_id) if set_type is not None and set_type in ["set", "segment"]: # existing set return ( Set(schema, user_id, set_id=set_id) .existing(filter_input) .json_for_appsync() ) else: raise RuntimeError("Unknown set") else: return Set(schema, user_id).new(filter_input, source_id).json_for_appsync() def get_prepared_filters(schema: str, set_id: int = None): # retrieve_schema(schema) if set_id is None: available_attributes = get_collection_attribute_names(schema, set_id) prepared_filters = [ get_filter(conf, schema).prepare_options() for conf in get_global_filter_configurations(for_frontend=True) if "attribute_name" not in conf or conf["attribute_name"] in available_attributes ] return prepared_filters set_type = get_collection_type(schema, set_id) if set_type in ["source", "enrichment"]: available_attributes = get_collection_attribute_names(schema, set_id) prepared_filters = [] for conf in get_global_filter_configurations(for_frontend=True): if ( "attribute_name" in conf and conf["attribute_name"] in available_attributes ): # hack collection_id into input parameters add_input_to_filter_conf(conf, collection_id=set_id) prepared_filters.append(get_filter(conf, schema).prepare_options()) return prepared_filters prepared_filters = [] if set_type in ["set", "segment"]: saved_filters = get_saved_filters(schema, set_id) saved_flt_fids = [flt["id"] for flt in saved_filters if flt["values"]] filter_map = get_global_filter_map_id() if set_type == "segment": # extract ids from flt 0 cid_set = set() for flt in saved_filters: if flt["id"] == 0: for cid in flt["values"]: prnt = get_parent(schema, cid) if prnt is not None: cid_set.add(prnt) break cid_set.update( get_related_collections( schema, set_id, only_types=["source", "enrichment", "algo"], include_segment_sources=True, ) ) available_attributes = set() for cid in cid_set: available_attributes.update(get_collection_attribute_names(schema, cid)) for conf in get_global_filter_configurations(): if ( "attribute_name" in conf and conf["attribute_name"] in available_attributes ): # only show filters with no previous settings if conf["fid"] not in saved_flt_fids: # hack collection_id into input parameters add_input_to_filter_conf(conf, collection_id=set_id) prepared_filters.append( get_filter(conf, schema).prepare_options() ) else: prepared_filters = [ get_filter(filter_map[int(saved["id"])], schema).prepare_options() for saved in saved_filters ] return prepared_filters raise RuntimeError("That's not how filters work") def get_filters(schema: str, set_id: int = None): prepared_filters = get_prepared_filters(schema, set_id) return [ jfa for jfa in [f.json_for_appsync() for f in prepared_filters] if jfa["options"] ] def get_grouped_filters(schema: str, set_id: int = None): """Return applicable analytics in groups""" prepared_filters = get_prepared_filters(schema, set_id) groups = get_group_map("filters") for flt in prepared_filters: jfa = flt.json_for_appsync() if jfa["options"]: groups[flt.conf["group_tag"]]["filters"].append(flt.json_for_appsync()) return [g for g in groups.values() if g["filters"]] def get_saved_filters(schema: str, set_id: int = None): query = f"SELECT filter_config_id, filter_type, filter_params FROM {schema}.set_filter WHERE set_id = %(set_id)s" res = run_query(query, {"set_id": set_id}) if res: return [{"id": r[0], "type": r[1], "values": r[2]} for r in res] else: return [] def save_segment(schema: str, user_id: str, set_id: int, new_name: str = None): """reassigns set as segment type and gives it new name, that's about it.""" set_id = int(set_id) current_type = get_collection_type(schema, set_id) if current_type == "set": collection = update_collection( schema, user_id, set_id, name=new_name, collection_type="segment" ) return collection raise RuntimeError("You can only save sets as segments") def re_pre_calculate_filter_config(schema: str, collection_id: int = None): """Re and/or Pre Calculates query data for affected charts It's assumed, that when collection_id is given, it's a new freshly added/modified collection, so the full data charts need to be recalculated as well :returns True, if all charts were calculated error free, False, if there were some errors. """ recalculating = [ get_filter(conf, schema, recalculate=True) for conf in get_global_filter_configurations() ] if collection_id is not None: available_attributes = get_collection_attribute_names(schema, collection_id) for conf in get_global_filter_configurations(): if ( "attribute_name" in conf and conf["attribute_name"] in available_attributes ): # hack collection_id into input parameters add_input_to_filter_conf(conf, collection_id=collection_id) recalculating.append(get_filter(conf, schema, recalculate=True)) if recalculating: return reduce( lambda x, y: x and y, [flt.wait_for_done() for flt in recalculating] ) return True def create_segment( schema: str, user_id: str, name: str, parent_collection_id, filter_id, filter_values: List, ): filter_input = [ { "id": "0", "values": [ str(parent_collection_id), ], }, {"id": str(filter_id), "values": [str(fv) for fv in filter_values]}, ] return ( Set(schema, user_id) .new_segment(filter_input, collection_id=parent_collection_id, name=name) .json_for_appsync() ) def create_audience_collection( schema: str, user_id: str, name: str = "", collection_ids: Optional[List] = None ): collection_ids = collection_ids or [] filter_input = [ {"id": "0", "values": [str(cid) for cid in collection_ids]}, ] return ( Set(schema, user_id) .new_audience( filter_input=filter_input, collection_id=collection_ids[0], name=name ) .json_for_appsync() )