Add AMD CPU support (CSV import + TechPowerUp scraper), fix alternatives accordion, untrack database.db
This commit is contained in:
@@ -637,6 +637,359 @@ def lookup_apple(cpu_key: str) -> dict:
|
||||
return {"lookup_status": "failed"}
|
||||
|
||||
|
||||
# ── AMD (TechPowerUp) ─────────────────────────────────────────────────────────
|
||||
|
||||
TPU_BASE = "https://www.techpowerup.com"
|
||||
|
||||
|
||||
def _amd_search_term(cpu_raw: str) -> str:
|
||||
"""Strip AMD prefix and parenthetical junk for TechPowerUp Google search."""
|
||||
s = cpu_raw.strip()
|
||||
s = re.sub(r'\s*\([^)]*\)', '', s).strip()
|
||||
s = re.sub(r'[-\s]*(processor|cpu|chip)$', '', s, flags=re.I).strip()
|
||||
return s
|
||||
|
||||
|
||||
def _tpu_find_url(html: str) -> Optional[str]:
|
||||
"""Extract the first techpowerup.com/cpu-specs/ URL from a search results page."""
|
||||
soup = BeautifulSoup(html, "lxml")
|
||||
|
||||
def _clean(href: str) -> str:
|
||||
m = re.search(r'/url\?q=(https://www\.techpowerup\.com[^&]+)', href)
|
||||
if m:
|
||||
href = m.group(1)
|
||||
return href
|
||||
|
||||
for a in soup.find_all("a", href=True):
|
||||
href = _clean(a["href"])
|
||||
if re.search(r'techpowerup\.com/cpu-specs/', href, re.I):
|
||||
# Strip any trailing Google redirect junk
|
||||
href = href.split("&")[0]
|
||||
return href
|
||||
return None
|
||||
|
||||
|
||||
def _google_find_tpu_url(term: str) -> Optional[str]:
|
||||
"""Search Google for the TechPowerUp CPU spec URL via Playwright."""
|
||||
clean_term = re.sub(r'\s*\([^)]*\)', '', term).strip()
|
||||
query = url_quote(f"{clean_term} site:techpowerup.com/cpu-specs")
|
||||
search_url = GOOGLE_SEARCH.format(query=query)
|
||||
log.info(f" → Google (TPU): {search_url}")
|
||||
|
||||
page = None
|
||||
html = None
|
||||
try:
|
||||
ctx = _get_pw_context()
|
||||
page = ctx.new_page()
|
||||
response = page.goto(search_url, wait_until="networkidle", timeout=30_000)
|
||||
status = response.status if response else 200
|
||||
|
||||
if status == 429 or _is_captcha_page(page):
|
||||
log.warning("=" * 60)
|
||||
log.warning(" ⚠ Google CAPTCHA / rate-limit detected!")
|
||||
log.warning(" The browser window is open — please solve the CAPTCHA manually.")
|
||||
log.warning(" Waiting up to 5 minutes for you to reach the search results…")
|
||||
log.warning("=" * 60)
|
||||
try:
|
||||
page.wait_for_selector("#search a, #rso a", timeout=300_000)
|
||||
log.info(" ✓ CAPTCHA solved — resuming.")
|
||||
except Exception:
|
||||
log.error(" Timed out waiting for CAPTCHA. Giving up.")
|
||||
return None
|
||||
elif status not in (200, 301, 302):
|
||||
log.warning(f" Google returned HTTP {status}")
|
||||
return None
|
||||
else:
|
||||
try:
|
||||
page.wait_for_selector("#search a, #rso a", timeout=10_000)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
html = page.content()
|
||||
|
||||
except Exception as exc:
|
||||
log.warning(f" Google (TPU) search error for '{term}': {exc}")
|
||||
return None
|
||||
finally:
|
||||
if page:
|
||||
try:
|
||||
page.close()
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
return _tpu_find_url(html)
|
||||
|
||||
|
||||
def _tpu_parse_specs(html: str, url: str) -> dict:
|
||||
"""Parse a TechPowerUp CPU spec page into our column dict."""
|
||||
data: dict = {"source_url": url}
|
||||
soup = BeautifulSoup(html, "lxml")
|
||||
|
||||
# CPU name — try the dedicated heading first
|
||||
for sel in ("h1.cpuname", "h1.cpu-name", "h1", ".cpu-name"):
|
||||
el = soup.select_one(sel)
|
||||
if el:
|
||||
data["full_name"] = el.get_text(strip=True)
|
||||
break
|
||||
|
||||
specs: dict[str, str] = {}
|
||||
|
||||
# Strategy 1: <dl> definition lists
|
||||
for dl in soup.select("dl"):
|
||||
for dt, dd in zip(dl.select("dt"), dl.select("dd")):
|
||||
k = dt.get_text(strip=True).rstrip(":")
|
||||
v = dd.get_text(" ", strip=True)
|
||||
if k and v and len(k) < 80:
|
||||
specs[k] = v
|
||||
|
||||
# Strategy 2: two-column <table> rows
|
||||
if len(specs) < 5:
|
||||
for table in soup.select("table"):
|
||||
for tr in table.select("tr"):
|
||||
cells = tr.select("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:
|
||||
specs.setdefault(k, v)
|
||||
|
||||
# Strategy 3: labeled divs/spans (TPU uses a section-based layout)
|
||||
if len(specs) < 5:
|
||||
for section in soup.select(".details, .specs, [class*='specification'], .cpu-details"):
|
||||
for row in section.select("li, div[class*='row']"):
|
||||
parts = row.find_all(["span", "div"], limit=2)
|
||||
if len(parts) >= 2:
|
||||
k = parts[0].get_text(strip=True).rstrip(":")
|
||||
v = parts[1].get_text(" ", strip=True)
|
||||
if k and v and len(k) < 80:
|
||||
specs.setdefault(k, v)
|
||||
|
||||
if specs:
|
||||
_tpu_map(specs, data)
|
||||
|
||||
return data
|
||||
|
||||
|
||||
def _tpu_map(specs: dict, data: dict) -> None:
|
||||
"""Map TechPowerUp spec key→value pairs to our cpu_specs columns."""
|
||||
s = lambda pattern: _spec_lookup(specs, pattern)
|
||||
|
||||
cn = s(r'codename|code.?name|die.?name')
|
||||
if cn:
|
||||
data["codename"] = cn
|
||||
lit = s(r'process\s*node|manufactur|process\s*tech|lithograph')
|
||||
if lit:
|
||||
data["lithography"] = lit
|
||||
ld = s(r'release\s*date|launch|announced')
|
||||
if ld:
|
||||
data["launch_date"] = ld
|
||||
|
||||
tc = s(r'^(?:#\s*)?cores?$|total\s*cores?|cpu\s*cores?')
|
||||
if tc:
|
||||
data["total_cores"] = _parse_int(tc)
|
||||
tt = s(r'^(?:#\s*)?threads?$|total\s*threads?')
|
||||
if tt:
|
||||
data["total_threads"] = _parse_int(tt)
|
||||
|
||||
base = s(r'base\s*(?:clock|freq(?:uency)?)|core\s*speed')
|
||||
if base:
|
||||
data["base_freq_ghz"] = _parse_freq_ghz(base)
|
||||
boost = s(r'boost\s*(?:clock|freq)|turbo\s*(?:clock|freq)|max\.?\s*boost')
|
||||
if boost:
|
||||
data["max_turbo_ghz"] = _parse_freq_ghz(boost)
|
||||
|
||||
l3 = s(r'l3\s*cache')
|
||||
if l3:
|
||||
data["l3_cache"] = l3
|
||||
l2 = s(r'l2\s*cache')
|
||||
if l2:
|
||||
data["l2_cache"] = l2
|
||||
|
||||
tdp = s(r'^tdp$|thermal\s*design\s*power|default\s*tdp')
|
||||
if tdp:
|
||||
m = re.search(r'(\d+)', tdp)
|
||||
if m:
|
||||
data["tdp_w"] = int(m.group(1))
|
||||
max_tdp = s(r'configurable\s*tdp|ctdp|max\.?\s*tdp|tdp.?up')
|
||||
if max_tdp:
|
||||
nums = re.findall(r'(\d+)', max_tdp)
|
||||
if nums:
|
||||
data["max_tdp_w"] = max(int(n) for n in nums)
|
||||
|
||||
pcie_ver = s(r'pci\s*-?e(?:xpress)?\s*(?:version|rev(?:ision)?|gen)?$|pcie\s*(?:ver|rev)')
|
||||
if pcie_ver:
|
||||
m = re.search(r'(\d+(?:\.\d+)?)', pcie_ver)
|
||||
if m:
|
||||
data["pcie_version"] = m.group(1)
|
||||
pcie_l = s(r'pci\s*-?e\s*lanes?|pcie\s*lanes?')
|
||||
if pcie_l:
|
||||
m = re.search(r'(\d+)', pcie_l)
|
||||
if m:
|
||||
data["pcie_lanes"] = int(m.group(1))
|
||||
|
||||
max_mem = s(r'max\.?\s*(?:memory|ram)|memory\s*size')
|
||||
if max_mem:
|
||||
m = re.search(r'(\d+)\s*GB', max_mem, re.I)
|
||||
if m:
|
||||
data["max_ram_gb"] = int(m.group(1))
|
||||
ram_t = s(r'memory\s*type|supported\s*memory|ram\s*type')
|
||||
if ram_t:
|
||||
data["ram_types"] = ram_t
|
||||
ram_spd = s(r'memory\s*(?:speed|freq)|ram\s*speed')
|
||||
if ram_spd:
|
||||
data["ram_speeds_mhz"] = ram_spd
|
||||
|
||||
igpu = s(r'gpu\s*(?:model|name)|integrated\s*graphics?|graphics?\s*model')
|
||||
if igpu:
|
||||
data["igpu_name"] = igpu
|
||||
igpu_base = s(r'gpu\s*base|graphics?\s*base\s*(?:clock|freq)')
|
||||
if igpu_base:
|
||||
m = re.search(r'(\d+)\s*MHz', igpu_base, re.I)
|
||||
if m:
|
||||
data["igpu_base_mhz"] = int(m.group(1))
|
||||
else:
|
||||
gv = re.search(r'(\d+[.,]\d*)\s*GHz', igpu_base, re.I)
|
||||
if gv:
|
||||
data["igpu_base_mhz"] = int(float(gv.group(1).replace(',', '.')) * 1000)
|
||||
igpu_boost = s(r'gpu\s*boost|graphics?\s*(?:boost|max|dynamic)\s*(?:clock|freq)')
|
||||
if igpu_boost:
|
||||
m = re.search(r'(\d+)\s*MHz', igpu_boost, re.I)
|
||||
if m:
|
||||
data["igpu_boost_mhz"] = int(m.group(1))
|
||||
else:
|
||||
gv = re.search(r'(\d+[.,]\d*)\s*GHz', igpu_boost, re.I)
|
||||
if gv:
|
||||
data["igpu_boost_mhz"] = int(float(gv.group(1).replace(',', '.')) * 1000)
|
||||
|
||||
|
||||
def lookup_amd(cpu_raw: str) -> dict:
|
||||
"""Fetch AMD CPU specs from TechPowerUp via Google → Playwright."""
|
||||
term = _amd_search_term(cpu_raw)
|
||||
log.info(f" → AMD lookup via TechPowerUp: '{term}'")
|
||||
|
||||
tpu_url = _google_find_tpu_url(term)
|
||||
_cpu_sleep()
|
||||
|
||||
if not tpu_url:
|
||||
log.warning(f" No TechPowerUp URL found for '{term}'")
|
||||
return {"lookup_status": "failed"}
|
||||
|
||||
log.info(f" → TechPowerUp: {tpu_url}")
|
||||
html = playwright_get(tpu_url)
|
||||
_cpu_sleep()
|
||||
|
||||
if not html:
|
||||
return {"lookup_status": "failed"}
|
||||
|
||||
result = _tpu_parse_specs(html, tpu_url)
|
||||
result["lookup_status"] = (
|
||||
"ok" if result.get("total_cores") or result.get("tdp_w") or result.get("full_name")
|
||||
else "failed"
|
||||
)
|
||||
return result
|
||||
|
||||
|
||||
# ── AMD CSV import ────────────────────────────────────────────────────────────
|
||||
|
||||
def import_amd_csv(conn: sqlite3.Connection, csv_path: str) -> None:
|
||||
"""Seed cpu_specs from the official AMD Processor Specifications CSV.
|
||||
|
||||
Sets lookup_status='csv' so the online TechPowerUp scraper can later
|
||||
upgrade entries to 'ok' with richer data (PCIe lanes, codename, etc.).
|
||||
Rows already marked 'ok' are never overwritten.
|
||||
"""
|
||||
import csv as _csv
|
||||
|
||||
log.info(f"═══ AMD CSV import from '{csv_path}' ═══")
|
||||
|
||||
def _parse_ghz(s: str) -> Optional[float]:
|
||||
m = re.search(r'(\d+(?:[.,]\d+)?)\s*GHz', s or '', re.I)
|
||||
return float(m.group(1).replace(',', '.')) if m else None
|
||||
|
||||
def _parse_mhz_or_ghz(s: str) -> Optional[int]:
|
||||
m = re.search(r'(\d+)\s*MHz', s or '', re.I)
|
||||
if m:
|
||||
return int(m.group(1))
|
||||
g = re.search(r'(\d+[.,]\d*)\s*GHz', s or '', re.I)
|
||||
return int(float(g.group(1).replace(',', '.')) * 1000) if g else None
|
||||
|
||||
def _parse_tdp(s: str) -> Optional[int]:
|
||||
m = re.search(r'(\d+)\s*W', s or '', re.I)
|
||||
return int(m.group(1)) if m else None
|
||||
|
||||
def _parse_ctdp_max(s: str) -> Optional[int]:
|
||||
# "15-54W" → 54
|
||||
nums = re.findall(r'(\d+)', s or '')
|
||||
return max(int(n) for n in nums) if nums else None
|
||||
|
||||
def _parse_pcie_ver(s: str) -> Optional[str]:
|
||||
m = re.search(r'(\d+(?:\.\d+)?)', s or '')
|
||||
return m.group(1) if m else None
|
||||
|
||||
def _clean_ram_types(s: str) -> str:
|
||||
# "DDR5 (FP8) , LPDDR5X (FP8)" → "DDR5, LPDDR5X"
|
||||
s = re.sub(r'\s*\([^)]*\)', '', s or '').strip()
|
||||
return re.sub(r'\s*,\s*', ', ', s)
|
||||
|
||||
rows_ok = 0
|
||||
rows_skip = 0
|
||||
|
||||
with open(csv_path, newline='', encoding='utf-8-sig') as f:
|
||||
reader = _csv.DictReader(f)
|
||||
for row in reader:
|
||||
name = (row.get("Name") or "").strip()
|
||||
if not name:
|
||||
continue
|
||||
|
||||
cpu_key = normalize_cpu_key(name)
|
||||
|
||||
existing = conn.execute(
|
||||
"SELECT lookup_status FROM cpu_specs WHERE cpu_key = ?", (cpu_key,)
|
||||
).fetchone()
|
||||
if existing and existing["lookup_status"] == "ok":
|
||||
rows_skip += 1
|
||||
continue
|
||||
|
||||
data = {
|
||||
"cpu_key": cpu_key,
|
||||
"cpu_model_raw": name,
|
||||
"vendor": "AMD",
|
||||
"full_name": name,
|
||||
"brand_name": (row.get("Series") or "").strip() or None,
|
||||
"codename": None,
|
||||
"lithography": (row.get("Processor Technology for CPU Cores") or "").strip() or None,
|
||||
"launch_date": (row.get("Launch Date") or "").strip() or None,
|
||||
"total_cores": _parse_int(row.get("# of CPU Cores") or ""),
|
||||
"performance_cores": None,
|
||||
"efficient_cores": None,
|
||||
"total_threads": _parse_int(row.get("# of Threads") or ""),
|
||||
"base_freq_ghz": _parse_ghz(row.get("Base Clock") or ""),
|
||||
"max_turbo_ghz": _parse_ghz(row.get("Max. Boost Clock") or ""),
|
||||
"l2_cache": (row.get("L2 Cache") or "").strip() or None,
|
||||
"l3_cache": (row.get("L3 Cache") or "").strip() or None,
|
||||
"tdp_w": _parse_tdp(row.get("Default TDP") or ""),
|
||||
"max_tdp_w": _parse_ctdp_max(row.get("AMD Configurable TDP (cTDP)") or ""),
|
||||
"pcie_version": _parse_pcie_ver(row.get("PCI Express® Version") or ""),
|
||||
"pcie_lanes": None,
|
||||
"max_ram_gb": None,
|
||||
"ram_types": _clean_ram_types(row.get("System Memory Type") or "") or None,
|
||||
"ram_speeds_mhz": (row.get("System Memory Specification") or "").strip() or None,
|
||||
"ecc_support": None,
|
||||
"igpu_name": (row.get("Graphics Model") or "").strip() or None,
|
||||
"igpu_base_mhz": _parse_mhz_or_ghz(row.get("Graphics Frequency") or ""),
|
||||
"igpu_boost_mhz": _parse_mhz_or_ghz(row.get("Graphics Boost Frequency") or ""),
|
||||
"source_url": "https://www.amd.com/en/products/specifications/processors",
|
||||
"lookup_status": "csv",
|
||||
"scraped_at": datetime.now().isoformat(),
|
||||
}
|
||||
|
||||
conn.execute(CPU_SPECS_INSERT, _safe_cpu_row(data))
|
||||
rows_ok += 1
|
||||
|
||||
conn.commit()
|
||||
log.info(f" Done: {rows_ok} inserted/updated, {rows_skip} skipped (already 'ok')")
|
||||
|
||||
|
||||
# ── Entry point ───────────────────────────────────────────────────────────────
|
||||
|
||||
def run(conn: sqlite3.Connection, reset: bool = False) -> None:
|
||||
@@ -700,12 +1053,24 @@ def run(conn: sqlite3.Connection, reset: bool = False) -> None:
|
||||
log.info(f"[{i}/{total}] {cpu_key} vendor={vendor}")
|
||||
log.info(f" raw: '{model_raw}'")
|
||||
|
||||
# Check for existing CSV-seeded data so we don't overwrite it on failure
|
||||
existing_status = None
|
||||
existing_row = conn.execute(
|
||||
"SELECT lookup_status FROM cpu_specs WHERE cpu_key = ?", (cpu_key,)
|
||||
).fetchone()
|
||||
if existing_row:
|
||||
existing_status = existing_row["lookup_status"]
|
||||
|
||||
try:
|
||||
if vendor == "Apple":
|
||||
result = lookup_apple(cpu_key)
|
||||
elif vendor == "AMD":
|
||||
log.info(f" Skipping AMD CPU (not supported): {cpu_key}")
|
||||
result = {"lookup_status": "skipped"}
|
||||
result = lookup_amd(model_raw)
|
||||
# If TechPowerUp failed but we already have CSV data, keep it
|
||||
if result.get("lookup_status") != "ok" and existing_status == "csv":
|
||||
log.info(f" TechPowerUp failed — keeping existing CSV data for {cpu_key}")
|
||||
done_fail += 1
|
||||
continue
|
||||
else:
|
||||
result = lookup_intel(model_raw)
|
||||
|
||||
|
||||
Reference in New Issue
Block a user