from __future__ import annotations from typing import Any from urllib.parse import parse_qsl, quote, urlencode, urljoin, urlsplit, urlunsplit from app.discovery.enums import MappingStatus def generate_search_url( *, category_url: str, criteria_items: list[dict[str, Any]], mappings: list[dict[str, Any]], ) -> tuple[str, list[str]]: """Build a source URL from VERIFIED mappings only. INTERNAL_ONLY keys are skipped.""" by_criterion = {item["criterion_key"]: item for item in mappings if item.get("status") == MappingStatus.VERIFIED.value} fragments: list[tuple[str, str]] = [] internal_only: list[str] = [] for criterion in criteria_items: key = str(criterion["criterion"]) mapping = by_criterion.get(key) if mapping is None: internal_only.append(key) continue fragments.extend(_encode_mapping(mapping, criterion)) fragments.sort() base = urlsplit(category_url) query = dict(parse_qsl(base.query, keep_blank_values=True)) extra: list[tuple[str, str]] = [] xf_values: list[str] = [] for name, value in fragments: if name == "xf": xf_values.append(value) continue if name in query: query[name] = value else: extra.append((name, value)) if xf_values: extra.append(("xf", "~".join(xf_values))) encoded = urlencode(list(query.items()) + extra, doseq=True, safe="+", quote_via=quote) url = urlunsplit((base.scheme, base.netloc, base.path, encoded, base.fragment)) return url, internal_only def _encode_mapping(mapping: dict[str, Any], criterion: dict[str, Any]) -> list[tuple[str, str]]: encoding = mapping.get("encoding") or {} param = str(encoding.get("url_param") or "xf") if encoding.get("encoding_type") == "RANGE_PARAMETERS" or param in {"bpmin", "bpmax"}: return _encode_range(encoding, criterion) if encoding.get("encoding_type") == "RANGE_TOKEN": return _encode_range_token(mapping, criterion) if param == "xf": return _encode_xf(mapping, criterion) if param in {"minp", "maxp", "prmin", "prmax", "asuch", "fs"}: return [(param, str(criterion.get("value")))] template = encoding.get("value_template") if template: return [(param, str(template).format(value=criterion.get("value"), **mapping))] raise ValueError(f"unsupported encoding for mapping {mapping.get('criterion_key')}") def _encode_range(encoding: dict[str, Any], criterion: dict[str, Any]) -> list[tuple[str, str]]: operator = str(criterion.get("operator") or "<=") value = criterion.get("value") min_param = str(encoding.get("min_parameter") or encoding.get("url_param") or "bpmin") max_param = str(encoding.get("max_parameter") or "bpmax") if operator == "between": if isinstance(value, dict): items: list[tuple[str, str]] = [] if value.get("min") is not None: items.append((min_param, str(value["min"]))) if value.get("max") is not None: items.append((max_param, str(value["max"]))) return items if isinstance(value, list) and len(value) == 2: return [(min_param, str(value[0])), (max_param, str(value[1]))] if operator in {">=", ">"}: return [(min_param, str(value))] return [(max_param, str(value))] def _encode_range_token(mapping: dict[str, Any], criterion: dict[str, Any]) -> list[tuple[str, str]]: filter_key = str(mapping["external_filter_key"]) looked_up = _lookup_option(mapping, criterion.get("value")) if looked_up is not None: return [("xf", f"{filter_key}_{_plus(looked_up)}")] operator = str(criterion.get("operator") or "<=") value = criterion.get("value") if operator == "between": if isinstance(value, dict): raw = f"{value.get('min') or ''}*{value.get('max') or ''}" elif isinstance(value, list) and len(value) == 2: raw = f"{value[0]}*{value[1]}" else: raw = f"*{value}" elif operator in {">=", ">"}: raw = f"{value}*" else: raw = f"*{value}" return [("xf", f"{filter_key}_{raw}")] def _lookup_option(mapping: dict[str, Any], value: Any) -> str | None: options = (mapping.get("encoding") or {}).get("options") or {} if value in options: return str(options[value]) if str(value) in options: return str(options[str(value)]) return None def _encode_xf(mapping: dict[str, Any], criterion: dict[str, Any]) -> list[tuple[str, str]]: filter_key = str(mapping["external_filter_key"]) raw = criterion.get("value") values = raw if isinstance(raw, list) else [raw] fragments: list[tuple[str, str]] = [] for value in values: if value is None or value == "": continue source_value = _xf_value(mapping, {**criterion, "value": value}) fragments.append(("xf", f"{filter_key}_{_plus(source_value)}")) if fragments: return fragments source_value = _xf_value(mapping, criterion) return [("xf", f"{filter_key}_{_plus(source_value)}")] def _xf_value(mapping: dict[str, Any], criterion: dict[str, Any]) -> str: looked_up = _lookup_option(mapping, criterion.get("value")) if looked_up is not None: return looked_up options = (mapping.get("encoding") or {}).get("options") or {} value = criterion.get("value") if options and value is not None and value != "": return str(value) return str(mapping.get("source_value") or _threshold_value(mapping, criterion)) def _threshold_value(mapping: dict[str, Any], criterion: dict[str, Any]) -> str: option = mapping.get("source_value") if option: return str(option) value = criterion.get("value") unit = mapping.get("unit") or "" operator = criterion.get("operator") if operator in {">=", ">"} and unit: return f"ab {value}{unit}" return str(value) def _plus(value: str) -> str: return value.replace(" ", "+") def decode_xf_roundtrip(url: str) -> list[tuple[str, str]]: parts = urlsplit(url) decoded: list[tuple[str, str]] = [] for name, value in parse_qsl(parts.query, keep_blank_values=True): if name != "xf": continue for part in value.replace("|", "~").split("~"): if "_" not in part: continue key, raw = part.split("_", 1) decoded.append((key, raw.replace("+", " "))) return decoded def same_origin_url(base: str, maybe_relative: str) -> str: return urljoin(base, maybe_relative)