智能沟通与询价管理 (旺旺交互架构设计)1. 核心漏斗过滤流程图Plaintext[ 输入源 ] 最终采购清单 (BOM) + 初始意向商家池 (来自模块二)
    │
    ▼
┌─────────────────────────────────────────────────────────┐
│              多轮次自动询价引擎 (RPA + LLM)             │
└─────────────────────────────────────────────────────────┘
    │
====│================ 轮次 1：货源基础摸底 ================│====
    │
    ├─> 发送话术: "老板，请问【商品标题】目前有现货吗？"
    │
    ├─> [ LLM 意图识别 ] ──(分析客服回复)──┐
    │                                  │
    ▼                                  ▼
【状态: R1_PASSED】                  【状态: R1_REJECTED】
 (识别结果: 有货)                     (识别结果: 无货/停产/下架)
 (保留，加入下一轮队列)                 (打上剔除标记，流程终止)
    │
====│================ 轮次 2：规格与起订量确认 ============│====
    │
    ├─> 发送话术: "我们需要【材质/规格】，起订量【MOQ】件能做吗？"
    │
    ├─> [ LLM 意图识别 ] ──(分析客服回复)──┐
    │                                  │
    ▼                                  ▼
【状态: R2_PASSED】                  【状态: R2_REJECTED】
 (识别结果: 支持该规格且满足起订)       (识别结果: 无法定制/量太少不接)
 (保留，加入下一轮队列)                 (打上剔除标记，流程终止)
    │
====│================ 轮次 3：核心商务谈判(价格与交期) ====│====
    │
    ├─> 发送话术: "如果按照这个数量，最低什么价格？交期大概要多久？"
    │
    ├─> [ LLM 实体提取 ] ──(解析回复并格式化)──┐
    │                                      │
    ▼                                      ▼
【状态: R3_PASSED】                      【状态: R3_PENDING】
 (提取到: 单价、总价、发货天数)             (识别结果: 需核算/含糊其辞)
 (将数据存入数据库，保留)                   (转入人工接管或进入定时重试队列)
    │
====│================ 轮次 4：合规与资质核验 ==============│====
    │
    ├─> 发送话术: "产品符合最新国标吗？能开增值税专用发票吗？"
    │
    ├─> [ LLM 意图识别 ] ──(分析客服回复)──┐
    │                                  │
    ▼                                  ▼
【状态: R4_PASSED】                  【状态: R4_WARNING / REJECTED】
 (识别结果: 合规且可开票)               (识别结果: 不合规/拒开票)
 (最终入库，输出至模块四)               (根据设置决定是降级备选还是直接剔除)
    │
    ▼
[ 输出源 ] 最终有效报价池 (模块四：内部比价看板的数据源)

核心模块
1、对话调度器 (Conversation Scheduler): 负责维护每一个商品-商家对话对的状态机制。它决定当前属于哪一轮，应该提取 BOM 表中的哪个字段，并调用对应的话术模板。

2、RPA 执行器 (Playwright Worker): 专门负责浏览器底层的自动化。它的任务极简：接收调度器传来的目标 URL 和文本 -> 在聊天框输入并发送 -> 抓取最新的聊天气泡文本返回。

3、AI 意图解析网关 (LLM Intent Gateway): 接收 RPA 抓取回来的客服回复，通过预设的 System Prompt 进行二分类或多分类判定（例如输出 YES, NO, UNCLEAR），或者执行信息提取（输出 JSON 格式的价格和交期）。

4、防风控休眠队列 (Anti-Ban Queue): 两轮对话之间不能秒回。必须引入随机休眠机制（如 10 - 45 秒的随机延迟），如果客服久久不回，需要将任务挂起，释放浏览器资源去处理下一个商家。

5、数据库状态机设计 (Supabase 表结构建议)为支撑多轮次流转，customer_service_chats 表需要增加状态控制字段。字段名类型说明idUUID唯一主键intent_idUUID关联的“采购意向单” IDmerchant_idString1688 商家/店铺 IDcurrent_roundInteger当前轮次 (1, 2, 3, 4)statusString状态枚举：WAITING_REPLY (等回复), R1_PASSED (首轮通过), REJECTED (被剔除), COMPLETED (全流程完成)extracted_dataJSONB存储提取到的结构化数据：{"price": 10.5, "delivery_days": 3, "is_guobiao": true}last_interacted_atTimestamp上次发送消息或收到回复的时间，用于控制超时重试这种漏斗式轮询设计非常依赖对客服回复时效性的处理。

