"""The `Item` dataclass: one priced product as it flows through the pipeline. `Item` exposes a `text` view (cleaned product description) and a `prompt` view (the question/answer string used during fine-tuning and evaluation). """ from __future__ import annotations import re from dataclasses import dataclass from typing import Optional, Self MIN_CHARS = 300 CEILING_CHARS = 160 * 7 PREFIX = "Price is $" QUESTION = "How much does this cost to the nearest dollar?" @dataclass class Item: title: str price: float description: str = "" features: list[str] | None = None details: dict[str, str] | None = None category: str = "Unknown" full: Optional[str] = None weight: Optional[float] = None summary: Optional[str] = None saved_prompt: Optional[str] = None id: Optional[int] = None @property def text(self) -> str: if self.summary: return clean_text(self.summary) if self.full: return clean_text(self.full)[:CEILING_CHARS].rsplit(" ", 1)[0] features = self.features or [] details = self.details or {} pieces = [self.title, self.description, " ".join(features), " ".join(details.values())] cleaned = clean_text(" ".join(piece for piece in pieces if piece)) return cleaned[:CEILING_CHARS].rsplit(" ", 1)[0] @property def prompt(self) -> str: if self.saved_prompt: return self.saved_prompt return f"{QUESTION}\n\n{self.text}\n\n{PREFIX}{round(self.price)}.00" @property def test_prompt(self) -> str: return self.prompt.split(PREFIX)[0] + PREFIX def is_valid(self) -> bool: return self.price > 0 and len(self.text) >= MIN_CHARS @classmethod def from_hub(cls, dataset_name: str) -> tuple[list[Self], list[Self], list[Self]]: from dealsight_intelligence.data.datasets import load_structured_items return load_structured_items(dataset_name) @classmethod def from_mapping(cls, row) -> Self: return cls( title=row.get("title") or "", description=row.get("description") or "", features=row.get("features") or [], details=row.get("details") or {}, price=float(row.get("price") or 0), category=row.get("category") or "Unknown", full=row.get("full"), weight=row.get("weight"), summary=row.get("summary"), saved_prompt=row.get("prompt"), id=row.get("id"), ) def clean_text(value: str) -> str: value = re.sub(r"[\r\n\t]+", " ", value) value = re.sub(r"\s+", " ", value) value = re.sub(r"\b[A-Z0-9]{8,}\b", " ", value) return value.strip()