scripts/local_hardware_items.py

from __future__ import annotations

import csv
import json
import re
from pathlib import Path
from typing import Any
from urllib.parse import urlparse

YEARS = set(range(2020, 2027))
PRODUCERS = {"Apple", "NVIDIA", "AMD"}
EVIDENCE_VALUES = {"quoted", "reconstructed", "bom_inferred"}
CONFIDENCE_VALUES = {"high", "medium", "low"}
ITEM_KINDS = {
    "complete_system",
    "configured_system",
    "component_based_workstation",
}
PRICE_KINDS = {
    "launch_msrp",
    "updated_msrp",
    "launch_config",
    "current_config",
    "quoted_system",
    "reconstructed_config",
    "component_bom",
    "historical_quote",
}
MEMORY_SYMBOLS = {
    "unified/coherent": "U",
    "single-GPU VRAM": "V",
    "aggregate multi-GPU VRAM": "Σ",
}
EXPECTED_SCHEMA = "../../schemas/local_hardware_item.schema.json"
ID_RE = re.compile(r"^[a-z0-9]+(?:-[a-z0-9]+)*$")

TOP_LEVEL_KEYS = {
    "$schema",
    "schema_version",
    "id",
    "producer",
    "name",
    "item_kind",
    "memory",
    "accelerators",
    "spec_sources",
    "price_history",
}
REQUIRED_TOP_LEVEL_KEYS = {
    "$schema",
    "schema_version",
    "id",
    "producer",
    "name",
    "item_kind",
    "memory",
    "price_history",
}
MEMORY_KEYS = {
    "capacity_gb",
    "type",
    "symbol",
    "bandwidth_gbps",
    "bandwidth_source_url",
    "bandwidth_note",
}
REQUIRED_MEMORY_KEYS = {"capacity_gb", "type", "symbol"}
ACCELERATOR_KEYS = {"vendor", "name", "count"}
SOURCE_KEYS = {"name", "url"}
PRICE_KEYS = {
    "id",
    "system_label",
    "start_year",
    "end_year",
    "catalog_order",
    "price_usd",
    "price_band",
    "price_kind",
    "evidence",
    "confidence",
    "source_url",
    "note",
}
MEMORY_PRICE_SOURCE_KEYS = {
    "pricing_id",
    "item_ids",
    "producer",
    "component_label",
    "pricing_method",
    "price_scope",
    "price_year",
    "priced_memory_gb",
    "memory_price_usd",
    "evidence",
    "confidence",
    "source_url",
    "note",
}
MEMORY_PRICE_METHODS = {
    "gpu_card_launch_price",
    "gpu_card_current_price",
    "same_device_config_delta",
}
MEMORY_PRICE_SCOPES = {
    "single_gpu_card",
    "multi_gpu_cards",
    "incremental_unified_memory",
}


class ItemValidationError(ValueError):
    pass


def _is_url(value: str) -> bool:
    parsed = urlparse(value)
    return parsed.scheme in {"http", "https"} and bool(parsed.netloc)


def _require_keys(
    errors: list[str],
    where: str,
    obj: dict[str, Any],
    required: set[str],
    allowed: set[str],
) -> None:
    missing = sorted(required - set(obj))
    extra = sorted(set(obj) - allowed)
    if missing:
        errors.append(f"{where}: missing required keys: {', '.join(missing)}")
    if extra:
        errors.append(f"{where}: unknown keys: {', '.join(extra)}")


