83 lines
3.1 KiB
Python
83 lines
3.1 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]] = []
|
|
for name, value in fragments:
|
|
if name in query and name != "xf":
|
|
query[name] = value
|
|
else:
|
|
extra.append((name, value))
|
|
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 param == "xf":
|
|
filter_key = str(mapping["external_filter_key"])
|
|
source_value = str(mapping.get("source_value") or _threshold_value(mapping, criterion))
|
|
return [("xf", f"{filter_key}_{_plus(source_value)}")]
|
|
if param in {"minp", "maxp", "prmin", "prmax"}:
|
|
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 _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" and "_" in value:
|
|
key, raw = value.split("_", 1)
|
|
decoded.append((key, raw.replace("+", " ")))
|
|
return decoded
|
|
|
|
|
|
def same_origin_url(base: str, maybe_relative: str) -> str:
|
|
return urljoin(base, maybe_relative)
|