import asyncio import logging import random import re from playwright.async_api import async_playwright, Page from playwright.async_api import TimeoutError as PlaywrightTimeoutError from selectolax.parser import HTMLParser from app.core.config import Settings from app.models.scrape import CargoRequest from app.utils.parse_util import extract_product_id, format_price from surugaya_common.browser_profile import DEFAULT_PAGE_HEADERS, stealth_launch_args from surugaya_common.browser_utils import clear_cart from surugaya_common.urls import CARGO_DETAIL_URL, product_detail_url logger = logging.getLogger(__name__) class CargoService: """购物车服务""" def __init__(self, settings: Settings): self._settings = settings async def get_cargo_info(self, request: CargoRequest) -> dict: """获取购物车信息 每次访问都会启动一个全新的 Playwright 浏览器实例， 创建全新的上下文，避免任何状态残留。 先将指定的商品加入购物车，然后再进入购物车页面解析信息。 """ async with async_playwright() as p: headless = self._settings.browser_headless_effective launch_args = stealth_launch_args() if not headless: launch_args.extend(["--start-maximized", "--window-position=0,0"]) # 启动全新的浏览器进程 browser = await p.chromium.launch( headless=headless, channel=self._settings.browser_channel, proxy=self._settings.proxy, args=launch_args, ) try: # 创建全新上下文 context = await browser.new_context( locale=self._settings.browser_locale, timezone_id=self._settings.browser_timezone_id, user_agent=self._settings.browser_user_agent, viewport=None if not headless else {"width": 1440, "height": 900}, ) if self._settings.browser_stealth_enabled: try: # launch_persistent_context 之外的裸 context 用 apply_stealth_async； # 共享工厂带日语 languages override，与 locale/Accept-Language 对齐。 from surugaya_common.stealth import create_stealth await create_stealth().apply_stealth_async(context) except Exception as e: logger.warning("应用 playwright-stealth 失败: %s", e) page = await context.new_page() await page.set_extra_http_headers(DEFAULT_PAGE_HEADERS) target_url = CARGO_DETAIL_URL # 先进入购物车页面 await page.goto(target_url, wait_until="domcontentloaded", timeout=15000) # 判断购物车是否有商品 cart_goods_num = await page.locator("table.item_box tbody tr.item").count() # 清空购物车 await asyncio.sleep(random.uniform(1, 3)) # 加入随机延迟，模拟人工操作间隔 if cart_goods_num > 0: await clear_cart(page) # 将待采购的商品加入购物车 item_list = request.payload.item_list for index, item in enumerate(item_list): product_id = item.id if not product_id: logger.warning("索引 %s 的商品未提供 id，跳过。", index) continue number = item.number or 1 if number <= 0: logger.warning("索引 %s 的商品数量为 0，跳过。", index) continue logger.debug("进入商品 %s 详情页，准备购买数量: %s...", product_id, number) detail_url = product_detail_url(product_id, item.tenpo_cd, item.branch_number) await page.goto(detail_url, wait_until="domcontentloaded", timeout=20000) # 判断是否缺货 out_of_stock_dom = page.locator("div.out-of-stock-text") is_out_of_stock = await out_of_stock_dom.is_visible() if is_out_of_stock: logger.warning("商品 %s 缺货，跳过", product_id) continue # 提取商品最大购买数量 try: quantity_select = page.locator("#quantity_selection") await quantity_select.wait_for(state="visible", timeout=5000) last_option_locator = quantity_select.locator("option").last buy_number_limit_str = await last_option_locator.get_attribute("value") buy_number_limit = int(buy_number_limit_str) if buy_number_limit_str else 0 logger.debug("商品 %s 最大购买数量: %s", product_id, buy_number_limit) if 0 < buy_number_limit < number: number = buy_number_limit except PlaywrightTimeoutError: logger.debug("商品 %s 没有数量选择框，默认购买数量: %s", product_id, number) try: btn_buy = page.locator("button.btn_buy").first await btn_buy.wait_for(state="visible", timeout=5000) if await btn_buy.is_disabled(): logger.warning("商品 %s 已售罄或无法购买（按钮被禁用），跳过", product_id) continue # 根据 number 循环点击加入购物车 for i in range(number): logger.debug("点击加入购物车按钮，第 %s 次", i + 1) await btn_buy.click() if i < number - 1: # 多次点击之间加入随机延迟，防止被识别为机器人或触发频率限制 await asyncio.sleep(random.uniform(0.8, 1.5)) except PlaywrightTimeoutError: logger.warning("商品 %s 找不到购买按钮(可能是已下架/缺货)，跳过", product_id) continue logger.info("再次访问购物车页面: %s", target_url) await page.goto(target_url, wait_until="domcontentloaded", timeout=15000) # 解析购物车 try: primary = page.locator("#container #main3 #primary") if not await primary.is_visible(timeout=10000): logger.warning("购物车商品元素未找到，跳过解析") raise Exception("购物车商品获取失败") result = await self.parse_cargo_info(page) except Exception as e: logger.warning("购物车价格元素未找到，跳过解析: %s", e) raise Exception("购物车价格获取失败") return result finally: await browser.close() async def parse_cargo_info(self, page: Page): """ 解析商品总额，运费和手续费金额等信息 """ logger.debug("解析购物车信息") total_goods_amount = 0 total_shipping_fee = 0 total_commission_fee = 0 items: list[dict] = [] total_price_dom = page.locator("#total_box div.total_price") try: await total_price_dom.first.wait_for(state="attached", timeout=20000) total_price_str = await total_price_dom.locator("p.price_name .total_price").first.text_content() if total_price_str: total_goods_amount = format_price(total_price_str.strip()) or 0 logger.debug("商品总额: %s", total_goods_amount) delivery_charge_html = await total_price_dom.locator("p.delivery_charge").first.inner_html() if delivery_charge_html: shipping_match = re.search(r"([\d,]+)(?=円の送料)", delivery_charge_html) if shipping_match: total_shipping_fee = int(shipping_match.group(1).replace(",", "")) logger.debug("运费: %s", total_shipping_fee) handling_match = re.search(r"通信販売手数料.*?:([\d,]+)(?=円)", delivery_charge_html) if handling_match: total_commission_fee = int(handling_match.group(1).replace(",", "")) logger.debug("手续费: %s", total_commission_fee) except PlaywrightTimeoutError: pass table = page.locator("#container #main3 #primary table.item_box") await table.wait_for(state="visible", timeout=10000) rows = table.locator("tr.item_row") for i in range(await rows.count()): row = rows.nth(i) logger.debug("解析商品 %s", i) link_node = row.locator("td.photo_box a").first href = await link_node.get_attribute("href") or "" img_node = row.locator("td.photo_box img").first goods_image = await img_node.get_attribute("src") or "" name_node = row.locator("td.item_detail .item_name a").first goods_name = (await name_node.text_content() or "").strip() price_node = row.locator("td.price").first price_text = (await price_node.text_content() or "").strip() goods_price = format_price(price_text) or 0 quantity = 1 qty_input = row.locator("input[name*='qty'], input[name*='quantity']").first if await qty_input.count(): qty_val = (await qty_input.get_attribute("value") or "").strip() if qty_val.isdigit(): quantity = int(qty_val) else: qty_select = row.locator("select[name*='qty'], select[name*='quantity'], select.quantity").first if await qty_select.count(): selected = qty_select.locator("option[selected]").first if await selected.count(): qty_val = (await selected.get_attribute("value") or "").strip() if qty_val.isdigit(): quantity = int(qty_val) items.append( { "goods_id": extract_product_id(href), "goods_name": goods_name, "goods_price": goods_price, "goods_image": goods_image, "quantity": quantity, } ) if total_goods_amount <= 0: total_goods_amount = sum(item["goods_price"] * item["quantity"] for item in items) return { "total_goods_amount": total_goods_amount, "total_shipping_fee": total_shipping_fee, "total_commission_fee": total_commission_fee, "items": items, } def _parse_cargo_html(self, html: str) -> dict: tree = HTMLParser(html) items = [] # 根据常见的骏河屋购物车结构进行解析 item_nodes = tree.css("table.cart_table tr.item_row, .cart-item") if not item_nodes: item_nodes = tree.css("div.item_box") # Fallback for node in item_nodes: # 解析商品ID id_node = node.css_first("input[name='goods_id'], .goods_id") goods_id = id_node.attributes.get("value", "") if id_node else "" if not goods_id: # 尝试从链接提取 link = node.css_first("a[href*='/product/detail/']") if link: href = link.attributes.get("href", "") import re match = re.search(r'/product/detail/([A-Za-z0-9]+)', href) if match: goods_id = match.group(1) if not goods_id: continue # 解析商品名称 name_node = node.css_first(".item_name, .title, p.name") goods_name = name_node.text().strip() if name_node else f"product-{goods_id}" # 解析商品价格 price_node = node.css_first(".price, .item_price, td.price") price_text = price_node.text().strip() if price_node else "0" goods_price = format_price(price_text) or 0 # 解析商品图片 img_node = node.css_first("img") goods_image = img_node.attributes.get("src", "") if img_node else "" items.append({ "goods_id": goods_id, "goods_name": goods_name, "goods_price": goods_price, "goods_image": goods_image, "number": 1 }) # 提取金额信息 total_goods_amount = 0 total_shipping_fee = 0 total_commission_fee = 0 # 寻找包含金额的总计区域 summary_nodes = tree.css(".summary_box, .total_box, table.total_table tr") for node in summary_nodes: text = node.text().replace(" ", "").replace("\n", "") if "商品合計" in text or "小計" in text: price_node = node.css_first(".price, td:last-child") if price_node: total_goods_amount = format_price(price_node.text().strip()) or 0 elif "送料" in text: price_node = node.css_first(".price, td:last-child") if price_node: total_shipping_fee = format_price(price_node.text().strip()) or 0 elif "通信販売手数料" in text or "代引手数料" in text or "手数料" in text: price_node = node.css_first(".price, td:last-child") if price_node: total_commission_fee = format_price(price_node.text().strip()) or 0 return { "items": items, "total_goods_amount": total_goods_amount, "total_shipping_fee": total_shipping_fee, "total_commission_fee": total_commission_fee }