607 lines
24 KiB
Python
607 lines
24 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
Nuvoo scraper
|
||
===============
|
||
Crawls https://nuvoo.com/sv-se/collections/laptops via the Shopify JSON API
|
||
and stores refurbished laptop listings in the shared SQLite database.
|
||
|
||
Usage (from main.py web trigger, or standalone):
|
||
from scrapers.scraper_nuvoo import run
|
||
conn = init_db(DB_PATH)
|
||
run(conn, skip_phase1=False, reset=False)
|
||
"""
|
||
|
||
import re
|
||
import json
|
||
import time
|
||
import random
|
||
import sqlite3
|
||
from typing import Optional
|
||
from datetime import datetime
|
||
|
||
from bs4 import BeautifulSoup
|
||
|
||
from . import shared as _shared
|
||
from .shared import (
|
||
log, _sleep, _make_session,
|
||
INSERT_SQL, _safe_row,
|
||
USER_AGENTS, map_condition_grade,
|
||
)
|
||
|
||
# ── Constants ─────────────────────────────────────────────────────────────────
|
||
|
||
NUVOO_BASE = "https://nuvoo.com"
|
||
NUVOO_LOCALE = "sv-se"
|
||
NUVOO_COLLECTION_URL = f"{NUVOO_BASE}/{NUVOO_LOCALE}/collections/laptops"
|
||
NUVOO_PRODUCTS_BASE = f"{NUVOO_BASE}/{NUVOO_LOCALE}/products"
|
||
NUVOO_MAX_PAGES = 30
|
||
|
||
# ── Nuvoo HTTP helper ─────────────────────────────────────────────────────────
|
||
|
||
def _nuvoo_get(url: str, retries: int = 3) -> Optional[object]:
|
||
"""Rate-limited GET for Nuvoo requests; resets shared session on error."""
|
||
for attempt in range(1, retries + 1):
|
||
_shared._session.headers["User-Agent"] = random.choice(USER_AGENTS)
|
||
try:
|
||
resp = _shared._session.get(url, timeout=20)
|
||
if resp.status_code == 200:
|
||
return resp
|
||
if resp.status_code == 429:
|
||
wait = 60 + random.uniform(10, 30)
|
||
log.warning(f"Rate-limited (429). Sleeping {wait:.0f}s (attempt {attempt}/{retries})")
|
||
time.sleep(wait)
|
||
elif resp.status_code in (404, 410):
|
||
log.info(f"Product gone ({resp.status_code}): {url}")
|
||
return None
|
||
else:
|
||
_shared._session.close()
|
||
_shared._session = _make_session()
|
||
log.warning(f"HTTP {resp.status_code} for {url} (attempt {attempt}/{retries})")
|
||
time.sleep(5 * attempt)
|
||
except Exception as exc:
|
||
log.warning(f"Request error attempt {attempt}/{retries}: {exc}")
|
||
_shared._session.close()
|
||
_shared._session = _make_session()
|
||
time.sleep(10 * attempt)
|
||
log.error(f"Gave up after {retries} attempts: {url}")
|
||
return None
|
||
|
||
|
||
# ── Phase 1: collect product handles from listing pages ───────────────────────
|
||
|
||
def _table_exists(conn: sqlite3.Connection, name: str) -> bool:
|
||
r = conn.execute(
|
||
"SELECT 1 FROM sqlite_master WHERE type='table' AND name=?", (name,)
|
||
).fetchone()
|
||
return r is not None
|
||
|
||
|
||
def collect_handles(conn: sqlite3.Connection) -> set[str]:
|
||
"""Paginate the Nuvoo laptop collection, re-queue every live product, return live handle set.
|
||
|
||
Uses INSERT OR REPLACE so existing queue entries have their done flag reset
|
||
to 0 — guaranteeing that re-runs always re-scrape and update every listing.
|
||
Returns the complete set of handles currently live on the site so the caller
|
||
can remove sold/delisted rows from the database.
|
||
"""
|
||
log.info("═══ Nuvoo Phase 1: Collecting product handles ═══")
|
||
|
||
page = 1
|
||
live_handles: set[str] = set()
|
||
handle_re = re.compile(r"/sv-se/products/([^?/#]+)")
|
||
|
||
while page <= NUVOO_MAX_PAGES:
|
||
url = NUVOO_COLLECTION_URL if page == 1 else f"{NUVOO_COLLECTION_URL}?page={page}"
|
||
log.info(f"Listing page {page}: {url}")
|
||
resp = _nuvoo_get(url)
|
||
if not resp:
|
||
log.warning(f" Failed to load listing page {page}, stopping.")
|
||
break
|
||
|
||
soup = BeautifulSoup(resp.text, "lxml")
|
||
|
||
handles_this_page: list[str] = []
|
||
for a in soup.find_all("a", href=handle_re):
|
||
m = handle_re.search(a["href"])
|
||
if m:
|
||
handle = m.group(1)
|
||
if handle not in handles_this_page:
|
||
handles_this_page.append(handle)
|
||
|
||
if not handles_this_page:
|
||
log.info(f" No product links found on page {page} – end of catalog.")
|
||
break
|
||
|
||
for handle in handles_this_page:
|
||
live_handles.add(handle)
|
||
product_url = f"{NUVOO_PRODUCTS_BASE}/{handle}.json"
|
||
conn.execute(
|
||
"INSERT OR REPLACE INTO scrape_queue (url, source, handle, done) VALUES (?, 'nuvoo', ?, 0)",
|
||
(product_url, handle),
|
||
)
|
||
conn.commit()
|
||
log.info(f" {len(handles_this_page)} products on page (running total: {len(live_handles)})")
|
||
|
||
if len(handles_this_page) < 15:
|
||
log.info(f" Only {len(handles_this_page)} products on page {page} — last page.")
|
||
break
|
||
|
||
page += 1
|
||
_sleep()
|
||
|
||
queued = conn.execute("SELECT COUNT(*) FROM scrape_queue WHERE source='nuvoo' AND done=0").fetchone()[0]
|
||
log.info(f"Phase 1 done. {queued} handles queued for scraping. {len(live_handles)} live products found.")
|
||
return live_handles
|
||
|
||
|
||
# ── Phase 2: fetch product JSON and insert rows ───────────────────────────────
|
||
|
||
def nuvoo_scrape_products(conn: sqlite3.Connection) -> None:
|
||
"""Work through scrape_queue (nuvoo rows), fetching product JSON and inserting rows."""
|
||
log.info("═══ Nuvoo Phase 2: Scraping product JSON ═══")
|
||
|
||
queue = conn.execute(
|
||
"SELECT url, handle FROM scrape_queue WHERE source='nuvoo' AND done=0 ORDER BY rowid"
|
||
).fetchall()
|
||
total = len(queue)
|
||
log.info(f"{total} products to scrape.")
|
||
|
||
for i, row in enumerate(queue, 1):
|
||
url = row["url"]
|
||
handle = row["handle"]
|
||
log.info(f"[{i}/{total}] {handle}")
|
||
|
||
resp = _nuvoo_get(url)
|
||
if not resp:
|
||
conn.execute("UPDATE scrape_queue SET done=-1 WHERE url=?", (url,))
|
||
conn.commit()
|
||
_sleep()
|
||
continue
|
||
|
||
try:
|
||
data = resp.json()
|
||
product = data.get("product", {})
|
||
|
||
# Also fetch the HTML product page to parse the Specifikation tab
|
||
html_url = f"{NUVOO_PRODUCTS_BASE}/{handle}"
|
||
resp_html = _nuvoo_get(html_url)
|
||
tab_specs = _parse_nuvoo_specs_tab(resp_html.text) if resp_html else {}
|
||
|
||
rows = _parse_nuvoo_product(product, handle, tab_specs)
|
||
for r in rows:
|
||
conn.execute(INSERT_SQL, _safe_row(r))
|
||
conn.execute("UPDATE scrape_queue SET done=1 WHERE url=?", (url,))
|
||
conn.commit()
|
||
log.info(f" ✓ {product.get('title', handle)} ({len(rows)} variant rows)")
|
||
except Exception as exc:
|
||
log.error(f" Error processing {url}: {exc}", exc_info=True)
|
||
conn.execute("UPDATE scrape_queue SET done=-1 WHERE url=?", (url,))
|
||
conn.commit()
|
||
|
||
_sleep()
|
||
|
||
done_n = conn.execute("SELECT COUNT(*) FROM scrape_queue WHERE source='nuvoo' AND done=1").fetchone()[0]
|
||
failed_n = conn.execute("SELECT COUNT(*) FROM scrape_queue WHERE source='nuvoo' AND done=-1").fetchone()[0]
|
||
log.info(f"Phase 2 done. ✓ {done_n} scraped ✗ {failed_n} failed.")
|
||
|
||
|
||
# ── Product parser ────────────────────────────────────────────────────────────
|
||
|
||
def _parse_nuvoo_product(product: dict, handle: str, tab_specs: dict | None = None) -> list[dict]:
|
||
"""
|
||
Parse a Shopify product dict into one dict row per condition grade.
|
||
Keeps the cheapest variant per grade as a separate row.
|
||
tab_specs, if provided, overrides body_html for screen/resolution/type.
|
||
"""
|
||
title = product.get("title", "")
|
||
vendor = product.get("vendor", "")
|
||
product_url = f"https://nuvoo.com/{NUVOO_LOCALE}/products/{handle}"
|
||
|
||
images = product.get("images", [])
|
||
image_urls = json.dumps([img["src"] for img in images if img.get("src")])
|
||
|
||
body_html = product.get("body_html", "")
|
||
specs = _parse_nuvoo_body_html(body_html)
|
||
|
||
# Tab specs take priority (body_html bullets rarely contain resolution)
|
||
if tab_specs:
|
||
for k, v in tab_specs.items():
|
||
if v:
|
||
specs[k] = v
|
||
|
||
condition_best: dict[str, dict] = {}
|
||
for v in product.get("variants", []):
|
||
option1 = v.get("option1", "") or ""
|
||
option2 = v.get("option2", "") or ""
|
||
option3 = v.get("option3", "") or ""
|
||
price_str = v.get("price", "0") or "0"
|
||
orig_str = v.get("compare_at_price", "") or ""
|
||
|
||
grade = map_condition_grade(option1, "nuvoo")
|
||
|
||
try:
|
||
price = int(float(price_str))
|
||
except (ValueError, TypeError):
|
||
price = None
|
||
|
||
try:
|
||
orig = int(float(orig_str)) if orig_str else None
|
||
except (ValueError, TypeError):
|
||
orig = None
|
||
|
||
if grade not in condition_best or (
|
||
price and condition_best[grade]["price_sek"] and price < condition_best[grade]["price_sek"]
|
||
):
|
||
condition_best[grade] = {
|
||
"condition_grade": grade,
|
||
"condition_raw": option1,
|
||
"price_sek": price,
|
||
"original_price_sek": orig,
|
||
"ram_source": option2,
|
||
"storage_source": option3,
|
||
}
|
||
|
||
base = _nuvoo_build_base_row(title, vendor, product_url, specs, image_urls)
|
||
|
||
rows = []
|
||
for grade, vdata in condition_best.items():
|
||
row = dict(base)
|
||
row["condition_grade"] = vdata["condition_grade"]
|
||
row["condition_detail"] = vdata["condition_raw"]
|
||
row["price_sek"] = vdata["price_sek"]
|
||
row["original_price_sek"] = vdata["original_price_sek"]
|
||
|
||
if not row.get("ram_gb"):
|
||
m = re.search(r"(\d+)\s*GB", vdata["ram_source"], re.I)
|
||
if m:
|
||
row["ram_gb"] = int(m.group(1))
|
||
if not row.get("storage_gb"):
|
||
m = re.search(r"(\d+)\s*(GB|TB)", vdata["storage_source"], re.I)
|
||
if m:
|
||
val = int(m.group(1))
|
||
unit = m.group(2).upper()
|
||
row["storage_gb"] = val * 1024 if unit == "TB" else val
|
||
|
||
row["url"] = f"{product_url}?grade={grade.lower()}"
|
||
rows.append(row)
|
||
|
||
return rows
|
||
|
||
|
||
def _nuvoo_build_base_row(
|
||
title: str, vendor: str, product_url: str, specs: dict, image_urls: str
|
||
) -> dict:
|
||
"""Build the shared column data for all condition variants of a product."""
|
||
row: dict = {
|
||
"url": product_url,
|
||
"title": title,
|
||
"brand": vendor,
|
||
"image_urls": image_urls,
|
||
"source": "nuvoo",
|
||
"scraped_at": datetime.now().isoformat(),
|
||
"touchscreen": 0,
|
||
"wired_ethernet": 1,
|
||
"has_4g_5g": 0,
|
||
"keyboard_backlit": 0,
|
||
"webcam": 1,
|
||
"optical_drive": 0,
|
||
}
|
||
|
||
s = specs
|
||
|
||
# Screen size: prefer direct tab field, else extract from combined screen bullet
|
||
if s.get("screen_size"):
|
||
raw_size = s["screen_size"]
|
||
size_m = re.search(r'\d+[,.]?\d*\s*(?:"|tum|inch|\u2033)', raw_size, re.I)
|
||
row["screen_size"] = size_m.group(0).strip() if size_m else raw_size.strip()
|
||
else:
|
||
screen_raw = s.get("screen", "")
|
||
screen_raw = re.sub(r"[-–\s]*sk[aä]rm\s*:?\s*$", "", screen_raw, flags=re.I).strip().rstrip(": \t")
|
||
size_m = re.search(r'\d+[,.]?\d*\s*(?:"|tum|inch|\u2033)', screen_raw, re.I)
|
||
row["screen_size"] = size_m.group(0).strip() if size_m else screen_raw
|
||
|
||
# Resolution: prefer direct tab field, else try to extract from body_html bullet
|
||
if s.get("screen_resolution"):
|
||
row["screen_resolution"] = s["screen_resolution"]
|
||
else:
|
||
combined_res = " ".join(filter(None, [s.get("screen"), s.get("screen_size")]))
|
||
res_m = re.search(r"(\d{3,4}\s*[×x]\s*\d{3,4}(?:\s*\([^)]+\))?)", combined_res)
|
||
if res_m:
|
||
row["screen_resolution"] = res_m.group(1).strip()
|
||
|
||
# Screen type and touchscreen: search all screen-related text
|
||
combined_screen = " ".join(filter(None, [
|
||
s.get("screen_type"), s.get("screen_size"), s.get("screen"), s.get("screen_resolution"),
|
||
])).lower()
|
||
if s.get("screen_type"):
|
||
raw_type = s["screen_type"].upper()
|
||
# Clean up Shopify values like "IPS-skärm" → "IPS"
|
||
type_m = re.search(r"(RETINA|OLED|IPS|TN|VA|LCD|LED)", raw_type)
|
||
row["screen_type"] = type_m.group(1) if type_m else raw_type
|
||
else:
|
||
for stype in ("retina", "oled", "ips", "tn", "va", "lcd", "led"):
|
||
if stype in combined_screen:
|
||
row["screen_type"] = stype.upper()
|
||
break
|
||
|
||
if re.search(r"pek|touch", combined_screen, re.I):
|
||
row["touchscreen"] = 1
|
||
|
||
cpu_raw = s.get("cpu", "")
|
||
# Strip parenthetical annotations (e.g. "(4 kärnor / 8 trådar)") from the
|
||
# stored name so filters and future lookups work against a clean model string.
|
||
# cpu_raw is kept intact below so cores/threads can still be parsed from it.
|
||
row["cpu_full"] = re.sub(r'\s*\([^)]*\)', '', cpu_raw).strip()
|
||
|
||
cpu_model_m = re.match(
|
||
r"(Apple\s+M\d[\w\s]*?(?:Pro|Max|Ultra)?|"
|
||
r"(?:Intel|AMD)\s+(?:Core(?:\s+Ultra)?|Ryzen|Celeron|Pentium|Xeon)\s+[\w\-]+(?:\s+PRO)?)",
|
||
cpu_raw, re.I,
|
||
)
|
||
if cpu_model_m:
|
||
model = cpu_model_m.group(1).strip()
|
||
model = re.sub(r"[-\s]*(processor|cpu|chip)$", "", model, flags=re.I).strip()
|
||
row["cpu_model"] = model
|
||
else:
|
||
fb_m = re.match(r"([\w\s\-]+?)\s+\d+[.,]\d+\s*GHz", cpu_raw)
|
||
if fb_m:
|
||
row["cpu_model"] = fb_m.group(1).strip()
|
||
elif cpu_raw:
|
||
model = cpu_raw.split("(")[0].strip()
|
||
model = re.sub(r"[-\s]*(processor|cpu|chip)$", "", model, flags=re.I).strip()
|
||
row["cpu_model"] = model
|
||
|
||
cores_m = re.search(r"(\d+)\s*kärnor", cpu_raw, re.I)
|
||
if cores_m:
|
||
row["cpu_cores"] = int(cores_m.group(1))
|
||
|
||
threads_m = re.search(r"(\d+)\s*trådar", cpu_raw, re.I)
|
||
if threads_m:
|
||
row["cpu_threads"] = int(threads_m.group(1))
|
||
elif row.get("cpu_cores"):
|
||
row["cpu_threads"] = row["cpu_cores"] * 2
|
||
|
||
ghz_vals = re.findall(r"(\d+[.,]\d+)\s*GHz", cpu_raw, re.I)
|
||
if ghz_vals:
|
||
row["cpu_base_ghz"] = float(ghz_vals[0].replace(",", "."))
|
||
|
||
turbo_m = re.search(r"turbo\s+(?:upp\s+till\s+)?(\d+[.,]\d+)\s*GHz", cpu_raw, re.I)
|
||
if turbo_m:
|
||
row["cpu_turbo_ghz"] = float(turbo_m.group(1).replace(",", "."))
|
||
|
||
gen_m = re.search(r"(\d{1,2})(?:th|:e|e)\s*(?:gen)?", cpu_raw, re.I)
|
||
if gen_m:
|
||
row["cpu_generation"] = f"{gen_m.group(1)}th"
|
||
else:
|
||
apple_m = re.search(r"Apple\s+(M\d(?:\s+(?:Pro|Max|Ultra))?)", title, re.I)
|
||
if apple_m:
|
||
row["cpu_generation"] = apple_m.group(1)
|
||
|
||
ram_raw = s.get("ram", "")
|
||
ram_m = re.search(r"(\d+)\s*GB", ram_raw, re.I)
|
||
if ram_m:
|
||
row["ram_gb"] = int(ram_m.group(1))
|
||
ram_type_m = re.search(r"(LPDDR\d+[A-Z]*x?|DDR\d+[A-Z]*)", ram_raw, re.I)
|
||
if ram_type_m:
|
||
row["ram_type"] = ram_type_m.group(1).upper()
|
||
|
||
stor_raw = s.get("storage", "")
|
||
stor_m = re.search(r"(\d+)\s*(GB|TB)", stor_raw, re.I)
|
||
if stor_m:
|
||
val = int(stor_m.group(1))
|
||
unit = stor_m.group(2).upper()
|
||
row["storage_gb"] = val * 1024 if unit == "TB" else val
|
||
if re.search(r"nvme", stor_raw, re.I):
|
||
row["storage_type"] = "NVMe SSD"
|
||
elif re.search(r"\bssd\b", stor_raw, re.I):
|
||
row["storage_type"] = "SSD"
|
||
|
||
row["gpu"] = s.get("gpu", "")
|
||
|
||
os_raw = s.get("os", "")
|
||
if os_raw:
|
||
row["operating_system"] = os_raw
|
||
elif "win 11 pro" in title.lower():
|
||
row["operating_system"] = "Windows 11 Pro"
|
||
elif "win 11 home" in title.lower():
|
||
row["operating_system"] = "Windows 11 Home"
|
||
elif "win 10" in title.lower():
|
||
row["operating_system"] = "Windows 10"
|
||
|
||
if re.search(r"\b4G\b|\b5G\b|LTE", title):
|
||
row["has_4g_5g"] = 1
|
||
|
||
weight_raw = s.get("weight", "")
|
||
wm = re.search(r"(\d+[.,]\d+)\s*kg", weight_raw, re.I)
|
||
if wm:
|
||
row["weight_kg"] = float(wm.group(1).replace(",", "."))
|
||
|
||
row["wifi"] = s.get("wifi", "")
|
||
row["bluetooth"] = s.get("bluetooth", "")
|
||
|
||
return row
|
||
|
||
|
||
def _parse_nuvoo_body_html(body_html: str) -> dict:
|
||
"""Extract structured specs from Nuvoo's product body_html bullet list."""
|
||
if not body_html:
|
||
return {}
|
||
|
||
soup = BeautifulSoup(body_html, "lxml")
|
||
specs: dict = {}
|
||
|
||
bullets = []
|
||
for li in soup.find_all("li"):
|
||
strong = li.find("strong")
|
||
if strong:
|
||
bullets.append(strong.get_text(strip=True))
|
||
else:
|
||
bullets.append(li.get_text(strip=True))
|
||
|
||
for b in bullets:
|
||
bl = b.lower()
|
||
if re.match(r"(intel|amd|apple)\s", bl) and not specs.get("cpu"):
|
||
# Nuvoo sometimes injects marketing copy after a colon, e.g.:
|
||
# "Intel Core i5-1135G7-processor: smidig prestanda ..."
|
||
# Truncate at the colon and strip trailing "-processor" / "-chip" suffixes.
|
||
cpu_val = b.split(":")[0].strip()
|
||
cpu_val = re.sub(r"[-\s]*(processor|cpu|chip)$", "", cpu_val, flags=re.I).strip()
|
||
specs["cpu"] = cpu_val
|
||
elif ('"' in b or "tum" in bl or "inch" in bl) and re.search(r'\d+[,.]?\d*\s*(?:"|tum|inch|\u2033)', b, re.I) and not specs.get("screen"):
|
||
# Take only the spec part before any marketing description after a colon
|
||
specs["screen"] = b.split(":")[0].strip()
|
||
elif re.search(r'retina|liquid\s+retina', bl) and not specs.get("screen"):
|
||
specs["screen"] = b.split(":")[0].strip()
|
||
elif re.match(r"\d+\s*gb.*ddr", bl) and not specs.get("ram"):
|
||
specs["ram"] = b
|
||
elif re.match(r"\d+\s*(gb|tb)\s*(nvme|ssd|hdd|emmc|sshd)?", bl) and not specs.get("storage"):
|
||
specs["storage"] = b
|
||
elif re.match(r"(nvidia|rtx|gtx|amd radeon rx|quadro)", bl) and not specs.get("gpu"):
|
||
specs["gpu"] = b
|
||
|
||
for b in bullets:
|
||
bl = b.lower()
|
||
if re.search(r"wi-fi|wifi|802\.11", bl) and not specs.get("wifi"):
|
||
specs["wifi"] = b
|
||
if re.search(r"bluetooth", bl) and not specs.get("bluetooth"):
|
||
specs["bluetooth"] = b
|
||
if re.search(r"windows\s+\d+|macos", bl, re.I) and not specs.get("os"):
|
||
specs["os"] = b
|
||
if re.search(r"\bkg\b", bl) and not specs.get("weight"):
|
||
specs["weight"] = b
|
||
|
||
all_text = soup.get_text(" ", strip=True)
|
||
if not specs.get("wifi"):
|
||
m = re.search(r"(Wi-Fi\s+\d+[^\s,.,]*(?:\s+\([^)]+\))?)", all_text, re.I)
|
||
if m:
|
||
specs["wifi"] = m.group(1)
|
||
if not specs.get("bluetooth"):
|
||
m = re.search(r"(Bluetooth\s+[\d.]+([\s\S]{0,20})?)", all_text, re.I)
|
||
if m:
|
||
parts = m.group(1).strip().split()[:3]
|
||
specs["bluetooth"] = " ".join(parts)
|
||
|
||
return specs
|
||
|
||
|
||
def _parse_nuvoo_specs_tab(html: str) -> dict:
|
||
"""Parse the 'Specifikation' tab (#tab-content-for-specs) from a Nuvoo
|
||
product page, returning a dict keyed by our internal field names.
|
||
|
||
Tries multiple HTML structures: table rows, definition lists, div pairs.
|
||
Maps Swedish and English labels to our field names.
|
||
"""
|
||
soup = BeautifulSoup(html, "lxml")
|
||
|
||
container = (
|
||
soup.find(id="tab-content-for-specs")
|
||
or soup.find(id=re.compile(r"specs?", re.I))
|
||
)
|
||
if not container:
|
||
for el in soup.find_all(["section", "div"],
|
||
id=re.compile(r"spec|spek", re.I)):
|
||
container = el
|
||
break
|
||
if not container:
|
||
return {}
|
||
|
||
raw: dict[str, str] = {}
|
||
|
||
# Strategy 1: HTML table
|
||
for tr in container.find_all("tr"):
|
||
cells = tr.find_all(["td", "th"])
|
||
if len(cells) >= 2:
|
||
k = cells[0].get_text(strip=True).rstrip(":")
|
||
v = cells[1].get_text(" ", strip=True)
|
||
if k and v and len(k) < 80:
|
||
raw.setdefault(k, v)
|
||
|
||
# Strategy 2: definition list <dl><dt>…</dt><dd>…</dd></dl>
|
||
if not raw:
|
||
dts = container.find_all("dt")
|
||
dds = container.find_all("dd")
|
||
for dt, dd in zip(dts, dds):
|
||
k = dt.get_text(strip=True).rstrip(":")
|
||
v = dd.get_text(" ", strip=True)
|
||
if k and v and len(k) < 80:
|
||
raw.setdefault(k, v)
|
||
|
||
# Strategy 3: div pairs with label/value class names
|
||
if not raw:
|
||
for div in container.find_all("div"):
|
||
lbl = div.find(class_=re.compile(r"label|key|term|name", re.I))
|
||
val = div.find(class_=re.compile(r"value|data|desc", re.I))
|
||
if lbl and val:
|
||
k = lbl.get_text(strip=True).rstrip(":")
|
||
v = val.get_text(" ", strip=True)
|
||
if k and v and len(k) < 80:
|
||
raw.setdefault(k, v)
|
||
|
||
if not raw:
|
||
return {}
|
||
|
||
# Map Swedish + English labels to our field names (first match wins)
|
||
_LABEL_MAP = [
|
||
(r"sk[aä]rmstorlek|display\s*size|screen\s*size", "screen_size"),
|
||
(r"uppl[oö]sning|resolution", "screen_resolution"),
|
||
(r"sk[aä]rmtyp|panel(?:\s*typ)?|display\s*type", "screen_type"),
|
||
(r"processor|cpu", "cpu"),
|
||
(r"ram|arbetsminne", "ram"),
|
||
(r"lagring|ssd|hdd|storage", "storage"),
|
||
(r"grafik(?:kort)?|gpu", "gpu"),
|
||
(r"operativsystem|os", "os"),
|
||
(r"vikt", "weight"),
|
||
(r"wi.?fi|wifi", "wifi"),
|
||
(r"bluetooth", "bluetooth"),
|
||
]
|
||
|
||
result: dict[str, str] = {}
|
||
for label, value in raw.items():
|
||
for pattern, key in _LABEL_MAP:
|
||
if re.search(pattern, label, re.I):
|
||
if key not in result:
|
||
result[key] = value
|
||
break
|
||
|
||
return result
|
||
|
||
|
||
# ── Entry point ───────────────────────────────────────────────────────────────
|
||
|
||
def run(conn: sqlite3.Connection, skip_phase1: bool = False, reset: bool = False) -> None:
|
||
"""Run the Nuvoo scrape (Phase 1 + Phase 2 + sold-item cleanup)."""
|
||
if reset:
|
||
conn.execute("DELETE FROM laptops WHERE source='nuvoo'")
|
||
conn.execute("DELETE FROM scrape_queue WHERE source='nuvoo'")
|
||
conn.commit()
|
||
log.info("Nuvoo rows cleared.")
|
||
|
||
live_handles: set[str] = set()
|
||
|
||
if skip_phase1:
|
||
pending = conn.execute("SELECT COUNT(*) FROM scrape_queue WHERE source='nuvoo' AND done=0").fetchone()[0]
|
||
log.info(f"--skip-phase1: {pending} handles pending in queue.")
|
||
else:
|
||
live_handles = collect_handles(conn)
|
||
|
||
nuvoo_scrape_products(conn)
|
||
|
||
# Remove any DB rows that are no longer listed on the site (sold / removed).
|
||
if live_handles:
|
||
sold_count = 0
|
||
for (url,) in conn.execute(
|
||
"SELECT url FROM laptops WHERE source='nuvoo'"
|
||
).fetchall():
|
||
handle_m = re.search(r'/products/([^?/#]+)', url)
|
||
if handle_m and handle_m.group(1) not in live_handles:
|
||
conn.execute("DELETE FROM laptops WHERE url=?", (url,))
|
||
sold_count += 1
|
||
if sold_count:
|
||
conn.commit()
|
||
log.info(f" Removed {sold_count} sold/unlisted product(s) from DB.")
|
||
else:
|
||
log.info(" No sold/unlisted products found.")
|
||
|
||
total = conn.execute("SELECT COUNT(*) FROM laptops WHERE source='nuvoo'").fetchone()[0]
|
||
log.info(f"Nuvoo done. {total} Nuvoo laptop rows in DB.")
|