diff --git a/XianyuAutoAsync.py b/XianyuAutoAsync.py index a879e1f..0c99f4d 100644 --- a/XianyuAutoAsync.py +++ b/XianyuAutoAsync.py @@ -119,21 +119,30 @@ class XianyuLive: logger.error(f"【{self.cookie_id}】获取自动确认发货设置失败: {self._safe_str(e)}") return True # 出错时默认启用 - def can_auto_delivery(self, item_id: str) -> bool: - """检查是否可以进行自动发货(防重复发货)""" + + + def can_auto_delivery(self, order_id: str) -> bool: + """检查是否可以进行自动发货(防重复发货)- 基于订单ID""" + if not order_id: + # 如果没有订单ID,则不进行冷却检查,允许发货 + return True + current_time = time.time() - last_delivery = self.last_delivery_time.get(item_id, 0) + last_delivery = self.last_delivery_time.get(order_id, 0) if current_time - last_delivery < self.delivery_cooldown: - logger.info(f"【{self.cookie_id}】商品 {item_id} 在冷却期内,跳过自动发货") + logger.info(f"【{self.cookie_id}】订单 {order_id} 在冷却期内,跳过自动发货") return False return True - def mark_delivery_sent(self, item_id: str): - """标记商品已发货""" - self.last_delivery_time[item_id] = time.time() - logger.debug(f"【{self.cookie_id}】标记商品 {item_id} 已发货") + def mark_delivery_sent(self, order_id: str): + """标记订单已发货 - 基于订单ID""" + if order_id: + self.last_delivery_time[order_id] = time.time() + logger.debug(f"【{self.cookie_id}】标记订单 {order_id} 已发货") + else: + logger.debug(f"【{self.cookie_id}】无订单ID,跳过发货标记") @@ -1062,8 +1071,10 @@ class XianyuLive: logger.error(f"发送自动发货通知异常: {self._safe_str(e)}") async def auto_confirm(self, order_id, retry_count=0): - """自动确认发货 - 使用加密模块""" + """自动确认发货 - 使用加密模块,不包含延时处理(延时已在_auto_delivery中处理)""" try: + logger.debug(f"【{self.cookie_id}】开始确认发货,订单ID: {order_id}") + # 导入超级混淆加密模块 from secure_confirm_ultra import SecureConfirm @@ -1083,8 +1094,48 @@ class XianyuLive: logger.error(f"【{self.cookie_id}】加密确认模块调用失败: {self._safe_str(e)}") return {"error": f"加密确认模块调用失败: {self._safe_str(e)}", "order_id": order_id} - async def _auto_delivery(self, item_id: str, item_title: str = None): - """自动发货功能""" + async def fetch_order_detail_info(self, order_id: str): + """获取订单详情信息""" + try: + logger.info(f"【{self.cookie_id}】开始获取订单详情: {order_id}") + + # 导入订单详情获取器 + from utils.order_detail_fetcher import fetch_order_detail_simple + + # 获取当前账号的cookie字符串 + cookie_string = self.cookie_value + logger.debug(f"【{self.cookie_id}】使用Cookie长度: {len(cookie_string) if cookie_string else 0}") + + # 异步获取订单详情(使用当前账号的cookie和无头模式) + result = await fetch_order_detail_simple(order_id, cookie_string, headless=True) + + if result: + logger.info(f"【{self.cookie_id}】订单详情获取成功: {order_id}") + logger.info(f"【{self.cookie_id}】页面标题: {result.get('title', '未知')}") + + # 获取解析后的规格信息 + spec_name = result.get('spec_name', '') + spec_value = result.get('spec_value', '') + + if spec_name and spec_value: + logger.info(f"【{self.cookie_id}】📋 规格名称: {spec_name}") + logger.info(f"【{self.cookie_id}】📝 规格值: {spec_value}") + print(f"🛍️ 【{self.cookie_id}】订单 {order_id} 规格信息: {spec_name} -> {spec_value}") + else: + logger.warning(f"【{self.cookie_id}】未获取到有效的规格信息") + print(f"⚠️ 【{self.cookie_id}】订单 {order_id} 规格信息获取失败") + + return result + else: + logger.warning(f"【{self.cookie_id}】订单详情获取失败: {order_id}") + return None + + except Exception as e: + logger.error(f"【{self.cookie_id}】获取订单详情异常: {self._safe_str(e)}") + return None + + async def _auto_delivery(self, item_id: str, item_title: str = None, order_id: str = None): + """自动发货功能 - 获取卡券规则,执行延时,确认发货,发送内容""" try: from db_manager import db_manager @@ -1199,12 +1250,43 @@ class XianyuLive: rule = delivery_rules[0] logger.info(f"找到匹配的发货规则: {rule['keyword']} -> {rule['card_name']} ({rule['card_type']})") - # 检查是否需要延时发货 + # 获取延时设置 delay_seconds = rule.get('card_delay_seconds', 0) + + # 执行延时(不管是否确认发货,只要有延时设置就执行) if delay_seconds and delay_seconds > 0: - logger.info(f"检测到延时发货设置: {delay_seconds}秒,开始延时...") + logger.info(f"检测到发货延时设置: {delay_seconds}秒,开始延时...") await asyncio.sleep(delay_seconds) - logger.info(f"延时发货完成,开始发送内容") + logger.info(f"延时完成") + + # 如果有订单ID,执行确认发货 + if order_id: + # 检查是否启用自动确认发货 + if not self.is_auto_confirm_enabled(): + logger.info(f"自动确认发货已关闭,跳过订单 {order_id}") + else: + # 检查确认发货冷却时间 + current_time = time.time() + should_confirm = True + + if order_id in self.confirmed_orders: + last_confirm_time = self.confirmed_orders[order_id] + if current_time - last_confirm_time < self.order_confirm_cooldown: + logger.info(f"订单 {order_id} 已在 {self.order_confirm_cooldown} 秒内确认过,跳过重复确认") + should_confirm = False + + if should_confirm: + logger.info(f"开始自动确认发货: 订单ID={order_id}") + confirm_result = await self.auto_confirm(order_id) + if confirm_result.get('success'): + self.confirmed_orders[order_id] = current_time + logger.info(f"🎉 自动确认发货成功!订单ID: {order_id}") + else: + logger.warning(f"⚠️ 自动确认发货失败: {confirm_result.get('error', '未知错误')}") + # 即使确认发货失败,也继续发送发货内容 + + # 开始处理发货内容 + logger.info(f"开始处理发货内容,规则: {rule['keyword']} -> {rule['card_name']} ({rule['card_type']})") delivery_content = None @@ -2024,50 +2106,17 @@ class XianyuLive: except Exception as parse_e: logger.debug(f"解析dynamicOperation JSON失败: {parse_e}") - # 如果成功获取到orderId,进行自动确认发货 + # 订单ID已提取,将在自动发货时进行确认发货处理 if order_id: - # 检查是否启用自动确认发货 - if not self.is_auto_confirm_enabled(): - logger.info(f'[{msg_time}] 【{self.cookie_id}】自动确认发货已关闭,跳过订单 {order_id}') - else: - # 检查是否已经确认过这个订单 - current_time = time.time() - if order_id in self.confirmed_orders: - last_confirm_time = self.confirmed_orders[order_id] - if current_time - last_confirm_time < self.order_confirm_cooldown: - logger.info(f'[{msg_time}] 【{self.cookie_id}】⏭️ 订单 {order_id} 已在 {self.order_confirm_cooldown} 秒内确认过,跳过重复确认') - else: - # 超过冷却时间,可以重新确认 - try: - logger.info(f'[{msg_time}] 【{self.cookie_id}】开始自动确认发货,订单ID: {order_id}') - confirm_result = await self.auto_confirm(order_id) - if confirm_result.get('success'): - self.confirmed_orders[order_id] = current_time - logger.info(f'[{msg_time}] 【{self.cookie_id}】🎉 自动确认发货成功!订单ID: {order_id}') - else: - logger.warning(f'[{msg_time}] 【{self.cookie_id}】⚠️ 自动确认发货失败: {confirm_result.get("error", "未知错误")}') - except Exception as confirm_e: - logger.error(f'[{msg_time}] 【{self.cookie_id}】自动确认发货异常: {self._safe_str(confirm_e)}') - else: - # 首次确认这个订单 - try: - logger.info(f'[{msg_time}] 【{self.cookie_id}】开始自动确认发货,订单ID: {order_id}') - confirm_result = await self.auto_confirm(order_id) - if confirm_result.get('success'): - self.confirmed_orders[order_id] = current_time - logger.info(f'[{msg_time}] 【{self.cookie_id}】🎉 自动确认发货成功!订单ID: {order_id}') - else: - logger.warning(f'[{msg_time}] 【{self.cookie_id}】⚠️ 自动确认发货失败: {confirm_result.get("error", "未知错误")}') - except Exception as confirm_e: - logger.error(f'[{msg_time}] 【{self.cookie_id}】自动确认发货异常: {self._safe_str(confirm_e)}') + logger.info(f'[{msg_time}] 【{self.cookie_id}】提取到订单ID: {order_id},将在自动发货时处理确认发货') else: logger.warning(f'[{msg_time}] 【{self.cookie_id}】❌ 未能提取到订单ID') except Exception as extract_e: logger.error(f"提取订单ID失败: {self._safe_str(extract_e)}") - # 检查是否可以进行自动发货(防重复) - if not self.can_auto_delivery(item_id): + # 检查是否可以进行自动发货(防重复)- 基于订单ID + if not self.can_auto_delivery(order_id): return # 构造用户URL @@ -2080,12 +2129,12 @@ class XianyuLive: logger.info(f"【{self.cookie_id}】准备自动发货: item_id={item_id}, item_title={item_title}") - # 调用自动发货方法 - delivery_content = await self._auto_delivery(item_id, item_title) + # 调用自动发货方法(包含自动确认发货) + delivery_content = await self._auto_delivery(item_id, item_title, order_id) if delivery_content: - # 标记已发货(防重复) - self.mark_delivery_sent(item_id) + # 标记已发货(防重复)- 基于订单ID + self.mark_delivery_sent(order_id) # 发送发货内容给买家 await self.send_msg(websocket, chat_id, send_user_id, delivery_content) @@ -2150,50 +2199,17 @@ class XianyuLive: except Exception as parse_e: logger.debug(f"解析dynamicOperation JSON失败: {parse_e}") - # 如果成功获取到orderId,进行自动确认发货 + # 订单ID已提取,将在自动发货时进行确认发货处理 if order_id: - # 检查是否启用自动确认发货 - if not self.is_auto_confirm_enabled(): - logger.info(f'[{msg_time}] 【{self.cookie_id}】自动确认发货已关闭,跳过订单 {order_id}') - else: - # 检查是否已经确认过这个订单 - current_time = time.time() - if order_id in self.confirmed_orders: - last_confirm_time = self.confirmed_orders[order_id] - if current_time - last_confirm_time < self.order_confirm_cooldown: - logger.info(f'[{msg_time}] 【{self.cookie_id}】⏭️ 订单 {order_id} 已在 {self.order_confirm_cooldown} 秒内确认过,跳过重复确认') - else: - # 超过冷却时间,可以重新确认 - try: - logger.info(f'[{msg_time}] 【{self.cookie_id}】开始自动确认发货,订单ID: {order_id}') - confirm_result = await self.auto_confirm(order_id) - if confirm_result.get('success'): - self.confirmed_orders[order_id] = current_time - logger.info(f'[{msg_time}] 【{self.cookie_id}】🎉 自动确认发货成功!订单ID: {order_id}') - else: - logger.warning(f'[{msg_time}] 【{self.cookie_id}】⚠️ 自动确认发货失败: {confirm_result.get("error", "未知错误")}') - except Exception as confirm_e: - logger.error(f'[{msg_time}] 【{self.cookie_id}】自动确认发货异常: {self._safe_str(confirm_e)}') - else: - # 首次确认这个订单 - try: - logger.info(f'[{msg_time}] 【{self.cookie_id}】开始自动确认发货,订单ID: {order_id}') - confirm_result = await self.auto_confirm(order_id) - if confirm_result.get('success'): - self.confirmed_orders[order_id] = current_time - logger.info(f'[{msg_time}] 【{self.cookie_id}】🎉 自动确认发货成功!订单ID: {order_id}') - else: - logger.warning(f'[{msg_time}] 【{self.cookie_id}】⚠️ 自动确认发货失败: {confirm_result.get("error", "未知错误")}') - except Exception as confirm_e: - logger.error(f'[{msg_time}] 【{self.cookie_id}】自动确认发货异常: {self._safe_str(confirm_e)}') + logger.info(f'[{msg_time}] 【{self.cookie_id}】提取到订单ID: {order_id},将在自动发货时处理确认发货') else: logger.warning(f'[{msg_time}] 【{self.cookie_id}】❌ 未能提取到订单ID') except Exception as extract_e: logger.error(f"提取订单ID失败: {self._safe_str(extract_e)}") - # 检查是否可以进行自动发货(防重复) - if not self.can_auto_delivery(item_id): + # 检查是否可以进行自动发货(防重复)- 基于订单ID + if not self.can_auto_delivery(order_id): return # 构造用户URL @@ -2206,12 +2222,12 @@ class XianyuLive: logger.info(f"【{self.cookie_id}】准备自动发货: item_id={item_id}, item_title={item_title}") - # 调用自动发货方法 - delivery_content = await self._auto_delivery(item_id, item_title) + # 调用自动发货方法(包含自动确认发货) + delivery_content = await self._auto_delivery(item_id, item_title, order_id) if delivery_content: - # 标记已发货(防重复) - self.mark_delivery_sent(item_id) + # 标记已发货(防重复)- 基于订单ID + self.mark_delivery_sent(order_id) # 发送发货内容给买家 await self.send_msg(websocket, chat_id, send_user_id, delivery_content) @@ -2300,50 +2316,17 @@ class XianyuLive: except Exception as parse_e: logger.debug(f"解析dynamicOperation JSON失败: {parse_e}") - # 如果成功获取到orderId,进行自动确认发货 + # 订单ID已提取,将在自动发货时进行确认发货处理 if order_id: - # 检查是否启用自动确认发货 - if not self.is_auto_confirm_enabled(): - logger.info(f'[{msg_time}] 【{self.cookie_id}】自动确认发货已关闭,跳过小刀成功订单 {order_id}') - else: - # 检查是否已经确认过这个订单 - current_time = time.time() - if order_id in self.confirmed_orders: - last_confirm_time = self.confirmed_orders[order_id] - if current_time - last_confirm_time < self.order_confirm_cooldown: - logger.info(f'[{msg_time}] 【{self.cookie_id}】⏭️ 订单 {order_id} 已在 {self.order_confirm_cooldown} 秒内确认过,跳过重复确认') - else: - # 超过冷却时间,可以重新确认 - try: - logger.info(f'[{msg_time}] 【{self.cookie_id}】小刀成功,开始自动确认发货,订单ID: {order_id}') - confirm_result = await self.auto_confirm(order_id) - if confirm_result.get('success'): - self.confirmed_orders[order_id] = current_time - logger.info(f'[{msg_time}] 【{self.cookie_id}】🎉 小刀成功,自动确认发货成功!订单ID: {order_id}') - else: - logger.warning(f'[{msg_time}] 【{self.cookie_id}】⚠️ 小刀成功,自动确认发货失败: {confirm_result.get("error", "未知错误")}') - except Exception as confirm_e: - logger.error(f'[{msg_time}] 【{self.cookie_id}】小刀成功,自动确认发货异常: {self._safe_str(confirm_e)}') - else: - # 首次确认这个订单 - try: - logger.info(f'[{msg_time}] 【{self.cookie_id}】小刀成功,开始自动确认发货,订单ID: {order_id}') - confirm_result = await self.auto_confirm(order_id) - if confirm_result.get('success'): - self.confirmed_orders[order_id] = current_time - logger.info(f'[{msg_time}] 【{self.cookie_id}】🎉 小刀成功,自动确认发货成功!订单ID: {order_id}') - else: - logger.warning(f'[{msg_time}] 【{self.cookie_id}】⚠️ 小刀成功,自动确认发货失败: {confirm_result.get("error", "未知错误")}') - except Exception as confirm_e: - logger.error(f'[{msg_time}] 【{self.cookie_id}】小刀成功,自动确认发货异常: {self._safe_str(confirm_e)}') + logger.info(f'[{msg_time}] 【{self.cookie_id}】小刀成功,提取到订单ID: {order_id},将在自动发货时处理确认发货') else: logger.warning(f'[{msg_time}] 【{self.cookie_id}】❌ 小刀成功但未能提取到订单ID') except Exception as extract_e: logger.error(f"提取订单ID失败: {self._safe_str(extract_e)}") - # 检查是否可以进行自动发货(防重复) - if not self.can_auto_delivery(item_id): + # 检查是否可以进行自动发货(防重复)- 基于订单ID + if not self.can_auto_delivery(order_id): return # 构造用户URL @@ -2356,12 +2339,12 @@ class XianyuLive: logger.info(f"【{self.cookie_id}】准备自动发货: item_id={item_id}, item_title={item_title}") - # 调用自动发货方法 - delivery_content = await self._auto_delivery(item_id, item_title) + # 调用自动发货方法(包含自动确认发货) + delivery_content = await self._auto_delivery(item_id, item_title, order_id) if delivery_content: - # 标记已发货(防重复) - self.mark_delivery_sent(item_id) + # 标记已发货(防重复)- 基于订单ID + self.mark_delivery_sent(order_id) # 发送发货内容给买家 await self.send_msg(websocket, chat_id, send_user_id, delivery_content) diff --git a/utils/order_detail_fetcher.py b/utils/order_detail_fetcher.py new file mode 100644 index 0000000..6541f81 --- /dev/null +++ b/utils/order_detail_fetcher.py @@ -0,0 +1,316 @@ +""" +闲鱼订单详情获取工具 +基于Playwright实现订单详情页面访问和数据提取 +""" + +import asyncio +import time +from typing import Optional, Dict, Any +from playwright.async_api import async_playwright, Browser, BrowserContext, Page +from loguru import logger + + +class OrderDetailFetcher: + """闲鱼订单详情获取器""" + + def __init__(self, cookie_string: str = None): + self.browser: Optional[Browser] = None + self.context: Optional[BrowserContext] = None + self.page: Optional[Page] = None + + # 请求头配置 + self.headers = { + "accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.7", + "accept-language": "en,zh-CN;q=0.9,zh;q=0.8,ru;q=0.7", + "cache-control": "no-cache", + "pragma": "no-cache", + "priority": "u=0, i", + "sec-ch-ua": "\"Not)A;Brand\";v=\"8\", \"Chromium\";v=\"138\", \"Google Chrome\";v=\"138\"", + "sec-ch-ua-mobile": "?0", + "sec-ch-ua-platform": "\"Windows\"", + "sec-fetch-dest": "document", + "sec-fetch-mode": "navigate", + "sec-fetch-site": "same-origin", + "sec-fetch-user": "?1", + "upgrade-insecure-requests": "1" + } + + # Cookie配置 - 支持动态传入 + self.cookie = cookie_string + + async def init_browser(self, headless: bool = True): + """初始化浏览器""" + try: + playwright = await async_playwright().start() + + # 启动浏览器 + self.browser = await playwright.chromium.launch( + headless=headless, + args=[ + '--no-sandbox', + '--disable-setuid-sandbox', + '--disable-dev-shm-usage', + '--disable-accelerated-2d-canvas', + '--no-first-run', + '--no-zygote', + '--disable-gpu' + ] + ) + + # 创建浏览器上下文 + self.context = await self.browser.new_context( + viewport={'width': 1920, 'height': 1080}, + user_agent='Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/138.0.0.0 Safari/537.36' + ) + + # 设置额外的HTTP头 + await self.context.set_extra_http_headers(self.headers) + + # 创建页面 + self.page = await self.context.new_page() + + # 设置Cookie + await self._set_cookies() + + logger.info("浏览器初始化成功") + return True + + except Exception as e: + logger.error(f"浏览器初始化失败: {e}") + return False + + async def _set_cookies(self): + """设置Cookie""" + try: + # 解析Cookie字符串 + cookies = [] + for cookie_pair in self.cookie.split('; '): + if '=' in cookie_pair: + name, value = cookie_pair.split('=', 1) + cookies.append({ + 'name': name.strip(), + 'value': value.strip(), + 'domain': '.goofish.com', + 'path': '/' + }) + + # 添加Cookie到上下文 + await self.context.add_cookies(cookies) + logger.info(f"已设置 {len(cookies)} 个Cookie") + + except Exception as e: + logger.error(f"设置Cookie失败: {e}") + + async def fetch_order_detail(self, order_id: str, timeout: int = 30) -> Optional[Dict[str, Any]]: + """ + 获取订单详情 + + Args: + order_id: 订单ID + timeout: 超时时间(秒) + + Returns: + 包含订单详情的字典,失败时返回None + """ + try: + if not self.page: + logger.error("浏览器未初始化") + return None + + # 构建订单详情URL + url = f"https://www.goofish.com/order-detail?orderId={order_id}&role=seller" + logger.info(f"开始访问订单详情页面: {url}") + + # 访问页面 + response = await self.page.goto(url, wait_until='networkidle', timeout=timeout * 1000) + + if not response or response.status != 200: + logger.error(f"页面访问失败,状态码: {response.status if response else 'None'}") + return None + + logger.info("页面加载成功,等待内容渲染...") + + # 等待页面完全加载 + await self.page.wait_for_load_state('networkidle') + + # 额外等待确保动态内容加载完成 + await asyncio.sleep(3) + + # 获取并解析SKU信息 + sku_info = await self._get_sku_content() + + # 获取页面标题 + title = await self.page.title() + + result = { + 'order_id': order_id, + 'url': url, + 'title': title, + 'sku_info': sku_info, # 包含解析后的规格信息 + 'spec_name': sku_info.get('spec_name', '') if sku_info else '', + 'spec_value': sku_info.get('spec_value', '') if sku_info else '', + 'timestamp': time.time() + } + + logger.info(f"订单详情获取成功: {order_id}") + if sku_info: + logger.info(f"规格信息 - 名称: {result['spec_name']}, 值: {result['spec_value']}") + return result + + except Exception as e: + logger.error(f"获取订单详情失败: {e}") + return None + + def _parse_sku_content(self, sku_content: str) -> Dict[str, str]: + """ + 解析SKU内容,根据冒号分割规格名称和规格值 + + Args: + sku_content: 原始SKU内容字符串 + + Returns: + 包含规格名称和规格值的字典,如果解析失败则返回空字典 + """ + try: + if not sku_content or ':' not in sku_content: + logger.warning(f"SKU内容格式无效或不包含冒号: {sku_content}") + return {} + + # 根据冒号分割 + parts = sku_content.split(':', 1) # 只分割第一个冒号 + + if len(parts) == 2: + spec_name = parts[0].strip() + spec_value = parts[1].strip() + + if spec_name and spec_value: + result = { + 'spec_name': spec_name, + 'spec_value': spec_value + } + logger.info(f"SKU解析成功 - 规格名称: {spec_name}, 规格值: {spec_value}") + return result + else: + logger.warning(f"SKU解析失败,规格名称或值为空: 名称='{spec_name}', 值='{spec_value}'") + return {} + else: + logger.warning(f"SKU内容分割失败: {sku_content}") + return {} + + except Exception as e: + logger.error(f"解析SKU内容异常: {e}") + return {} + + async def _get_sku_content(self) -> Optional[Dict[str, str]]: + """获取并解析SKU内容""" + try: + # 等待SKU元素出现 + sku_selector = '.sku--u_ddZval' + + # 检查元素是否存在 + sku_element = await self.page.query_selector(sku_selector) + + if sku_element: + # 获取元素文本内容 + sku_content = await sku_element.text_content() + if sku_content: + sku_content = sku_content.strip() + logger.info(f"找到SKU原始内容: {sku_content}") + print(f"🛍️ SKU原始内容: {sku_content}") + + # 解析SKU内容 + parsed_sku = self._parse_sku_content(sku_content) + if parsed_sku: + print(f"📋 规格名称: {parsed_sku['spec_name']}") + print(f"📝 规格值: {parsed_sku['spec_value']}") + return parsed_sku + else: + logger.warning("SKU内容解析失败") + return {} + else: + logger.warning("SKU元素内容为空") + return {} + else: + logger.warning("未找到SKU元素") + + # 尝试获取页面的所有class包含sku的元素 + all_sku_elements = await self.page.query_selector_all('[class*="sku"]') + if all_sku_elements: + logger.info(f"找到 {len(all_sku_elements)} 个包含'sku'的元素") + for i, element in enumerate(all_sku_elements): + class_name = await element.get_attribute('class') + text_content = await element.text_content() + logger.info(f"SKU元素 {i+1}: class='{class_name}', text='{text_content}'") + + return {} + + except Exception as e: + logger.error(f"获取SKU内容失败: {e}") + return {} + + async def close(self): + """关闭浏览器""" + try: + if self.page: + await self.page.close() + if self.context: + await self.context.close() + if self.browser: + await self.browser.close() + logger.info("浏览器已关闭") + except Exception as e: + logger.error(f"关闭浏览器失败: {e}") + + async def __aenter__(self): + """异步上下文管理器入口""" + await self.init_browser() + return self + + async def __aexit__(self, exc_type, exc_val, exc_tb): + """异步上下文管理器出口""" + await self.close() + + +# 便捷函数 +async def fetch_order_detail_simple(order_id: str, cookie_string: str = None, headless: bool = True) -> Optional[Dict[str, Any]]: + """ + 简单的订单详情获取函数 + + Args: + order_id: 订单ID + cookie_string: Cookie字符串,如果不提供则使用默认值 + headless: 是否无头模式 + + Returns: + 订单详情字典或None + """ + fetcher = OrderDetailFetcher(cookie_string) + try: + if await fetcher.init_browser(headless=headless): + return await fetcher.fetch_order_detail(order_id) + finally: + await fetcher.close() + return None + + +# 测试代码 +if __name__ == "__main__": + async def test(): + # 测试订单ID + test_order_id = "2856024697612814489" + + print(f"🔍 开始获取订单详情: {test_order_id}") + + result = await fetch_order_detail_simple(test_order_id, headless=False) + + if result: + print("✅ 订单详情获取成功:") + print(f"📋 订单ID: {result['order_id']}") + print(f"🌐 URL: {result['url']}") + print(f"📄 页面标题: {result['title']}") + print(f"🛍️ SKU内容: {result['sku_content']}") + else: + print("❌ 订单详情获取失败") + + # 运行测试 + asyncio.run(test())