diff --git a/backend/Dockerfile b/backend/Dockerfile index a9b991e..7a8c461 100644 --- a/backend/Dockerfile +++ b/backend/Dockerfile @@ -18,11 +18,13 @@ RUN uv sync --frozen --no-dev --no-install-project FROM base AS runtime COPY . . -RUN uv sync --frozen --no-dev \ +ENV PLAYWRIGHT_BROWSERS_PATH=/ms-playwright +RUN uv sync --frozen --extra live --no-dev \ + && playwright install --with-deps chromium \ && sed -i 's/\r$//' /app/scripts/startup.sh \ && chmod +x /app/scripts/startup.sh \ && useradd --create-home --uid 1000 app \ - && chown -R app:app /app + && chown -R app:app /app /ms-playwright USER app EXPOSE 8000 diff --git a/backend/app/discovery/urlgen.py b/backend/app/discovery/urlgen.py index e7fa2d3..5b73888 100644 --- a/backend/app/discovery/urlgen.py +++ b/backend/app/discovery/urlgen.py @@ -27,11 +27,17 @@ def generate_search_url( base = urlsplit(category_url) query = dict(parse_qsl(base.query, keep_blank_values=True)) extra: list[tuple[str, str]] = [] + xf_values: list[str] = [] for name, value in fragments: - if name in query and name != "xf": + if name == "xf": + xf_values.append(value) + continue + if name in query: query[name] = value else: extra.append((name, value)) + if xf_values: + extra.append(("xf", "~".join(xf_values))) 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 @@ -40,11 +46,15 @@ def generate_search_url( 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 encoding.get("encoding_type") == "RANGE_PARAMETERS" or param in {"bpmin", "bpmax"}: + return _encode_range(encoding, criterion) + if encoding.get("encoding_type") == "RANGE_TOKEN": + return _encode_range_token(mapping, criterion) if param == "xf": filter_key = str(mapping["external_filter_key"]) - source_value = str(mapping.get("source_value") or _threshold_value(mapping, criterion)) + source_value = _xf_value(mapping, criterion) return [("xf", f"{filter_key}_{_plus(source_value)}")] - if param in {"minp", "maxp", "prmin", "prmax"}: + if param in {"minp", "maxp", "prmin", "prmax", "asuch", "fs"}: return [(param, str(criterion.get("value")))] template = encoding.get("value_template") if template: @@ -52,6 +62,63 @@ def _encode_mapping(mapping: dict[str, Any], criterion: dict[str, Any]) -> list[ raise ValueError(f"unsupported encoding for mapping {mapping.get('criterion_key')}") +def _encode_range(encoding: dict[str, Any], criterion: dict[str, Any]) -> list[tuple[str, str]]: + operator = str(criterion.get("operator") or "<=") + value = criterion.get("value") + min_param = str(encoding.get("min_parameter") or encoding.get("url_param") or "bpmin") + max_param = str(encoding.get("max_parameter") or "bpmax") + if operator == "between": + if isinstance(value, dict): + items: list[tuple[str, str]] = [] + if value.get("min") is not None: + items.append((min_param, str(value["min"]))) + if value.get("max") is not None: + items.append((max_param, str(value["max"]))) + return items + if isinstance(value, list) and len(value) == 2: + return [(min_param, str(value[0])), (max_param, str(value[1]))] + if operator in {">=", ">"}: + return [(min_param, str(value))] + return [(max_param, str(value))] + + +def _encode_range_token(mapping: dict[str, Any], criterion: dict[str, Any]) -> list[tuple[str, str]]: + filter_key = str(mapping["external_filter_key"]) + looked_up = _lookup_option(mapping, criterion.get("value")) + if looked_up is not None: + return [("xf", f"{filter_key}_{_plus(looked_up)}")] + operator = str(criterion.get("operator") or "<=") + value = criterion.get("value") + if operator == "between": + if isinstance(value, dict): + raw = f"{value.get('min') or ''}*{value.get('max') or ''}" + elif isinstance(value, list) and len(value) == 2: + raw = f"{value[0]}*{value[1]}" + else: + raw = f"*{value}" + elif operator in {">=", ">"}: + raw = f"{value}*" + else: + raw = f"*{value}" + return [("xf", f"{filter_key}_{raw}")] + + +def _lookup_option(mapping: dict[str, Any], value: Any) -> str | None: + options = (mapping.get("encoding") or {}).get("options") or {} + if value in options: + return str(options[value]) + if str(value) in options: + return str(options[str(value)]) + return None + + +def _xf_value(mapping: dict[str, Any], criterion: dict[str, Any]) -> str: + looked_up = _lookup_option(mapping, criterion.get("value")) + if looked_up is not None: + return looked_up + return str(mapping.get("source_value") or _threshold_value(mapping, criterion)) + + def _threshold_value(mapping: dict[str, Any], criterion: dict[str, Any]) -> str: option = mapping.get("source_value") if option: @@ -72,8 +139,12 @@ 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) + if name != "xf": + continue + for part in value.replace("|", "~").split("~"): + if "_" not in part: + continue + key, raw = part.split("_", 1) decoded.append((key, raw.replace("+", " "))) return decoded diff --git a/backend/app/services/catalog_service.py b/backend/app/services/catalog_service.py index 3029d60..27d519c 100644 --- a/backend/app/services/catalog_service.py +++ b/backend/app/services/catalog_service.py @@ -100,7 +100,7 @@ class CatalogService: SourceCategoryRecord.product_category_id == product_category_id, ) ).first() - if source_category is not None and source_category.taxonomy_status == TaxonomyStatus.NOT_LEARNED.value: + if source_category is not None and self._should_learn(source_category, source_id): from app.services.discovery_service import TaxonomyService TaxonomyService(self.session).discover(source_id, source_category.external_category_key) @@ -172,7 +172,33 @@ class CatalogService: if definition.id in seen: continue criteria.append(self._criterion_out(definition, None, None, [], None)) - return jsonable({"product_category_id": product_category_id, "source": source_id, "criteria": criteria}) + return jsonable( + { + "product_category_id": product_category_id, + "source": source_id, + "taxonomy_status": None if source_category is None else source_category.taxonomy_status, + "criteria": criteria, + } + ) + + def _should_learn(self, source_category: SourceCategoryRecord, source_id: str) -> bool: + if source_category.taxonomy_status == TaxonomyStatus.NOT_LEARNED.value: + return True + if source_category.taxonomy_status != TaxonomyStatus.READY.value: + return False + from app.settings import get_settings + + if get_settings().geizhals_use_fixtures or source_id != "geizhals": + return False + keys = { + row.external_filter_key + for row in self.session.scalars( + select(SourceFilterDefinitionRecord).where( + SourceFilterDefinitionRecord.source_category_id == source_category.id + ) + ) + } + return keys <= {"17023", "price"} def list_templates(self) -> list[dict[str, Any]]: rows, _total = self.catalog.list_profiles(200, 0) diff --git a/backend/app/services/discovery_service.py b/backend/app/services/discovery_service.py index 2bc0137..e81dea3 100644 --- a/backend/app/services/discovery_service.py +++ b/backend/app/services/discovery_service.py @@ -14,6 +14,7 @@ from app.discovery.diff import diff_run, next_lifecycle from app.discovery.enums import ( CandidateStatus, DiffStatus, + FilterType, LifecycleStatus, MappingStatus, MatchStatus, @@ -99,7 +100,20 @@ class TaxonomyService: if catalog.get_source(source_id) is None: raise AppError(404, "NOT_FOUND", "source not found", {"source_id": source_id}) adapter = self.adapter or discovery_adapter(source_id) - payload = adapter.discover_taxonomy(category_key) + try: + payload = adapter.discover_taxonomy(category_key) + except AppError as exc: + category = self.session.scalars( + select(SourceCategoryRecord).where( + SourceCategoryRecord.source_id == source_id, + SourceCategoryRecord.external_category_key == category_key, + ) + ).first() + if category is not None and exc.code == "SOURCE_BLOCKED": + category.taxonomy_status = TaxonomyStatus.BLOCKED.value + category.last_taxonomy_scan_at = now_utc() + self.session.flush() + raise now = now_utc() category = self.session.scalars( select(SourceCategoryRecord).where( @@ -148,7 +162,7 @@ class TaxonomyService: label=str(raw.get("label") or raw["external_filter_key"]), filter_type=filter_type, canonical_unit=raw.get("canonical_unit"), - status=MappingStatus.VERIFIED.value, + status=MappingStatus.LEARNED_UNVERIFIED.value, ) self.session.add(filt) self.session.flush() @@ -161,9 +175,14 @@ class TaxonomyService: self._sync_options(filt, raw.get("options") or []) if raw.get("range"): self._sync_range(filt, raw["range"]) - category.taxonomy_status = ( - TaxonomyStatus.READY.value if payload.get("filters") else TaxonomyStatus.BROKEN.value - ) + self._sync_learned_mapping(category, filt, raw) + requested = str(payload.get("taxonomy_status") or "") + if requested in {item.value for item in TaxonomyStatus}: + category.taxonomy_status = requested + else: + category.taxonomy_status = ( + TaxonomyStatus.READY.value if payload.get("filters") else TaxonomyStatus.BROKEN.value + ) self.session.flush() return self.filters(source_id, category.external_category_key) @@ -275,6 +294,89 @@ class TaxonomyService: if removed: self._break_mappings(filt.source_category_id, filt.external_filter_key) + def _sync_learned_mapping( + self, + category: SourceCategoryRecord, + filt: SourceFilterDefinitionRecord, + raw: dict[str, Any], + ) -> None: + encoding = raw.get("encoding") or {"url_param": "xf"} + definition = upsert_definition( + self.session, + { + "key": f"{category.source_id}:{category.external_category_key}:{filt.external_filter_key}", + "origin": "SOURCE", + "source": category.source_id, + "product_category_id": category.product_category_id, + "label": filt.label, + "value_type": "number" + if filt.filter_type in {FilterType.NUMERIC_RANGE.value, FilterType.NUMERIC_THRESHOLD.value} + else "string", + "unit": filt.canonical_unit, + "category": category.product_category_id, + }, + ) + mapping = self.session.scalars( + select(CriterionMappingRecord).where( + CriterionMappingRecord.source_category_id == category.id, + CriterionMappingRecord.external_filter_key == filt.external_filter_key, + ) + ).first() + options = raw.get("options") or [] + source_value = None + if options: + source_value = str(options[0].get("query_value") or options[0].get("source_value")) + now = now_utc() + if mapping is None: + mapping = CriterionMappingRecord( + id=uuid4(), + criterion_definition_id=definition.id, + criterion_key=definition.key, + source_id=category.source_id, + source_category_id=category.id, + external_filter_key=filt.external_filter_key, + source_value=source_value, + encoding=encoding, + status=MappingStatus.LEARNED_UNVERIFIED.value, + definition_hash=sha256_text(canonical_json({"encoding": encoding, "source_value": source_value})), + created_at=now, + updated_at=now, + ) + self.session.add(mapping) + self.session.flush() + else: + mapping.criterion_definition_id = definition.id + mapping.criterion_key = definition.key + mapping.source_value = source_value + mapping.encoding = encoding + mapping.definition_hash = sha256_text(canonical_json({"encoding": encoding, "source_value": source_value})) + mapping.updated_at = now + probe_value = source_value or 0 + if options and options[0].get("normalized_value") is not None: + probe_value = options[0]["normalized_value"] + operator = (options[0].get("operator") if options else None) or "<=" + if filt.filter_type == FilterType.NUMERIC_RANGE.value: + operator = ">=" + probe_value = 1000 + verified = verify_roundtrip( + category_url=category.source_url or f"https://geizhals.de/?cat={category.external_category_key}", + criterion={ + "criterion": definition.key, + "operator": operator, + "value": probe_value, + }, + mapping={ + "criterion_key": definition.key, + "external_filter_key": filt.external_filter_key, + "source_value": source_value, + "encoding": encoding, + }, + ) + if mapping.status != MappingStatus.DISABLED.value and verified: + mapping.status = MappingStatus.VERIFIED.value + elif mapping.status == MappingStatus.BROKEN.value: + mapping.status = MappingStatus.LEARNED_UNVERIFIED.value + def _sync_range(self, filt: SourceFilterDefinitionRecord, raw: dict[str, Any]) -> None: row = self.session.get(SourceFilterRangeRecord, filt.id) if row is None: diff --git a/backend/app/settings.py b/backend/app/settings.py index bd04c9b..d0947fa 100644 --- a/backend/app/settings.py +++ b/backend/app/settings.py @@ -16,6 +16,8 @@ class Settings(BaseSettings): engine_build: str = "dev" clock_now: datetime | None = None cors_allowed_origins: str = "http://localhost:5173,http://127.0.0.1:5173" + geizhals_use_fixtures: bool = False + geizhals_playwright: bool = True @property def cors_origins(self) -> list[str]: diff --git a/backend/app/sources/geizhals/adapter.py b/backend/app/sources/geizhals/adapter.py index dcbc3d0..b140afd 100644 --- a/backend/app/sources/geizhals/adapter.py +++ b/backend/app/sources/geizhals/adapter.py @@ -7,6 +7,10 @@ from urllib.parse import parse_qs, urlparse from app.discovery.enums import FilterType from app.errors import AppError +from app.settings import get_settings +from app.sources.geizhals.live_page import load_category_page, load_result_page +from app.sources.geizhals.result_parser import parse_result_html, with_page +from app.sources.geizhals.taxonomy_parser import parse_taxonomy_html FIXTURE_DIR = Path(__file__).resolve().parents[3] / "fixtures" / "geizhals" @@ -14,12 +18,31 @@ FIXTURE_DIR = Path(__file__).resolve().parents[3] / "fixtures" / "geizhals" class GeizhalsAdapter: source_key = "geizhals" - def __init__(self, fixture_dir: Path | None = None, live: bool = False) -> None: + def __init__(self, fixture_dir: Path | None = None, live: bool | None = None) -> None: self.fixture_dir = fixture_dir or FIXTURE_DIR - self.live = live + settings = get_settings() + self.live = (not settings.geizhals_use_fixtures) if live is None else live def discover_taxonomy(self, category_ref: str) -> dict[str, Any]: - payload = self._load(f"{category_ref}-taxonomy.json") + if not self.live: + return self._load(f"{category_ref}-taxonomy.json") + url = f"https://geizhals.de/?cat={category_ref}&promode=true" + html, mode, status = load_category_page( + url, + category_ref=category_ref, + playwright=get_settings().geizhals_playwright, + ) + payload = parse_taxonomy_html(html, category_ref=category_ref, source_url=url) + if payload.get("blocked") or not payload.get("filters"): + raise AppError( + 422, + "TAXONOMY_EMPTY", + "Geizhals-Taxonomie konnte nicht aus der Kategorieseite gelesen werden", + {"url": url, "http_status": status, "mode": mode}, + ) + payload["discovery_mode"] = mode + payload["http_status"] = status + payload["taxonomy_status"] = "DEGRADED" if mode == "recorded_html" else "READY" return payload def build_query(self, criteria: list[dict[str, Any]], mappings: list[dict[str, Any]]) -> dict[str, Any]: @@ -34,13 +57,32 @@ class GeizhalsAdapter: return {"url": url, "internal_only": internal_only} def fetch_results(self, query: dict[str, Any], pagination: dict[str, Any]) -> dict[str, Any]: - if self.live: - raise AppError(422, "LIVE_FETCH_DISABLED", "live Geizhals fetch is not enabled in this process") page = int(pagination.get("page") or 1) - category = parse_qs(urlparse(query["url"]).query).get("cat", ["nb"])[0] - return self._load(f"{category}-search-page{page}.json") + if not self.live: + category = parse_qs(urlparse(query["url"]).query).get("cat", ["nb"])[0] + return self._load(f"{category}-search-page{page}.json") + url = with_page(str(query["url"]), page) + html, mode, status, blocked = load_result_page(url, playwright=get_settings().geizhals_playwright) + if blocked: + return { + "page": page, + "page_count": page, + "result_count": 0, + "http_status": status or 403, + "blocked": True, + "challenge": "captcha" if status in {401, 403} else None, + "items": [], + "discovery_mode": mode, + } + payload = parse_result_html(html, page_url=url) + payload["page"] = page + payload["http_status"] = status + payload["discovery_mode"] = mode + return payload def parse_results(self, response: dict[str, Any]) -> list[dict[str, Any]]: + if response.get("html") and not response.get("items"): + return list(parse_result_html(str(response["html"]), page_url=str(response.get("url") or "")).get("items") or []) return list(response.get("items") or []) def normalize_result(self, result: dict[str, Any]) -> dict[str, Any]: diff --git a/backend/app/sources/geizhals/live_page.py b/backend/app/sources/geizhals/live_page.py new file mode 100644 index 0000000..043ea50 --- /dev/null +++ b/backend/app/sources/geizhals/live_page.py @@ -0,0 +1,104 @@ +from __future__ import annotations + +from pathlib import Path + +from app.errors import AppError +from app.sources.geizhals.taxonomy_parser import is_challenge_html + +FIXTURE_DIR = Path(__file__).resolve().parents[3] / "fixtures" / "geizhals" +BROWSER_HEADERS = { + "User-Agent": ( + "Mozilla/5.0 (Windows NT 10.0; Win64; x64) " + "AppleWebKit/537.36 (KHTML, like Gecko) Chrome/128.0.0.0 Safari/537.36" + ), + "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8", + "Accept-Language": "de-DE,de;q=0.9,en;q=0.8", +} +EXPAND_LABELS = ( + "Immer alle Filter anzeigen", + "weitere Filter anzeigen", +) + + +def load_result_page(url: str, *, playwright: bool = True) -> tuple[str, str, int, bool]: + status, html = 0, "" + try: + status, html = _fetch_http(url) + if not _blocked(status, html): + return html, "live_http", status, False + except Exception: + status, html = 0, "" + if playwright: + try: + status, html = _fetch_playwright(url) + if not _blocked(status, html): + return html, "live_playwright", status, False + except Exception: + pass + if html and not is_challenge_html(html) and status and status < 400: + return html, "live_http", status, False + return html, "blocked", status or 403, True + + +def load_category_page(url: str, *, category_ref: str, playwright: bool = True) -> tuple[str, str, int]: + status, html = 0, "" + try: + status, html = _fetch_http(url) + if not _blocked(status, html): + return html, "live_http", status + except Exception: + status, html = 0, "" + if playwright: + try: + status, html = _fetch_playwright(url) + if not _blocked(status, html): + return html, "live_playwright", status + except Exception: + pass + recorded = FIXTURE_DIR / f"{category_ref}-category.html" + if recorded.exists(): + return recorded.read_text(encoding="utf-8"), "recorded_html", 0 + raise AppError( + 423, + "SOURCE_BLOCKED", + "Geizhals blockiert den Taxonomie-Abruf (403/Challenge)", + {"url": url, "http_status": status}, + ) + + +def _fetch_http(url: str) -> tuple[int, str]: + import httpx + + response = httpx.get(url, headers=BROWSER_HEADERS, follow_redirects=True, timeout=25.0) + return response.status_code, response.text + + +def _fetch_playwright(url: str) -> tuple[int, str]: + try: + from playwright.sync_api import sync_playwright + except ImportError as exc: + raise AppError( + 423, + "SOURCE_BLOCKED", + "Geizhals verlangt einen Browser-Abruf, Playwright ist nicht installiert", + ) from exc + with sync_playwright() as playwright: + browser = playwright.chromium.launch(headless=True) + page = browser.new_page(locale="de-DE", extra_http_headers=BROWSER_HEADERS) + response = page.goto(url, wait_until="domcontentloaded", timeout=45000) + page.wait_for_timeout(4000) + for label in EXPAND_LABELS: + button = page.get_by_text(label, exact=False) + if button.count(): + button.first.click(timeout=5000) + page.wait_for_timeout(1500) + html = page.content() + status = response.status if response is not None else 200 + browser.close() + return status, html + + +def _blocked(status: int, html: str) -> bool: + if status in {401, 403, 429}: + return True + return is_challenge_html(html) diff --git a/backend/app/sources/geizhals/result_parser.py b/backend/app/sources/geizhals/result_parser.py new file mode 100644 index 0000000..4caefd5 --- /dev/null +++ b/backend/app/sources/geizhals/result_parser.py @@ -0,0 +1,285 @@ +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() + elif child.tag == "dd" and pending: + pairs[pending] = re.sub(r"\s+", " ", node_text(child)).strip() + 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 diff --git a/backend/app/sources/geizhals/taxonomy_parser.py b/backend/app/sources/geizhals/taxonomy_parser.py new file mode 100644 index 0000000..6c91659 --- /dev/null +++ b/backend/app/sources/geizhals/taxonomy_parser.py @@ -0,0 +1,474 @@ +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", +} +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) + ] + filters.sort(key=lambda item: (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)) + 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 == "a": + href = node.attrs.get("href") or "" + label = node.attrs.get("aria-label") or _node_text(node) + if _usable_option(node, label, allow_histogram=allow_histogram): + for key, query_value in _xf_tokens(href): + if required_key is not None and key != required_key: + continue + if required_key is None and not _has_structural_label(node): + continue + group_label = _declared_label(_filter_root(node)) or _nearest_group_label(node) + group = groups.setdefault(key, _empty_group(key, group_label, None)) + if group_label and (group["label"].startswith("Filter ") or group["label"] == key): + group["label"] = group_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" + 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", + "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" + groups[name] = { + "external_filter_key": name, + "label": label, + "filter_type": FilterType.TEXT.value, + "canonical_unit": None, + "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 _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 = re.sub(r"\(\d[\d.]*\s*(produkte)?\)\s*$", "", label, flags=re.IGNORECASE).strip() + cleaned = re.sub(r"\s+", " ", cleaned) + 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, + "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 = re.sub(r"\s+", " ", label).strip() + if not cleaned or _skip_option(cleaned): + return False + if "überspringen" in cleaned.lower(): + return False + if not allow_histogram and "produkte" in cleaned.lower(): + return False + if len(cleaned.split()) > 8: + return False + return True + + +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 diff --git a/backend/fixtures/geizhals/nb-category.html b/backend/fixtures/geizhals/nb-category.html new file mode 100644 index 0000000..c7cc3bc --- /dev/null +++ b/backend/fixtures/geizhals/nb-category.html @@ -0,0 +1,132 @@ + + +
+| Hersteller: | ++ Apple + ASUS + Lenovo + | +
|---|---|
| Display-Größe: | ++ 13.x" + 14.x" + 15.x" + 16.x" + | +
| Display-Auflösung: | ++ 1920x1080 (Full HD) + 2560x1600 (WQXGA) + | +
| CPU-Kerne: | ++ 8 (Octa-Core) + 10 + 14-Core + | +
| CPU-Kerne ab: | ++ ab 8 + ab 10 + | +
| CPU-Threads ab: | ++ ab 12 + ab 16 + | +
| CPU-Turbotakt: | ++ ab 4.5GHz + ab 5GHz + | +
| RAM ab: | ++ ab 16GB + ab 32GB + ab 64GB + | +
| RAM-Typ: | ++ DDR5 + LPDDR5 + | +
| Speichertyp: | ++ SSD + M.2 + | +
| SSD-Kapazität: | ++ ab 1TB + ab 2TB + | +
| Grafik-Typ: | ++ dedizierte Grafik (dGPU) + | +
| GPU - Speichergröße ab: | ++ ab 2GB + ab 4GB + ab 8GB + | +
| GPU - Speichergröße: | ++ 8GB + 12GB + | +
| USB-C mit DisplayPort gesamt (inkl. Thunderbolt): | ++ ab 1x + | +
Source: {source || "–"}
Kategorie: {category || "–"}
Status: {selected?.taxonomy_status || "NOT_LEARNED"}
+Gelernte Filter: {filters.data?.filters.length ?? 0}
{discover.isError ?{(discover.error as Error).message}
: null} {!category ? ( diff --git a/frontend/src/pages/WizardPage.tsx b/frontend/src/pages/WizardPage.tsx index f4da244..93bf698 100644 --- a/frontend/src/pages/WizardPage.tsx +++ b/frontend/src/pages/WizardPage.tsx @@ -18,6 +18,7 @@ export function WizardPage() { const [sourceOverride, setSourceOverride] = useStateQuelle: {source}
) : null} + {(criteria.data?.taxonomy_status || sourceRow?.taxonomy_status) === "DEGRADED" ? ( +Die Taxonomie stammt aus einer aufgezeichneten Kategorieseite. Live-Lernen war blockiert.
+ ) : null} {criteria.isError ?+ {(criteria.data?.criteria || []).length} Kriterien aus der Quelle. Wähle die relevanten aus und setze + Grenzwerte. +
+ {visible.map((item) => { const on = selected.includes(item.criterion_definition_id); return (