Create filter_utils.py

appStore/filter_utils.py

@@ -0,0 +1,116 @@
+import re
+import json
+import streamlit as st
+from appStore.crs_utils import lookup_crs_value
+from appStore.search import hybrid_search
+from appStore.prep_data import remove_duplicates, extract_year
+
+def parse_budget(value):
+    """
+    Safely parse a budget value from string to float. Returns 0.0 on error.
+    """
+    try:
+        return float(value)
+    except:
+        return 0.0
+
+def filter_results(results, country_filter, region_filter, end_year_range, crs_filter, budget_filter,
+                   region_df, iso_code_to_sub_region, clean_country_code_fn, get_country_name_fn):
+    """
+    Filter search results by country, region, end_year range, CRS code, and budget threshold.
+
+    Args:
+        results (list): List of results from the vector DB.
+        country_filter (str): Selected country name to filter by (or "All/Not allocated").
+        region_filter (str): Selected region name to filter by (or "All/Not allocated").
+        end_year_range (tuple): (min_end_year, max_end_year) from a Streamlit slider.
+        crs_filter (str): The chosen CRS code from the dropdown (or "All/Not allocated").
+        budget_filter (float): Minimum budget in million EUR.
+        region_df (pd.DataFrame): Region lookup DataFrame.
+        iso_code_to_sub_region (dict): Mapping from alpha-2 country code to sub-region name.
+        clean_country_code_fn (callable): Function that normalizes country code strings.
+        get_country_name_fn (callable): Function that returns the country name from alpha-2 code.
+
+    Returns:
+        list: Filtered list of results that match all criteria.
+    """
+    filtered = []
+    for r in results:
+        metadata = r.payload.get('metadata', {})
+        country = metadata.get('country', "[]")
+        year_str = metadata.get('end_year')
+        if year_str:
+            extracted = extract_year(year_str)
+            try:
+                end_year_val = int(extracted) if extracted != "Unknown" else 0
+            except ValueError:
+                end_year_val = 0
+        else:
+            end_year_val = 0
+
+        # Parse country field as a JSON list or a single string
+        if country.strip().startswith("["):
+            try:
+                parsed_country = json.loads(country.replace("'", '"'))
+                if isinstance(parsed_country, str):
+                    country_list = [parsed_country]
+                else:
+                    country_list = parsed_country
+            except json.JSONDecodeError:
+                country_list = []
+        else:
+            country_list = [country.strip()]
+
+        c_list = [clean_country_code_fn(code) for code in country_list if len(clean_country_code_fn(code)) == 2]
+        resolved_names = [get_country_name_fn(code, region_df) for code in c_list]
+
+        # Check if any of the codes match the chosen region
+        country_in_region = any(
+            iso_code_to_sub_region.get(code, "Not allocated") == region_filter
+            for code in c_list
+        )
+
+        # CRS filtering
+        crs_key = metadata.get("crs_key", "").strip()
+        crs_key_clean = re.sub(r'\.0$', '', str(crs_key))
+        new_crs_value = lookup_crs_value(crs_key_clean)
+        new_crs_value_clean = re.sub(r'\.0$', '', str(new_crs_value).strip())
+        crs_combined = f"{crs_key_clean}: {new_crs_value_clean}" if crs_key_clean else ""
+
+        if crs_filter != "All/Not allocated" and crs_combined:
+            if crs_filter != crs_combined:
+                continue
+
+        # Budget filtering
+        budget_value = parse_budget(metadata.get('total_project', "0"))
+        if budget_value < (budget_filter * 1e6):
+            continue
+
+        year_ok = True if end_year_val == 0 else (end_year_range[0] <= end_year_val <= end_year_range[1])
+
+        if ((country_filter == "All/Not allocated" or (country_filter in resolved_names))
+            and (region_filter == "All/Not allocated" or country_in_region)
+            and year_ok):
+            filtered.append(r)
+
+    return filtered
+
+@st.cache_data(show_spinner=False)
+def get_crs_options(_client, collection_name):
+    """
+    Return a sorted list of all CRS code+description combos found in the collection.
+    Used to populate the 'CRS' dropdown in the app.
+    """
+    results = hybrid_search(_client, "", collection_name)
+    all_results = results[0] + results[1]
+    crs_set = set()
+    for res in all_results:
+        metadata = res.payload.get('metadata', {})
+        raw_crs_key = metadata.get("crs_key", "")
+        crs_key_clean = re.sub(r'\.0$', '', str(raw_crs_key).strip())
+        if crs_key_clean:
+            new_value_raw = lookup_crs_value(crs_key_clean)
+            new_value_clean = re.sub(r'\.0$', '', str(new_value_raw).strip())
+            crs_combined = f"{crs_key_clean}: {new_value_clean}"
+            crs_set.add(crs_combined)
+    return sorted(crs_set)