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 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 = [ "--disable-blink-features=AutomationControlled", "--disable-infobars", "--no-sandbox", "--disable-setuid-sandbox", "--disable-dev-shm-usage", "--disable-gpu", "--use-mock-keychain", "--password-store=basic", ] 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: from playwright_stealth import Stealth await 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( { "Accept-Language": "ja-JP,ja;q=0.9,en-US;q=0.8,en;q=0.7", "Cache-Control": "no-cache", "Pragma": "no-cache", } ) target_url = "https://www.suruga-ya.jp/cargo/detail" # 先进入购物车页面 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 self.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 = f"https://www.suruga-ya.jp/product/detail/{product_id}" 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 } async def clear_cart(self, page): """ 清空购物车 """ cart_goods_num = await page.locator("table.item_box tbody tr.item").count() if cart_goods_num < 1: return is_delete = 0 # 购物车商品数量超过3个后会出现"全削除"按钮 if cart_goods_num > 3: try: delete_button = page.locator("#total_box .delbtn-all").get_by_role("button", name="全削除") # 3. 显式等待按钮在页面上可见（容错处理：比如给它最多 3000 毫秒的加载时间） # 如果按钮是一开始就在静态 HTML 里的，这一步也可以省略，直接用下面的 is_visible() await delete_button.wait_for(state="visible", timeout=3000) if await delete_button.is_visible(): logger.debug("检测到‘全削除’按钮存在，正在执行点击...") await delete_button.click() is_delete = 1 except Exception: logger.debug("未检测到‘全削除’按钮。") if is_delete == 1: return # 遍历删除 item_selector = "table.item_box tbody tr.item" remove_selector = "table.item_box a.remove_product" while True: cart_goods_num = await page.locator(item_selector).count() if cart_goods_num == 0: break remove_link = page.locator(remove_selector).first try: await remove_link.click() # 删除按钮不会触发弹窗，直接等待购物车条目数减少即可。 await page.wait_for_function( """({selector, expectedCount}) => { return document.querySelectorAll(selector).length < expectedCount; }""", arg={"selector": item_selector, "expectedCount": cart_goods_num}, timeout=15_000, ) await page.wait_for_timeout(random.randint(800, 1500)) except PlaywrightTimeoutError: latest_count = await page.locator(item_selector).count() if latest_count < cart_goods_num: continue await page.wait_for_timeout(1000)