咸鱼自动收货源码解析：从订单轮询到自动确认的Python实现

简介这份源码面向闲鱼卖家及二手交易平台开发者用于搭建自动收货系统解决手动确认收货效率低、交易周期长的问题。其核心逻辑是生成运费保价页面买家确认开通并支付运费后系统自动触发发货与收货流程后台可自由配置价格、规则和条件适配不同交易场景。资源包共133个文件以73个PHP脚本为主体辅以JSON配置、Markdown说明、CSS样式、JS脚本及图片素材压缩包约1.21MB运行环境为PHP 7.4无需数据库部署门槛较低。后台入口为域名/admin默认账号admin、密码123456便于快速上手调试。目前已有1374人学习下载适合具备基础PHP部署能力、希望提升闲鱼交易自动化水平的卖家或开发者参考可借此理解自动收货流程的触发机制与后台规则配置思路。1. 从手动点收货到脚本自动跑这套源码到底解决什么问题做二手交易的人都有一个共同的痛买家确认收货这件事平台不会替你催卖家只能一遍遍刷订单列表。单量少的时候无所谓一旦一天几十单光点“确认收货”就能点到手指发麻。这套咸鱼自动收货源码要解决的就是把这个重复动作交给程序去跑。它面向的是手里有批量订单、需要定时巡检并自动完成收货确认的开发者或小团队不是给普通买家用的傻瓜工具。源码本身是一套可独立部署的自动化脚本核心逻辑围绕订单轮询、状态判断、收货请求模拟三个环节展开。你拿到手之后需要自己配置账号凭证、调整轮询间隔、对接通知渠道才能跑成一条完整的流水线。适合谁适合能看懂 Python 或 JavaScript、愿意花半小时配环境、并且对平台接口有一定认知的开发者。如果你连命令行都没碰过这套东西会让你在第一步就卡住。2. 技术选型与目录结构先看清这套源码的骨架2.1 为什么是脚本而不是完整 Web 服务这套源码没有做成带前端界面的管理系统而是以脚本形式交付原因很直接自动收货是一个低频、短时、无需人工干预的任务做成 Web 服务反而增加部署成本和被探测的风险。常见做法是用 Python 写主逻辑配合 requests 或 httpx 发请求用 schedule 或 APScheduler 做定时调度。也有部分实现走 Node.js 路线用 axios 加 node-cron。两种路线在功能上等价区别在于你团队更熟悉哪套生态。我一般会建议选 Python因为后续要加日志分析、异常告警、数据落库时Python 的库生态更顺手。源码包解压后典型目录结构如下xianyu-auto-receive/ ├── config/ │ ├── settings.yaml # 账号、轮询间隔、通知配置 │ └── cookies.json # 登录态凭证需自行抓取 ├── core/ │ ├── order_fetcher.py # 订单列表拉取 │ ├── status_checker.py # 订单状态判断 │ └── receiver.py # 收货请求提交 ├── utils/ │ ├── logger.py # 日志封装 │ ├── notifier.py # 通知推送钉钉/飞书/邮件 │ └── retry.py # 重试装饰器 ├── main.py # 入口调度循环 └── requirements.txt这个结构的好处是职责分离core/只管业务动作utils/管通用能力config/管可变参数。你要改轮询频率只动settings.yaml要换通知渠道只动notifier.py要调整收货判断逻辑只动status_checker.py。不会出现改一处崩全身的情况。2.2 核心依赖与运行环境requirements.txt里通常包含这些包包名用途备注requests发 HTTP 请求也可换 httpx 支持异步pyyaml解析配置文件比 JSON 好写注释schedule定时调度轻量适合单机loguru日志输出比 logging 少写很多样板代码tenacity重试控制处理网络抖动Python 版本建议 3.9 以上3.11 更稳。安装命令python -m venv venv source venv/bin/activate # Windows 用 venv\Scripts\activate pip install -r requirements.txt提示不要用系统自带的 Python 直接装依赖虚拟环境能避免版本冲突。我见过太多人因为全局环境里有个旧版 requests 导致 SSL 报错排查半天。2.3 配置文件的关键字段settings.yaml是整套源码的“控制面板”字段设计决定了你能调多少东西account: cookie_file: config/cookies.json user_agent: Mozilla/5.0 ... polling: interval_seconds: 300 # 轮询间隔建议不低于 180 秒 max_orders_per_round: 20 # 单轮最多处理订单数 receive_delay_seconds: 5 # 每单收货后停顿避免触发风控 notification: enable: true channel: dingtalk # 可选 dingtalk / feishu / email webhook: https://... logging: level: INFO file: logs/auto_receive.loginterval_seconds是最关键的参数。设得太短请求频率高容易触发平台限流设得太长收货不及时影响卖家回款周期。常见做法是 300 秒起步观察一周没有异常再考虑降到 180 秒。receive_delay_seconds是每单之间的停顿别设成 0连续请求是典型的机器行为特征。3. 核心模块拆解一个订单从拉取到收货的完整链路3.1 订单拉取order_fetcher.py这个模块负责从平台接口拿到当前账号下的订单列表。核心逻辑是构造请求、解析响应、过滤出待收货订单。import requests import json from utils.logger import logger from utils.retry import retry_on_failure class OrderFetcher: def __init__(self, cookie_file, user_agent): self.session requests.Session() self.session.headers.update({User-Agent: user_agent}) self._load_cookies(cookie_file) def _load_cookies(self, path): 从 JSON 文件加载登录态格式为 [{name, value, domain}, ...] with open(path, r, encodingutf-8) as f: cookies json.load(f) for c in cookies: self.session.cookies.set(c[name], c[value], domainc.get(domain)) retry_on_failure(max_retries3, delay2) def fetch_orders(self, page1, page_size20): 拉取订单列表返回原始 JSON url https://api.example.com/order/list params {page: page, pageSize: page_size, status: pending_receive} resp self.session.get(url, paramsparams, timeout10) resp.raise_for_status() data resp.json() logger.info(f拉取到 {len(data.get(items, []))} 条订单) return data def get_pending_orders(self): 过滤出状态为待收货的订单 ID 列表 data self.fetch_orders() orders data.get(items, []) pending [o[orderId] for o in orders if o.get(status) pending_receive] logger.info(f待收货订单{len(pending)} 条) return pending逻辑说明_load_cookies把浏览器抓到的登录态注入 session这是所有后续请求能通过鉴权的前提。fetch_orders带了重试装饰器网络抖动时自动重试三次避免因为一次超时就漏掉整轮订单。get_pending_orders做了一层过滤只返回状态匹配的订单 ID把业务判断和网络请求分开。参数方面page_size不要设太大20 到 50 之间比较稳妥太大容易触发接口分页限制。3.2 状态判断status_checker.py拿到订单 ID 之后不能无脑收货。有些订单可能已经收货、已退款、或者处于纠纷状态需要二次确认。class StatusChecker: def __init__(self, fetcher): self.fetcher fetcher def is_receivable(self, order_id): 判断单个订单是否满足自动收货条件 detail self.fetcher.fetch_order_detail(order_id) status detail.get(status) refund_flag detail.get(refundStatus) dispute_flag detail.get(disputeStatus) if status ! pending_receive: logger.debug(f订单 {order_id} 状态为 {status}跳过) return False if refund_flag and refund_flag ! none: logger.warning(f订单 {order_id} 存在退款申请跳过) return False if dispute_flag and dispute_flag ! none: logger.warning(f订单 {order_id} 存在纠纷跳过) return False return True这段代码的价值在于“后悔药”机制自动收货最大的风险是收了一个有问题的订单钱打给卖家之后才发现货不对。加上退款和纠纷判断等于给脚本装了一道保险。refundStatus和disputeStatus的字段名需要根据实际接口调整不同平台叫法不一样抓包确认后再写死。3.3 收货提交receiver.py真正执行收货动作的模块也是最需要控制频率的地方。import time from utils.logger import logger from utils.notifier import send_notification class Receiver: def __init__(self, session, delay_seconds5): self.session session self.delay delay_seconds def receive_order(self, order_id): 提交收货请求返回是否成功 url https://api.example.com/order/receive payload {orderId: order_id, action: confirm} try: resp self.session.post(url, jsonpayload, timeout10) result resp.json() if result.get(success): logger.info(f订单 {order_id} 收货成功) return True else: logger.error(f订单 {order_id} 收货失败{result.get(message)}) return False except Exception as e: logger.exception(f订单 {order_id} 请求异常{e}) return False def batch_receive(self, order_ids): 批量收货每单之间强制停顿 success, fail 0, 0 for oid in order_ids: if self.receive_order(oid): success 1 else: fail 1 time.sleep(self.delay) # 关键强制间隔 send_notification(f本轮收货完成成功 {success}失败 {fail}) return success, failbatch_receive里的time.sleep(self.delay)是整套源码里最不能省的一行。连续快速请求是机器行为最明显的特征加上随机化的停顿比如 3 到 8 秒之间随机比固定值更自然。send_notification在每轮结束后推送结果方便你不在电脑前也能知道跑没跑成功。3.4 调度入口main.py把所有模块串起来形成定时循环。import schedule import time from core.order_fetcher import OrderFetcher from core.status_checker import StatusChecker from core.receiver import Receiver from config.loader import load_settings def job(): settings load_settings() fetcher OrderFetcher(settings[account][cookie_file], settings[account][user_agent]) checker StatusChecker(fetcher) receiver Receiver(fetcher.session, settings[polling][receive_delay_seconds]) pending fetcher.get_pending_orders() receivable [oid for oid in pending if checker.is_receivable(oid)] if receivable: receiver.batch_receive(receivable) else: print(本轮无待收货订单) if __name__ __main__: interval load_settings()[polling][interval_seconds] schedule.every(interval).seconds.do(job) job() # 启动时先跑一次 while True: schedule.run_pending() time.sleep(1)这个入口的设计是“启动即跑一次然后按间隔循环”。schedule库足够轻量单机跑完全没问题。如果你要部署到服务器长期运行建议用 systemd 或 supervisor 托管崩了能自动拉起。4. 避坑与常见问题跑起来之后最容易翻车的几个点4.1 登录态失效脚本静默失败现象日志里一直显示“拉取到 0 条订单”但手动打开 App 明明有待收货。原因cookie 过期了接口返回的是登录页 HTML 而不是 JSONresp.json()解析失败被吞掉。解决在fetch_orders里加一层判断如果响应内容不是 JSON 或者包含“login”关键字直接抛异常并触发通知。我一般会在_load_cookies之后加一个check_login()方法启动时先验证登录态。4.2 轮询间隔太短触发限流现象跑了几轮之后接口返回 429 或直接封 IP。原因interval_seconds设成了 60 甚至 30请求频率超出正常用户行为。解决间隔不低于 180 秒单轮订单数控制在 20 以内每单之间加随机停顿。如果已经被限流换 IP 之后把间隔调到 600 秒观察一天。4.3 收货成功但状态没更新现象日志显示收货成功但订单列表里还是待收货。原因接口返回的success字段可能是异步的实际状态更新有延迟或者收货请求被平台风控拦截返回了假成功。解决收货后不要立即信任返回值下一轮轮询时检查该订单是否还在待收货列表里如果还在标记为异常并通知人工处理。4.4 多账号跑在同一进程里互相干扰现象两个账号的 cookie 串了A 账号收了 B 账号的单。原因requests.Session是共享的cookie 没有按账号隔离。解决每个账号实例化独立的OrderFetcher和Receiver不要复用 session。如果账号多用多进程而不是多线程避免 GIL 和 cookie 竞争。4.5 日志文件无限增长撑爆磁盘现象跑了一个月logs/auto_receive.log几个 G。原因loguru 默认不轮转。解决在logger.py里加rotation100 MB和retention7 days自动切割和清理。这个坑不致命但很烦磁盘满了之后所有写入操作都会失败。5. 二次开发与改造从能跑到好用5.1 加一层订单落库方便对账源码默认只打日志不存数据。实际用起来你肯定想知道“这个月自动收了多少单、失败了多少”。最省事的做法是加 SQLiteimport sqlite3 def init_db(pathdata/orders.db): conn sqlite3.connect(path) conn.execute( CREATE TABLE IF NOT EXISTS receive_log ( order_id TEXT PRIMARY KEY, status TEXT, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) ) conn.commit() return conn def log_receive(conn, order_id, status): conn.execute(INSERT OR REPLACE INTO receive_log (order_id, status) VALUES (?, ?), (order_id, status)) conn.commit()在batch_receive里每处理一单就写一条后面用 SQL 查成功率、查某天的记录都很方便。SQLite 单文件、零配置适合这种轻量场景。5.2 把通知渠道做成可插拔notifier.py里如果写死了钉钉换飞书就要改代码。改成策略模式class Notifier: def __init__(self, channel, webhook): self.channel channel self.webhook webhook def send(self, message): if self.channel dingtalk: self._send_dingtalk(message) elif self.channel feishu: self._send_feishu(message) elif self.channel email: self._send_email(message) def _send_dingtalk(self, msg): import requests requests.post(self.webhook, json{ msgtype: text, text: {content: msg} })这样配置文件里改一个channel字段就能切换不用动核心逻辑。5.3 验证改造是否生效的笨办法改完代码别急着扔到服务器上跑。我一般会先用一个测试账号手动造几条待收货订单把interval_seconds设成 30 秒盯着日志看三轮。确认拉取、判断、收货、通知四个环节都正常再把间隔调回去。这个习惯帮我省过很多次“上线才发现通知没发出去”的尴尬。5.4 一个容易忽略的细节时区如果部署在海外服务器上created_at默认是 UTC对账时会对不上。在init_db之后加一句conn.execute(PRAGMA timezone localtime)或者在写入时手动转时区。这个坑不报错但会让你在查数据时怀疑人生。从那以后我每次改完调度相关的代码都强制走一遍“测试账号 短间隔 盯三轮日志”的流程再也没出现过上线后才发现轮询没生效的情况。希望帮到你。本文还有配套的精品资源点击获取