def validate_items(
    items: list[dict[str, Any]],
    *,
    item_paths: dict[str, Path] | None = None,
    bands: list[tuple[str, str, int, int | None]],
) -> None:
    errors: list[str] = []
    band_map = {key: (lo, hi) for key, _, lo, hi in bands}
    item_ids: set[str] = set()
    catalog_keys: set[tuple[Any, ...]] = set()
    catalog_orders: set[int] = set()

    for item in items:
        if not isinstance(item, dict):
            errors.append("item root must be an object")
            continue

        item_id = item.get("id", "<missing-id>")
        where = str(item_id)
        _require_keys(
            errors,
            where,
            item,
            REQUIRED_TOP_LEVEL_KEYS,
            TOP_LEVEL_KEYS,
        )

        if item.get("$schema") != EXPECTED_SCHEMA:
            errors.append(f"{where}: $schema must be {EXPECTED_SCHEMA}")
        if item.get("schema_version") != 1:
            errors.append(f"{where}: schema_version must be 1")
        if not isinstance(item.get("id"), str) or not ID_RE.match(item["id"]):
            errors.append(f"{where}: id must be a lowercase slug")
        elif item["id"] in item_ids:
            errors.append(f"{where}: duplicate item id")
        else:
            item_ids.add(item["id"])

        if item_paths and item.get("id") in item_paths:
            stem = item_paths[item["id"]].stem
            if stem != item["id"]:
                errors.append(f"{where}: id must match filename {stem}.json")

        if item.get("producer") not in PRODUCERS:
            errors.append(f"{where}: invalid producer {item.get('producer')!r}")
        if item.get("item_kind") not in ITEM_KINDS:
            errors.append(f"{where}: invalid item_kind {item.get('item_kind')!r}")
        if not isinstance(item.get("name"), str) or not item.get("name"):
            errors.append(f"{where}: name must be a non-empty string")

        memory = item.get("memory")
        if not isinstance(memory, dict):
            errors.append(f"{where}: memory must be an object")
            continue

        _require_keys(errors, f"{where}.memory", memory, REQUIRED_MEMORY_KEYS, MEMORY_KEYS)
        memory_type = memory.get("type")
        expected_symbol = MEMORY_SYMBOLS.get(memory_type)
        if expected_symbol is None:
            errors.append(f"{where}: invalid memory.type {memory_type!r}")
        elif memory.get("symbol") != expected_symbol:
            errors.append(
                f"{where}: memory.symbol must be {expected_symbol!r} for {memory_type!r}"
            )
        if not isinstance(memory.get("capacity_gb"), int) or memory.get("capacity_gb", 0) <= 0:
            errors.append(f"{where}: memory.capacity_gb must be a positive integer")
        bandwidth = memory.get("bandwidth_gbps")
        if bandwidth is not None and (
            not isinstance(bandwidth, (int, float)) or bandwidth <= 0
        ):
            errors.append(f"{where}: memory.bandwidth_gbps must be positive or null")
        if bandwidth is not None and "bandwidth_source_url" not in memory:
            errors.append(f"{where}: memory.bandwidth_source_url is required with bandwidth_gbps")
        if bandwidth is None and "bandwidth_source_url" in memory:
            errors.append(f"{where}: memory.bandwidth_source_url requires bandwidth_gbps")
        if "bandwidth_source_url" in memory and not _is_url(memory["bandwidth_source_url"]):
            errors.append(f"{where}: memory.bandwidth_source_url must be an http(s) URL")
        bandwidth_note = memory.get("bandwidth_note")
        if bandwidth_note is not None and (
            not isinstance(bandwidth_note, str) or not bandwidth_note
        ):
            errors.append(f"{where}: memory.bandwidth_note must be a non-empty string")
        if memory_type == "aggregate multi-GPU VRAM" and bandwidth is not None and not bandwidth_note:
            errors.append(f"{where}: aggregate bandwidth requires memory.bandwidth_note")

        for index, accelerator in enumerate(item.get("accelerators", [])):
            if not isinstance(accelerator, dict):
                errors.append(f"{where}.accelerators[{index}]: must be an object")
                continue
            _require_keys(
                errors,
                f"{where}.accelerators[{index}]",
                accelerator,
                {"vendor", "name"},
                ACCELERATOR_KEYS,
            )
            if "count" in accelerator and (
                not isinstance(accelerator["count"], int) or accelerator["count"] <= 0
            ):
                errors.append(f"{where}.accelerators[{index}]: count must be positive")

        for index, source in enumerate(item.get("spec_sources", [])):
            if not isinstance(source, dict):
                errors.append(f"{where}.spec_sources[{index}]: must be an object")
                continue
            _require_keys(
                errors,
                f"{where}.spec_sources[{index}]",
                source,
                {"name", "url"},
                SOURCE_KEYS,
            )
            if "url" in source and not _is_url(source["url"]):
                errors.append(f"{where}.spec_sources[{index}]: url must be an http(s) URL")

        price_history = item.get("price_history")
        if not isinstance(price_history, list) or not price_history:
            errors.append(f"{where}: price_history must be a non-empty array")
            continue

        event_ids: set[str] = set()
        covered_years: set[int] = set()
        previous_start = -1

        for index, event in enumerate(price_history):
            event_where = f"{where}.price_history[{index}]"
            if not isinstance(event, dict):
                errors.append(f"{event_where}: must be an object")
                continue

            _require_keys(errors, event_where, event, PRICE_KEYS, PRICE_KEYS)
            event_id = event.get("id")
            if not isinstance(event_id, str) or not ID_RE.match(event_id):
                errors.append(f"{event_where}: id must be a lowercase slug")
            elif event_id in event_ids:
                errors.append(f"{event_where}: duplicate price history id")
            else:
                event_ids.add(event_id)

            start = event.get("start_year")
            end = event.get("end_year")
            if not isinstance(start, int) or start not in YEARS:
                errors.append(f"{event_where}: start_year must be in 2020-2026")
                continue
            if not isinstance(end, int) or end not in YEARS:
                errors.append(f"{event_where}: end_year must be in 2020-2026")
                continue
            if start > end:
                errors.append(f"{event_where}: start_year must be <= end_year")
                continue
            if start < previous_start:
                errors.append(f"{event_where}: price_history must be sorted by start_year")
            previous_start = start
            years = set(range(start, end + 1))
            overlap = years & covered_years
            if overlap:
                errors.append(
                    f"{event_where}: overlapping price years: "
                    + ", ".join(str(year) for year in sorted(overlap))
                )
            covered_years |= years

            catalog_order = event.get("catalog_order")
            if not isinstance(catalog_order, int) or catalog_order <= 0:
                errors.append(f"{event_where}: catalog_order must be a positive integer")
            elif catalog_order in catalog_orders:
                errors.append(f"{event_where}: duplicate catalog_order {catalog_order}")
            else:
                catalog_orders.add(catalog_order)

            price = event.get("price_usd")
            if not isinstance(price, int) or price <= 0:
                errors.append(f"{event_where}: price_usd must be a positive integer")
            price_band = event.get("price_band")
            if price_band not in band_map:
                errors.append(f"{event_where}: invalid price_band {price_band!r}")
            elif isinstance(price, int):
                lo, hi = band_map[price_band]
                if price < lo or (hi is not None and price >= hi):
                    errors.append(
                        f"{event_where}: ${price:,} does not fit price_band {price_band}"
                    )

            if event.get("price_kind") not in PRICE_KINDS:
                errors.append(f"{event_where}: invalid price_kind {event.get('price_kind')!r}")
            if event.get("evidence") not in EVIDENCE_VALUES:
                errors.append(f"{event_where}: invalid evidence {event.get('evidence')!r}")
            if event.get("confidence") not in CONFIDENCE_VALUES:
                errors.append(f"{event_where}: invalid confidence {event.get('confidence')!r}")
            if not isinstance(event.get("source_url"), str) or not _is_url(event["source_url"]):
                errors.append(f"{event_where}: source_url must be an http(s) URL")
            if not isinstance(event.get("system_label"), str) or not event.get("system_label"):
                errors.append(f"{event_where}: system_label must be non-empty")
            if not isinstance(event.get("note"), str) or not event.get("note"):
                errors.append(f"{event_where}: note must be non-empty")

            key = (
                item.get("producer"),
                event.get("system_label"),
                start,
                end,
                price,
                memory.get("capacity_gb"),
            )
            if key in catalog_keys:
                errors.append(f"{event_where}: duplicate generated catalog row")
            catalog_keys.add(key)

    if errors:
        raise ItemValidationError("\n".join(f"- {error}" for error in errors))


