import re from dataclasses import dataclass, field from typing import List from bs4 import BeautifulSoup from loguru import logger from client.session import GemmaSession @dataclass class SearchResult: product_id: str columns: List[str] = field(default_factory=list) def search( session: GemmaSession, titolo: str = "", tipo: str = "contiene", distributore_id: str = "", pagina: int = 1, ) -> tuple[List[str], List[SearchResult], int, int]: """Cerca prodotti. Restituisce (headers, results, totale, totale_pagine).""" params = { "pagina": pagina, "tipoRicercaTitolo": tipo, "ordineVisione": "tutto", } if titolo: params["titolo"] = titolo if distributore_id: params["distributoreIdList"] = distributore_id else: params["_distributoreIdList"] = "1" params["_supportoIdList"] = "1" params["_tipologiaCodeList"] = "1" params["_statoPrestitoCodiceList"] = "1" r = session.get("/GSV/mvc/icrTitoli/ricercaParamCustom", params=params) return _parse_results(r.text) def _parse_results(html: str) -> tuple[List[str], List[SearchResult], int, int]: soup = BeautifulSoup(html, "html.parser") # Totale risultati — Risultati: 466 total = 0 span = soup.find("span", string=re.compile(r"Risultati:\s*\d+")) if span: m = re.search(r"(\d+)", span.get_text()) if m: total = int(m.group(1)) # Totale pagine — cerca il testo "di N" ovunque nell'HTML total_pages = 1 for tag in soup.find_all(string=re.compile(r'^\s*di\s+\d+\s*$')): m = re.search(r'(\d+)', tag) if m: total_pages = int(m.group(1)) break table = soup.find("table", id="idTableElencoTitoli") if not table: logger.warning("Tabella #idTableElencoTitoli non trovata nella risposta") return [], [], total, total_pages headers = [] thead = table.find("thead") if thead: headers = [th.get_text(strip=True) for th in thead.find_all("th")] tbody = table.find("tbody") if not tbody: return headers, [], total, total_pages results = [] for tr in tbody.find_all("tr"): cells = [td.get_text(strip=True) for td in tr.find_all("td")] if not cells: continue if len(cells) == 1 and "nessun" in cells[0].lower(): break product_id = _extract_id(tr) results.append(SearchResult(product_id=product_id, columns=cells)) if total == 0: total = len(results) logger.info(f"Trovati {len(results)} risultati (totale: {total}, pagine: {total_pages})") return headers, results, total, total_pages def _extract_id(tr) -> str: if tr.get("data-id"): return tr["data-id"] onclick = tr.get("onclick", "") if onclick: m = re.search(r'\b(\d{4,})\b', onclick) if m: return m.group(1) hidden = tr.find("input", {"type": "hidden"}) if hidden and str(hidden.get("value", "")).isdigit(): return hidden["value"] first_td = tr.find("td") if first_td: text = first_td.get_text(strip=True) if text.isdigit(): return text logger.warning(f"ID non trovato per riga: {tr.get('class', '')}") return ""