简介：这份源码面向闲鱼卖家及二手交易平台开发者，用于搭建自动收货系统，解决手动确认收货效率低、交易周期长的问题。其核心逻辑是生成运费保价页面，买家确认开通并支付运费后，系统自动触发发货与收货流程，后台可自由配置价格、规则和条件，适配不同交易场景。资源包共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里通常包含这些包：

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.log"

interval_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", encoding="utf-8") as f: cookies = json.load(f) for c in cookies: self.session.cookies.set(c["name"], c["value"], domain=c.get("domain")) @retry_on_failure(max_retries=3, delay=2) def fetch_orders(self, page=1, page_size=20): """拉取订单列表，返回原始 JSON""" url = "https://api.example.com/order/list" params = {"page": page, "pageSize": page_size, "status": "pending_receive"} resp = self.session.get(url, params=params, timeout=10) 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_seconds=5): 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, json=payload, timeout=10) 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, fail

batch_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 而不是 JSON，resp.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里加rotation="100 MB"和retention="7 days"，自动切割和清理。这个坑不致命但很烦，磁盘满了之后所有写入操作都会失败。

5. 二次开发与改造：从能跑到好用

5.1 加一层订单落库，方便对账

源码默认只打日志，不存数据。实际用起来，你肯定想知道“这个月自动收了多少单、失败了多少”。最省事的做法是加 SQLite：

import sqlite3 def init_db(path="data/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'")或者在写入时手动转时区。这个坑不报错，但会让你在查数据时怀疑人生。

从那以后我每次改完调度相关的代码，都强制走一遍“测试账号 + 短间隔 + 盯三轮日志”的流程，再也没出现过上线后才发现轮询没生效的情况。希望帮到你。

本文还有配套的精品资源，点击获取