55 lines
1.8 KiB
Python
55 lines
1.8 KiB
Python
from __future__ import annotations
|
|
|
|
import csv
|
|
from dataclasses import dataclass
|
|
from decimal import Decimal, InvalidOperation
|
|
from pathlib import Path
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class RawCatalogItem:
|
|
sku: str
|
|
name: str
|
|
unit: str
|
|
price: Decimal
|
|
|
|
|
|
def load_raw_catalog(path: Path) -> tuple[RawCatalogItem, ...]:
|
|
if not path.is_file():
|
|
raise FileNotFoundError(f"Catalog file does not exist: {path}")
|
|
|
|
items: list[RawCatalogItem] = []
|
|
with path.open("r", encoding="utf-8-sig", newline="") as file:
|
|
reader = csv.DictReader(file)
|
|
expected = {"sku", "name", "unit", "price"}
|
|
if set(reader.fieldnames or ()) != expected:
|
|
raise ValueError(
|
|
f"Unexpected catalog columns: {reader.fieldnames!r}; expected {sorted(expected)!r}"
|
|
)
|
|
|
|
seen_skus: set[str] = set()
|
|
for line_number, row in enumerate(reader, start=2):
|
|
sku = (row.get("sku") or "").strip()
|
|
name = (row.get("name") or "").strip()
|
|
unit = (row.get("unit") or "").strip()
|
|
price_text = (row.get("price") or "").strip()
|
|
|
|
if not sku or not name or not unit or not price_text:
|
|
raise ValueError(f"Catalog line {line_number} contains an empty required field")
|
|
if sku in seen_skus:
|
|
raise ValueError(f"Duplicate SKU at catalog line {line_number}: {sku}")
|
|
|
|
try:
|
|
price = Decimal(price_text)
|
|
except InvalidOperation as exc:
|
|
raise ValueError(
|
|
f"Invalid price at catalog line {line_number}: {price_text!r}"
|
|
) from exc
|
|
|
|
seen_skus.add(sku)
|
|
items.append(RawCatalogItem(sku=sku, name=name, unit=unit, price=price))
|
|
|
|
if not items:
|
|
raise ValueError("Catalog is empty")
|
|
return tuple(items)
|