问题处理：
问题1、在实际执行中，很多 1688 客服可能需要几个小时甚至半天才会回复第一轮的“有没有现货”。
方法：采用**“消息队列定时巡检”（Polling + Message Queue）**
1. 核心机制：指数退避轮询 (Exponential Backoff Polling)商家回复的时间是不确定的，可能秒回，也可能半天后回。因此，巡检绝不能是固定频率（比如每 5 分钟），而应该采用**“指数退避”**策略，越往后检查的间隔越长，以节省宝贵的浏览器资源。第 1 次巡检： 发送消息后 3 分钟（防秒回遗漏）。第 2 次巡检： 距离上次 15 分钟。第 3 次巡检： 距离上次 1 小时。从第 4 次巡检开始：每 1 小时巡检 1 次。超时判定： 超过 48 小时未回复，标记为 TIMEOUT_REJECTED，自动剔除。2. 数据库扩展 (Supabase 表结构升级)为了支持定时巡检，我们需要在之前的 customer_service_chats 表中增加几个关键的调度控制字段：新增字段名类型说明next_check_timeTimestamp核心！ 下一次允许 Playwright 去打开这个聊天框检查的时间。retry_countInteger当前轮次已轮询的次数（初始为 0），用于计算“指数退避”的间隔。last_message_hashString上次抓取到的最后一条消息的哈希值（或消息 ID）。用于对比是否有“新回复”。

3. 巡检 Worker 的标准工作流你需要编写一个单独的 Node.js 脚本（例如 cron_worker.ts），并使用 node-cron 或 PM2 定时执行（例如每 2 分钟运行一次）。

核心代码逻辑：
你需要安装以下 Python 库：
pip install supabase playwright playwright-stealth
playwright install chromium

Python 巡检 Worker 核心代码 (cron_worker.py)
import asyncio
import hashlib
from datetime import datetime, timezone, timedelta
from supabase import create_client, Client
from playwright.async_api import async_playwright, Page
from playwright_stealth import stealth_async

# ================= 配置区 =================
SUPABASE_URL = "your_supabase_url"
SUPABASE_KEY = "your_supabase_service_role_key"
MAX_RETRIES = 10  # 最大轮询次数

# 初始化 Supabase 客户端
supabase: Client = create_client(SUPABASE_URL, SUPABASE_KEY)

# ================= 辅助函数 =================

def calculate_next_check_time(retry_count: int) -> str:
    """计算指数退避的下一次检查时间"""
    # 基础等待 3 分钟，每次重试等待时间倍增，例如：3, 6, 12, 24, 48 分钟...
    wait_minutes = 3 * (2 ** retry_count)
    next_time = datetime.now(timezone.utc) + timedelta(minutes=wait_minutes)
    return next_time.isoformat()

def get_msg_hash(text: str) -> str:
    """计算消息哈希值"""
    return hashlib.md5(text.encode('utf-8')).hexdigest()

async def scrape_latest_message(page: Page) -> dict:
    """
    抓取最新消息（需根据 1688 实际 DOM 结构替换选择器）
    假设返回字典格式：{"text": "老板有货", "sender": "merchant", "hash": "..."}
    """
    # 伪代码：等待消息列表加载
    await page.wait_for_selector('.chat-message-list', timeout=10000)
    
    # 伪代码：获取最后一条消息的文本和发送者属性
    # 这里需要你根据实际页面的 class 或 attribute 来手写提取逻辑
    latest_text = "模拟抓取到的最后一条消息文本" 
    sender = "merchant" # 假设是商家发来的
    
    return {
        "text": latest_text,
        "sender": sender,
        "hash": get_msg_hash(latest_text)
    }

async def check_intent_with_llm(text: str, current_round: int) -> dict:
    """调用大模型解析意图 (此处为伪代码)"""
    # 这里接入你喜欢的 LLM API (如 OpenAI, 智谱, DeepSeek 等)
    print(f"请求 LLM 分析消息: {text}, 当前轮次: {current_round}")
    # 假设 LLM 返回通过
    return {"passed": True, "reason": "有现货"}

# ================= 核心工作流 =================

