147 lines
5.3 KiB
Python
147 lines
5.3 KiB
Python
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
from urllib.parse import parse_qs, urlsplit
|
|
|
|
import httpx
|
|
|
|
from app.discovery.enums import FilterType
|
|
from app.sources.geizhals.live_page import BROWSER_HEADERS
|
|
from app.sources.geizhals.taxonomy_parser import _group_sort, _option_from_link
|
|
|
|
TYPE_MAP = {
|
|
"enum": FilterType.ENUM.value,
|
|
"substring": FilterType.ENUM.value,
|
|
"range_from": FilterType.NUMERIC_THRESHOLD.value,
|
|
"range_to": FilterType.NUMERIC_THRESHOLD.value,
|
|
}
|
|
GROUP_RULES = (
|
|
("Anschlüsse", ("anschluss", "usb", "thunderbolt", "docking")),
|
|
("Display", ("display", "pixeldichte", "screen-to-body", "bildwiederhol", "farbraum", "farbräume")),
|
|
("CPU", ("cpu", "ai-rechen")),
|
|
("RAM erweitert", ("ram",)),
|
|
("Speicher", ("speicher", "ssd", "hdd", "flash", "m.2", "cardreader", "laufwerk")),
|
|
("Grafik", ("grafik", "gpu", "igpu", "directx")),
|
|
("Wireless", ("wlan", "wifi", "bluetooth", "wireless")),
|
|
("Mobilfunk", ("mobilfunk", "navigation")),
|
|
("Akku", ("akku", "netzteil")),
|
|
("Abmessungen", ("breite", "tiefe", "höhe", "volumen", "gewicht", "verhältnis")),
|
|
("Eingabe", ("eingabe", "tastatur", "touchpad", "webcam", "kamera")),
|
|
("Betriebssystem", ("betriebssystem", "sprachversion", "chrome auto")),
|
|
)
|
|
|
|
|
|
def categorylist_api_url(source_url: str) -> str:
|
|
parts = urlsplit(source_url)
|
|
query = parse_qs(parts.query)
|
|
category = (query.get("cat") or [""])[0] or "nb"
|
|
hloc = (query.get("hloc") or ["de"])[0]
|
|
return (
|
|
"https://geizhals.de/api/gh0/categorylist/"
|
|
f"?cat={category}&xf=&sort=t&hloc={hloc}&v=e&filters_only=1"
|
|
)
|
|
|
|
|
|
def fetch_taxonomy_api(source_url: str, *, category_ref: str) -> dict[str, Any] | None:
|
|
url = categorylist_api_url(source_url)
|
|
try:
|
|
response = httpx.get(
|
|
url,
|
|
headers={**BROWSER_HEADERS, "Accept": "application/json"},
|
|
follow_redirects=True,
|
|
timeout=25.0,
|
|
)
|
|
if response.status_code >= 400:
|
|
return None
|
|
return parse_taxonomy_api(
|
|
response.json(),
|
|
category_ref=category_ref,
|
|
source_url=source_url,
|
|
)
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
def parse_taxonomy_api(payload: dict[str, Any], *, category_ref: str, source_url: str) -> dict[str, Any]:
|
|
raw_filters = ((payload.get("response") or {}).get("filters") or [])
|
|
filters = [_filter_from_api(item) for item in raw_filters if isinstance(item, dict)]
|
|
filters = [item for item in filters if item["options"] or item.get("range")]
|
|
if not any(item["external_filter_key"] == "price" for item in filters):
|
|
filters.append(_price_filter())
|
|
filters.sort(key=lambda item: (item["group_sort"], item["group_label"].lower(), item["label"].lower()))
|
|
return {
|
|
"source": "geizhals",
|
|
"external_category_key": category_ref,
|
|
"name": "Notebooks" if category_ref == "nb" else category_ref,
|
|
"canonical_category": "notebook" if category_ref == "nb" else None,
|
|
"source_url": source_url,
|
|
"filters": filters,
|
|
"blocked": False,
|
|
"discovery_mode": "live_api",
|
|
"taxonomy_status": "READY",
|
|
}
|
|
|
|
|
|
def _filter_from_api(raw: dict[str, Any]) -> dict[str, Any]:
|
|
key = str(raw.get("id") or "")
|
|
label = str(raw.get("title") or key)
|
|
source_type = str(raw.get("type") or "enum")
|
|
filter_type = TYPE_MAP.get(source_type, FilterType.ENUM.value)
|
|
options: list[dict[str, Any]] = []
|
|
encoding_options: dict[str, str] = {}
|
|
for value in raw.get("values") or []:
|
|
query = str(value.get("id") or "")
|
|
if "_" in query:
|
|
query = query.split("_", 1)[1].replace("+", " ")
|
|
option = _option_from_link(str(value.get("name") or query), query)
|
|
if source_type == "range_from":
|
|
option["operator"] = ">="
|
|
elif source_type == "range_to":
|
|
option["operator"] = "<="
|
|
options.append(option)
|
|
encoding_options[option["source_value"]] = query
|
|
if option.get("normalized_value") is not None:
|
|
encoding_options[str(option["normalized_value"])] = query
|
|
group_label = _group_label(label)
|
|
return {
|
|
"external_filter_key": key,
|
|
"label": label,
|
|
"filter_type": filter_type,
|
|
"canonical_unit": next((item.get("unit") for item in options if item.get("unit")), None),
|
|
"group_label": group_label,
|
|
"group_sort": _group_sort(group_label),
|
|
"options": options,
|
|
"encoding": {
|
|
"url_param": "xf",
|
|
"encoding_type": "OPTION_TOKEN",
|
|
"options": encoding_options,
|
|
},
|
|
}
|
|
|
|
|
|
def _price_filter() -> dict[str, Any]:
|
|
return {
|
|
"external_filter_key": "price",
|
|
"label": "Preis",
|
|
"filter_type": FilterType.NUMERIC_RANGE.value,
|
|
"canonical_unit": "EUR",
|
|
"group_label": "Allgemein",
|
|
"group_sort": _group_sort("Allgemein"),
|
|
"options": [],
|
|
"range": {"minimum": 0, "maximum": None, "step": 1, "unit": "EUR"},
|
|
"encoding": {
|
|
"url_param": "bpmin",
|
|
"min_parameter": "bpmin",
|
|
"max_parameter": "bpmax",
|
|
"encoding_type": "RANGE_PARAMETERS",
|
|
},
|
|
}
|
|
|
|
|
|
def _group_label(title: str) -> str:
|
|
text = title.casefold()
|
|
for group, needles in GROUP_RULES:
|
|
if any(needle in text for needle in needles):
|
|
return group
|
|
return "Allgemein"
|