HTML 77.2%
TypeScript 10.5%
Python 9.6%
JavaScript 2.5%
1"""Manufacturer spec pages → claims on hardware entities (tier 1). One class per manufacturer domain (source provenance), module `spec_pages`.23 * apple_specs Mac Studio / Mac mini / MacBook Pro / MacBook Air / iMac tech-spec pages. The "Chip" and "Memory" rows are parsed per4 model column: chip name + "NNNGB/s memory bandwidth" (+ configurable chip variants with their own bandwidth), memory sizes5 (+ the chip variant a size requires, in parentheses). Chips get `memory_bandwidth_gbs` (the highest stated for that chip)6 and `memory_bandwidth_options_gbs`; memory *configurations* are claimed on a product entity ("Mac mini (Apple M6)",7 kind computer) that `uses` the chip — the same chip ships with different memory options per product, so chip-level8 `memory_gb` from five pages would flip-flop. Only what the page states; the iMac page (older template) is parsed from text.9 * nvidia_specs H100 / H200 (two-column spec tables SXM | NVL), DGX B200 and DGX Spark (key/value tables): memory, bandwidth, TDP,10 Tensor Core TFLOPS (NVIDIA publishes them with sparsity → `compute_*_tflops` + `compute_sparsity: true`).1112Entities are addressed with `{"registry_hardware": key}` when the key exists in registry/hardware.yaml (+ hardware.d fragments); a page13value that differs from the curated registry is left to the writer's conflict handling (never edited silently).14"""15from __future__ import annotations1617import re18from typing import Any1920from aiatlas.registry import load, org_ref21from aiatlas.sdk.connector import BaseConnector, Parsed, RunContext22from aiatlas.sdk.facts import EntityRef, Facts, Target23from aiatlas.sdk.fetch import FetchResult2425APPLE_PAGES = {"mac-studio": "Mac Studio", "mac-mini": "Mac mini", "macbook-pro": "MacBook Pro", "macbook-air": "MacBook Air", "imac": "iMac"}26CHIP = re.compile(r"\bApple\s*(M\d+(?:\s*(?:Pro|Max|Ultra))?)\s*chip", re.IGNORECASE)27CHIP_SHORT = re.compile(r"\b(M\d+(?:\s*(?:Pro|Max|Ultra))?)\b")28BANDWIDTH = re.compile(r"(\d+(?:\.\d+)?)\s*(GB|TB)/s memory bandwidth", re.IGNORECASE)29GB = re.compile(r"(\d+)\s*GB\b")30PAREN = re.compile(r"\(([^)]*)\)")313233def hardware_ref(key: str, fallback_name: str, *, manufacturer: str, kind: str, scheme: str = "spec_key") -> EntityRef:34 entry = next((h for h in load("hardware") if h["key"] == key), None)35 org = org_ref(manufacturer)36 if entry:37 return EntityRef(entity_type="hardware", name=entry["name"], identifiers={"registry_hardware": key}, aliases=list(entry.get("aliases", [])),38 slug_hint=key, organization=org)39 return EntityRef(entity_type="hardware", name=fallback_name, identifiers={scheme: key}, slug_hint=key, organization=org, attributes={"kind": kind})404142def _norm_chip(s: str) -> str:43 s = re.sub(r"\s+", " ", s.replace("\xa0", " ")).strip()44 m = CHIP_SHORT.search(s)45 return f"Apple {m.group(1)}".replace(" ", " ") if m else s464748def _bw_gbs(num: str, unit: str) -> float:49 v = float(num)50 return round(v * 1000 if unit.upper() == "TB" else v, 1)515253class _Specs(BaseConnector):54 version = "1"55 parser_version = "1"56 interval_seconds = 8640057 min_interval_seconds = 12 * 360058 max_interval_seconds = 14 * 8640059 tier = 160 priority = 261 concurrency = 1626364# ================================================================================================ Apple65class AppleSpecsConnector(_Specs):66 name = "apple_specs"67 label = "Apple — Mac tech specs (chips, memory bandwidth, memory configurations)"68 description = "apple.com/<product>/specs/ pages: Apple Silicon chip per model column, unified memory options and memory bandwidth."69 source_key = "apple.com"70 rate_per_min = 1071 expected_min_records = 57273 async def discover(self, ctx: RunContext) -> list[Target]:74 return [Target(url=f"https://www.apple.com/{slug}/specs/", doc_type="spec_page", key=f"apple:{slug}", min_bytes=20000, priority=1,75 meta={"product": product}) for slug, product in APPLE_PAGES.items()]7677 async def extract(self, ctx: RunContext, target: Target, res: FetchResult, parsed: Parsed) -> Facts:78 facts = Facts()79 html = parsed.html80 if not html:81 return facts82 product = target.meta.get("product") or (html.title or "").split(" - ")[0].strip()83 apple = org_ref("apple")84 facts.entities.append(apple)85 chips: dict[str, set[float]] = {} # chip name → bandwidths stated86 products: dict[str, dict[str, Any]] = {} # chip name → {"sizes": set, "bw": {size: bw}}87 rows = html.css(".techspecs-row")88 if rows:89 self._parse_rows(rows, chips, products)90 else:91 self._parse_text(html.text, chips, products)92 if not chips:93 ctx.log.warning("apple specs: no chip rows found", extra={"url": target.url})94 for chip, bws in chips.items():95 ref = self._chip_ref(facts, chip)96 if bws:97 facts.claim(ref, "memory_bandwidth_gbs", max(bws), unit="GB/s")98 if len(bws) > 1:99 facts.claim(ref, "memory_bandwidth_options_gbs", sorted(bws), unit="GB/s")100 # no chip-level `spec_url`: the same chip is described on several product pages (the registry keeps its curated one)101 facts.relate(apple, "manufactures", ref)102 for chip, info in products.items():103 if not info["sizes"]:104 continue105 chip_ref = self._chip_ref(facts, chip)106 key = f"{target.key.split(':', 1)[1] if target.key else product.lower()}:{_key(chip)}"107 pref = facts.entity("hardware", f"{product} ({chip})", identifiers={"apple_product_chip": key}, organization=apple, aliases=[f"{product} {chip}"])108 facts.claim(pref, "kind", "computer")109 facts.claim(pref, "product_line", product)110 facts.claim(pref, "memory_gb", sorted(info["sizes"]), unit="GB")111 facts.claim(pref, "memory_type", "unified memory")112 bws = set(chips.get(chip) or set()) | set(info["bw"].values())113 if bws:114 facts.claim(pref, "memory_bandwidth_gbs", max(bws), unit="GB/s")115 if info["bw"]:116 facts.claim(pref, "memory_configurations", [{"memory_gb": s, "memory_bandwidth_gbs": b} for s, b in sorted(info["bw"].items())])117 facts.claim(pref, "spec_url", target.url)118 facts.relate(pref, "uses", chip_ref)119 facts.relate(apple, "manufactures", pref)120 facts.document_entity = apple121 facts.document_title = html.title122 return facts123124 def _chip_ref(self, facts: Facts, chip: str) -> EntityRef:125 key = _key(chip)126 for e in facts.entities:127 if e.entity_type == "hardware" and (e.identifiers.get("registry_hardware") == key or e.identifiers.get("apple_chip") == key):128 return e129 ref = hardware_ref(key, chip, manufacturer="apple", kind="soc", scheme="apple_chip")130 facts.entities.append(ref)131 return ref132133 def _parse_rows(self, rows: list[Any], chips: dict[str, set[float]], products: dict[str, dict[str, Any]]) -> None:134 column_chips: list[str | None] = []135 for row in rows:136 head_node = row.css_first(".techspecs-rowheader")137 head = re.sub(r"\d+$", "", head_node.text(strip=True)).strip().lower() if head_node else ""138 cols = row.css(".techspecs-column")139 if head == "chip":140 column_chips = []141 for col in cols:142 lines = _column_lines(col)143 chip = next((f"Apple {m.group(1)}" for t in lines for m in [CHIP.search(t)] if m), None)144 chip = _norm_chip(chip) if chip else None145 column_chips.append(chip)146 if not chip:147 continue148 chips.setdefault(chip, set())149 products.setdefault(chip, {"sizes": set(), "bw": {}})150 for t in lines:151 m = BANDWIDTH.search(t)152 if not m:153 continue154 bw = _bw_gbs(m.group(1), m.group(2))155 cm = CHIP_SHORT.search(re.sub(r"^(?:or|and)\s+", "", t.strip()))156 target_chip = _norm_chip(cm.group(1)) if cm and " with " in f" {t} " else chip157 chips.setdefault(target_chip, set()).add(bw)158 products.setdefault(target_chip, {"sizes": set(), "bw": {}})159 elif head == "memory" and column_chips:160 for i, col in enumerate(cols):161 chip = column_chips[i] if i < len(column_chips) else None162 if not chip:163 continue164 products.setdefault(chip, {"sizes": set(), "bw": {}})165 for t in _column_lines(col):166 self._memory_item(t, chip, chips, products)167168 @staticmethod169 def _memory_item(text: str, column_chip: str, chips: dict[str, set[float]], products: dict[str, dict[str, Any]]) -> None:170 """'24GB or 32GB (170GB/s memory bandwidth)' · '36GB (M5 Max with 32-core GPU)' · 'or 48GB (M5 Pro or M5 Max with 40-core GPU)'."""171 outside = PAREN.sub("", text)172 sizes = [int(g) for g in GB.findall(outside)]173 if not sizes:174 return175 parens = PAREN.findall(text)176 bw_m = next((BANDWIDTH.search(p) for p in parens if BANDWIDTH.search(p)), None)177 chip_targets = {_norm_chip(c) for p in parens for c in CHIP_SHORT.findall(p)}178 chip_targets = {c for c in chip_targets if re.fullmatch(r"Apple M\d+( Pro| Max| Ultra)?", c)} or {column_chip}179 for chip in chip_targets:180 info = products.setdefault(chip, {"sizes": set(), "bw": {}})181 info["sizes"].update(sizes)182 if bw_m:183 bw = _bw_gbs(bw_m.group(1), bw_m.group(2))184 chips.setdefault(chip, set()).add(bw)185 for s in sizes:186 info["bw"][s] = bw187188 def _parse_text(self, text: str, chips: dict[str, set[float]], products: dict[str, dict[str, Any]]) -> None:189 """Older template (iMac): sequential text. Safe only when the page describes a single chip."""190 found = {f"Apple {m.group(1)}" for m in CHIP.finditer(text)}191 if len(found) != 1:192 for chip in found:193 chips.setdefault(chip, set())194 return195 chip = next(iter(found))196 bws = {_bw_gbs(m.group(1), m.group(2)) for m in BANDWIDTH.finditer(text)}197 chips[chip] = bws198 info = products.setdefault(chip, {"sizes": set(), "bw": {}})199 mem = re.search(r"\nMemory\n(.*?)\n(?:Display|Storage)\n", text, flags=re.DOTALL)200 if mem:201 info["sizes"].update(int(g) for g in re.findall(r"(\d+)GB(?: or \d+GB)? unified memory", mem.group(1)))202 info["sizes"].update(int(g) for g in re.findall(r"(\d+)GB or \d+GB unified memory", mem.group(1)))203 info["sizes"].update(int(g) for g in re.findall(r"\d+GB or (\d+)GB unified memory", mem.group(1)))204205206def _column_lines(col: Any) -> list[str]:207 """Text lines of one spec column (both Apple templates: <strong class=title>/<li> and <p class=techspecs-subheader|copy>)."""208 out: list[str] = []209 for node in col.css("li, p, strong.title"):210 if node.tag == "p" and node.css_first("strong.title"):211 continue # the <strong> itself is collected212 t = re.sub(r"\s+", " ", node.text(separator=" ", strip=True).replace("\xa0", " ")).strip()213 if t and (not out or out[-1] != t):214 out.append(t)215 return out216217218def _key(chip: str) -> str:219 return re.sub(r"[^a-z0-9]+", "-", chip.lower()).strip("-")220221222# ================================================================================================ NVIDIA223NV_PAGES = [224 ("https://www.nvidia.com/en-us/data-center/h100/", "nvidia:h100", {"H100 SXM": "nvidia-h100-sxm", "H100 NVL": "nvidia-h100-nvl"}),225 ("https://www.nvidia.com/en-us/data-center/h200/", "nvidia:h200", {"H200 SXM": "nvidia-h200", "H200 NVL": "nvidia-h200-nvl"}),226 ("https://www.nvidia.com/en-us/data-center/dgx-b200/", "nvidia:dgx-b200", {"__single__": "nvidia-dgx-b200"}),227 ("https://www.nvidia.com/en-us/products/workstations/dgx-spark/", "nvidia:dgx-spark", {"__single__": "nvidia-dgx-spark"}),228]229NUM = re.compile(r"(\d{1,3}(?:,\d{3})+|\d+(?:\.\d+)?)")230TFLOPS_ROWS = {"fp64": "compute_fp64_tflops", "fp64 tensor core": "compute_fp64_tensor_tflops", "fp32": "compute_fp32_tflops", "tf32 tensor core": "compute_tf32_tflops",231 "bfloat16 tensor core": "compute_bf16_tflops", "fp16 tensor core": "compute_fp16_tflops", "fp8 tensor core": "compute_fp8_tflops",232 "int8 tensor core": "compute_int8_tops", "fp4 tensor core": "compute_fp4_tflops"}233234235class NvidiaSpecsConnector(_Specs):236 name = "nvidia_specs"237 label = "NVIDIA — data-center GPU and DGX system specifications"238 description = "H100, H200, DGX B200 and DGX Spark spec tables: memory, bandwidth, TDP, Tensor Core throughput, form factor."239 source_key = "nvidia.com"240 rate_per_min = 10241 expected_min_records = 4242243 async def discover(self, ctx: RunContext) -> list[Target]:244 return [Target(url=url, doc_type="spec_page", key=key, min_bytes=20000, priority=1, meta={"columns": cols}) for url, key, cols in NV_PAGES]245246 async def extract(self, ctx: RunContext, target: Target, res: FetchResult, parsed: Parsed) -> Facts:247 facts = Facts()248 html = parsed.html249 if not html:250 return facts251 nvidia = org_ref("nvidia")252 facts.entities.append(nvidia)253 columns: dict[str, str] = target.meta.get("columns") or {}254 for table in html.tables:255 rows = [r for r in table["rows"] if r]256 headers = table["headers"]257 if not headers and rows and rows[0] and rows[0][0] == "" and len(rows[0]) >= 3:258 headers, rows = rows[0], rows[1:]259 if headers and len(headers) >= 3 and headers[0] == "":260 names = [re.sub(r"[¹²³⁴*]+", "", h).strip() for h in headers[1:]]261 keys = [columns.get(n) for n in names]262 if not any(keys):263 continue264 for i, key in enumerate(keys):265 if not key:266 continue267 ref = self._ref(facts, key, names[i])268 for row in rows:269 if len(row) > i + 1:270 self._spec(facts, ref, row[0], row[i + 1])271 self._finish(facts, nvidia, ref, target.url)272 elif "__single__" in columns and rows and all(len(r) >= 2 for r in rows[:5]):273 ref = self._ref(facts, columns["__single__"], columns["__single__"])274 for row in rows:275 if len(row) >= 2:276 self._spec(facts, ref, row[0], row[1])277 self._finish(facts, nvidia, ref, target.url)278 break279 facts.document_entity = nvidia280 facts.document_title = html.title281 return facts282283 def _ref(self, facts: Facts, key: str, name: str) -> EntityRef:284 for e in facts.entities:285 if e.identifiers.get("registry_hardware") == key or e.identifiers.get("nvidia_product") == key:286 return e287 ref = hardware_ref(key, f"NVIDIA {name}", manufacturer="nvidia", kind="gpu", scheme="nvidia_product")288 facts.entities.append(ref)289 return ref290291 def _finish(self, facts: Facts, nvidia: EntityRef, ref: EntityRef, url: str) -> None:292 facts.claim(ref, "spec_url", url)293 facts.relate(nvidia, "manufactures", ref)294 if ref.identifiers.get("registry_hardware") == "nvidia-dgx-b200":295 b200 = hardware_ref("nvidia-b200", "NVIDIA B200", manufacturer="nvidia", kind="gpu")296 facts.entities.append(b200)297 facts.relate(ref, "uses", b200, attributes={"count": 8})298299 def _spec(self, facts: Facts, ref: EntityRef, label: str, value: str) -> None:300 key = re.sub(r"[¹²³⁴*]+", "", label).strip().lower()301 key = re.sub(r"\s+\d$", "", key)302 val = value.strip()303 if not val or val in ("-", "—"):304 return305 if key in ("gpu memory", "system memory") and "GB" in val:306 m = re.search(r"(\d{1,3}(?:,\d{3})*|\d+)\s*GB", val)307 if m:308 facts.claim(ref, "memory_gb", int(m.group(1).replace(",", "")), unit="GB")309 t = re.search(r"\b(HBM\d[eE]?|LPDDR\d[xX]?|GDDR\d[xX]?)\b", val)310 if t:311 facts.claim(ref, "memory_type", t.group(1))312 bw = re.search(r"(\d+(?:\.\d+)?)\s*(TB|GB)/s", val)313 if bw:314 facts.claim(ref, "memory_bandwidth_gbs", _bw_gbs(bw.group(1), bw.group(2)), unit="GB/s")315 elif key in ("gpu memory bandwidth", "memory bandwidth"):316 bw = re.search(r"(\d+(?:\.\d+)?)\s*(TB|GB)/s", val)317 if bw:318 facts.claim(ref, "memory_bandwidth_gbs", _bw_gbs(bw.group(1), bw.group(2)), unit="GB/s")319 elif "thermal design power" in key or key in ("tdp", "gb10 tdp"):320 watts = [int(w) for w in re.findall(r"(\d{2,4})\s*W", val.replace(",", ""))]321 if watts:322 facts.claim(ref, "tdp_watts", max(watts), unit="W")323 if len(watts) > 1 or "up to" in val.lower() or "configurable" in val.lower():324 facts.claim(ref, "tdp_note", val[:120])325 elif key in ("system power usage", "power supply"):326 m = re.search(r"(\d+(?:\.\d+)?)\s*(kW|W)", val.replace(",", ""))327 if m:328 watts = float(m.group(1)) * (1000 if m.group(2) == "kW" else 1)329 facts.claim(ref, "system_power_watts" if "usage" in key else "power_supply_watts", int(watts), unit="W")330 elif key in TFLOPS_ROWS:331 m = NUM.search(val)332 if m:333 num = float(m.group(1).replace(",", ""))334 if re.search(r"\bP(?:eta)?FLOPS", val, re.IGNORECASE):335 num *= 1000336 facts.claim(ref, TFLOPS_ROWS[key], int(num) if num.is_integer() else num, unit="TFLOPS" if "tops" not in TFLOPS_ROWS[key] else "TOPS")337 if "*" in label or "²" in label or "sparsity" in val.lower():338 facts.claim(ref, "compute_sparsity", True)339 elif key == "performance":340 for prec, num, unit in re.findall(r"(FP\d)\s*Tensor Core:\s*(\d+(?:\.\d+)?)\s*(PFLOPS|TFLOPS)", val):341 tflops = float(num) * (1000 if unit == "PFLOPS" else 1)342 facts.claim(ref, f"compute_{prec.lower()}_tflops", int(tflops), unit="TFLOPS")343 facts.claim(ref, "compute_sparsity", True)344 elif key.startswith("tensor performance"):345 m = re.search(r"(\d+(?:\.\d+)?)\s*(PFLOP|TFLOP)S?\s*(FP\d)", val, re.IGNORECASE)346 if m:347 tflops = float(m.group(1)) * (1000 if m.group(2).upper() == "PFLOP" else 1)348 facts.claim(ref, f"compute_{m.group(3).lower()}_tflops", int(tflops), unit="TFLOPS")349 elif key == "form factor":350 facts.claim(ref, "form_factor", val[:80])351 elif key == "interconnect":352 m = re.search(r"NVLink[^\d]*(\d+)\s*GB/s", val)353 if m:354 facts.claim(ref, "nvlink_bandwidth_gbs", int(m.group(1)), unit="GB/s")355 elif key == "gpu":356 m = re.match(r"(\d+)x\s+(.*)", val)357 if m:358 facts.claim(ref, "gpu_count", int(m.group(1)))359 facts.claim(ref, "gpu_model", m.group(2)[:80])360 elif "architecture" in val.lower():361 facts.claim(ref, "gpu_architecture", val[:80])362 elif key == "architecture":363 facts.claim(ref, "architecture", val[:80])364 elif key == "cpu":365 facts.claim(ref, "cpu", val[:160])366 elif key in ("storage",):367 facts.claim(ref, "storage", val[:160])368 elif key == "multi-instance gpus":369 facts.claim(ref, "mig_support", val[:80])370 elif key == "nvidia nvlink bandwidth":371 m = re.search(r"(\d+(?:\.\d+)?)\s*TB/s", val)372 if m:373 facts.claim(ref, "nvlink_bandwidth_gbs", int(float(m.group(1)) * 1000), unit="GB/s")374375376CONNECTORS = [AppleSpecsConnector, NvidiaSpecsConnector]377