104 lines
3.2 KiB
Python
104 lines
3.2 KiB
Python
from __future__ import annotations
|
|
|
|
import csv
|
|
from dataclasses import dataclass
|
|
from decimal import Decimal, InvalidOperation
|
|
from pathlib import Path
|
|
|
|
from app.aliases import ProductAliasIndex
|
|
from app.attributes import BrandIndex, ParsedAttributes, parse_attributes
|
|
from app.normalization import meaningful_tokens, normalize_text, tokenize
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class RawCatalogItem:
|
|
sku: str
|
|
name: str
|
|
unit: str
|
|
price: Decimal
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class CatalogItem:
|
|
sku: str
|
|
name: str
|
|
unit: str
|
|
price: Decimal
|
|
normalized_name: str
|
|
search_tokens: tuple[str, ...]
|
|
product_type: str
|
|
attributes: ParsedAttributes
|
|
|
|
|
|
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)
|
|
|
|
|
|
def build_catalog(
|
|
raw_items: tuple[RawCatalogItem, ...],
|
|
alias_index: ProductAliasIndex,
|
|
brand_index: BrandIndex,
|
|
) -> tuple[CatalogItem, ...]:
|
|
result: list[CatalogItem] = []
|
|
for raw in raw_items:
|
|
normalized = normalize_text(raw.name)
|
|
product_type = alias_index.detect_catalog_type(raw.name)
|
|
attributes = parse_attributes(
|
|
raw.name,
|
|
product_type,
|
|
unit=raw.unit,
|
|
catalog_item=True,
|
|
brand_index=brand_index,
|
|
)
|
|
result.append(
|
|
CatalogItem(
|
|
sku=raw.sku,
|
|
name=raw.name,
|
|
unit=raw.unit,
|
|
price=raw.price,
|
|
normalized_name=normalized,
|
|
search_tokens=meaningful_tokens(
|
|
tokenize(normalized, already_normalized=True)
|
|
),
|
|
product_type=product_type,
|
|
attributes=attributes,
|
|
)
|
|
)
|
|
return tuple(result)
|