288 lines
9.2 KiB
Python
288 lines
9.2 KiB
Python
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
|