"""PSREF (psref.lenovo.com) spec-sheet enrichment. Fills gaps our own outlet scrape never reports (chassis/security/ports/certs) by resolving each catalog product's Machine Type (first 4 chars of `product_code`) against Lenovo's undocumented PSREF JSON API and caching the flattened spec sheet, keyed by MT (a spec sheet covers a whole product line, not one SKU). Field-by-field research (which PSREF fields are flat-safe, which are genuinely ambiguous, the two deterministic joins) lives in the `project_psref_feasibility` memory — this module builds against that directly. Response shape below was confirmed against the real API this session (not guessed from the memory's prose description), e.g.: Suggest: {"data": [{"ProductKey": "...", "MachineType": "21JR", "info": {"page": "https://psref.lenovo.com/l/..."}}]} LoadSpecData: {"data": {"GeneralSpecData": [{"GeneralSpecJson": ""}]}} GeneralSpecJson parses to {"data": {"SpecData": [ {"L1": "Performance", "L2": [{"L2": "Chipset", "Features": [ {"FName": "Chipset", "Optional": "0", "FVs": [ {"FVGs": [{"FVGItem": [{"Att": "DefaultDescription", "AttV": "AMD SoC..."}]}]} ]} ]}]} ]}} ToS: Lenovo's Terms of Use bans "high volume" automated harvesting. This is 2 calls per unique MT, once, cached forever (~120 calls total as of this writing) — flagged to the user before this runs against the live site for real, same as the memory's own caveat. """ import html import json import re import time import urllib.parse from typing import Any, Dict, List, Optional, Tuple from backend.config import settings from backend.db.connection import db_connection from backend.logging_config import _log, current_time from backend.scraper.api import _request # Polite delay between PSREF calls — this is a one-off backfill, not a hot path. PSREF_REQUEST_DELAY = 0.4 SpecField = Dict[str, Any] # {"l1", "l2", "num_fvs", "optional", "values": [attrs, ...]} # --------------------------------------------------------------------------- # Text helpers # --------------------------------------------------------------------------- def _clean_text(s: Optional[str]) -> str: """Some PSREF description fields (e.g. Docking) embed a live tag pointing at their own accessories page — converted to `[[text|url]]` (not markdown's `[text](url)`: some of these real URLs contain literal parentheses, e.g. the BIOS Simulator link's "...(21Q6,21Q7)" segment, which breaks naive markdown-link parsing) so the frontend (lib/specValue.tsx) can render a real clickable link, instead of dead "please visit here" text with the href silently discarded. Any other/nested tags (a `` almost always wraps the anchor's own text) are stripped as plain formatting noise.""" if not s: return "" s = html.unescape(s) s = re.sub(r']*href="([^"]*)"[^>]*>(.*?)', r"[[\2|\1]]", s, flags=re.I | re.S) s = re.sub(r"<[^>]+>", "", s) s = re.sub(r"\s+", " ", s).strip() return s def _fv_text(attrs: Dict[str, str]) -> str: """Render one FV group's attrs as display text. Single-attribute features use the generic "DefaultDescription" key; composite ones (Weight, Dimensions) carry a "Models" qualifier alongside the real value — skip it here, it's metadata, not part of the visible value.""" if "DefaultDescription" in attrs: return attrs["DefaultDescription"] parts = [v for k, v in attrs.items() if k != "Models"] return ", ".join(parts) # --------------------------------------------------------------------------- # LoadSpecData parsing # --------------------------------------------------------------------------- def _flatten_spec_data(spec_data: List[dict]) -> Dict[str, SpecField]: """Flatten PSREF's L1 -> L2 -> Feature -> FVs -> FVGs -> FVGItem nesting into {FName: {l1, l2, num_fvs, optional, values: [attrs_dict, ...]}}. Keyed by the Feature name (FName) directly, not nested under category — resolve_specs_for_laptop looks fields up by name; which display category each ends up in is decided there, independent of PSREF's own L1/L2 taxonomy.""" flat: Dict[str, SpecField] = {} for l1 in spec_data: l1_name = l1.get("L1", "") for l2 in l1.get("L2", []): l2_name = l2.get("L2", "") for feature in l2.get("Features", []): fname = feature.get("FName") if not fname: continue values: List[Dict[str, str]] = [] for fv in feature.get("FVs", []): for fvg in fv.get("FVGs", []): attrs: Dict[str, str] = {} for item in fvg.get("FVGItem", []): att = item.get("Att") attv = _clean_text(item.get("AttV")) if att and attv: attrs[att] = attv if attrs: values.append(attrs) if not values: continue flat[fname] = { "l1": l1_name, "l2": l2_name, "num_fvs": len(values), "optional": feature.get("Optional") == "1", "values": values, } return flat def resolve_machine_type(mt: str) -> Optional[Dict[str, str]]: """Resolve a Machine Type to a PSREF `ProductKey` + spec-sheet page URL via the `Suggest` search endpoint. 100% hit rate across all 119 MTs ever seen in the catalog (per the feasibility sweep) — a miss here means a genuinely new MT PSREF hasn't indexed yet, not a bug.""" url = ( f"{settings.PSREF_BASE_URL}/api/search/DefinitionFilterAndSearch/Suggest" f"?kw={urllib.parse.quote(mt)}&limit=6&IsPreviewProduct=true&SearchType=Normal" ) try: response = _request("GET", url, headers=settings.PSREF_HEADERS, timeout=15) if response.status_code != 200: _log(f"[PSREF] Suggest HTTP {response.status_code} for MT {mt}") return None results = response.json().get("data") or [] exact = next((r for r in results if r.get("MachineType") == mt), None) item = exact or (results[0] if results else None) if not item or not item.get("ProductKey"): return None return { "product_key": item["ProductKey"], "page_url": (item.get("info") or {}).get("page"), } except Exception as e: _log(f"[PSREF] Suggest error for MT {mt}: {e}") return None def fetch_load_spec_data(product_key: str, mt: str) -> Optional[Dict[str, SpecField]]: """Fetch and flatten the canonical per-product-family spec sheet (the same data that powers PSREF's own Spec tab and Compare feature).""" url = f"{settings.PSREF_BASE_URL}/api/product/Compare/LoadSpecData?ProductKey={urllib.parse.quote(product_key)}" try: response = _request("GET", url, headers=settings.PSREF_HEADERS, timeout=15) if response.status_code != 200: _log(f"[PSREF] LoadSpecData HTTP {response.status_code} for MT {mt}") return None general = ((response.json().get("data") or {}).get("GeneralSpecData")) or [] if not general: _log(f"[PSREF] LoadSpecData: no GeneralSpecData for MT {mt} ({product_key})") return None raw_json = general[0].get("GeneralSpecJson") if not raw_json: return None spec_list = ((json.loads(raw_json).get("data") or {}).get("SpecData")) or [] flat = _flatten_spec_data(spec_list) if not flat: _log(f"[PSREF] LoadSpecData: empty SpecData for MT {mt} ({product_key})") return None return flat except Exception as e: _log(f"[PSREF] LoadSpecData error for MT {mt}: {e}") return None # --------------------------------------------------------------------------- # Backfill # --------------------------------------------------------------------------- def backfill_psref_specs(machine_types: Optional[List[str]] = None) -> Dict[str, int]: """One-off backfill, same shape as enrichment.py's enrich_specs(): resolve + fetch + cache per Machine Type not already in psref_specs (or an explicit list). Spec sheets don't change post-announce, so this never re-fetches an MT that's already cached — call with explicit `machine_types` to force a refresh of specific ones.""" with db_connection() as conn: cursor = conn.cursor() if machine_types: targets = list(dict.fromkeys(machine_types)) else: cursor.execute(""" SELECT DISTINCT substr(product_code, 1, 4) as mt FROM products WHERE substr(product_code, 1, 4) NOT IN (SELECT machine_type FROM psref_specs) """) targets = [row[0] for row in cursor.fetchall() if row[0]] if not targets: _log("[PSREF] No machine types need enrichment.") return {"enriched": 0, "failed": 0} _log(f"[PSREF] Backfilling {len(targets)} machine type(s)...") enriched = 0 failed = 0 max_failed = getattr(settings, "PSREF_MAX_FAILED", 3) for i, mt in enumerate(targets): if failed > max_failed: remaining = len(targets) - i _log( f"[PSREF] {failed} failures exceeded PSREF_MAX_FAILED ({max_failed}) — " f"PSREF may be down. Aborting this run, {remaining} machine type(s) left " "uncached (picked up again next backfill)." ) break try: resolved = resolve_machine_type(mt) time.sleep(PSREF_REQUEST_DELAY) if not resolved: failed += 1 _log(f"[PSREF] Could not resolve MT {mt} via Suggest.") continue spec_data = fetch_load_spec_data(resolved["product_key"], mt) time.sleep(PSREF_REQUEST_DELAY) if spec_data is None: failed += 1 continue cursor.execute(""" INSERT OR REPLACE INTO psref_specs (machine_type, product_key, page_url, spec_data, updated_at) VALUES (?, ?, ?, ?, ?) """, ( mt, resolved["product_key"], resolved["page_url"], json.dumps(spec_data, ensure_ascii=False), current_time(), )) conn.commit() enriched += 1 _log(f"[PSREF] Cached spec data for MT {mt} ({len(spec_data)} fields).") except Exception as e: failed += 1 _log(f"[PSREF] Error enriching MT {mt}: {e}") _log(f"[PSREF] Backfill done. Enriched: {enriched} | Failed: {failed}") return {"enriched": enriched, "failed": failed} # --------------------------------------------------------------------------- # Read-time resolution — outlet-vs-PSREF precedence, joins, wording rules. # All from project_psref_feasibility memory's "How to apply" section — build # against that directly, don't re-derive the field classification here. # --------------------------------------------------------------------------- SpecRow = Dict[str, str] # {"label": ..., "value": ..., "field"?: ...} — "field" only set by _emit_enum SpecGroup = Dict[str, Any] # {"category": ..., "rows": [SpecRow, ...]} # Display order for specs_extra/reference_specs groups. Processor/Memory/Storage # (split out of the former combined "Performance" category so each PSREF # extra/join lands next to the outlet field it augments)/Graphics/Display/ # Connectivity/Physical are existing Product-Details categories PSREF rows merge # into; Platform/Ports/Security/Durability & Certifications are new ones. _CATEGORY_ORDER = [ "Processor", "Memory", "Storage", "Graphics", "Display", "Connectivity", "Physical", "Platform", "Ports", "Security", "Durability & Certifications", ] # Flat/all-variant singular fields PSREF is the sole source for (outlet never # states these) — routed through the universal num_fvs==1 -> extra / >1 -> # reference rule, no per-field hardcoding of which is which. _FLAT_FIELDS: List[Tuple[str, str]] = [ ("Chipset", "Platform"), ("Audio Chip", "Platform"), ("Speakers", "Platform"), ("Microphone", "Platform"), ("Card Reader", "Platform"), ("Docking", "Platform"), ("Max Memory", "Memory"), ("Max Storage Support", "Storage"), ("Storage Slot", "Storage"), ("Security Chip", "Security"), ("Physical Locks", "Security"), ("ThinkShield", "Security"), ("Screen-to-Body Ratio", "Display"), ("Monitor Support", "Platform"), ("Case Material", "Physical"), ("Dimensions (WxDxH)", "Physical"), ("Mil-Spec Test", "Durability & Certifications"), ] # Enumeration fields — a laptop has several at once. Joined into one row per # field (dynamic loop over whatever PSREF returns, never a hardcoded expected # list) rather than exploded into many same-label rows. _ENUM_FIELDS: List[Tuple[str, str]] = [ ("Standard Ports", "Ports"), ("Optional Ports", "Ports"), ("BIOS Security", "Security"), ("Other Security", "Security"), ("Green Certifications", "Durability & Certifications"), ("Other Certifications", "Durability & Certifications"), ] # Fields outlet already states (even conditionally) or that memory's sweep found # majority-ambiguous and not worth pulling — never surfaced from PSREF, not even # as a fallback in the unmapped-field catch-all. "Processor"/"Display" feed the # two deterministic joins below and are excluded from standalone surfacing for # that reason, not because they're ambiguous. _EXCLUDE_FNAMES = { "Processor Family", "Processor", "Operating System", "Graphics", "Camera", "Battery", "Battery Life", "Power Adapter", "Display", "Touchscreen", "Keyboard", "Keyboard Backlight", "Weight", "Case Color", "Fingerprint Reader", "WLAN + Bluetooth", "SIM Card", "System Management", "Base Warranty", "Memory Type", "Storage Type", "Memory Slots", # handled specially, see _emit_memory_slots "Smart Card Reader", # handled specially, see _emit_smart_card_reader "WWAN", # handled specially, see _emit_wwan "Bundled Accessories", # dropped: PSREF describes retail-box contents; outlet units ship with no bundled accessories, so this field is just wrong here } # Fallback bucket for PSREF fields not named above (item 9: "any unmapped field # still surfaces, never silently dropped") — keyed by PSREF's own L2 subcategory. _L2_FALLBACK_CATEGORY = { "Memory": "Memory", "Storage": "Storage", "Removable Storage": "Platform", "Multi-Media": "Platform", "Input Device": "Physical", "Mechanical": "Physical", "Network": "Connectivity", "Ports": "Ports", "Docking": "Platform", "Security": "Security", "Operating Environment": "Platform", "Sustainability": "Durability & Certifications", "Green Certifications": "Durability & Certifications", "Other Certifications": "Durability & Certifications", "Display": "Display", } def _emit_singular( field: SpecField, category: str, label: str, extra: Dict[str, List[SpecRow]], reference: Dict[str, List[SpecRow]], ) -> None: """Universal rule for every singular field: check the live num_fvs for this specific product. ==1 -> confirmed, flat row. >1 -> all documented variants, joined (never a bare em dash — see feedback_no_em_dashes memory).""" texts = [t for t in (_fv_text(v) for v in field["values"]) if t] if not texts: return if len(texts) == 1: extra.setdefault(category, []).append({"label": label, "value": texts[0]}) else: uniq = list(dict.fromkeys(texts)) reference.setdefault(category, []).append({"label": label, "value": " or ".join(uniq)}) # Per-item label extraction for enumeration fields — one row per item (a laptop # has several ports/certs/security-features at once; joining them into one # comma-separated row loses the "separated" granularity Product Details/Compare # use everywhere else). Ports and certifications have a clean extractable # type/brand name; fields without one (BIOS Security, Other Security, Bundled # Accessories) fall back to the field name repeated per item — still correct # since rendering keys off (label, index), never label alone. _PORT_LABEL_PATTERNS: List[Tuple[re.Pattern, str]] = [ (re.compile(r"thunderbolt", re.I), "Thunderbolt / USB-C"), (re.compile(r"usb-?c|usb4", re.I), "USB-C"), (re.compile(r"usb-?a\b", re.I), "USB-A"), (re.compile(r"hdmi", re.I), "HDMI"), (re.compile(r"displayport|\bdp\b", re.I), "DisplayPort"), (re.compile(r"ethernet|rj-?45", re.I), "Ethernet"), (re.compile(r"headphone|microphone|audio jack|combo jack", re.I), "Audio Jack"), (re.compile(r"smart card", re.I), "Smart Card Reader"), (re.compile(r"card reader|sd card", re.I), "Card Reader"), (re.compile(r"nano-?sim|sim card", re.I), "SIM Slot"), (re.compile(r"\bvga\b", re.I), "VGA"), ] _CERT_LABEL_PATTERNS: List[Tuple[re.Pattern, str]] = [ (re.compile(r"energy star", re.I), "ENERGY STAR"), (re.compile(r"epeat", re.I), "EPEAT"), (re.compile(r"\berp\b", re.I), "ErP"), (re.compile(r"rohs", re.I), "RoHS"), (re.compile(r"\btco\b", re.I), "TCO Certified"), (re.compile(r"eyesafe", re.I), "Eyesafe"), (re.compile(r"t[üu]v", re.I), "TÜV Rheinland"), (re.compile(r"mil-std", re.I), "MIL-STD"), ] def _enum_item_label(field_name: str, text: str) -> str: patterns = _PORT_LABEL_PATTERNS if "port" in field_name.lower() else _CERT_LABEL_PATTERNS for pattern, label in patterns: if pattern.search(text): return label return field_name def _emit_enum(field: SpecField, category: str, label: str, extra: Dict[str, List[SpecRow]]) -> None: """One row per item (Product Details/Compare's "separated" convention), each tagged with the source PSREF field name via "field" — Product Details ignores it, but Compare's fixed-row-per-laptop table can't show a variable-length per-item list per column, so it groups back by "field" into one compact joined row instead (see lib/compare.ts's psrefFieldJoined).""" texts = [t for t in (_fv_text(v) for v in field["values"]) if t] for text in texts: extra.setdefault(category, []).append( {"label": _enum_item_label(label, text), "value": text, "field": label} ) def _emit_memory_slots( laptop: Dict[str, Any], spec_data: Dict[str, SpecField], extra: Dict[str, List[SpecRow]], reference: Dict[str, List[SpecRow]], ) -> None: """Always emits a row. Never guesses stick count/size. If outlet's Memory string already states a soldered amount, pair it with PSREF's slot-count + DDR generation (not outlet's own socketed-capacity figure, which would falsely imply the exact populated stick size).""" mem = laptop.get("memory") or "" field = spec_data.get("Memory Slots") soldered = re.search(r"(\d+\s*GB)\s*Soldered", mem, re.I) if soldered and field: psref_text = " ".join(_fv_text(v) for v in field["values"]) slot = re.search(r"\b(one|two|three|four|\d+)\s+(DDR\d)\s+SO-?DIMM\s+slot", psref_text, re.I) if slot: value = f"{soldered.group(1)} Soldered + {slot.group(1)} {slot.group(2)} SO-DIMM slot" else: value = mem.strip() extra.setdefault("Memory", []).append({"label": "Memory Slots", "value": value}) return if field: _emit_singular(field, "Memory", "Memory Slots", extra, reference) def _emit_smart_card_reader(spec_data: Dict[str, SpecField], extra: Dict[str, List[SpecRow]]) -> None: """Decisive Yes/No, never a hedge — a buyer can't retrofit one either way, so an occasionally-wrong "No" beats a technically-safer "may vary".""" field = spec_data.get("Smart Card Reader") if not field: return if field["num_fvs"] == 1: text = _fv_text(field["values"][0]).strip().lower() value = "No" if text.startswith("no") else "Yes" else: value = "No" extra.setdefault("Security", []).append({"label": "Smart Card Reader", "value": value}) def _emit_wwan(laptop: Dict[str, Any], spec_data: Dict[str, SpecField], extra: Dict[str, List[SpecRow]]) -> None: """Three real states. Outlet's `SIM` key present is itself a fully confirmed signal (same reliability as Fingerprint Reader's presence/absence) — its absence means "Built-in" is ruled out, not "unknown".""" full_specs = laptop.get("full_specs") or {} if full_specs.get("SIM"): value = "Built-in" else: field = spec_data.get("WWAN") texts = " ".join(_fv_text(v) for v in field["values"]).lower() if field else "" value = "Has a WWAN slot" if re.search(r"slot|antenna|upgradable", texts) else "No" extra.setdefault("Connectivity", []).append({"label": "WWAN", "value": value}) # --- Deterministic joins ---------------------------------------------------- def _normalize_cpu(s: str) -> str: s = html.unescape(s or "") s = re.sub(r"[®™©]", "", s) # PSREF appends a parenthetical codename (e.g. "Core Ultra 5 225H (ARL)") to # disambiguate 'Processor Name' entries on spec sheets that span multiple # silicon generations under one MT — outlet's own Processor string never has # this, so left unstripped it silently breaks the substring match below for # every such entry (found while testing the iGPU fallback against MT 21SS, # a Meteor/Arrow/Lunar Lake mixed sheet; same fix benefits the pre-existing # Cache/Cores-Threads/NPU-TOPS joins in _processor_join_rows, not just this). s = re.sub(r"\([^)]*\)", "", s) s = re.sub(r"\s+", " ", s).strip().lower() return s _PROCESSOR_JOIN_SKIP_ATTRS = { "Processor Name", "Cores", "Threads", "Base Frequency", "Max Frequency", "Cache", "Processor Graphics", } def _match_processor(spec_data: Dict[str, SpecField], outlet_processor: Optional[str]) -> Optional[Dict[str, str]]: """Match outlet's known Processor string against PSREF's per-SKU Processor Name to unlock Cache/Cores-Threads/NPU/iGPU-TOPS — none of which outlet reports. Deterministic: the CPU SKU token is exact, not fuzzy.""" field = spec_data.get("Processor") if not field or not outlet_processor: return None outlet_norm = _normalize_cpu(outlet_processor) for attrs in field["values"]: name = attrs.get("Processor Name") if name and _normalize_cpu(name) in outlet_norm: return attrs return None # PSREF's per-CPU 'Processor Graphics' attr is only ever the chip's *own* # integrated GPU — it never tells us whether the specific laptop *also* has a # discrete GPU bolted on, and CPU SKU alone doesn't reliably predict that. Real # catalog data proves it both ways for the same exact SKU: AMD Ryzen 5 7535HS # ships integrated-only on several ThinkPad/ThinkBook MTs but with an RTX 3050 # on the IdeaPad Gaming 3 (82SBR000R0); Ryzen 9 7945HX (82WM, Legion Pro 5) # ships an RTX 4070 despite PSREF listing its Processor Graphics as plain # 'AMD Radeon 610M'; Core Ultra 7 155H ships an RTX 1000 Ada on the ThinkPad # P16v (21KX) mobile workstation despite PSREF listing 'Intel Arc Graphics' as # its Processor Graphics. Even U/V (15-28W ultraportable) tiers aren't a # guaranteed exception — a config pairing one with a discrete GPU may simply # not exist yet in this catalog. That's why this function is only ever used to # populate the *integrated*-GPU field (laptop["igpu-name"]) — a fact that's # unconditionally true regardless of what else the laptop has — never to claim # a laptop has no discrete GPU. Discrete-GPU presence is sourced exclusively # from outlet's own 'Graphic Card' field (see get_laptops_data in # enrichment.py); if outlet doesn't report one, dgpu-name simply stays # unknown rather than this function guessing "Integrated" and being wrong. def infer_igpu_from_psref(psref_spec_data_json: Optional[str], outlet_processor: Optional[str]) -> Optional[str]: """Resolve a laptop's integrated GPU name from PSREF's per-SKU Processor Graphics attr (the same join _match_processor already does for Cache/ Cores-Threads/NPU-TOPS), so it works for any CPU SKU PSREF has indexed — not a fixed table. Confirmed against a live call for MT 21JK (ThinkPad E14 Gen 5 Intel): Core i3-1315U -> 'Intel® UHD Graphics', Core i5-1335U/ i7-1355U/i7-1365U -> 'Intel® Iris® Xe Graphics' — exact match to what outlet itself reports for every sibling SKU in the same catalog, so this is read as ground truth, not guessed. See the module-level comment above for why this is safe to call for *any* CPU tier despite that not being true of the single-field dGPU-vs-iGPU question.""" if not psref_spec_data_json or not outlet_processor: return None try: spec_data: Dict[str, SpecField] = json.loads(psref_spec_data_json) except (TypeError, ValueError): return None attrs = _match_processor(spec_data, outlet_processor) if not attrs: return None gfx = attrs.get("Processor Graphics") return f"Integrated {gfx}" if gfx else None def _processor_join_rows(attrs: Dict[str, str]) -> List[SpecRow]: rows: List[SpecRow] = [] if attrs.get("Cache"): rows.append({"label": "Cache", "value": attrs["Cache"]}) cores, threads = attrs.get("Cores"), attrs.get("Threads") if cores and threads: rows.append({"label": "Cores / Threads", "value": f"{cores} / {threads}"}) for key, val in attrs.items(): if key in _PROCESSOR_JOIN_SKIP_ATTRS: continue if re.search(r"npu|tops", key, re.I): rows.append({"label": key, "value": val}) return rows def _parse_inches(s: str) -> Optional[float]: m = re.search(r"(\d+(?:\.\d+)?)", s or "") return float(m.group(1)) if m else None def _extract_resolution(s: str) -> Optional[str]: m = re.search(r"(\d{3,4}\s*x\s*\d{3,4})", s or "", re.I) return re.sub(r"\s+", "", m.group(1)).lower() if m else None def _match_display(spec_data: Dict[str, SpecField], laptop: Dict[str, Any]) -> Optional[Dict[str, str]]: """Match on Size+Resolution+Touch+(Type when known) — the combination we already know live, high-confidence against false matches within one MT's product line — to unlock Contrast Ratio/Viewing Angle/panel Certifications.""" field = spec_data.get("Display") if not field: return None target_size = laptop.get("screen-size") target_res = _extract_resolution(laptop.get("resolution") or "") if target_size is None or not target_res: return None target_touch = bool(laptop.get("touch-screen")) target_type = (laptop.get("panel-type") or "").strip().lower() or None for attrs in field["values"]: size = _parse_inches(attrs.get("Size", "")) res = _extract_resolution(attrs.get("Resolution", "")) if size is None or res is None or abs(size - target_size) > 0.05 or res != target_res: continue touch_text = (attrs.get("Touch") or "").lower() touch = bool(touch_text) and "non-touch" not in touch_text if touch != target_touch: continue psref_type = (attrs.get("Type") or "").strip().lower() if target_type and psref_type and psref_type != target_type: continue return attrs return None def _display_join_rows(attrs: Dict[str, str]) -> List[SpecRow]: rows: List[SpecRow] = [] if attrs.get("Contrast Ratio"): rows.append({"label": "Contrast Ratio", "value": attrs["Contrast Ratio"]}) viewing_angle = attrs.get("Viewing Angle (L/R/U/D)") if viewing_angle: rows.append({"label": "Viewing Angle", "value": viewing_angle}) key_features = attrs.get("Key Features") if key_features: rows.append({"label": "Certified", "value": key_features}) return rows def _finalize(groups: Dict[str, List[SpecRow]]) -> List[SpecGroup]: ordered = [c for c in _CATEGORY_ORDER if c in groups] ordered += sorted(c for c in groups if c not in _CATEGORY_ORDER) return [{"category": c, "rows": groups[c]} for c in ordered if groups[c]] def resolve_specs_for_laptop( laptop: Dict[str, Any], psref_row: Optional[Dict[str, Any]] ) -> Tuple[List[SpecGroup], List[SpecGroup], Optional[str]]: """Run at read time (not baked into the backfill), so it always reflects live outlet fields. Returns (specs_extra, reference_specs, reference_specs_url).""" if not psref_row or not psref_row.get("spec_data"): return [], [], None try: spec_data: Dict[str, SpecField] = json.loads(psref_row["spec_data"]) except (TypeError, ValueError): return [], [], None extra: Dict[str, List[SpecRow]] = {} reference: Dict[str, List[SpecRow]] = {} handled = set(_EXCLUDE_FNAMES) for fname, category in _FLAT_FIELDS: field = spec_data.get(fname) handled.add(fname) if field: _emit_singular(field, category, fname, extra, reference) for fname, category in _ENUM_FIELDS: field = spec_data.get(fname) handled.add(fname) if field: _emit_enum(field, category, fname, extra) _emit_memory_slots(laptop, spec_data, extra, reference) _emit_smart_card_reader(spec_data, extra) _emit_wwan(laptop, spec_data, extra) processor_match = _match_processor(spec_data, laptop.get("processor")) if processor_match: for row in _processor_join_rows(processor_match): extra.setdefault("Processor", []).append(row) display_match = _match_display(spec_data, laptop) if display_match: for row in _display_join_rows(display_match): extra.setdefault("Display", []).append(row) # Item 9: any unmapped PSREF field still surfaces rather than being silently # dropped, bucketed by its own L2 subcategory (defaulting to Platform). for fname, field in spec_data.items(): if fname in handled: continue category = _L2_FALLBACK_CATEGORY.get(field.get("l2"), "Platform") _emit_singular(field, category, fname, extra, reference) return _finalize(extra), _finalize(reference), psref_row.get("page_url")