1. 背景与核心概念在二手电商运营中尤其是像闲鱼这样的平台卖家常常面临一个核心痛点如何高效处理海量的商品咨询与订单尤其是在非工作时间。手动回复“在吗”、“还有货吗”等重复性问题不仅耗时耗力还容易错过潜在订单。因此一个能够实现自动回复、智能对话乃至自动发货的系统成为了提升运营效率、解放卖家时间的利器。本文将围绕“闲鱼自动发货与AI回复系统”的搭建思路展开。请注意本文旨在探讨技术实现的原理、合法合规的自动化方案以及开源工具的应用不涉及任何破解、绕过平台限制或使用未经授权的第三方客户端。所有讨论均基于模拟用户操作如浏览器自动化或利用平台官方接口如果存在且允许的合法合规方式。核心概念解析自动回复指系统能够自动响应买家的消息。这可以分为两个层次关键词回复预设规则如买家消息包含“价格”、“包邮”等关键词则自动回复预设的答案。AI智能回复利用大语言模型如GPT、文心一言、通义千问等理解买家问题的意图生成更自然、更个性化的回复甚至进行多轮对话。自动发货指在买家付款后系统能自动完成发货流程。这通常涉及订单监控定时或实时检查是否有新订单产生。物流信息获取从自己的ERP系统、电子面单平台或预设的物流模板中获取运单号。发货操作将运单号填入闲鱼订单并点击“发货”按钮。系统搭建意味着我们需要一个7x24小时运行的后台服务。它需要部署在服务器上例如云服务器ECS、轻量应用服务器或者性能足够的个人电脑不推荐因网络和电力不稳定。为什么需要了解这些对于个人卖家或小团队手动操作尚可应付。但当商品数量增多、咨询量变大时自动化系统能显著降低人力成本避免因回复不及时导致的客户流失实现“躺赚”式的被动收入流。对于开发者而言这是一个典型的“业务流程自动化RPA”与“AI应用落地”的结合项目具有很高的学习和实践价值。2. 环境准备与版本说明在开始搭建之前我们需要准备好运行环境。本文将采用一种主流且灵活的技术栈PythonSelenium用于Web自动化 FastAPI用于提供API服务 ChatGPT API用于AI回复。你也可以根据熟悉的技术替换相应组件。基础环境操作系统Ubuntu 20.04/22.04 LTS推荐用于生产环境Windows 10/11 或 macOS可用于开发测试。Python版本 3.8 或以上。本文示例使用 Python 3.9。版本管理工具Git可选用于代码管理。服务器一台具有公网IP的云服务器如阿里云ECS、腾讯云CVM。配置建议1核2G以上用于长期运行服务。核心依赖库我们将创建一个requirements.txt文件来管理依赖。# requirements.txt # Web 自动化与浏览器驱动 selenium4.15.0 webdriver-manager4.0.1 # Web 框架用于构建监控API或管理后台 fastapi0.104.1 uvicorn[standard]0.24.0 # HTTP 请求库用于调用AI接口或其它API httpx0.25.1 # 环境变量管理 python-dotenv1.0.0 # 定时任务调度 apscheduler3.10.4 # 数据处理 pandas2.1.3 # 可选用于处理订单数据 # AI 接口调用 (以OpenAI为例) openai1.3.0重要说明浏览器与驱动Selenium 需要控制真实的浏览器如Chrome。在服务器上我们需要安装无头Headless模式的Chrome。AI模型选择除了OpenAI你可以替换为国内可访问的模型如百度千帆文心一言、阿里灵积通义千问、智谱AI等只需更换对应的API调用代码。环境变量敏感信息如API密钥、账号密码等务必通过.env文件管理不要硬编码在代码中。3. 系统架构与原理拆解一个完整的系统可以分为以下几个核心模块它们协同工作实现从消息监听、智能处理到订单履行的闭环。3.1 整体架构图文字描述1. 触发层 - 定时任务每隔X分钟检查一次闲鱼PC端网页或模拟App端的消息和订单。 - 消息队列理想情况下可通过逆向工程监听WebSocket复杂度高风险大本文不采用。我们主要使用定时轮询。 2. 数据获取层 - Selenium自动化模拟登录闲鱼PC网页定位消息列表和订单列表元素获取未读消息和新订单数据。 - API调用如果未来开放直接调用闲鱼官方API目前对普通卖家限制严格。 3. 智能处理层 - 规则引擎对消息进行关键词匹配触发固定回复。 - AI引擎将消息内容发送给大语言模型API获取生成的回复文本。需要设计提示词Prompt来约束AI的回复风格和内容。 4. 执行层 - 消息回复通过Selenium定位对话框输入AI生成的回复文本并点击发送。 - 订单发货通过Selenium在订单详情页填入从其他系统获取的物流单号点击发货。 5. 支撑层 - 配置管理管理商品、回复话术、AI参数等。 - 日志与监控记录所有操作和异常便于排查问题。 - 任务调度管理定时任务的启停和周期。3.2 关键技术原理1. Selenium Web自动化原理Selenium通过WebDriver协议与真实浏览器如ChromeDriver通信。WebDriver直接控制浏览器执行点击、输入、跳转等操作并能获取页面DOM元素的内容。这模拟了真实用户的所有操作。优点完全模拟真人不易被简单的反爬机制拦截。缺点速度慢资源消耗大浏览器环境需要维护。2. 大语言模型LLM提示工程为了让AI回复更符合“闲鱼卖家”的身份我们需要精心设计提示词Prompt。例如你是一个专业的闲鱼二手商品卖家性格热情、耐心、简洁。请根据用户关于商品【{商品标题}】的提问进行友好、专业的回复。只回复对话内容本身不要添加任何解释。 商品信息{商品详细信息} 用户问题{用户消息} 历史对话{最近3条对话历史}通过这样的Prompt可以引导AI生成更贴合场景的回复。3. 状态管理与防重复系统需要记录哪些消息已回复、哪些订单已处理避免重复操作。通常会在本地数据库如SQLite或文件中记录一个处理状态的日志。4. 核心模块代码实现接下来我们将分步骤实现核心模块。请注意以下代码为示例原型在生产环境中需要添加完善的错误处理、日志记录和容错机制。4.1 项目结构创建首先创建项目目录结构。mkdir xianyu-auto-system cd xianyu-auto-system touch main.py config.py ai_replier.py msg_monitor.py order_processor.py requirements.txt .env.example mkdir logs4.2 环境配置与浏览器初始化config.py用于集中管理配置。# config.py import os from dotenv import load_dotenv from selenium import webdriver from selenium.webdriver.chrome.options import Options from selenium.webdriver.chrome.service import Service from webdriver_manager.chrome import ChromeDriverManager load_dotenv() # 加载 .env 文件中的环境变量 class Config: # 闲鱼账号强烈建议使用小号测试 XIANYU_USERNAME os.getenv(XIANYU_USERNAME) XIANYU_PASSWORD os.getenv(XIANYU_PASSWORD) # AI 配置 (以OpenAI为例) OPENAI_API_KEY os.getenv(OPENAI_API_KEY) OPENAI_BASE_URL os.getenv(OPENAI_BASE_URL, https://api.openai.com/v1) AI_MODEL os.getenv(AI_MODEL, gpt-3.5-turbo) # 系统配置 CHECK_INTERVAL int(os.getenv(CHECK_INTERVAL, 300)) # 检查间隔默认5分钟 HEADLESS os.getenv(HEADLESS, True).lower() true # 是否使用无头模式 # 日志路径 LOG_FILE logs/xianyu_auto.log staticmethod def init_driver(): 初始化Chrome WebDriver chrome_options Options() if Config.HEADLESS: chrome_options.add_argument(--headless) # 无头模式服务器运行必备 chrome_options.add_argument(--no-sandbox) chrome_options.add_argument(--disable-dev-shm-usage) chrome_options.add_argument(--disable-gpu) chrome_options.add_argument(--window-size1920,1080) # 添加其他参数以避免被检测为自动化工具部分 chrome_options.add_experimental_option(excludeSwitches, [enable-automation]) chrome_options.add_experimental_option(useAutomationExtension, False) chrome_options.add_argument(--disable-blink-featuresAutomationControlled) # 使用 webdriver-manager 自动管理驱动版本 service Service(ChromeDriverManager().install()) driver webdriver.Chrome(serviceservice, optionschrome_options) # 执行CDP命令隐藏WebDriver特征 driver.execute_cdp_cmd(Page.addScriptToEvaluateOnNewDocument, { source: Object.defineProperty(navigator, webdriver, { get: () undefined }); }) return driver.env.example文件使用时复制为.env并填写真实信息。# .env.example XIANYU_USERNAMEyour_username XIANYU_PASSWORDyour_password OPENAI_API_KEYsk-your_openai_api_key_here CHECK_INTERVAL300 HEADLESSTrue4.3 消息监控与获取模块msg_monitor.py负责登录闲鱼并获取未读消息。# msg_monitor.py import time from selenium.webdriver.common.by import By from selenium.webdriver.support.ui import WebDriverWait from selenium.webdriver.support import expected_conditions as EC from config import Config import logging logging.basicConfig(levellogging.INFO, format%(asctime)s - %(name)s - %(levelname)s - %(message)s) logger logging.getLogger(__name__) class MessageMonitor: def __init__(self, driver): self.driver driver self.wait WebDriverWait(driver, 20) def login(self): 模拟登录闲鱼PC端此处为示例实际登录逻辑可能因页面改版失效 logger.info(正在尝试登录闲鱼...) self.driver.get(https://login.taobao.com/) # 淘宝统一登录入口 time.sleep(3) # 等待并点击密码登录示例定位实际需根据页面调整 try: pwd_login_btn self.wait.until( EC.element_to_be_clickable((By.XPATH, //a[contains(text(), 密码登录)])) ) pwd_login_btn.click() time.sleep(1) except Exception as e: logger.warning(未找到密码登录标签页可能已默认显示。) # 输入用户名和密码 username_input self.wait.until( EC.presence_of_element_located((By.ID, fm-login-id)) ) password_input self.driver.find_element(By.ID, fm-login-password) username_input.clear() username_input.send_keys(Config.XIANYU_USERNAME) time.sleep(0.5) password_input.send_keys(Config.XIANYU_PASSWORD) time.sleep(1) # 处理滑块验证这是一个难点可能需要更复杂的识别方案此处仅示意 # 在实际项目中可能需要接入打码平台或使用图像识别库 logger.warning(请注意如需滑块验证请手动完成。自动化处理滑块验证涉及复杂识别本文不展开。) time.sleep(10) # 留出时间手动操作 # 点击登录按钮 login_btn self.driver.find_element(By.XPATH, //button[typesubmit]) login_btn.click() time.sleep(5) # 登录后跳转到闲鱼消息页 self.driver.get(https://message.taobao.com/) # 淘宝消息中心闲鱼消息在此 time.sleep(5) logger.info(登录流程执行完毕。) def fetch_unread_messages(self): 获取未读消息列表示例逻辑需根据实际页面结构调整 messages [] try: # 假设未读消息有一个特定的CSS类例如 ‘unread’ # 实际中需要分析闲鱼消息页面的HTML结构 unread_elements self.driver.find_elements(By.CSS_SELECTOR, .message-item.unread) for elem in unread_elements[:5]: # 每次最多处理5条避免超时 try: sender elem.find_element(By.CSS_SELECTOR, .sender-name).text preview elem.find_element(By.CSS_SELECTOR, .message-preview).text link elem.find_element(By.CSS_SELECTOR, a).get_attribute(href) messages.append({ sender: sender, preview: preview, link: link, element: elem # 保存元素方便后续点击 }) except Exception as e: logger.error(f解析单个消息元素失败: {e}) continue except Exception as e: logger.error(f获取未读消息列表失败: {e}) return messages4.4 AI智能回复模块ai_replier.py负责调用大模型生成回复。# ai_replier.py import openai from config import Config import logging import json logger logging.getLogger(__name__) # 初始化OpenAI客户端如果使用其他平台替换此部分 client openai.OpenAI( api_keyConfig.OPENAI_API_KEY, base_urlConfig.OPENAI_BASE_URL ) class AIReplier: def __init__(self, product_infoNone): self.product_info product_info or 这是一个二手商品功能完好详情见描述。 def generate_reply(self, user_message, conversation_historyNone): 调用AI生成回复 :param user_message: 用户发送的消息 :param conversation_history: 最近的对话历史列表格式 [{role:user,content:...}, {role:assistant,content:...}] :return: AI生成的回复文本 prompt f 你是一个闲鱼二手商品卖家请用热情、简洁、专业的口吻回复买家。 商品信息{self.product_info} 买家提问{user_message} 请直接给出回复内容不要添加任何额外的解释或标记。 messages [{role: system, content: prompt}] if conversation_history: # 将历史对话并入上下文注意长度限制 messages.extend(conversation_history[-6:]) # 只保留最近3轮对话 try: response client.chat.completions.create( modelConfig.AI_MODEL, messagesmessages, max_tokens150, temperature0.7, # 控制创造性0.7较为平衡 ) reply response.choices[0].message.content.strip() logger.info(fAI生成回复成功。问题『{user_message}』 - 回复『{reply}』) return reply except openai.APIError as e: logger.error(f调用AI API失败: {e}) # 降级策略返回一个预设的通用回复 return 您好我现在暂时无法处理复杂问题。您可以查看商品描述或稍后联系我。 except Exception as e: logger.error(f生成回复时发生未知错误: {e}) return 收到您的消息稍后回复您。4.5 订单处理模块order_processor.py负责监控并处理新订单。# order_processor.py import time from selenium.webdriver.common.by import By from selenium.webdriver.support.ui import WebDriverWait from selenium.webdriver.support import expected_conditions as EC import logging logger logging.getLogger(__name__) class OrderProcessor: def __init__(self, driver): self.driver driver self.wait WebDriverWait(driver, 20) def check_new_orders(self): 检查是否有待发货的新订单示例逻辑 new_orders [] try: # 导航到闲鱼“我卖出的”页面 self.driver.get(https://trade.taobao.com/trade/itemlist/list_sold_items.htm) time.sleep(5) # 寻找状态为“等待卖家发货”的订单 # 实际XPath需要根据闲鱼页面实时调整 order_elements self.driver.find_elements(By.XPATH, //div[contains(class, order-item) and .//span[contains(text(), 等待卖家发货)]]) for elem in order_elements[:5]: # 每次处理最多5个订单 try: order_id elem.find_element(By.CSS_SELECTOR, .order-id).text buyer_name elem.find_element(By.CSS_SELECTOR, .buyer-name).text product_name elem.find_element(By.CSS_SELECTOR, .product-name).text new_orders.append({ order_id: order_id, buyer: buyer_name, product: product_name, element: elem }) except Exception as e: logger.error(f解析订单元素失败: {e}) continue except Exception as e: logger.error(f检查新订单失败: {e}) return new_orders def ship_order(self, order_info, tracking_number): 执行发货操作 :param order_info: 订单信息字典 :param tracking_number: 物流单号 try: # 点击进入订单详情页 order_info[element].click() time.sleep(3) # 切换到新打开的标签页 window_handles self.driver.window_handles if len(window_handles) 1: self.driver.switch_to.window(window_handles[-1]) time.sleep(2) # 点击“发货”按钮 ship_btn self.wait.until( EC.element_to_be_clickable((By.XPATH, //button[contains(text(), 发货)])) ) ship_btn.click() time.sleep(2) # 选择物流方式并填入单号这里以“其他物流”为例 # 实际页面可能有多步操作 other_logistics self.wait.until( EC.element_to_be_clickable((By.XPATH, //div[text()其他物流])) ) other_logistics.click() time.sleep(1) tracking_input self.wait.until( EC.presence_of_element_located((By.XPATH, //input[placeholder请输入物流单号])) ) tracking_input.clear() tracking_input.send_keys(tracking_number) time.sleep(1) # 确认发货 confirm_btn self.driver.find_element(By.XPATH, //button[contains(text(), 确认发货)]) confirm_btn.click() time.sleep(3) logger.info(f订单 {order_info[order_id]} 发货成功物流单号{tracking_number}) # 关闭详情页标签页切回主窗口 self.driver.close() self.driver.switch_to.window(window_handles[0]) except Exception as e: logger.error(f订单 {order_info.get(order_id)} 发货失败: {e}) # 错误处理截图保存 self.driver.save_screenshot(flogs/ship_error_{order_info.get(order_id, unknown)}.png)4.6 主调度程序main.py将以上模块串联起来形成定时任务。# main.py import time import schedule from datetime import datetime from config import Config from msg_monitor import MessageMonitor from ai_replier import AIReplier from order_processor import OrderProcessor import logging logging.basicConfig( levellogging.INFO, format%(asctime)s - %(name)s - %(levelname)s - %(message)s, handlers[ logging.FileHandler(Config.LOG_FILE), logging.StreamHandler() ] ) logger logging.getLogger(__name__) # 全局状态记录防止重复处理 processed_messages set() processed_orders set() def job(): 定时执行的任务 logger.info( 开始执行定时检查任务 ) driver None try: driver Config.init_driver() monitor MessageMonitor(driver) processor OrderProcessor(driver) ai AIReplier(product_info【示例商品】九成新无线蓝牙耳机音质好续航强。) # 1. 登录仅在首次或会话过期时执行这里简化处理 # 实际应增加登录状态检查 monitor.login() # 2. 检查并回复消息 unread_msgs monitor.fetch_unread_messages() for msg in unread_msgs: msg_key f{msg[sender]}_{msg[preview][:20]} if msg_key in processed_messages: continue # 点击进入聊天窗口简化示意 # driver.get(msg[link]) # time.sleep(3) # 实际需要定位输入框并发送消息 # reply_text ai.generate_reply(msg[preview]) # send_reply(driver, reply_text) # 需要实现send_reply函数 logger.info(f待回复消息来自: {msg[sender]}, 内容预览: {msg[preview]}) processed_messages.add(msg_key) time.sleep(2) # 避免操作过快 # 3. 检查并处理订单 new_orders processor.check_new_orders() for order in new_orders: if order[order_id] in processed_orders: continue # 假设从其他系统获取了物流单号这里用模拟数据 fake_tracking_number fYT{int(time.time())}CN # processor.ship_order(order, fake_tracking_number) # 实际发货测试时注释掉 logger.info(f检测到待发货订单: {order[order_id]} - {order[product]}。模拟发货单号: {fake_tracking_number}) processed_orders.add(order[order_id]) time.sleep(3) except Exception as e: logger.error(f定时任务执行过程中发生严重错误: {e}, exc_infoTrue) finally: if driver: driver.quit() logger.info( 定时检查任务结束 \n) if __name__ __main__: logger.info(闲鱼自动发货回复系统启动...) # 立即执行一次 job() # 每隔 CHECK_INTERVAL 秒执行一次 schedule.every(Config.CHECK_INTERVAL).seconds.do(job) while True: schedule.run_pending() time.sleep(1)5. 部署与运行5.1 服务器环境准备以Ubuntu为例安装Python3和pip:sudo apt update sudo apt install python3-pip python3-venv -y安装Chrome浏览器无头模式:wget -q -O - https://dl-ssl.google.com/linux/linux_signing_key.pub | sudo apt-key add - sudo sh -c echo deb [archamd64] http://dl.google.com/linux/chrome/deb/ stable main /etc/apt/sources.list.d/google.list sudo apt update sudo apt install google-chrome-stable -y # 检查版本 google-chrome --version克隆或上传代码到服务器:git clone 你的代码仓库地址 # 或使用scp上传 cd xianyu-auto-system创建虚拟环境并安装依赖:python3 -m venv venv source venv/bin/activate pip install -r requirements.txt配置环境变量:cp .env.example .env nano .env # 使用vim或nano编辑填入真实的账号和API密钥5.2 运行与测试首次运行测试非无头模式便于调试: 在.env中设置HEADLESSFalse然后在服务器上运行python main.py此时会弹出图形界面的Chrome需要服务器有桌面环境或使用VNC。观察登录、消息获取等流程是否正常。请务必准备好手动处理滑块验证。生产环境运行无头模式: 在.env中设置HEADLESSTrue。使用nohup或systemd让程序在后台运行。nohup python main.py output.log 21 查看日志tail -f logs/xianyu_auto.log6. 常见问题与排查思路问题现象可能原因排查思路与解决方案Selenium无法启动Chrome1. Chrome未安装。2. ChromeDriver版本不匹配。3. 服务器内存不足。1. 运行google-chrome --version确认安装。2.webdriver-manager通常能自动匹配检查日志。3. 使用free -h检查内存考虑升级配置。登录时出现滑块验证无法通过闲鱼/淘宝的风控机制检测到自动化行为。1. 这是最大难点。可尝试增加随机延迟、模拟人类移动鼠标轨迹、使用更稳定的Cookie登录需定期更新。2. 考虑接入第三方打码平台进行识别。3.终极方案研究是否有合法的官方API或合作渠道。获取不到消息或订单列表页面HTML结构发生变化。1. 使用浏览器开发者工具重新分析页面元素更新代码中的CSS选择器或XPath。2. 增加等待时间time.sleep()或使用更智能的WebDriverWait。3. 在关键步骤后截图 (driver.save_screenshot(debug.png)) 辅助分析。AI回复内容不相关或违规Prompt设计不佳或AI模型理解偏差。1. 优化Prompt更明确地限定身份、场景和回复格式。2. 在Prompt中加入禁止项如“不要透露价格”、“不要使用营销话术”。3. 对AI回复内容进行二次过滤和检查。重复回复或重复发货状态记录 (processed_messages,processed_orders) 未持久化服务重启后丢失。1. 将已处理状态存入数据库如SQLite或文件每次启动时加载。2. 使用更精确的唯一标识如“消息ID”而非“发送者预览”。程序运行一段时间后崩溃1. 内存/资源泄漏。2. 网络波动。3. 浏览器会话过期。1. 确保在finally块中调用driver.quit()。2. 添加全面的异常捕获和重试机制。3. 实现会话心跳或定期重新登录。服务器IP被闲鱼封禁同一IP高频访问。1. 大幅增加检查间隔如30分钟一次。2. 使用代理IP池需谨慎确保合规。3.最重要的控制操作频率模拟真实用户行为。7. 最佳实践与工程建议合规与风险第一严格遵守平台规则频繁、机械的自动化操作可能违反闲鱼用户协议存在账号被封风险。务必使用备用小号进行测试和运营。控制操作频率将检查间隔 (CHECK_INTERVAL) 设置得足够长如5-10分钟避免秒级请求。人工兜底AI回复无法处理复杂议价、纠纷等问题。系统应标记这类对话提醒人工介入。代码健壮性异常处理每个网络请求、页面操作、API调用都必须有try...except并记录详细日志。重试机制对于临时性失败如网络超时应实现指数退避的重试逻辑。状态持久化使用轻量级数据库SQLite记录任务状态、消息日志、订单日志便于排查和恢复。可维护性配置外化所有变量如URL、选择器、时间间隔都应放在配置文件或环境变量中。模块化设计如本文所示将登录、监控、AI、订单处理分离方便单独调试和替换。日志分级使用logging模块区分INFO、WARNING、ERROR级别并输出到文件。性能与扩展单浏览器实例避免为每个任务创建新浏览器应复用同一个Driver实例注意会话管理。异步处理考虑使用asyncio和aiohttp来处理多个AI请求提升并发能力。任务队列如果商品和订单量极大可引入Redis或RabbitMQ作为任务队列实现分布式处理。安全密钥管理绝不在代码中硬编码密码和API Key。使用.env文件并在生产环境使用密钥管理服务。权限最小化运行程序的系统用户应具有最小必要权限。输入检查对AI返回的内容进行敏感词过滤避免自动发出违规信息。8. 总结与展望通过本文的拆解我们实现了一个基于Python Selenium和大型语言模型的闲鱼自动化系统原型。它涵盖了从环境搭建、原理分析、模块编码到部署运行的全流程。核心价值在于提供了一套合法合规前提下提升电商运营效率的技术思路和实现参考。关键点回顾自动化基础Selenium WebDriver 是模拟网页操作的核心工具。智能核心通过设计良好的Prompt调用大模型API可以实现拟人化的自动回复。系统集成将消息监控、智能回复、订单处理通过定时任务串联形成一个闭环系统。风险控制必须高度重视频率控制、异常处理和平台规则这是项目能否长期运行的关键。下一步可以探索的方向多渠道集成将这套逻辑适配到其他电商平台或社交软件。知识库增强为AI回复接入商品知识库使其回答更精准。可视化后台使用Flask或Django开发一个Web管理后台方便配置话术、查看日志、管理任务。RPA平台考虑使用更专业的RPA平台如影刀、UiPath来构建流程可能更稳定且易于维护。自动化工具是一把双刃剑它能极大提升效率但也需在平台规则和用户体验的边界内谨慎使用。希望本文提供的技术方案能为你打开思路在实际应用中请务必以测试为先小步迭代平稳运行。