import json import re import threading import urllib.parse import time import math from typing import Optional, Dict, List, Any from backend.config import settings from backend.logging_config import _log, current_time from backend.db.connection import get_db_connection, db_connection from backend.scraper.api import _request from backend.scraper.psref import resolve_specs_for_laptop, infer_igpu_from_psref # --------------------------------------------------------------------------- # Spec Parsing & Extraction Helpers # --------------------------------------------------------------------------- def clean_model(name: str) -> str: """Clean model names for better UI readability.""" if not name: return "Unknown Model" cleaned = re.sub(r'(?i)^\s*(notebook|lenovo)', '', name) cleaned = re.sub(r'(?i)^\s*(workstation)', 'ThinkPad', cleaned) return cleaned.strip() def parse_processor_brand(proc_str: Optional[str]) -> Optional[str]: if not proc_str: return None match = re.search(r'(?i)Intel|AMD|MediaTek', proc_str) return match.group(0) if match else None def parse_processor_range(proc_str: Optional[str]) -> Optional[str]: if not proc_str: return None s = proc_str.replace("®", "").replace("™", "") s_lower = s.lower() # 1. Intel Core Ultra if "core ultra" in s_lower or "ultra" in s_lower: for num in ["9", "7", "5", "3"]: if f"ultra {num}" in s_lower or f"ultra {num}" in s_lower: return f"Intel Core Ultra {num}" return "Intel Core Ultra" # 2. AMD Ryzen AI if "ryzen ai" in s_lower: for num in ["9", "7", "5", "3"]: if f"ai {num}" in s_lower: return f"AMD Ryzen AI {num}" return "AMD Ryzen AI" # 3. AMD Ryzen if "ryzen" in s_lower: for num in ["9", "7", "5", "3"]: if f"ryzen {num}" in s_lower or f"ryzen r{num}" in s_lower: return f"AMD Ryzen {num}" return "AMD Ryzen" # 4. Intel Core iX if "core" in s_lower and any(f"i{num}" in s_lower for num in ["3", "5", "7", "9"]): for num in ["9", "7", "5", "3"]: if f"i{num}" in s_lower: return f"Intel Core i{num}" # 5. Intel Core X (e.g. Core 5 210H) if "core" in s_lower: for num in ["9", "7", "5", "3"]: if re.search(r'\bcore\s+' + num + r'\b', s_lower): return f"Intel Core {num}" # 6. Others for other in ["Celeron", "Pentium", "Athlon", "Xeon", "Atom"]: if other.lower() in s_lower: return other return "Others" def parse_memory_size(mem_str: Optional[str]) -> Optional[str]: if not mem_str: return None match = re.search(r'\d+\s*GB', mem_str, re.IGNORECASE) return match.group(0).replace(" ", "") if match else None def parse_storage_size(store_str: Optional[str]) -> Optional[str]: if not store_str: return None match = re.search(r'([\d\.]+)\s*(GB|TB)', store_str, re.IGNORECASE) if match: return f"{match.group(1)}{match.group(2)}".upper() return None def parse_storage_type(store_str: Optional[str]) -> Optional[str]: if not store_str: return None s = store_str.lower() if "and" in s and "drives" in s: return "Multi" if "hard drive" in s or "rpm" in s: return "HDD" if "embedded multi media card" in s or "emmc" in s: return "eMMC" if "solid state" in s or "ssd" in s: if "pcie" in s or "nvme" in s or "m.2" in s: return "SSD (NVMe)" if "sata" in s: return "SSD (SATA)" return "SSD" return None def parse_gpu_type(gpu_str: Optional[str]) -> Optional[str]: if not gpu_str: return None s = gpu_str.lower() if "integrated" in s or "intel iris" in s or "intel uhd" in s or "intel graphics" in s or "radeon graphics" in s: return "Integrated" if "rtx" in s or "gtx" in s or "nvidia" in s or "dedicated" in s or "discrete" in s or "geforce" in s or "quadro" in s or "t500" in s or "t600" in s or "t1000" in s or "arc a" in s or "rx " in s: return "Dedicated" if "radeon" in s or "intel" in s: return "Integrated" return "Integrated" def parse_gpu_name(gpu_str: Optional[str]) -> Optional[str]: if not gpu_str: return None s = gpu_str.replace("®", "").replace("™", "") s = re.sub(r'\s+', ' ', s).strip() match = re.search(r'(GeForce\s+RTX\s*\d{4}(?:\s*Ti)?(?:\s*[A-Z]\b)?|RTX\s*\d{4}(?:\s*Ti)?(?:\s*[A-Z]\b)?|GeForce\s+GTX\s*\d{4}(?:\s*Ti)?|GeForce\s+MX\d{3,5})', s, re.IGNORECASE) if match: val = match.group(1).strip() val = re.sub(r'(?i)geforce', 'GeForce', val) val_lower = val.lower() if "geforce" not in val_lower: val = "GeForce " + val if "nvidia" not in val.lower(): val = "NVIDIA " + val return val match = re.search(r'(Radeon\s+\d{3,4}M)', s, re.IGNORECASE) if match: return f"AMD {match.group(1).strip()}" match = re.search(r'(Radeon\s+RX\s*\d{4}(?:\s*S|\s*XT)?)', s, re.IGNORECASE) if match: return f"AMD {match.group(1).strip()}" if "arc" in s.lower(): match = re.search(r'(Arc\s+\d{3,4}\w*)', s, re.IGNORECASE) if match: return f"Intel {match.group(1).strip()}" return "Intel Arc Graphics" if "iris" in s.lower(): if "max" in s.lower(): return "Intel Iris Xe Max" return "Intel Iris Xe" if "uhd" in s.lower(): return "Intel UHD Graphics" if "intel graphics" in s.lower(): return "Intel Graphics" if "radeon graphics" in s.lower(): return "AMD Radeon Graphics" cleaned = re.sub(r'(?i)^\s*(integrated|discrete|dedicated)\s*', '', s) return cleaned.strip() _CPU_SKU_RE = re.compile(r'i[3579]-\d{4,5}[A-Z]{0,3}', re.IGNORECASE) # Last-resort iGPU table for when PSREF hasn't been backfilled yet for a given # MT (infer_igpu_from_psref returns None) — some DB copies have a populated # psref_specs cache (confirmed against data/lenovo_tracker.db: 110 machine # types as of 2026-07-20) and won't need this at all for most rows; others # (freshly restored snapshots, a new machine type PSREF hasn't indexed) will. # Covers the 6 13th Gen Intel SKUs originally found shipping with an entirely empty outlet # 'Graphic Card' field. Values are cross-checked against this catalog's own # other rows sharing the same SKU where any exist (I3-1315U, I7-1355U, # I5-13420H, I7-13620H all have a live sibling row with 'Graphic Card' # populated matching the value here); I5-1335U and I7-1365U have no live # sibling anywhere in the catalog, so those two are sourced from Lenovo's own # PSREF spec sheet for ThinkPad E14 Gen 5 (Intel), which lists i5-1335U/ # i7-1355U/i7-1365U as the same Iris Xe tier (unlike i3-1315U, UHD). # Deliberately a fixed table, not a live EU-count/P-suffix heuristic — a quick # real-data check found H-series branding doesn't split cleanly on P-suffix # alone (i7-13700H ships Iris Xe despite no "P", while i7-13620H ships UHD), # so only SKUs actually confirmed are listed here rather than guessed at scale. _INTEL_IGPU_FALLBACK: Dict[str, str] = { "I3-1315U": "Integrated Intel® UHD Graphics", "I5-1335U": "Integrated Intel® Iris® Xe Graphics", "I7-1355U": "Integrated Intel® Iris® Xe Graphics", "I7-1365U": "Integrated Intel® Iris® Xe Graphics", "I5-13420H": "Integrated Intel® UHD Graphics", "I7-13620H": "Integrated Intel® UHD Graphics", } def infer_gpu_from_processor(proc_str: Optional[str]) -> Optional[str]: """Last-resort iGPU fallback — see _INTEL_IGPU_FALLBACK above for which SKUs this covers and why.""" if not proc_str: return None s = proc_str.replace("®", "").replace("™", "") match = _CPU_SKU_RE.search(s) if not match: return None return _INTEL_IGPU_FALLBACK.get(match.group(0).upper()) def resolve_igpu_dgpu(gpu_str: Optional[str], proc_str: Optional[str], psref_spec_data_json: Optional[str]) -> "tuple[Optional[str], Optional[str]]": """Split outlet's single 'Graphic Card' field — present, entirely absent, or (when a laptop has both) only ever describing the dGPU, never the iGPU alongside it, per real outlet convention — into independent igpu/dgpu raw marketing strings (same "not yet cleaned" convention as 'graphic-card' itself — callers apply parse_gpu_name/clean_gpu_model same as they already do for gpu_str, so downstream detail like VRAM/GDDR-gen/"for 13th Gen Intel Processors" isn't lost the way it would be if this returned pre-shortened names). dGPU presence is trusted exclusively from outlet's own field, never inferred: there's no general signal that reliably predicts it (same exact CPU SKU ships integrated-only on some MTs and with a discrete GPU on others — see psref.py's infer_igpu_from_psref module comment for the concrete catalog examples that ruled out a CPU-tier heuristic). iGPU, by contrast, is a fixed fact of the CPU itself regardless of what else the laptop has, so it's resolved from the CPU whenever outlet's field doesn't already state it directly — this is what lets a buyer see genuine hybrid graphics (better idle/battery life) vs. a dGPU-only config, a distinction outlet's single conflated field can never make.""" gpu_type = parse_gpu_type(gpu_str) if gpu_str else None dgpu_raw = gpu_str if gpu_type == "Dedicated" else None if gpu_type == "Integrated": igpu_raw = gpu_str else: igpu_raw = infer_igpu_from_psref(psref_spec_data_json, proc_str) or infer_gpu_from_processor(proc_str) return igpu_raw, dgpu_raw def parse_processor_generation(proc_str: Optional[str]) -> Optional[str]: if not proc_str: return None s = proc_str.replace("®", "").replace("™", "") s_lower = s.lower() gen_match = re.search(r'(\d+)(?:th|st|nd|rd)\s+Generation', s, re.IGNORECASE) if gen_match: return f"{gen_match.group(1)}th Gen" if "core ultra" in s_lower or "ultra" in s_lower: model_match = re.search(r'(?:Ultra\s+[3579]\s+)?(\d{3})[U|H|V|HX|X]?', s, re.IGNORECASE) if model_match: first_digit = model_match.group(1)[0] if first_digit == "1": return "Core Ultra Series 1" elif first_digit == "2": return "Core Ultra Series 2" return "Core Ultra" if "intel core" in s_lower: model_match = re.search(r'(?:Core\s+[3579]\s+)?(\d{3})[U|H|V|HX]?', s, re.IGNORECASE) if model_match: first_digit = model_match.group(1)[0] if first_digit == "1": return "Core Series 1" elif first_digit == "2": return "Core Series 2" if "ryzen" in s_lower: if "ryzen ai" in s_lower: model_match = re.search(r'(?:AI\s+[3579]\s+)?(\d{3})', s, re.IGNORECASE) if model_match: first_digit = model_match.group(1)[0] if first_digit == "3": return "Ryzen AI 300 Series" return "Ryzen AI" model_match = re.search(r'Ryzen\s+[3579]\s+R?(\d{4})', s, re.IGNORECASE) if model_match: first_digit = model_match.group(1)[0] if first_digit == "5": return "Ryzen 5000 Series" elif first_digit == "6": return "Ryzen 6000 Series" elif first_digit == "7": return "Ryzen 7000 Series" elif first_digit == "8": return "Ryzen 8000 Series" return None def parse_processor_series(proc_str: Optional[str]) -> Optional[str]: if not proc_str: return None s = proc_str.replace("®", "").replace("™", "") match = re.search(r'\b\d{3,4}(HX|HS|H|U|V)\b', s, re.IGNORECASE) if match: suffix = match.group(1).upper() if suffix in ("H", "HS"): return "H/HS-Series" return f"{suffix}-Series" if re.search(r'\b\d{3,5}HX\b|\bHX\b', s, re.IGNORECASE): return "HX-Series" if re.search(r'\b\d{3,5}HS\b|\bHS\b', s, re.IGNORECASE): return "H/HS-Series" if re.search(r'\b\d{3,5}H\b|\bH\b', s, re.IGNORECASE): return "H/HS-Series" if re.search(r'\b\d{3,5}U\b|\bU\b', s, re.IGNORECASE): return "U-Series" if re.search(r'\b\d{3,5}V\b|\bV\b', s, re.IGNORECASE): return "V-Series" return "Other" def clean_cpu_model(proc_str: Optional[str]) -> Optional[str]: if not proc_str: return None s = proc_str.replace("®", "").replace("™", "") s = re.sub(r'\s+', ' ', s).strip() # Strip generation prefix if present s = re.sub(r'(?i)^\s*\d+(?:th|st|nd|rd)\s+Gen(?:eration)?\s+', '', s) # Extract portion before 'Processor' match = re.search(r'(.*?)\s+Processor', s, re.IGNORECASE) if match: s = match.group(1).strip() return s def clean_gpu_model(gpu_str: Optional[str]) -> Optional[str]: if not gpu_str: return None s = gpu_str.replace("®", "").replace("™", "") s = re.sub(r'\s+', ' ', s).strip() s = re.sub(r'(?i)^\s*(Integrated|Discrete|Dedicated)\s+', '', s) s = re.sub(r'(?i)geforce', 'GeForce', s) # Simplify dedicated GPU laptop suffixes while keeping VRAM (e.g. Laptop GPU 8GB GDDR6 -> 8GB) s = re.sub(r'(?i)\s+Laptop\s+GPU\s+(\d+GB)\s+GDDR\d+', r' \1', s) return s def get_product_cpu_gpu(product: dict) -> tuple[str, str]: # Try to extract cpu and gpu from product first cpu = None gpu = None # Check specs sub-dict or full_specs if present in product dict product_specs = product.get("specs") or product.get("full_specs") if isinstance(product_specs, dict): cpu_raw = product_specs.get("Processor") gpu_raw = product_specs.get("Graphic Card") if cpu_raw: cpu = clean_cpu_model(cpu_raw) if gpu_raw: gpu = clean_gpu_model(gpu_raw) # Check classification list classification = product.get("classification") if not cpu or not gpu: if classification and isinstance(classification, list): for item in classification: if isinstance(item, dict): if item.get("a") == "Processor": cpu = clean_cpu_model(item.get("b")) elif item.get("a") == "Graphic Card": gpu = clean_gpu_model(item.get("b")) # If not found, check DB if not cpu or not gpu: code = product.get("productCode") or product.get("product_code") if code: try: conn = get_db_connection() cursor = conn.cursor() cursor.execute("SELECT specs FROM products WHERE product_code = ?", (code,)) row = cursor.fetchone() if row and row[0]: specs_dict = json.loads(row[0]) if not cpu: cpu = clean_cpu_model(specs_dict.get("Processor")) if not gpu: gpu = clean_gpu_model(specs_dict.get("Graphic Card")) except Exception: pass finally: if 'conn' in locals(): conn.close() return cpu or "Unknown CPU", gpu or "Unknown GPU" def parse_ddr_gen(mem_str: Optional[str]) -> Optional[str]: if not mem_str: return None s = mem_str.upper() for gen in ["LPDDR5X", "LPDDR5", "DDR5", "LPDDR4X", "LPDDR4", "DDR4"]: if gen in s: return gen if "DDR3" in s: return "DDR3" return None def parse_screen_size(disp_str: Optional[str]) -> Optional[float]: if not disp_str: return None # A minority of outlet Display strings lead with a centimeter figure ahead of # the parenthesized inch size — e.g. "39.62cms (15.6) FHD…" (should be 15.6) # or "35.56cms (14) WUXGA…" (should be 14). Confirmed against every distinct # live Display string: 2/58 do this. The generic regex below would otherwise # grab the cm number first and misparse it as an inch size (fable audit # B-SCR-6), so prefer the parenthesized figure whenever "cm" is present. if re.search(r'(?i)cm', disp_str): paren_match = re.search(r'\(([\d.]+)\)', disp_str) if paren_match: try: return float(paren_match.group(1)) except ValueError: pass match = re.search(r'(\d{2}\.?\d?)', disp_str) if match: try: return float(match.group(1)) except ValueError: pass return None def parse_resolution(disp_str: Optional[str]) -> Optional[str]: if not disp_str: return None match = re.search(r'(\d+)\s*x\s*(\d+)', disp_str) if match: return f"{match.group(1)}x{match.group(2)}" standards = { "WUXGA": "1920x1200", "WQXGA": "2560x1600", "WQUXGA": "3840x2400", "2.2K": "2240x1400", "2.5K": "2560x1600", "2.8K": "2880x1800", "2K": "2160x1350", "3K": "3072x1920", "FHD+": "1920x1200", "UHD+": "3840x2400", "FHD": "1920x1080", "UHD": "3840x2160", "QHD": "2560x1600", "HD+": "1600x900", "HD": "1366x768" } for k, v in standards.items(): if k in disp_str: return v return None def parse_aspect_ratio(resolution: Optional[str]) -> Optional[str]: if not resolution: return None match = re.search(r'(\d+)\s*x\s*(\d+)', resolution) if not match: return None try: w = int(match.group(1)) h = int(match.group(2)) except ValueError: return None if h == 0: return None ratio = w / h if abs(ratio - 16/10) < 0.01: return "16:10" if abs(ratio - 16/9) < 0.01: return "16:9" if abs(ratio - 3/2) < 0.01: return "3:2" if abs(ratio - 4/3) < 0.01: return "4:3" if abs(ratio - 5/4) < 0.01: return "5:4" if abs(ratio - 21/9) < 0.05: return "21:9" if abs(ratio - 32/9) < 0.05: return "32:9" gcd = math.gcd(w, h) return f"{w // gcd}:{h // gcd}" def parse_battery_capacity(battery_str: Optional[str]) -> Optional[float]: if not battery_str: return None match = re.search(r'(\d+(?:\.\d+)?)\s*Wh', battery_str, re.IGNORECASE) if match: try: return float(match.group(1)) except ValueError: return None return None def parse_weight_kg(weight_str: Optional[str]) -> Optional[float]: """Real format e.g. '1.6kgs / 3.52lbs' — extract just the kg figure.""" if not weight_str: return None match = re.search(r'(\d+(?:\.\d+)?)\s*kg', weight_str, re.IGNORECASE) if match: try: return float(match.group(1)) except ValueError: return None return None # --------------------------------------------------------------------------- # Data Handlers (Retrieving parsed DB results) # --------------------------------------------------------------------------- # Full-catalog build (product_code=None) reparses every row from scratch — regex # parsing of raw spec text, PSREF joins, etc. — which is real per-request CPU cost # that was previously redone on every hit to GET /api/laptops even though the # underlying data only changes once per scrape cycle (settings.POLL_INTERVAL, # 60s default). Cache the built full-catalog list, TTL'd to POLL_INTERVAL, so # concurrent/rapid requests within one scrape cycle reuse the same build instead # of redoing it. The single-product path (product_code set) stays uncached — it's # already cheap via the SQL `WHERE`, so caching it would add complexity (a # per-code cache) for no real win. See DECISION_LOG.md's 2026-07-20 entry. _catalog_cache_lock = threading.Lock() _catalog_cache: Dict[str, Any] = {"data": None, "built_at": 0.0} def _invalidate_catalog_cache() -> None: """Test-only seam — production never needs to call this; the TTL handles staleness on its own. Exists so tests that swap `settings.DB_FILE` between cases (see test_enrichment.py) can't read a stale cached build from a previous test's DB.""" with _catalog_cache_lock: _catalog_cache["data"] = None _catalog_cache["built_at"] = 0.0 def get_laptops_data(product_code: Optional[str] = None) -> List[Dict[str, Any]]: """Fetch laptops from DB and parse specifications dynamically. `product_code=None` (the full catalog) is served from a short TTL cache — see the comment above `_catalog_cache`. Pass a `product_code` to bypass the cache and read the single row directly.""" if product_code is not None: return _build_laptops_data(product_code=product_code) with _catalog_cache_lock: fresh = ( _catalog_cache["data"] is not None and time.time() - _catalog_cache["built_at"] < settings.POLL_INTERVAL ) if fresh: return _catalog_cache["data"] data = _build_laptops_data(product_code=None) with _catalog_cache_lock: _catalog_cache["data"] = data _catalog_cache["built_at"] = time.time() return data def _build_laptops_data(product_code: Optional[str] = None) -> List[Dict[str, Any]]: """Fetch laptops from DB and parse specifications dynamically. With `product_code`, filters to that single row in SQL rather than fetching and parsing the full catalog just to discard everything else — backs the single-product endpoint (`GET /api/laptops?code=`) used by Product Details. """ if not settings.DB_FILE.exists(): return [] with db_connection() as conn: cursor = conn.cursor() if product_code: cursor.execute(""" SELECT p.*, (SELECT ph.price_delta FROM price_history ph WHERE ph.product_code = p.product_code ORDER BY ph.timestamp DESC, ph.id DESC LIMIT 1) as latest_price_delta, psref.product_key as psref_product_key, psref.page_url as psref_page_url, psref.spec_data as psref_spec_data FROM products p LEFT JOIN psref_specs psref ON substr(p.product_code, 1, 4) = psref.machine_type WHERE p.product_code = ? """, (product_code,)) else: cursor.execute(""" SELECT p.*, (SELECT ph.price_delta FROM price_history ph WHERE ph.product_code = p.product_code ORDER BY ph.timestamp DESC, ph.id DESC LIMIT 1) as latest_price_delta, psref.product_key as psref_product_key, psref.page_url as psref_page_url, psref.spec_data as psref_spec_data FROM products p LEFT JOIN psref_specs psref ON substr(p.product_code, 1, 4) = psref.machine_type ORDER BY p.active DESC, p.last_seen DESC """) rows = cursor.fetchall() laptops = [] for row in rows: specs = {} if row['specs']: try: specs = json.loads(row['specs']) except Exception as e: _log(f"Error parsing specs for {row['product_code']}: {e}") proc = specs.get('Processor', '') mem = specs.get('Memory', '') store = specs.get('Storage', '') disp = specs.get('Display', '') gpu_str = specs.get('Graphic Card') # See resolve_igpu_dgpu's docstring: independently-sourced iGPU # (CPU-derived, PSREF live join then the fixed fallback table) and # dGPU (outlet's own field only, never guessed) raw strings — kept # raw (not parse_gpu_name'd here) same as gpu_str/'graphic-card', # so VRAM/GDDR-gen/generation-suffix detail survives for callers # that want it (the GPU filter facet, Product Details/Compare's # spec rows); igpu-name/dgpu-name below are the parsed short form # for space-constrained displays (browse table, ProductCard, search). igpu_raw, dgpu_raw = resolve_igpu_dgpu(gpu_str, proc, row["psref_spec_data"]) p_name = row['product_name'] or "Lenovo Laptop" full_specs = {} if row['full_specs']: try: full_specs = json.loads(row['full_specs']) except Exception: pass # Calculate cart hold status dynamically in_cart_hold = False hold_expires_in_seconds = 0 if row['active'] == 0 and row['pending_removal_since'] is not None: try: fmt = "%Y-%m-%d %H:%M:%S" pending_ts = time.mktime(time.strptime(row['pending_removal_since'], fmt)) elapsed = int(time.time() - pending_ts) db_debounce = row['debounce_duration'] if db_debounce is None: db_debounce = settings.TIER0_DEBOUNCE_SECONDS remaining = db_debounce - elapsed if remaining > 0: in_cart_hold = True hold_expires_in_seconds = remaining except Exception as e: _log(f"Error calculating hold expiration for {row['product_code']}: {e}") res = parse_resolution(disp) laptop = { "available": row['active'] == 1, "in_cart_hold": in_cart_hold, "hold_expires_in_seconds": hold_expires_in_seconds, "brand": "Lenovo", "model": clean_model(p_name), "processor": proc, "processor-brand": parse_processor_brand(proc), "processor-range": parse_processor_range(proc), "processor-generation": parse_processor_generation(proc), "processor-series": parse_processor_series(proc), "memory": mem, "memory-size": parse_memory_size(mem), # "Memory on Package" (Intel Lunar Lake/Meteor Lake-class LPDDR5X # fused directly onto the CPU package) is Lenovo's alternate wording # for non-upgradeable memory, seen alongside plain "(Soldered)" in # real catalog data — arguably even more permanently fixed than a # soldered-to-motherboard chip, not less. A bare "soldered" substring # check missed it entirely, mislabeling it "Upgradeable" (backwards). "memory-soldered": any(kw in mem.lower() for kw in ("soldered", "on package")) if mem else False, "ddr-gen": parse_ddr_gen(mem), "storage": store, "storage-size": parse_storage_size(store), "storage-type": parse_storage_type(store), "display": disp, "screen-size": parse_screen_size(disp), "panel-type": "IPS" if disp and "ips" in disp.lower() else "OLED" if disp and "oled" in disp.lower() else "TN" if disp and "tn" in disp.lower() else None, "touch-screen": "touch" in disp.lower() and "non-touch" not in disp.lower() if disp else False, "resolution": res, "screen-aspect-ratio": parse_aspect_ratio(res), "operating-system": specs.get('Operating System'), "graphic-card": gpu_str, "gpu-type": parse_gpu_type(gpu_str), "gpu-name": parse_gpu_name(gpu_str), "igpu": igpu_raw, "dgpu": dgpu_raw, "igpu-name": parse_gpu_name(igpu_raw), "dgpu-name": parse_gpu_name(dgpu_raw), "orig-price": row['original_price'], "price": row['current_price'], "price-delta": row['latest_price_delta'] if row['latest_price_delta'] is not None else 0.0, "percentage-savings": row['save_percent'], "product-condition": row['condition'], "product-number": row['product_code'], "url": f"https://www.lenovo.com/in/outletin/en/p/{row['product_code']}", "warranty": specs.get('Warranty') or full_specs.get('Warranty'), "weight": specs.get('Weight') or full_specs.get('Weight'), "weight-kg": parse_weight_kg(specs.get('Weight') or full_specs.get('Weight')), "thumbnail_url": row['thumbnail_url'], "rating_star": row['rating_star'], "comment_count": row['comment_count'], "first_seen": row['first_seen'], "last_seen": row['last_seen'], "removed_at": row['removed_at'], # Extended spec fields from compare API "camera": full_specs.get('Camera'), "wlan": full_specs.get('WIFI'), "bluetooth": full_specs.get('Bluetooth'), "battery": full_specs.get('Battery'), "battery-capacity": parse_battery_capacity(full_specs.get('Battery')), "keyboard": full_specs.get('Keyboard'), "fingerprint-reader": full_specs.get('Fingerprint Reader'), "pointing-device": full_specs.get('Pointing Device'), "ac-adapter": full_specs.get('AC Adapter / Power Supply'), "color": full_specs.get('Color'), "specs": specs, "full_specs": full_specs, } psref_row = None if row["psref_spec_data"]: psref_row = { "product_key": row["psref_product_key"], "page_url": row["psref_page_url"], "spec_data": row["psref_spec_data"], } specs_extra, reference_specs, reference_specs_url = resolve_specs_for_laptop(laptop, psref_row) laptop["specs_extra"] = specs_extra laptop["reference_specs"] = reference_specs laptop["reference_specs_url"] = reference_specs_url laptops.append(laptop) # Strip internal scraper metadata — these columns are not for the frontend _INTERNAL_COLS = ("stability_count", "pending_state", "pending_removal_since", "debounce_duration") for item in laptops: for col in _INTERNAL_COLS: item.pop(col, None) return laptops # D-00014's PO Notes: "Similar models are matched objectively by brand, size, and # +/-15% price range proximity." Brand is a no-op (every catalog entry is Lenovo). SIMILAR_PRICE_TOLERANCE_PCT = 15 def get_similar_laptops(product_code: str, limit: int = 6) -> Optional[List[Dict[str, Any]]]: """Server-side port of frontend-v2's `findSimilarLaptops` (formerly `lib/productDetails.ts`, D-00014: same screen-size + +/-15% price + in-stock, sorted by price proximity) — backs `GET /api/laptops?similar_to=`. Filtering here instead of on the frontend means Product Details no longer needs the full ~4MB catalog just to find ~6 matches (see DECISION_LOG.md's 2026-07-20 entry). Matching logic is duplicated rather than shared across the Python/TS boundary — deliberate, since D-00014's rule is small and stable; see the same log entry for the tradeoff this accepts. Returns `None` if `product_code` itself doesn't match any product (caller's cue to 404, same as `?code=`'s not-found handling) — distinct from a real empty match list (`[]`, a valid "no similar laptops" result for a found product), same not-found/empty distinction `?code=` already makes. """ target_rows = _build_laptops_data(product_code=product_code) if not target_rows: return None target = target_rows[0] target_size = target["screen-size"] target_price = target["price"] lo = target_price * (1 - SIMILAR_PRICE_TOLERANCE_PCT / 100) hi = target_price * (1 + SIMILAR_PRICE_TOLERANCE_PCT / 100) candidates = [ l for l in get_laptops_data() if l["product-number"] != target["product-number"] and l["available"] is True and l["screen-size"] == target_size and lo <= l["price"] <= hi ] candidates.sort(key=lambda l: abs(l["price"] - target_price)) return candidates[:limit] def get_price_history(product_code: str) -> List[Dict[str, Any]]: """Fetch price history data for a product code.""" if not settings.DB_FILE.exists(): return [] with db_connection() as conn: cursor = conn.cursor() cursor.execute(""" SELECT timestamp, price, save_percent FROM price_history WHERE product_code = ? ORDER BY timestamp ASC """, (product_code,)) rows = cursor.fetchall() return [dict(row) for row in rows] # Minimum price_history points a product needs to be eligible as the landing # page's featured chart subject (a 1-point history is a flat, uninformative line). FEATURED_PRICE_HISTORY_MIN_POINTS = 3 def get_featured_price_history_subject() -> Optional[Dict[str, Any]]: """Picks the product with the richest price-history record to showcase on the landing page's live-catalog chart (backs GET /api/price_history/featured). Deliberately considers historical (inactive, no-longer-listed) products too, not just currently-active ones — this is what makes the landing page's "Nothing's ever lost" claim concrete rather than just a copy claim. If no product clears FEATURED_PRICE_HISTORY_MIN_POINTS yet (early-catalog / low-history state), falls back to whichever product has the most points at all, as long as it still clears PriceHistoryChart's own 2-point rendering floor — showing the best real chart available beats hiding the landing page card entirely. Returns None only when no product has even 2 points.""" if not settings.DB_FILE.exists(): return None with db_connection() as conn: cursor = conn.cursor() cursor.execute(""" SELECT product_code, COUNT(*) as pts FROM price_history GROUP BY product_code HAVING COUNT(*) >= ? ORDER BY pts DESC, MAX(timestamp) DESC LIMIT 1 """, (FEATURED_PRICE_HISTORY_MIN_POINTS,)) row = cursor.fetchone() if not row: cursor.execute(""" SELECT product_code, COUNT(*) as pts FROM price_history GROUP BY product_code HAVING COUNT(*) >= 2 ORDER BY pts DESC, MAX(timestamp) DESC LIMIT 1 """) row = cursor.fetchone() if not row: return None product_code = row["product_code"] cursor.execute( "SELECT product_name, active FROM products WHERE product_code = ?", (product_code,), ) product = cursor.fetchone() if not product: return None return { "code": product_code, "name": product["product_name"], "active": bool(product["active"]), "data": get_price_history(product_code), } # --------------------------------------------------------------------------- # Spec Enrichment & Ghost listings cleanup tasks # --------------------------------------------------------------------------- def clean_ghost_listings() -> List[str]: """Verify inventoryStatus of active laptops and mark out-of-stock items as inactive.""" with db_connection() as conn: cursor = conn.cursor() cursor.execute("SELECT product_code, current_price FROM products WHERE active = 1") rows = cursor.fetchall() codes = [row['product_code'] for row in rows] prices = {row['product_code']: row['current_price'] for row in rows} if not codes: return [] compare_req = [{"categoryCode": "laptops", "productNumber": codes}] url = "https://openapi.lenovo.com/in/outletin/en/product/compare/getCompareData?compareReq=" + urllib.parse.quote(json.dumps(compare_req)) cleaned_codes = [] try: response = _request("GET", url, headers=settings.COMPARE_HEADERS, timeout=15) if response.status_code == 200: data = response.json() categories = data.get("data", []) items = categories[0].get("productList", []) if categories else [] for item in items: prod_code = item.get("productNumber") inv_status = item.get("inventoryStatus") # 1 = available, 2 = unavailable if prod_code and inv_status == 2: now_str = current_time() cursor.execute(""" UPDATE products SET active = 0, removed_at = ? WHERE product_code = ? AND active = 1 """, (now_str, prod_code)) if cursor.rowcount > 0: cleaned_codes.append(prod_code) # This is an admin-triggered manual cleanup (POST # /api/clean_ghosts), not the debounced auto-scan path in # processor.py — correctly stays immediate/undebounced. # Routed through the shared helper (not a duplicate raw # INSERT) so there's exactly one stock_history insertion # point to keep the CatalogState cursor in sync with. from backend.scraper.processor import _insert_stock_event _insert_stock_event(cursor, prod_code, "removed", now_str, prices.get(prod_code)) _log(f"[Ghost Cleanup] Product {prod_code} is out of stock. Marked active=0.") else: _log(f"Ghost cleanup API error: HTTP {response.status_code}") except Exception as e: _log(f"Error executing ghost cleanup API query: {e}") result = cleaned_codes # Resync the in-memory cursor from whatever actually committed — same reasoning # as process_scanned_products (backend/scraper/processor.py): must happen after # the `with db_connection()` block has exited (its own commit already fired), # not before, so a rolled-back transaction can never leave the cursor drifted. if result: from backend.scraper.catalog_state import catalog_state catalog_state.init_from_db() return result def get_stock_history(product_code: str) -> List[Dict[str, Any]]: """Fetch stock history data for a product code.""" if not settings.DB_FILE.exists(): return [] with db_connection() as conn: cursor = conn.cursor() cursor.execute(""" SELECT timestamp, event_type, price FROM stock_history WHERE product_code = ? ORDER BY timestamp ASC """, (product_code,)) rows = cursor.fetchall() return [dict(row) for row in rows] def get_last_updated_seconds_ago() -> Optional[int]: """Seconds since the most recent product row was touched by the scraper. Backs the header's Live Monitoring pill (D-00086) — `last_seen` is bumped on every scan for every currently-visible product, so its max is a real freshness signal, not a fabricated one. """ if not settings.DB_FILE.exists(): return None with db_connection() as conn: cursor = conn.cursor() cursor.execute("SELECT MAX(last_seen) as latest FROM products") row = cursor.fetchone() latest = row['latest'] if row else None if not latest: return None try: latest_ts = time.mktime(time.strptime(latest, "%Y-%m-%d %H:%M:%S")) except ValueError: return None return max(0, int(time.time() - latest_ts)) def get_activity(limit: int = 200, skus: Optional[List[str]] = None) -> List[Dict[str, Any]]: """Unified stock + price event feed backing /updates (Phase 6). Rows are never hard-deleted from stock_history/price_history, so the LEFT JOIN to products for display fields (name/thumbnail) is always safe. Each source table is independently ORDER BY timestamp DESC LIMIT ?'d *before* the UNION ALL, so at most 2*limit rows ever reach the outer sort/join — the previous version unioned the two full tables first and sorted that combined set on every call (fable audit B-PERF-2), which only gets more expensive as both tables grow (already unbounded — see B-DB-4).""" if not settings.DB_FILE.exists(): return [] sku_filter = "" sku_params: List[Any] = [] if skus: placeholders = ",".join("?" for _ in skus) sku_filter = f"WHERE product_code IN ({placeholders})" sku_params = list(skus) params = sku_params + [limit] + sku_params + [limit] + [limit] with db_connection() as conn: cursor = conn.cursor() cursor.execute(f""" SELECT combined.id, combined.row_type, combined.product_code, combined.event_type, combined.timestamp, combined.price, combined.old_price, combined.price_delta, products.product_name, products.thumbnail_url FROM ( SELECT * FROM ( SELECT id, 'stock' as row_type, product_code, event_type, timestamp, price, NULL as old_price, NULL as price_delta FROM stock_history {sku_filter} ORDER BY timestamp DESC, id DESC LIMIT ? ) UNION ALL SELECT * FROM ( SELECT id, 'price' as row_type, product_code, NULL as event_type, timestamp, price, price - price_delta as old_price, price_delta FROM price_history {sku_filter} ORDER BY timestamp DESC, id DESC LIMIT ? ) ) combined LEFT JOIN products ON products.product_code = combined.product_code ORDER BY combined.timestamp DESC, combined.id DESC LIMIT ? """, params) rows = cursor.fetchall() return [dict(row) for row in rows] def enrich_specs(codes: Optional[List[str]] = None) -> Dict[str, Any]: """Fetch full specs from the Lenovo Compare API for active products and cache them in DB. Used by the manual /api/enrich_specs endpoint as a backfill tool. Sends all codes in a single Compare API request (API supports 166+ codes per request). """ with db_connection() as conn: cursor = conn.cursor() if codes: placeholders = ",".join("?" * len(codes)) cursor.execute(f"SELECT product_code FROM products WHERE product_code IN ({placeholders})", codes) else: cursor.execute("SELECT product_code FROM products WHERE active = 1 AND (full_specs IS NULL OR full_specs = '')") all_codes = [row[0] for row in cursor.fetchall()] if not all_codes: _log("[Enrich] No products need enrichment.") return {"enriched": 0, "failed": 0} enriched_count = 0 failed_count = 0 # Send all codes in a single request — API supports 166+ codes per request compare_req = [{"categoryCode": "laptops", "productNumber": all_codes}] url = ("https://openapi.lenovo.com/in/outletin/en/product/compare/getCompareData?compareReq=" + urllib.parse.quote(json.dumps(compare_req))) try: _log(f"[Enrich] Fetching full_specs for {len(all_codes)} products in one request...") response = _request("GET", url, headers=settings.COMPARE_HEADERS, timeout=30) if response.status_code != 200: _log(f"[Enrich] HTTP {response.status_code}") failed_count = len(all_codes) else: data = response.json() categories = data.get("data", []) items = categories[0].get("productList", []) if categories else [] returned_codes = set() for item in items: prod_code = item.get("productNumber") if not prod_code: continue returned_codes.add(prod_code) classification = item.get("classification", []) full_specs_dict = { s["a"]: s["b"] for s in classification if "a" in s and "b" in s } if item.get("inventoryStatus") is not None: full_specs_dict["__inventoryStatus"] = item["inventoryStatus"] full_specs_json = json.dumps(full_specs_dict, ensure_ascii=False) cursor.execute( "UPDATE products SET full_specs = ? WHERE product_code = ?", (full_specs_json, prod_code) ) if cursor.rowcount > 0: enriched_count += 1 _log(f"[Enrich] Updated full_specs for {prod_code} ({len(full_specs_dict)} fields)") # Count codes not returned by API as failed failed_count = len(set(all_codes) - returned_codes) if failed_count: _log(f"[Enrich] {failed_count} codes not returned by Compare API.") except Exception as e: _log(f"[Enrich] Request error: {e}") failed_count = len(all_codes) _log(f"[Enrich] Done. Enriched: {enriched_count} | Failed: {failed_count}") return {"enriched": enriched_count, "failed": failed_count} def verify_and_enrich(products: List[Dict[str, Any]]) -> List[Dict[str, Any]]: """Single Compare API call per cycle that both filters ghosts and enriches full_specs. Replaces the old filter_out_ghosts() + post-cycle enrich_specs() pattern. Sends ALL product codes from the Search API in ONE Compare API request, then: 1. Filters out ghost products (inventoryStatus == 2) 2. Writes full_specs for any product that doesn't have them yet Result: 4 Lenovo API requests per cycle (3 search + 1 compare) with instant ghost detection. """ if not products: return [] codes = [p["productCode"] for p in products if p.get("productCode")] if not codes: return products # Determine which products are missing full_specs in DB so we can backfill missing_specs_codes: set = set() try: with db_connection() as conn: cursor = conn.cursor() placeholders = ",".join("?" * len(codes)) cursor.execute( f"SELECT product_code FROM products WHERE product_code IN ({placeholders}) AND (full_specs IS NULL OR full_specs = '')", codes ) missing_specs_codes = {row[0] for row in cursor.fetchall()} except Exception as e: _log(f"[Verify+Enrich] DB query for missing specs failed: {e}") # Single Compare API request for ALL codes compare_req = [{"categoryCode": "laptops", "productNumber": codes}] url = ("https://openapi.lenovo.com/in/outletin/en/product/compare/getCompareData?compareReq=" + urllib.parse.quote(json.dumps(compare_req))) unavailable_codes: set = set() enriched_count = 0 try: _log(f"[Verify+Enrich] Checking {len(codes)} products (ghost filter + spec enrichment)...") response = _request("GET", url, headers=settings.COMPARE_HEADERS, timeout=30) if response.status_code != 200: _log(f"[Verify+Enrich] Compare API error HTTP {response.status_code}. Aborting cycle — fail-closed.") return None data = response.json() categories = data.get("data", []) items = categories[0].get("productList", []) if categories else [] specs_to_write: List[tuple] = [] for item in items: prod_code = item.get("productNumber") if not prod_code: continue inv_status = item.get("inventoryStatus") # 1 = available, 2 = unavailable if inv_status == 2: unavailable_codes.add(prod_code) # Spec enrichment — only for products missing full_specs # (also picks up brand-new products just inserted in this cycle) if prod_code in missing_specs_codes: classification = item.get("classification", []) full_specs_dict = { s["a"]: s["b"] for s in classification if "a" in s and "b" in s } if inv_status is not None: full_specs_dict["__inventoryStatus"] = inv_status if full_specs_dict: specs_to_write.append((json.dumps(full_specs_dict, ensure_ascii=False), prod_code)) # Write enriched specs to DB if specs_to_write: try: with db_connection() as conn: cursor = conn.cursor() cursor.executemany( "UPDATE products SET full_specs = ? WHERE product_code = ?", specs_to_write ) enriched_count = cursor.rowcount except Exception as e: _log(f"[Verify+Enrich] DB write for full_specs failed: {e}") ghost_count = len(unavailable_codes) if ghost_count > 0: _log(f"[Verify+Enrich] Ghosts filtered ({ghost_count}): {', '.join(sorted(unavailable_codes))}") _log(f"[Verify+Enrich] Done. Ghosts filtered: {ghost_count} | Specs enriched: {enriched_count}/{len(missing_specs_codes)}") # Phase 9: persist the cycle's listed-vs-purchasable snapshot + change-only # ghost log (ghosts never enter the products table, so this is the only # queryable record of them). Never lets a stats failure abort the cycle. try: record_scrape_cycle_stats(products, unavailable_codes) except Exception as e: _log(f"[Verify+Enrich] Stats recording failed (non-fatal): {e}") except Exception as e: _log(f"[Verify+Enrich] Request error: {e}. Aborting cycle — fail-closed.") return None # Return products with ghosts removed return [p for p in products if p.get("productCode") not in unavailable_codes] def record_scrape_cycle_stats(products: List[Dict[str, Any]], unavailable_codes: set) -> None: """Persist per-cycle catalog stats (scrape_stats, single row) and a change-only ghost log (ghost_events + ghosts_current) — migration 12. "Change-only" per user request (2026-07-14): a product gets one 'appeared' row when it first shows up as a ghost and one 'resolved' row when it stops being one, not a row per ghost per cycle. Name/price are captured from the Search API payload since ghost products are filtered out before ever reaching the products table. """ now = time.strftime("%Y-%m-%d %H:%M:%S") listed = len({p.get("productCode") for p in products if p.get("productCode")}) ghost_count = len(unavailable_codes) purchasable = listed - ghost_count by_code = {p.get("productCode"): p for p in products if p.get("productCode")} with db_connection() as conn: cursor = conn.cursor() cursor.execute( """INSERT INTO scrape_stats (id, listed_count, purchasable_count, ghost_count, updated_at) VALUES (1, ?, ?, ?, ?) ON CONFLICT(id) DO UPDATE SET listed_count = excluded.listed_count, purchasable_count = excluded.purchasable_count, ghost_count = excluded.ghost_count, updated_at = excluded.updated_at""", (listed, purchasable, ghost_count, now), ) cursor.execute("SELECT product_code FROM ghosts_current") previous = {row[0] for row in cursor.fetchall()} for code in unavailable_codes - previous: p = by_code.get(code, {}) name = p.get("productName") try: price = float(p.get("finalPrice", 0)) or None except (TypeError, ValueError): price = None cursor.execute( "INSERT INTO ghost_events (timestamp, product_code, product_name, price, event_type) VALUES (?, ?, ?, ?, 'appeared')", (now, code, name, price), ) cursor.execute( "INSERT OR REPLACE INTO ghosts_current (product_code, product_name, price, first_ghosted_at) VALUES (?, ?, ?, ?)", (code, name, price, now), ) # Only codes still listed by the Search API can be declared resolved — # a ghost that vanished from Lenovo entirely is 'resolved' too (it's no # longer a listed-but-uncartable trap either way). for code in previous - unavailable_codes: cursor.execute("SELECT product_name, price FROM ghosts_current WHERE product_code = ?", (code,)) row = cursor.fetchone() cursor.execute( "INSERT INTO ghost_events (timestamp, product_code, product_name, price, event_type) VALUES (?, ?, ?, ?, 'resolved')", (now, code, row[0] if row else None, row[1] if row else None), ) cursor.execute("DELETE FROM ghosts_current WHERE product_code = ?", (code,)) def get_scrape_stats() -> Optional[Dict[str, Any]]: """Last scrape cycle's listed-vs-purchasable snapshot, or None before the first cycle (or on a pre-migration-12 DB). Backs GET /api/stats.""" with db_connection() as conn: cursor = conn.cursor() try: cursor.execute( "SELECT listed_count, purchasable_count, ghost_count, updated_at FROM scrape_stats WHERE id = 1" ) except Exception: return None row = cursor.fetchone() if not row: return None return { "listedCount": row[0], "purchasableCount": row[1], "ghostCount": row[2], "updatedAt": row[3], }