diff --git a/clients_protocol/stores.py b/clients_protocol/stores.py new file mode 100644 index 0000000..8371462 --- /dev/null +++ b/clients_protocol/stores.py @@ -0,0 +1,186 @@ +"""Data definitions for store clients.""" + +from collections.abc import Iterator, Sequence +from fin_defs import AssetAmount +from frozendict import frozendict +import abc +import dataclasses +import datetime +import enum +import logging + +from . import AbstractClient + +DEFAULT_MAX_RESULTS = 1 + +URL = str + +@dataclasses.dataclass(frozen=True, order=True, slots=True) +class SellerInfo: + name: str + url: str + address: str + country_alpha2: str + company_registration_number: int | None = None + phone_number: str | None = None + email_address: str | None = None + validated: bool | None = None + creation_date: datetime.date | None = None + + def __post_init__(self): + assert self.name.strip() == self.name + assert self.name != '' + assert self.url.strip() == self.url + assert self.address.strip() == self.address + assert len(self.country_alpha2) == 2, 'Invalid ISO 3166-1 alpha-2 country code' + + +class StoreOfferProperty(enum.Enum): + AUCTION = enum.auto() + SOLD_OUT = enum.auto() + DISCOUNT = enum.auto() + + +@dataclasses.dataclass(frozen=True, order=True, slots=True) +class StoreOffer: + title: str + description: str + listed_price: AssetAmount + source: str + image_sources: Sequence[str] = () + seller: SellerInfo | None = None + date_published: datetime.datetime | None = None + date_updated: datetime.datetime | None = None + properties: frozenset[StoreOfferProperty] = dataclasses.field( + default_factory=frozenset, + ) + metadata: frozendict[str, str] = dataclasses.field(default_factory=frozendict) + price_shipping_estimate: AssetAmount | None = None + + def __post_init__(self): + assert self.title.strip() == self.title, self.title + # assert '\n' not in self.title, self.title + assert self.description is not None + assert self.source.strip() == self.source, self.source + assert None not in self.metadata + assert None not in self.image_sources + assert isinstance(self.metadata, frozendict), 'Metadata must be a frozendict' + assert isinstance(self.image_sources, tuple), 'image_sources must be a tuple' + assert ( + isinstance(self.listed_price, AssetAmount) or self.listed_price is None + ), f'listed_price must be an AssetAmount, but was {self.listed_price}' + assert ( + isinstance(self.price_shipping_estimate, AssetAmount) + or self.price_shipping_estimate is None + ), ( + f'price_shipping_estimate must be an AssetAmount, but was {self.price_shipping_estimate}' + ) + + @property + def is_auction(self) -> bool: + return StoreOfferProperty.AUCTION in self.properties + + +ALLOWED_CATEGORIES = {None, 'video-games'} +ALLOWED_REGION_NAMES = frozenset({None, 'japan', 'pal', 'ntsc', 'english asian'}) + + +@dataclasses.dataclass(frozen=True, order=True, slots=True) +class Product: + name: str + category: str + platform: str | None = None + edition: str | None = None + region: str | None = None + + def __post_init__(self): + assert self.category in ALLOWED_CATEGORIES, f'Unknown category: {self.category}' + assert self.region in ALLOWED_REGION_NAMES, f'Unknown region: {self.region}' + + @staticmethod + def videogame(name, **kwargs): + return Product(name, category='video-games', **kwargs) + + +@dataclasses.dataclass(frozen=True) +class ProductPrices: + """Reference prices for a specific product. + + Fields: + - product: Product that this object describes prices for. + - loose: Price for the product itself without any packaging. + - complete: Price for the product with all of its packaging and in-box extracts. + - new: Price of an unopened sealed product, with all of its packaging and in-box extracts. + - graded: Price of an unopened sealed product that have been graded by a professional. + - source: URL indicating the source of the pricing information. + """ + + product: Product + loose: AssetAmount + complete: AssetAmount + new: AssetAmount + graded: AssetAmount + source: str + + +class ItemSaleStatus(enum.Enum): + SOLD = enum.auto() + + +class OfferNoLongerAvailableError(Exception): + pass + + +class StoreOfferAccessClient(AbstractClient, abc.ABC): + """Client for accessing specific store offers on specific websites.""" + + @abc.abstractmethod + def supports_url(self, url: URL) -> bool: + """Check if this client supports the given URL. + + This method determines whether the provided URL is supported by this + client. + + Args: + url: The URL to check for support. + + Returns: + True if the URL is supported, False otherwise. + + Performance: + Lightweight. Implementations should not send any HTTP requests. + """ + + @abc.abstractmethod + def get_store_offer(self, url: URL) -> StoreOffer: + """Downloads the description of the store offer.""" + + +class StoreOfferSearchClient(AbstractClient, abc.ABC): + """Client for searching for store offers on specific websites.""" + + @abc.abstractmethod + def search_for_games( + self, + max_results: int = DEFAULT_MAX_RESULTS, + ) -> Iterator[StoreOffer]: + """Searches for all games offers on the respective website.""" + + @abc.abstractmethod + def search_for_product( + self, + search_term: str, + max_results: int = DEFAULT_MAX_RESULTS, + ) -> Iterator[StoreOffer]: + """Searches for the given named product.""" + + +class ReferencePriceClient(AbstractClient, abc.ABC): + """Client for accessing reference prices for specific products.""" + + @abc.abstractmethod + def get_prices( + self, + product: Product, + ) -> ProductPrices | None: + """Gets reference prices for the given product."""