def load_items(
    items_dir: Path,
    *,
    bands: list[tuple[str, str, int, int | None]],
) -> list[dict[str, Any]]:
    items: list[dict[str, Any]] = []
    item_paths: dict[str, Path] = {}
    for path in sorted(items_dir.glob("*.json")):
        item = json.loads(path.read_text(encoding="utf-8"))
        item_id = item.get("id", path.stem) if isinstance(item, dict) else path.stem
        item_paths[str(item_id)] = path
        items.append(item)
    validate_items(items, item_paths=item_paths, bands=bands)
    return items


def catalog_from_items(items: list[dict[str, Any]]) -> list[dict[str, Any]]:
    ordered_rows: list[tuple[int, dict[str, Any]]] = []
    for item in items:
        memory = item["memory"]
        for event in item["price_history"]:
            bandwidth = memory.get("bandwidth_gbps")
            ordered_rows.append(
                (
                    event["catalog_order"],
                    {
                        "item_id": item["id"],
                        "price_event_id": event["id"],
                        "producer": item["producer"],
                        "system": event["system_label"],
                        "start_year": event["start_year"],
                        "end_year": event["end_year"],
                        "price_usd": event["price_usd"],
                        "price_band": event["price_band"],
                        "price_kind": event["price_kind"],
                        "memory_gb": memory["capacity_gb"],
                        "memory_type": memory["type"],
                        "symbol": memory["symbol"],
                        "bandwidth_gbps": bandwidth if bandwidth is not None else "",
                        "bandwidth_source_url": memory.get("bandwidth_source_url", ""),
                        "bandwidth_note": memory.get("bandwidth_note", ""),
                        "price_per_memory_gb": round(
                            event["price_usd"] / memory["capacity_gb"], 2
                        ),
                        "evidence": event["evidence"],
                        "confidence": event["confidence"],
                        "source_url": event["source_url"],
                        "note": event["note"],
                    },
                )
            )
    return [row for _, row in sorted(ordered_rows, key=lambda pair: pair[0])]


def load_catalog_from_items(
    items_dir: Path,
    *,
    bands: list[tuple[str, str, int, int | None]],
) -> list[dict[str, Any]]:
    return catalog_from_items(load_items(items_dir, bands=bands))


