surugaya-api/surugaya_api/product.py

import json
from dataclasses import dataclass
from typing import Optional

import aiohttp
from bs4 import BeautifulSoup

from surugaya_api.consts import SURU_COOKIE_STRING

import inspect

from surugaya_api.japan_prefectures import JapanesePrefectures


def get_or_create_aiohttp_session(func):
    async def wrapper(*args, **kwargs):
        sig = inspect.signature(func)
        bound_args = sig.bind(*args, **kwargs)
        bound_args.apply_defaults()

        aiohttp_session = bound_args.arguments.get("aiohttp_session")
        is_created = False

        if not aiohttp_session:
            is_created = True
            aiohttp_session = aiohttp.ClientSession()
            bound_args.arguments["aiohttp_session"] = aiohttp_session

        result = await func(*bound_args.arguments.values())

        if is_created:
            await aiohttp_session.close()

        return result

    return wrapper


@dataclass
class ProductStock:
    condition: str
    stock: str
    price: int
    on_sale_price: Optional[int]

    @property
    def is_on_sale(self) -> bool:
        return self.on_sale_price is not None

    @staticmethod
    def from_item_price(item_price):
        data: dict = json.loads(item_price.select_one("input").attrs["data-zaiko"])

        return ProductStock(
            condition=item_price.select_one("label").attrs["data-label"].strip(),
            stock=data["zaiko"],
            price=data["baika"],
            on_sale_price=data.get("price_sale"),
        )


@dataclass
class Product:
    id: int
    name: str
    main_image_href: str
    stock: list[ProductStock]

    @property
    def in_stock(self):
        return not self.stock


@get_or_create_aiohttp_session
async def load_product(product_id, aiohttp_session=None):
    async with aiohttp_session.get(
        url="https://www.suruga-ya.jp/product/detail/" + str(product_id),
        headers={
            "Cookie": SURU_COOKIE_STRING,
        },
    ) as response:
        page = await response.text()
        page_bs = BeautifulSoup(page, features="html.parser")

        product = Product(
            id=int(product_id),
            name=page_bs.select_one("#item_title").text.strip(),
            main_image_href=page_bs.select_one(".main-pro-img").attrs["src"],
            stock=[
                ProductStock.from_item_price(item_price)
                for item_price in page_bs.select(".item-price")
            ],
        )

    return product


@dataclass
class MarketplaceListing:
    product_id: int
    item_price: int
    condition: str
    product_link: str
    store_name: str
    store_id: int
    special_info: str
    shipping_price: int

    @property
    def total_price(self) -> int:
        return self.item_price + self.shipping_price


@get_or_create_aiohttp_session
async def get_shipping_price_for_store(
    store_id,
    to_prefecture=JapanesePrefectures.KANAGAWA,
    aiohttp_session: Optional[aiohttp.ClientSession] = None,
) -> int:
    response = await aiohttp_session.post(
        url="https://www.suruga-ya.jp/product-other/shipping-fee",
        headers={
            "Accept": "application/json, text/javascript, */*; q=0.01",
            "Content-Type": "application/x-www-form-urlencoded; charset=UTF-8",
            "Origin": "https://www.suruga-ya.jp",
            "Cookie": SURU_COOKIE_STRING,
        },
        data={
            "tenpo_cd": store_id,
        },
    )

    response_json = await response.json()
    shipping_fees = response_json["data"]

    if not shipping_fees["list_pref_fee"]:
        shipping_price = shipping_fees["shipping"]["national_fee"]
    else:
        shipping_filtered = [
            x
            for x in shipping_fees["list_pref_fee"]
            if x["prefecture"] == to_prefecture.value
        ]

        if not shipping_filtered:
            shipping_price = shipping_fees["shipping"]["national_fee"]
        else:
            shipping_price = shipping_filtered[0]["fee"]

    return int(shipping_price)


@get_or_create_aiohttp_session
async def get_marketplace_availability(
    product_id,
    shipping_to_prefecture=JapanesePrefectures.KANAGAWA,
    aiohttp_session: Optional[aiohttp.ClientSession] = None,
):
    async with aiohttp_session.get(
        f"https://www.suruga-ya.jp/product/other/{product_id}"
    ) as response:
        if response.status == 301 or response.status == 302:
            # TODO: Handle redirect to /product/detail if only one store is available
            return []

        page = await response.text()
        page_bs = BeautifulSoup(page, features="html.parser")

        items = page_bs.select("#tbl_all tr.item")
        listings = []

        for item in items:
            if item.select_one("td:nth-child(3) a").attrs["href"] == "/":
                # Main online store, not processing
                continue

            item_price = (
                item.select_one("td:nth-child(1) .title_product > strong")
                .text.replace(",", "")
                .replace("円", "")
            )

            condition = item.select_one("td:nth-child(2) .title_product").text
            product_link = item.select_one("td:nth-child(2) > a").attrs["href"]

            store_id = item.select_one("td:nth-child(3) a").attrs["href"].split("/")[-1]
            store_name = item.select_one("td:nth-child(3) a").text
            shipping_price = await get_shipping_price_for_store(
                store_id, shipping_to_prefecture, aiohttp_session
            )

            special_info = "\n".join(
                [x.text for x in item.select("td:nth-child(4) .padT5 div")]
            )

            listings.append(
                MarketplaceListing(
                    product_id=int(product_id),
                    item_price=int(item_price),
                    condition=condition,
                    product_link=product_link,
                    store_name=store_name,
                    store_id=int(store_id),
                    special_info=special_info,
                    shipping_price=shipping_price,
                )
            )

        return listings