172 lines
6.5 KiB
Python
172 lines
6.5 KiB
Python
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)
|