from __future__ import annotations import re from typing import Any from urllib.parse import parse_qsl, urlencode, urljoin, urlsplit from app.sources.geizhals.taxonomy_parser import is_challenge_html, node_text, parse_html_tree PRODUCT_ID_RE = re.compile(r"-a(\d+)\.html", re.IGNORECASE) PRICE_RE = re.compile(r"(\d{1,3}(?:\.\d{3})+|\d+),(\d{2})") OFFER_RE = re.compile(r"(\d+)\s*Angebote?", re.IGNORECASE) PRODUCT_COUNT_RE = re.compile(r"(\d[\d.]*)\s*Produkte", re.IGNORECASE) CAPACITY_RE = re.compile(r"(\d+(?:[.,]\d+)?)\s*(TB|GB)", re.IGNORECASE) WEIGHT_RE = re.compile(r"(\d+(?:[.,]\d+)?)\s*kg", re.IGNORECASE) AVAILABILITY_BY_CODE = {"0": "out_of_stock", "1": "limited", "2": "in_stock"} AVAILABILITY_BY_LABEL = { "lagernd": "in_stock", "kurzfristig lieferbar": "limited", "kurzfristig": "limited", "nicht verfügbar": "out_of_stock", } def parse_result_html(html: str, *, page_url: str) -> dict[str, Any]: if is_challenge_html(html): return {"items": [], "page_count": 1, "result_count": 0, "blocked": True} root = parse_html_tree(html) cards = _find_cards(root) items: list[dict[str, Any]] = [] seen: set[str] = set() for card in cards: item = _parse_card(card, page_url) if item is None or item["source_external_id"] in seen: continue seen.add(item["source_external_id"]) items.append(item) page_count = _page_count(root) result_count = _result_count(root) or len(items) return { "items": items, "page_count": page_count, "result_count": result_count, "blocked": False, } def with_page(url: str, page: int) -> str: parts = urlsplit(url) query = [(name, value) for name, value in parse_qsl(parts.query, keep_blank_values=True) if name != "pg"] if page > 1: query.append(("pg", str(page))) return parts._replace(query=urlencode(query, doseq=True)).geturl() def _find_cards(root: Any) -> list[Any]: cards: list[Any] = [] def walk(node: Any) -> None: classes = node.attrs.get("class", "") if "galleryview__item" in classes or "listview__item" in classes: cards.append(node) return for child in node.children: if not isinstance(child, str): walk(child) walk(root) return cards def _parse_card(card: Any, page_url: str) -> dict[str, Any] | None: link = _first_link(card, "name-link") or _first_product_link(card) if link is None: return None href = link.attrs.get("href") or "" match = PRODUCT_ID_RE.search(href) if match is None: return None title = (link.attrs.get("title") or node_text(link)).strip() if not title: return None source_url = urljoin(page_url, href.split("#")[0]) raw_attributes = _definition_list(card) sku = _first_text_by_class(card, "galleryview__mpn") or _first_text_by_class(card, "listview__mpn") price_text = _first_text_by_class(card, "price") offers_text = _first_text_by_class(card, "offercount") or "" availability_node = _first_by_attr(card, "data-availability") manufacturer, model = _split_title(title) return { "source_external_id": match.group(1), "source_url": source_url, "source_title": re.sub(r"\s+", " ", title), "manufacturer": manufacturer, "model": model, "manufacturer_sku": sku, "gtin_ean": None, "best_price": _parse_price(price_text), "currency": "EUR" if price_text and "€" in price_text else None, "availability": _availability(availability_node), "seller_count": _seller_count(offers_text), "raw_attributes": raw_attributes, "normalized_attributes": _normalize_attributes(raw_attributes), } def _first_link(node: Any, fragment: str) -> Any | None: classes = node.attrs.get("class", "") if node.tag == "a" and fragment in classes: return node for child in node.children: if not isinstance(child, str): found = _first_link(child, fragment) if found is not None: return found return None def _first_product_link(node: Any) -> Any | None: if node.tag == "a" and PRODUCT_ID_RE.search(node.attrs.get("href") or ""): href = node.attrs.get("href") or "" if "#ratings" not in href and "#offerlist" not in href: return node for child in node.children: if not isinstance(child, str): found = _first_product_link(child) if found is not None: return found return None def _first_text_by_class(node: Any, fragment: str) -> str | None: found = _first_by_class(node, fragment) if found is None: return None text = node_text(found).strip() return re.sub(r"\s+", " ", text) or None def _first_by_class(node: Any, fragment: str) -> Any | None: if fragment in node.attrs.get("class", ""): return node for child in node.children: if not isinstance(child, str): found = _first_by_class(child, fragment) if found is not None: return found return None def _first_by_attr(node: Any, name: str) -> Any | None: if name in node.attrs: return node for child in node.children: if not isinstance(child, str): found = _first_by_attr(child, name) if found is not None: return found return None def _definition_list(node: Any) -> dict[str, str]: pairs: dict[str, str] = {} def walk(current: Any) -> None: if current.tag == "dl": pending: str | None = None for child in current.children: if isinstance(child, str): continue if child.tag == "dt": pending = re.sub(r"\s+", " ", node_text(child)).strip().rstrip(":") elif child.tag == "dd" and pending: value = re.sub(r"\s+", " ", node_text(child)).strip() if value: pairs[pending] = value pending = None return for child in current.children: if not isinstance(child, str): walk(child) walk(node) return pairs def _split_title(title: str) -> tuple[str | None, str | None]: parts = title.split(None, 1) if not parts: return None, None if len(parts) == 1: return parts[0], None return parts[0], parts[1] def _parse_price(text: str | None) -> str | None: if not text: return None match = PRICE_RE.search(text.replace("\xa0", " ")) if match is None: return None whole = match.group(1).replace(".", "") return f"{whole}.{match.group(2)}" def _seller_count(text: str) -> int | None: match = OFFER_RE.search(text) if match is None: return None return int(match.group(1)) def _availability(node: Any | None) -> str: if node is None: return "unknown" code = AVAILABILITY_BY_CODE.get(node.attrs.get("data-availability", "")) if code: return code label = (node.attrs.get("aria-label") or node_text(node)).strip().casefold() return AVAILABILITY_BY_LABEL.get(label, "unknown") def _normalize_attributes(raw: dict[str, str]) -> dict[str, Any]: normalized: dict[str, Any] = {} for label, value in raw.items(): key = label.casefold() capacity = _capacity_gb(value) if capacity is not None and key in {"ram", "arbeitsspeicher"}: normalized["ram_gb"] = capacity elif capacity is not None and key in {"speicher", "ssd", "festplatte"}: normalized["storage_gb"] = capacity weight = _weight_kg(value) if weight is not None and "gewicht" in key: normalized["weight_kg"] = weight return normalized def _capacity_gb(text: str) -> float | int | None: match = CAPACITY_RE.search(text) if match is None: return None value = float(match.group(1).replace(",", ".")) if match.group(2).upper() == "TB": value *= 1024 return int(value) if value.is_integer() else value def _weight_kg(text: str) -> float | None: match = WEIGHT_RE.search(text) if match is None: return None return float(match.group(1).replace(",", ".")) def _page_count(root: Any) -> int: pages = {1} def walk(node: Any) -> None: href = node.attrs.get("href") or "" if "pg=" in href and ("pagination" in node.attrs.get("class", "") or _in_pagination(node)): for name, value in parse_qsl(urlsplit(href).query): if name == "pg" and value.isdigit(): pages.add(int(value)) for child in node.children: if not isinstance(child, str): walk(child) walk(root) return max(pages) def _in_pagination(node: Any) -> bool: current = node while current is not None: classes = current.attrs.get("class", "") ident = current.attrs.get("id", "") if "pagination" in classes or ident == "pagination": return True current = current.parent return False def _result_count(root: Any) -> int | None: counts = [ int(match.replace(".", "")) for match in PRODUCT_COUNT_RE.findall(node_text(root)) ] return max(counts) if counts else None