from __future__ import annotations import re from html.parser import HTMLParser from typing import Any from urllib.parse import parse_qsl, unquote, urlparse from app.discovery.enums import FilterType SKIP_OPTION_LABELS = { "alle", "entfernen", "überspringen", "aktualisieren", "abonnieren", "filter zurücksetzen", "nur beliebteste filter anzeigen", "immer alle filter anzeigen", "filter überspringen", "einfache filter", "erweiterte filter", "weitere", "weniger", "alle anzeigen", "weniger anzeigen", } PRODUCT_COUNT_RE = re.compile(r"\(\d[\d.]*\s*(produkte)?\)\s*$", re.IGNORECASE) OPTION_TAGS = {"a", "button", "label", "input", "span"} GROUP_ORDER = ( "Allgemein", "Display", "CPU", "RAM", "RAM erweitert", "Speicher", "Grafik", "Anschlüsse", "Wireless", "Mobilfunk", "Eingabe", "Betriebssystem", "Akku", "Abmessungen", "Sonstiges", ) RANGE_MIN_NAMES = {"bpmin", "minp", "prmin"} RANGE_MAX_NAMES = {"bpmax", "maxp", "prmax"} TEXT_NAMES = {"asuch", "fs"} HEADING_TAGS = {"th", "h2", "h3", "h4", "h5", "legend", "dt", "caption", "label", "strong"} NUMBER_RE = re.compile(r"(-?\d+(?:[.,]\d+)?)") THRESHOLD_RE = re.compile(r"^(ab|bis)\s+", re.IGNORECASE) UNIT_RE = re.compile(r"(GB|TB|GHz|MHz|Hz|cd/m²|ppi|kg|\"|″|mm|%|€|EUR)$", re.IGNORECASE) TYPE_MAP = { "text": FilterType.ENUM.value, "color": FilterType.ENUM.value, "text_range": FilterType.NUMERIC_THRESHOLD.value, "numeric_range": FilterType.NUMERIC_RANGE.value, } class _Node: def __init__(self, tag: str, attrs: dict[str, str]) -> None: self.tag = tag self.attrs = attrs self.children: list[_Node | str] = [] self.parent: _Node | None = None class _TreeBuilder(HTMLParser): VOID = { "area", "base", "br", "col", "embed", "hr", "img", "input", "link", "meta", "param", "source", "track", "wbr", } def __init__(self) -> None: super().__init__(convert_charrefs=True) self.root = _Node("document", {}) self._stack = [self.root] def handle_starttag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None: node = _Node(tag, {key: value or "" for key, value in attrs}) parent = self._stack[-1] node.parent = parent parent.children.append(node) if tag not in self.VOID: self._stack.append(node) def handle_endtag(self, tag: str) -> None: for index in range(len(self._stack) - 1, 0, -1): if self._stack[index].tag == tag: del self._stack[index:] break def handle_data(self, data: str) -> None: text = data.strip() if text: self._stack[-1].children.append(text) def parse_html_tree(html: str) -> _Node: builder = _TreeBuilder() builder.feed(html) return builder.root def node_text(node: _Node) -> str: return _node_text(node) def parse_taxonomy_html(html: str, *, category_ref: str, source_url: str) -> dict[str, Any]: if is_challenge_html(html): return { "source": "geizhals", "external_category_key": category_ref, "name": category_ref, "source_url": source_url, "filters": [], "blocked": True, } builder = _TreeBuilder() builder.feed(html) groups: dict[str, dict[str, Any]] = {} _collect_declared_filters(builder.root, groups) _collect_xf_links(builder.root, groups) _collect_named_inputs(builder.root, groups) filters = [ group for group in groups.values() if _usable_group(group) ] for group in filters: section = group.get("group_label") or "Allgemein" group["group_label"] = section group["group_sort"] = _group_sort(section) filters.sort(key=lambda item: (item["group_sort"], item["group_label"].lower(), item["label"].lower(), item["external_filter_key"])) return { "source": "geizhals", "external_category_key": category_ref, "name": _category_name(builder.root, category_ref), "canonical_category": "notebook" if category_ref == "nb" else None, "source_url": source_url, "filters": filters, "blocked": False, } def is_challenge_html(html: str) -> bool: sample = html.lower() markers = ( "cf-mitigated", "cf-challenge", "just a moment", "enable javascript and cookies", "checking your browser", "verification successful. waiting for geizhals", ) return any(marker in sample for marker in markers) def _collect_declared_filters(node: _Node, groups: dict[str, dict[str, Any]]) -> None: filter_id = node.attrs.get("data-filter-id") if filter_id: source_type = node.attrs.get("data-filter-type") or "" label = _declared_label(node) or f"Filter {filter_id}" group = groups.setdefault(filter_id, _empty_group(filter_id, label, source_type)) section = _section_label(node) if section: group["group_label"] = section if label and group["label"].startswith("Filter "): group["label"] = label if source_type: group["filter_type"] = TYPE_MAP.get(source_type, group["filter_type"]) if source_type == "numeric_range": group["encoding"]["encoding_type"] = "RANGE_TOKEN" group["encoding"]["value_template"] = "{min}*{max}" group.setdefault("range", {"minimum": None, "maximum": None, "step": None, "unit": group.get("canonical_unit")}) _collect_xf_links(node, groups, required_key=filter_id, allow_histogram=source_type == "numeric_range") return for child in node.children: if isinstance(child, _Node): _collect_declared_filters(child, groups) def _collect_xf_links( node: _Node, groups: dict[str, dict[str, Any]], required_key: str | None = None, allow_histogram: bool = False, ) -> None: if node.tag in OPTION_TAGS: label = node.attrs.get("aria-label") or node.attrs.get("title") or _node_text(node) if _usable_option(node, label, allow_histogram=allow_histogram): for key, query_value in _node_xf_tokens(node): if required_key is not None and key != required_key: continue if required_key is None and not _has_structural_label(node): continue filter_label = _declared_label(_filter_root(node)) or _nearest_group_label(node) group = groups.setdefault(key, _empty_group(key, filter_label, None)) section = _section_label(node) if section: group["group_label"] = section if filter_label and (group["label"].startswith("Filter ") or group["label"] == key): group["label"] = filter_label option = _option_from_link(label, query_value) existing = {item["source_value"] for item in group["options"]} if option["source_value"] not in existing: group["options"].append(option) group["encoding"].setdefault("options", {}) group["encoding"]["options"][option["source_value"]] = query_value if option.get("normalized_value") is not None: group["encoding"]["options"][str(option["normalized_value"])] = query_value if group["filter_type"] not in {FilterType.NUMERIC_RANGE.value, FilterType.NUMERIC_THRESHOLD.value}: group["filter_type"] = _classify_options(group["options"]) if group["canonical_unit"] is None: group["canonical_unit"] = option.get("unit") for child in node.children: if isinstance(child, _Node): _collect_xf_links(child, groups, required_key=required_key, allow_histogram=allow_histogram) def _collect_named_inputs(node: _Node, groups: dict[str, dict[str, Any]]) -> None: inputs: list[_Node] = [] _find_inputs(node, inputs) names = {item.attrs.get("name", ""): item for item in inputs if item.attrs.get("name")} if names.keys() & RANGE_MIN_NAMES or names.keys() & RANGE_MAX_NAMES: min_name = next((name for name in RANGE_MIN_NAMES if name in names), "bpmin") max_name = next((name for name in RANGE_MAX_NAMES if name in names), "bpmax") label = _nearest_group_label(names.get(min_name) or names.get(max_name) or node) or "Preis" section = _section_label(names.get(min_name) or names.get(max_name) or node) or "Allgemein" groups["price"] = { "external_filter_key": "price", "label": "Preis" if "preis" not in label.lower() else label, "filter_type": FilterType.NUMERIC_RANGE.value, "canonical_unit": "EUR", "group_label": section, "group_sort": _group_sort(section), "options": [], "range": {"minimum": 0, "maximum": None, "step": 1, "unit": "EUR"}, "encoding": { "url_param": min_name, "min_parameter": min_name, "max_parameter": max_name, "encoding_type": "RANGE_PARAMETERS", }, } for name in TEXT_NAMES: if name not in names: continue label = _nearest_group_label(names[name]) or "Suchbegriffe" section = _section_label(names[name]) or "Allgemein" groups[name] = { "external_filter_key": name, "label": label, "filter_type": FilterType.TEXT.value, "canonical_unit": None, "group_label": section, "group_sort": _group_sort(section), "options": [], "encoding": {"url_param": name, "encoding_type": "TEXT_PARAMETER", "value_template": "{value}"}, } def _find_inputs(node: _Node, found: list[_Node]) -> None: if node.tag == "input": found.append(node) for child in node.children: if isinstance(child, _Node): _find_inputs(child, found) def _node_xf_tokens(node: _Node) -> list[tuple[str, str]]: tokens = _xf_tokens(node.attrs.get("href") or "") if tokens: return tokens for attr in ("data-xf", "data-filter-value", "data-value", "value"): raw = node.attrs.get(attr) or "" if attr == "value" and node.attrs.get("name") not in {"xf", ""} and node.tag != "input": continue parsed = _xf_raw_token(raw) if parsed: return [parsed] return [] def _xf_raw_token(raw: str) -> tuple[str, str] | None: value = unquote(raw).replace("|", "~").split("~")[0] if "_" not in value: return None key, token = value.split("_", 1) if key.isdigit() and token: return key, token.replace("+", " ") return None def _xf_tokens(href: str) -> list[tuple[str, str]]: if not href or href.startswith("#") or href.startswith("javascript:"): return [] query = urlparse(href).query tokens: list[tuple[str, str]] = [] for name, value in parse_qsl(query, keep_blank_values=True): if name != "xf" or not value: continue for part in unquote(value).replace("|", "~").split("~"): if "_" not in part: continue key, raw = part.split("_", 1) if key.isdigit() and raw: tokens.append((key, raw.replace("+", " "))) return tokens def _option_from_link(label: str, query_value: str) -> dict[str, Any]: cleaned = PRODUCT_COUNT_RE.sub("", re.sub(r"\s+", " ", label)).strip() operator = None if cleaned.lower().startswith("ab "): operator = ">=" elif cleaned.lower().startswith("bis "): operator = "<=" number = _first_number(cleaned) unit = _unit(cleaned) return { "source_value": cleaned or query_value, "query_value": query_value, "normalized_value": number, "operator": operator, "unit": unit, } def _classify_options(options: list[dict[str, Any]]) -> str: if not options: return FilterType.UNKNOWN.value labels = [str(item["source_value"]).lower() for item in options] if all(item in {"ja", "nein", "egal", "true", "false"} for item in labels): return FilterType.BOOLEAN.value if all(THRESHOLD_RE.match(str(item["source_value"])) for item in options): return FilterType.NUMERIC_THRESHOLD.value return FilterType.ENUM.value def _empty_group(key: str, label: str | None, source_type: str | None) -> dict[str, Any]: encoding_type = "OPTION_TOKEN" if source_type == "numeric_range": encoding_type = "RANGE_TOKEN" return { "external_filter_key": key, "label": label or f"Filter {key}", "filter_type": TYPE_MAP.get(source_type or "", FilterType.ENUM.value), "canonical_unit": None, "group_label": None, "group_sort": _group_sort("Allgemein"), "options": [], "encoding": { "url_param": "xf", "encoding_type": encoding_type, "options": {}, **({"value_template": "{min}*{max}"} if encoding_type == "RANGE_TOKEN" else {}), }, } def _declared_label(node: _Node | None) -> str | None: if node is None: return None attr_id = node.attrs.get("id", "") if attr_id.startswith("filter-element__title-label"): return _clean_label(_node_text(node)) or None for child in node.children: if isinstance(child, _Node): found = _declared_label(child) if found: return found return None def _filter_root(node: _Node) -> _Node | None: current: _Node | None = node while current is not None: if current.attrs.get("data-filter-id"): return current current = current.parent return None def _nearest_group_label(node: _Node | None) -> str | None: current = node while current is not None: declared = _declared_label(current) if declared: return declared if current.tag in HEADING_TAGS: text = _clean_label(_node_text(current)) if text: return text previous = _previous_element(current) while previous is not None: if previous.tag in HEADING_TAGS: text = _clean_label(_node_text(previous)) if text: return text previous = _previous_element(previous) labeled = current.attrs.get("aria-label") or current.attrs.get("data-filter-name") or current.attrs.get("title") if labeled: text = _clean_label(labeled) if text and len(text.split()) <= 6: return text current = current.parent return None def _has_structural_label(node: _Node) -> bool: if _filter_root(node) is not None: return True current: _Node | None = node while current is not None: if current.tag in {"th", "td", "tr", "table", "dt", "dd"}: return _nearest_group_label(node) is not None current = current.parent return False def _usable_option(node: _Node, label: str, *, allow_histogram: bool) -> bool: classes = f"{node.attrs.get('class', '')} {node.attrs.get('id', '')}".lower() if "skip-link" in classes or "add-to-columns" in classes: return False if "bucket" in classes and not allow_histogram: return False cleaned = PRODUCT_COUNT_RE.sub("", re.sub(r"\s+", " ", label)).strip() if not cleaned or _skip_option(cleaned): return False if "überspringen" in cleaned.lower() or "aufklappen" in cleaned.lower() or "zuklappen" in cleaned.lower(): return False if cleaned.lower().startswith("weitere"): return False if not allow_histogram and re.fullmatch(r"\d[\d.]*\s*produkte", cleaned, re.IGNORECASE): return False if len(cleaned.split()) > 12: return False return True def _section_label(node: _Node | None) -> str | None: current = node while current is not None: classes = current.attrs.get("class", "").lower() if "filter-row" in classes or current.attrs.get("data-filter-row-id"): text = _strip_section_suffix( current.attrs.get("data-filter-row-name") or current.attrs.get("aria-label") or _row_heading(current) ) if text: return text labeled = current.attrs.get("aria-label") or "" if re.search(r"(auf|zu)klappen", labeled, re.IGNORECASE): text = _strip_section_suffix(labeled) if text: return text current = current.parent return None def _row_heading(node: _Node) -> str: for child in node.children: if isinstance(child, _Node) and child.tag in {"button", "h2", "h3", "h4", "legend", "strong"}: text = _clean_label(_node_text(child)) if text: return text return _clean_label(node.attrs.get("aria-label") or "") def _strip_section_suffix(text: str | None) -> str: cleaned = _clean_label(text or "") return re.sub(r"\s*(aufklappen|zuklappen|einklappen)$", "", cleaned, flags=re.IGNORECASE).strip() def _group_sort(label: str) -> int: try: return GROUP_ORDER.index(label) except ValueError: return len(GROUP_ORDER) def _usable_group(group: dict[str, Any]) -> bool: if group["filter_type"] == FilterType.TEXT.value: return True if group.get("range") and group["filter_type"] == FilterType.NUMERIC_RANGE.value: return True if not group.get("options") and not group.get("range"): return False label = str(group.get("label") or "") if label.startswith("Filter ") and len(group.get("options") or []) < 2: return False if len(label.split()) > 8: return False return True def _previous_element(node: _Node) -> _Node | None: if node.parent is None: return None siblings = [child for child in node.parent.children if isinstance(child, _Node)] try: index = siblings.index(node) except ValueError: return None return siblings[index - 1] if index > 0 else None def _node_text(node: _Node) -> str: parts: list[str] = [] for child in node.children: if isinstance(child, str): parts.append(child) else: parts.append(_node_text(child)) return " ".join(part for part in parts if part).strip() def _clean_label(text: str) -> str: cleaned = re.sub(r"\s+", " ", text).strip().rstrip(":") cleaned = re.sub(r"\s*(überspringen|entfernen)$", "", cleaned, flags=re.IGNORECASE).strip() if not cleaned or _skip_option(cleaned): return "" return cleaned def _skip_option(label: str) -> bool: return label.strip().lower() in SKIP_OPTION_LABELS def _first_number(text: str) -> float | int | None: match = NUMBER_RE.search(text.replace(",", ".")) if match is None: return None raw = match.group(1) value = float(raw) return int(value) if value.is_integer() else value def _unit(text: str) -> str | None: match = UNIT_RE.search(text.replace(" ", "")) if match is None: if "kg" in text.lower(): return "kg" return None unit = match.group(1) if unit in {'"', "″"}: return "inch" return unit def _category_name(root: _Node, fallback: str) -> str: titles: list[str] = [] def walk(node: _Node) -> None: if node.tag in {"h1", "title"}: text = _node_text(node) if text: titles.append(text) for child in node.children: if isinstance(child, _Node): walk(child) walk(root) if titles: return titles[0].split("»")[0].strip() or fallback return "Notebooks" if fallback == "nb" else fallback