def load_memory_price_sources(
    path: Path,
    items: list[dict[str, Any]],
) -> list[dict[str, Any]]:
    item_by_id = {item["id"]: item for item in items}
    errors: list[str] = []
    rows: list[dict[str, Any]] = []
    pricing_ids: set[str] = set()

    if not path.exists():
        return []

    with path.open(newline="", encoding="utf-8") as f:
        reader = csv.DictReader(f)
        if set(reader.fieldnames or []) != MEMORY_PRICE_SOURCE_KEYS:
            expected = ", ".join(sorted(MEMORY_PRICE_SOURCE_KEYS))
            actual = ", ".join(reader.fieldnames or [])
            raise ItemValidationError(
                f"{path}: unexpected memory price fields: {actual}; expected {expected}"
            )

        for index, row in enumerate(reader, start=2):
            where = f"{path.name}:{index}"
            row_errors: list[str] = []
            _require_keys(
                row_errors,
                where,
                row,
                MEMORY_PRICE_SOURCE_KEYS,
                MEMORY_PRICE_SOURCE_KEYS,
            )

            pricing_id = row.get("pricing_id", "")
            if not ID_RE.match(pricing_id):
                row_errors.append(f"{where}: pricing_id must be a lowercase slug")
            elif pricing_id in pricing_ids:
                row_errors.append(f"{where}: duplicate pricing_id {pricing_id}")
            else:
                pricing_ids.add(pricing_id)

            item_ids = [item_id.strip() for item_id in row.get("item_ids", "").split(";")]
            item_ids = [item_id for item_id in item_ids if item_id]
            if not item_ids:
                row_errors.append(f"{where}: item_ids must name at least one item")

            for item_id in item_ids:
                if item_id not in item_by_id:
                    row_errors.append(f"{where}: unknown item_id {item_id}")
                elif item_by_id[item_id]["producer"] != row.get("producer"):
                    row_errors.append(
                        f"{where}: producer does not match item {item_id}"
                    )

            if row.get("pricing_method") not in MEMORY_PRICE_METHODS:
                row_errors.append(
                    f"{where}: invalid pricing_method {row.get('pricing_method')!r}"
                )
            if row.get("price_scope") not in MEMORY_PRICE_SCOPES:
                row_errors.append(f"{where}: invalid price_scope {row.get('price_scope')!r}")
            if row.get("evidence") not in EVIDENCE_VALUES:
                row_errors.append(f"{where}: invalid evidence {row.get('evidence')!r}")
            if row.get("confidence") not in CONFIDENCE_VALUES:
                row_errors.append(f"{where}: invalid confidence {row.get('confidence')!r}")
            if not isinstance(row.get("source_url"), str) or not _is_url(row["source_url"]):
                row_errors.append(f"{where}: source_url must be an http(s) URL")
            if not row.get("component_label"):
                row_errors.append(f"{where}: component_label must be non-empty")
            if not row.get("note"):
                row_errors.append(f"{where}: note must be non-empty")

            try:
                price_year = int(row.get("price_year", ""))
            except ValueError:
                row_errors.append(f"{where}: price_year must be an integer")
                price_year = 0
            if price_year and not 2010 <= price_year <= max(YEARS):
                row_errors.append(f"{where}: price_year must be between 2010 and {max(YEARS)}")

            try:
                priced_memory_gb = float(row.get("priced_memory_gb", ""))
            except ValueError:
                row_errors.append(f"{where}: priced_memory_gb must be numeric")
                priced_memory_gb = 0
            if priced_memory_gb <= 0:
                row_errors.append(f"{where}: priced_memory_gb must be positive")

            try:
                memory_price_usd = float(row.get("memory_price_usd", ""))
            except ValueError:
                row_errors.append(f"{where}: memory_price_usd must be numeric")
                memory_price_usd = 0
            if memory_price_usd <= 0:
                row_errors.append(f"{where}: memory_price_usd must be positive")

            if row_errors:
                errors.extend(row_errors)
                continue

            for item_id in item_ids:
                rows.append({
                    "pricing_id": pricing_id,
                    "item_id": item_id,
                    "producer": row["producer"],
                    "item_name": item_by_id[item_id]["name"],
                    "component_label": row["component_label"],
                    "pricing_method": row["pricing_method"],
                    "price_scope": row["price_scope"],
                    "price_year": price_year,
                    "priced_memory_gb": int(priced_memory_gb)
                    if priced_memory_gb.is_integer() else priced_memory_gb,
                    "memory_price_usd": int(memory_price_usd)
                    if memory_price_usd.is_integer() else memory_price_usd,
                    "memory_price_per_gb": round(memory_price_usd / priced_memory_gb, 2),
                    "evidence": row["evidence"],
                    "confidence": row["confidence"],
                    "source_url": row["source_url"],
                    "note": row["note"],
                })

    if errors:
        raise ItemValidationError("\n".join(f"- {error}" for error in errors))
    return rows