async def process_task(task: dict, page: Page):
    task_id = task['id']
    merchant_id = task['merchant_id']
    current_round = task['current_round']
    
    print(f"开始处理任务 [{task_id}], 商家: {merchant_id}")
    
    try:
        # 1. 打开聊天窗口 (加载预先存好的 Cookie state)
        await page.goto(f"https://wangwang.1688.com/.../chat?merchantId={merchant_id}")
        await page.wait_for_load_state("networkidle")
        
        # 2. 抓取最新消息
        latest_message = await scrape_latest_message(page)
        
        # 3. 判断是否有新回复
        if latest_message['hash'] != task.get('last_message_hash') and latest_message['sender'] == 'merchant':
            print(">>> 发现商家新回复！")
            
            # 4. LLM 意图识别
            llm_result = await check_intent_with_llm(latest_message['text'], current_round)
            
            if llm_result['passed']:
                next_round = current_round + 1
                next_msg_text = f"这是第 {next_round} 轮的话术..." # 应该从预设话术库获取
                
                # 发送下一轮消息
                await page.fill('#chat-input', next_msg_text)
                await page.click('#send-btn')
                
                # 存入 Supabase，进入下一轮
                supabase.table('customer_service_chats').update({
                    "current_round": next_round,
                    "retry_count": 0,
                    "next_check_time": calculate_next_check_time(0),
                    "last_message_hash": get_msg_hash(next_msg_text) # 更新为自己发出的最后一条消息的 hash
                }).eq('id', task_id).execute()
                print(f"任务 [{task_id}] 已推进至第 {next_round} 轮。")
                
            else:
                # 不符合条件，拒绝/剔除
                supabase.table('customer_service_chats').update({
                    "status": "REJECTED",
                    "extracted_data": {"reject_reason": llm_result['reason']}
                }).eq('id', task_id).execute()
                print(f"任务 [{task_id}] 不符合条件，已剔除。")
                
        else:
            print(">>> 商家暂未回复新消息。")
            new_retry_count = task['retry_count'] + 1
            
            if new_retry_count > MAX_RETRIES:
                # 超过最大重试次数，标记超时
                supabase.table('customer_service_chats').update({
                    "status": "TIMEOUT_REJECTED"
                }).eq('id', task_id).execute()
                print(f"任务 [{task_id}] 超时未回复，已放弃。")
            else:
                # 指数退避，推迟下一次检查时间
                supabase.table('customer_service_chats').update({
                    "retry_count": new_retry_count,
                    "next_check_time": calculate_next_check_time(new_retry_count)
                }).eq('id', task_id).execute()
                print(f"任务 [{task_id}] 推迟至下一次检查。重试次数: {new_retry_count}")

    except Exception as e:
        print(f"处理任务 [{task_id}] 时发生错误: {e}")

async def main_worker_loop():
    print("🤖 1688 自动巡检 Worker 启动...")
    
    async with async_playwright() as p:
        # 启动浏览器 (带上防风控参数和你的用户数据目录以保持登录状态)
        # 注意: Mac mini M4 上的 Chrome 路径可能不同，可以不指定 executable_path 让 PW 自动下载
        browser = await p.chromium.launch_persistent_context(
            user_data_dir="./user_data", 
            headless=False, # 建议初期调试保持 False，稳定后可在服务器上改为 True
            args=["--disable-blink-features=AutomationControlled"]
        )
        
        page = await browser.new_page()
        # 注入 stealth 插件绕过检测
        await stealth_async(page)
        
        while True:
            now_iso = datetime.now(timezone.utc).isoformat()
            
            # 从数据库拉取到期需要检查的任务
            response = supabase.table('customer_service_chats')\
                .select('*')\
                .eq('status', 'WAITING_REPLY')\
                .lte('next_check_time', now_iso)\
                .order('next_check_time')\
                .limit(5)\
                .execute()
                
            tasks = response.data
            
            if not tasks:
                print(f"[{datetime.now().strftime('%H:%M:%S')}] 队列为空，休眠 30 秒...")
                await asyncio.sleep(30)
                continue
                
            print(f"[{datetime.now().strftime('%H:%M:%S')}] 抓取到 {len(tasks)} 个待巡检任务。")
            
            for task in tasks:
                await process_task(task, page)
                # 两个任务之间随机休息一下，模拟真人换网页
                await asyncio.sleep(2)
            
            # 一轮跑完，稍微休息再进入下一轮循环
            await asyncio.sleep(10)

if __name__ == "__main__":
    # 运行异步主循环
    asyncio.run(main_worker_loop())