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