clients-protocol/clients_protocol/stores.py

"""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."""