简介：这款京东京粉鲸鱼转链转发工具面向京东联盟推广者及社群运营人员，用于解决手动转链、分群发送效率低的问题。软件支持定时自动发单、采集群后批量转发到多个群，并能配置多个京东联盟推广位ID；同时提供替换词、前后缀消息、屏蔽词等规则，可灵活适配不同群组的发送风格，还支持同时登录多个微信账号，适合高频群发场景的电商从业者。压缩包整体约129.5MB，共15个文件，类型覆盖dll动态库、exe程序、html配置页面、json数据配置、gif界面演示等，其中dll与exe负责主程序和运行依赖，html/json用于界面与规则配置，附带gif可直观了解操作效果，文件结构清晰。已有512人学习/下载，说明该工具在同类场景中有一定实用价值。压缩包内额外附带了微信3.1.0安装包，这是其运行所需的重要依赖，同时提供常用配置文件，下载后可在匹配环境中快速部署，体验自动发单、多群批量转发及联盟参数设置等完整功能，省去额外寻找依赖的麻烦。

1. 转链转发工具到底在解决什么：从复制口令到群内自动推送

一个做京东推广的群主，最烦的事不是选品，而是每天要把几十上百个商品链接手动转到自己的推广位，再复制文案、发到好几个群里。操作本身不难，但架不住量大：每个链接要打开联盟后台、粘贴、转链、复制、再换群粘贴，一套下来慢的三分钟，快的也要一分钟。如果有人把链接直接丢进群里让你帮忙转，你还得先跟对方解释半天什么是转链。京东京粉鲸鱼转链转发工具解决的就是这一整段重复劳动：它把「解析链接、生成推广链接、拼文案、推到群」做成了流水线，链接进来、成品出去，全程不用人碰。适合每天转发量在几十条以上的个人推广者和三五个人的小团队，不适合只是偶尔发一两条的普通用户——因为搭建和调试本身需要投入一个下午。

2. 转链转发的底层逻辑：从skuId到推广链接，参数为什么一个都不能少

2.1 转链的本质：把普通链接换成带pid和positionId的推广链接

普通人打开京东商品页拿到的那条链接，谁点进去都和推广者没关系。平台要确认「这个单算谁的」，靠的是链接里几个关键参数：appKey 对应的联盟账号、pid（代表推广位）、positionId（代表推广位置）。转链就是把一条无归属的商品链接，重写成带这些参数的推广链接。

我一般用一个类比来解释：原始链接是「产品本身」，推广链接是「产品 + 你的工牌」。商品还是那个商品，但结账的时候系统会看工牌认人。所以转链工具的核心工作，就是找到商品的真实 ID（skuId），然后向京东联盟的推广接口申请一条新链接，把 pid 和 positionId 挂上去。返回的新链接通常是一个短链（形如 u.jd.com 开头的跳转），用户点开照样是原商品页，但购买归属已经记在你名下了。

这里有个新手最容易理解错的地方：转链不是把原链接改几个参数。京东联盟接口返回的是一个全新的短链接，原链接可以废弃。如果你手工把原链接的 URL 参数里塞一个 pid，大概率是无效的，因为平台校验的是整条链接的签名和来源。这也是为什么必须走 API，而不能靠字符串拼接。

2.2 链接入参的三种来源：PC链接、短链、京口令的解析差异

用户丢进群里的商品链接，形态五花八门。最常见的有三种，处理方式完全不同：

第一种是 PC 端商品链接，形如 item.jd.com 开头，后面带一串数字。这种最好办，URL 里直接能看到 skuId；有时候是 wareId，本质同一个东西，转链接口两个参数都认。

第二种是短链，常见于分享按钮生成的 m.jd.com 或 u.jd.com 短地址。这种不能直接解析，必须先发起一次网络请求，让服务器返回 302 跳转，跟着重定向走到最终的商品页 URL，再从里面提取 skuId。写代码时一定要设置 allow_redirects=True，否则你拿到的只是中间跳转地址。

第三种是京口令，也就是微信里那种「京东超市丨xx商品，某东价xx元」带口令的文本。口令本体是一段密文，转链之前需要先调用京东联盟的口令解析接口还原出 skuId。这是三种来源里唯一必须走接口的，工具里要单独做分支处理：识别文本里是否包含「€」或「￥」之类的口令特征符号，有就先进解析流程。

2.3 技术选型：Python 3 + 京东联盟API + webhook 推送的组成

这个工具的技术栈不复杂，我用的是 Python 3 + requests 库，调度用 cron 或系统计划任务，推送走webhook。选 Python 是因为京东联盟官方示例以 Python 为主，处理字符串和 JSON 都顺手；requests 库做重定向跟随和 POST 请求很成熟，不需要引重型框架。

整个工具由四个模块组成：入参解析模块负责把三种形态的链接还原成 skuId；转链模块负责调推广链接接口和优惠券接口；文案格式化模块把商品名、券后价、推广链接拼成适合发群的文本；推送模块把文案 POST 到群机器人的 webhook 地址。每个模块独立成函数，方便单独测试。我用最简单的方式组织代码：一个 main.py 跑主流程，一个 config.py 放常量，不引入数据库，转链结果直接追加写入本地 CSV 作为留底——对个人使用来说，文件即数据库。

选型上还有一个坑要提前说：京东联盟开放平台的接口分两套，一套是老的「京东联盟 API」，一套是「京东联盟 SaaS 接口」，签名方式不同。日常用的转链、解析、优惠券接口都在 SaaS 接口域下，鉴权方式是 appKey + 签名串，不是 OAuth 的 access_token。第一次对接的人很容易拿着另一套文档的鉴权方式调接口，结果一直报签名错误，这个我在第 3 章会展开讲。

3. 搭建环境与鉴权：appKey、secret、PID、推广位，先跑通第一个转链请求

3.1 申请联盟账号并整理调用参数

动手写代码之前，先把账号侧的几样东西准备好，缺一样都跑不通。你需要登录京东联盟开放平台，创建一个应用，拿到 appKey 和 secretKey；然后在推广管理里创建至少一个推广位，拿到对应的 pid；pid 的格式一般是三段数字加短横线，中间那一段代表推广位 ID，也叫 positionId，很多接口调用里其实只需要传 positionId 就能定位推广位。

我习惯把这些信息集中放在一个配置文件中，不硬编码在代码里。下面是 config.py 的写法：

# config.py APP_KEY = "你的appKey" SECRET_KEY = "你的secretKey" PID = "你的完整pid，格式类似 123456_789_12345678" POSITION_ID = "推广位ID，即pid中间那一段" # webhook地址，由群机器人提供 WEBHOOK_URL = "https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=xxxx"

注意 POSITION_ID 看起来像是 PID 的一段，但它们是两个独立参数。接口传参时 PID 传完整串，POSITION_ID 只传中间那段，两者不一致会导致佣金无法正确归属。我见过有人把这两个值填反，转出来的链接佣金比例显示正常，但后台跟单数据一直为空，排查了两天才发现是这里的问题。所以建议在配置文件里把这两个字段分开写，并加注释说明。

3.2 用 requests 库构造第一个转链请求，验证鉴权

京东联盟 SaaS 接口的签名规则可以概括为：把请求参数（不含 sign 本身）按 key 的字典序升序排列，拼成 key1value1key2value2 的字符串，再拼接 secretKey，做 MD5，转大写，得到签名串。下面是通用签名函数和一次转链调用的完整示例：

import hashlib import requests import time import json def gen_sign(params: dict, secret: str) -> str: """按京东联盟SaaS接口规则生成签名""" # 1. 过滤掉空值和sign本身 filtered = {k: v for k, v in params.items() if v != "" and k != "sign"} # 2. 按key字典序升序排列 sorted_keys = sorted(filtered.keys()) # 3. 拼接成 key1value1key2value2 格式 src = "".join(f"{k}{filtered[k]}" for k in sorted_keys) # 4. 拼接secret并做MD5，转大写 sign = hashlib.md5((src + secret).encode("utf-8")).hexdigest().upper() return sign def build_params(payload: dict) -> dict: """封装公共参数，SaaS接口必须带appKey和timestamp""" params = { "appKey": APP_KEY, "timestamp": str(int(time.time() * 1000)), # 毫秒级时间戳 "format": "json", "v": "1.0", } params.update(payload) params["sign"] = gen_sign(params, SECRET_KEY) return params

这段代码里有两个细节容易错。第一，timestamp 必须用毫秒，不是秒；用秒的话接口会报时间戳无效。第二，拼接签名时参数值直接拼进去，不要加引号、不要加等号，很多人按别的平台签名习惯拼成 key=value 格式，签名永远对不上。过滤空值那一步也很关键，因为如果某个参数是空字符串，它会参与拼接导致签名不一致，而拼签名的时候这个参数实际又没传。

接下来用「京东联盟」的推广链接生成接口做一次真实调用，参数传 skuId、pid、positionId：

# trans.py 核心转链函数 def get_promotion_link(sku_id: str) -> dict: """传入商品ID，返回推广链接和佣金信息""" payload = { "method": "jd.union.open.promotion.common.get", "promotionCodeReq": json.dumps({ "materialId": f"https://item.jd.com/{sku_id}.html", "pid": PID, "positionId": POSITION_ID, "couponUrl": "", # 可空，有券时填券链接 "subUnionId": "", # 子渠道标识，可空 }, ensure_ascii=False) } params = build_params(payload) resp = requests.post("https://api.jd.com/routerjson", data=params, timeout=10) result = resp.json() if result.get("code") == "0": data = result["data"] # data是JSON字符串，需要再次解析 real = json.loads(data) if isinstance(data, str) else data return { "short_url": real["shortURL"], "click_url": real["clickURL"], "commission_rate": real.get("commissionRate", "0"), } else: raise RuntimeError(f"转链失败: {result.get('msg')}")

京东联盟的接口返回有一个很坑的结构：外层 HTTP 响应是 JSON，但 data 字段本身是字符串，里面嵌套了另一层 JSON。代码里做了兼容处理，isinstance(data, str) 判断后二次解析，不然你会拿到一串字符串而不是字典。另外注意 materialId 参数传的是完整商品 URL，不是纯 skuId；有些接口变体支持直接传 skuId，但用 URL 最稳。返回值里的 shortURL 是最终要发到群里的链接，commissionRate 是佣金比例，后面拼文案要用。

3.3 封装统一签名函数，为批量转链做准备

跑通单次请求后，不要急着写业务逻辑，先把签名函数和请求函数封装成独立模块。原因是批量转链时一次要处理几十个链接，每个链接都要走一遍「参数组装 → 签名 → POST → 解析」的流程，公共逻辑不抽出来，代码会膨胀得很厉害。

我的做法是建一个 jd_client.py，把签名和请求封装成一个 JdClient 类，后续所有接口调用都走这个类的 call 方法：

# jd_client.py import hashlib, requests, time, json class JdClient: def __init__(self, app_key: str, secret: str): self.app_key = app_key self.secret = secret self.endpoint = "https://api.jd.com/routerjson" def _sign(self, params: dict) -> str: filtered = {k: v for k, v in params.items() if v != "" and k != "sign"} sorted_keys = sorted(filtered.keys()) src = "".join(f"{k}{filtered[k]}" for k in sorted_keys) return hashlib.md5((src + self.secret).encode("utf-8")).hexdigest().upper() def call(self, method: str, biz_params: dict) -> dict: """通用请求方法，method为接口名，biz_params为业务参数（已序列化为字符串）""" params = { "appKey": self.app_key, "timestamp": str(int(time.time() * 1000)), "format": "json", "v": "1.0", "method": method, } params.update(biz_params) params["sign"] = self._sign(params) resp = requests.post(self.endpoint, data=params, timeout=15) result = resp.json() if result.get("code") != "0": raise RuntimeError(f"接口 {method} 调用失败: {result.get('msg')}") data = result["data"] return json.loads(data) if isinstance(data, str) else data

封装之后，上层业务代码不需要关心签名细节，只需要传接口名和业务参数。后面加优惠券查询、京口令解析，都只是多写一行 call 调用的事。还有一点值得养成习惯：把每次调用的入参和返回原始 JSON 追加写到 logs.txt，排查问题时比 debugger 好用得多，因为京东联盟接口的错误提示有时候很含糊，有完整请求记录才能比对出是参数问题还是签名问题。

4. 写一个能用的转链转发工具：从单链接转链到推送到群机器人

4.1 解析用户发来的商品链接，还原真实商品 ID

有了基础客户端，下一步处理「用户丢进来什么」的问题。群里收到的消息可能是纯链接、带文案的链接、短链或京口令。我设计了一个 parse_input 函数，接收原始文本，先判断类型再走不同的解析路径：

# parser.py import re, requests def parse_input(text: str) -> str: """从任意文本中提取商品skuId，支持PC链接、短链、京口令""" # 1. 优先检测京口令：口令文本通常以【】包裹 kw_match = re.search(r"【(.+?)】", text) if kw_match and ("京东" in kw_match.group(1) or "￥" in text or "€" in text): return parse_command(text) # 2. 提取URL url_match = re.search(r"https?://[^\s]+", text) if not url_match: raise ValueError("未找到有效链接或口令") url = url_match.group(0).strip() # 3. 如果是短链，跟随重定向拿到真实路径 if "u.jd.com" in url or "3.cn" in url: resp = requests.get(url, allow_redirects=True, timeout=10) url = resp.url # 4. 从最终URL中提取skuId sku_match = re.search(r"(?:sku|ware)[Ii]?[Dd]?=(\d+)", url) if not sku_match: sku_match = re.search(r"/(\d{6,})\.html", url) if not sku_match: raise ValueError(f"无法从链接中解析商品ID: {url}") return sku_match.group(1)

这段代码的思路是层层兜底：先看是不是京口令，再看有没有 URL，短链就展开，最后用正则去两个最常见的位置抓 skuId。注意第 4 步的两个正则，第一个匹配 item.jd.com/?skuId=123 这种查询参数形式，第二个匹配 item.jd.com/123.html 这种路径形式，很多商品链接是后者。真实场景里还有 https://item.m.jd.com/product/123.html 这种移动端结构，需要再加一个正则兼容，我在后面的避坑章节会单独说。

parse_command 函数调用的是京口令解析接口，核心参数是口令文本：

# parser.py from jd_client import JdClient def parse_command(client: JdClient, text: str) -> str: """调用京口令解析接口，还原商品ID""" payload = { "method": "jd.union.open.promotion.command.get", "promotionCodeReq": json.dumps({ "command": text.strip(), "pid": PID, }, ensure_ascii=False) } result = client.call("jd.union.open.promotion.command.get", payload) # 返回结构里是解析后的链接 url = result.get("data", {}).get("clickURL", "") sku_match = re.search(r"/(\d{6,})\.html", url) if not sku_match: raise ValueError("京口令解析结果中未找到商品ID") return sku_match.group(1)

京口令解析接口是我在实际调试中觉得最不稳的一环，有时候同样的口令，间隔几分钟调用返回的链接结构会有差异。所以解析成功之后不要直接信任结果，把返回的 URL 存到日志里，万一后面转链失败方便回头排查。

4.2 调用推广链接生成接口，并同步查询优惠券

拿到 skuId 后，可以开始真正的转链。但这里有一个影响最终收益的细节：很多商品有隐藏优惠券，不领券直接下单，单价高、佣金也按高价计算，而用户看到没有券可能就不买了。所以工具在转链时要同时查询优惠券，如果有券，要把优惠券信息拼进推广文案里。

我写的 transform 函数，把转链和查券合并在一段流程里：

# trans.py def transform(client: JdClient, sku_id: str) -> dict: """执行转链，并查询该商品是否有优惠券""" # 1. 转链 promo = client.call("jd.union.open.promotion.common.get", { "promotionCodeReq": json.dumps({ "materialId": f"https://item.jd.com/{sku_id}.html", "pid": PID, "positionId": POSITION_ID, }, ensure_ascii=False) }) short_url = promo["shortURL"] commission_rate = promo.get("commissionRate", "0") # 2. 查券 coupon_info = None try: coupon = client.call("jd.union.open.coupon.query", { "couponReq": json.dumps({ "skuIds": [int(sku_id)], }, ensure_ascii=False) }) if coupon and isinstance(coupon, list): coupon_info = coupon[0].get("couponInfo", {}) except Exception: pass # 查券失败不影响转链结果 return { "short_url": short_url, "commission_rate": commission_rate, "coupon": coupon_info, "sku_id": sku_id, }

查券接口返回的优惠券信息通常包含面额、门槛、使用时间。把这些数据整理出来，文案里写「领券立减 xx 元」比光秃秃丢一条链接的转化率高很多。注意代码里查券用 try-except 包裹，但不打印异常——这是因为部分商品确实没有券，接口会返回空列表，这不算错误；有些商品有券但券已过期，接口返回状态码异常也是常态。不要让查券的失败阻断转链主流程。

4.3 格式化推广文案并通过 webhook 推送到鲸鱼群

转链结果拿到手，最后一步是拼文案、发群。文案模板我调过很多版，目前用得比较顺手的结构是：商品名放第一行（从商品信息接口取，或者用转链接口返回的 goodsName），第二行写价格和券信息，第三行放短链，末尾跟一句引导语。下面是格式化函数和推送函数：

# push.py import requests def format_message(goods_name: str, price: str, coupon_desc: str, short_url: str) -> str: """拼接推广文案，结构固定，方便后续改模板""" lines = [ f"【京东】{goods_name}", f"价格：{price} 元", ] if coupon_desc: lines.append(f"优惠券：{coupon_desc}") lines.append(f"下单链接：{short_url}") lines.append("需要的自取，有券先领券再下单") return "\n".join(lines) def push_to_group(webhook_url: str, content: str) -> bool: """推送文本消息到群机器人webhook地址""" payload = { "msgtype": "text", "text": { "content": content, "mentioned_list": [], # 不@任何人 } } resp = requests.post(webhook_url, json=payload, timeout=10) result = resp.json() if result.get("errcode") == 0: return True else: print(f"推送失败: {result}") return False

推送这一步看起来简单，但有三个容易翻车的细节。第一，webhook 地址不要写死在脚本里，放到 config.py 或环境变量中；群机器人地址相当于群的一个入口，泄漏后任何人都能往里发消息。第二，mentioned_list 默认为空数组，不要省略这个字段，有些群机器人版本对这个字段有要求。第三，推送结果一定要检查 errcode，不能只看 HTTP 状态码——webhook 接口即使返回 200，也可能 errcode 非 0，表示消息被限流或内容包含违规词。

主流程把它们串起来：

# main.py from jd_client import JdClient from parser import parse_input from trans import transform from push import format_message, push_to_group import config def handle_message(raw_text: str): client = JdClient(config.APP_KEY, config.SECRET_KEY) sku_id = parse_input(raw_text) # 1. 解析入参 result = transform(client, sku_id) # 2. 转链 + 查券 msg = format_message( goods_name=result.get("goods_name", "商品"), price=result.get("price", "见详情"), coupon_desc=result.get("coupon_desc", ""), short_url=result["short_url"], ) push_to_group(config.WEBHOOK_URL, msg) # 3. 推送到群 if __name__ == "__main__": # 本地测试：读入一行文本，模拟用户发来的消息 test_text = "https://item.jd.com/100012345678.html" handle_message(test_text)

这段主流程不是完整的线上服务，没有监听逻辑，是最小可运行的骨架。真实使用时，你还需要把 handle_message 接到消息源上——比如定时读取某个文件里新增的链接、或者用消息钩子监听群里的 @ 消息。骨架的价值在于，任何一条链路出错，你都能通过单独跑对应函数定位问题，而不是面对一个黑匣子。

5. 转链转发工具避坑指南：佣金为0、链接失效、机器人被踢

5.1 现象：转出来的链接佣金比例显示为 0

明明选了商品，转链接口也返回了短链，但链接里的佣金比例是 0，发到群里用户下单后你一分钱拿不到。原因通常是两类：一类是这个商品本身没有开通 CPS 推广，属于自营但不参与联盟计划；另一类是推广位和商品类目不匹配，部分类目商品只有在特定推广位下才有佣金。

解决方法分两步。第一步先排除商品自身问题，用商品信息接口查询该商品的「是否可推广」字段，如果标识为不可推广，直接放弃这个链接，不要转。第二步检查 pid 的推广位类型，到联盟后台看该推广位绑定的媒体类型，如果是「自购返利」类型，转到部分商品会按自购单处理，佣金归自己但不计入推广效果数据。我通常的做法是准备两个推广位，一个用于常规分享，一个用于自购测试，避免混用。

5.2 现象：短链接解析后报错「您输入的链接格式不正确」

群里用户发的短链，程序跟随重定向后拿到的最终 URL 不是商品页，而是京东首页或者活动页。这是因为部分短链是「活动链接」，跳转后可能先到 H5 活动页再二次跳转到商品页，而 requests 库默认的多级重定向在这个场景下可能只跟了第一跳就被页面里的脚本拦截了。

解决方法是不要只依赖一次 GET 请求，而是手动做两轮请求：第一轮请求拿到中间跳转地址，如果该地址中含 redirect 或 jump 之类的关键字，继续请求第二次，直到最终 URL 里出现 item.jd.com 或 skuId 为止。另外注意带上 User-Agent 请求头，京东的活动页对无 UA 的请求会直接返回聚合页而不是商品页。

提示：如果你自己用浏览器打开短链能正常跳到商品页，但脚本不行，首先改 UA，其次加 Cookie 模拟访客身份。

5.3 现象：同一条商品链接，不同人的工具转出来的佣金不一样

同一个商品，你的工具转出来佣金比例是 5%，别人的是 3%，这不是接口返回随机，而是参数差异。京东联盟的佣金比例和三个因素相关：推广位所在的媒体类型、是否携带子渠道标识 subUnionId、以及领取的优惠券是否属于「渠道专享券」。

最容易被忽视的是 subUnionId。它相当于在一个推广位下面再细分流量来源，很多工具没传这个参数也能正常转链，但佣金比例会取推广位的默认值；如果该推广位之前绑定过渠道专享券活动，不传 subUnionId 就拿不到渠道专享佣金。我的建议是给自己固定一个 subUnionId 命名规则，比如 20240101A，一直不变，这样报表里也能按 subUnionId 区分数据来源。

5.4 现象：推送的推广链接在微信群里被折叠成灰色链接

辛苦拼好的文案推到群里，商品链接被微信折叠成一行灰色小字，用户不敢点，转化率断崖下跌。这不是京东的问题，是微信对短链接域名的风控策略。u.jd.com 和 3.cn 都是被折叠的高频域名，尤其当你的文案里连续出现多条相似短链时，更容易触发折叠。

目前有效的方式是把短链接嵌在「阅读原文」或者小程序卡片里，但群机器人无法直接发小程序，所以在纯文本场景下，我的做法是给链接前面加一个引导词，比如「点开链接看详情」，不要直接裸发链接；同时控制同一条群里短链出现的频率，同一短链在一小时内重复推送超过两次，折叠概率明显上升。如果条件允许，给工具加上「自动间隔发送」逻辑，让同一条商品在两个群里的推送时间至少错开五分钟。

5.5 现象：同一把 key 频繁请求报「qps 超限」

工具上线后转发量大增，某个时间点突然连续收到 qps 超限的异常，导致一批链接没有转成功。京东联盟接口对单 appKey 的调用频率有限制，常见的是每秒几次、每分钟几百次，具体数值和账号等级挂钩。批量场景下，循环里不加休眠地连续调用，几乎必然触发。

解决方法是增加一个轻量级的限速器，在每次调用前检查距上次调用的时间，不足 0.3 秒就 sleep 补齐；同时把并发改成串行。更重要的一点是，把转链操作集中到一个时间窗口执行，不要实时响应每个入群链接——用户往群里丢一条，程序转一条，这种模式在高峰期很容易把 QPS 打满。改成「积攒 30 秒内的请求，合并后串行处理」，限流问题基本消失。

# throttle.py import time class RateLimiter: """简单限速器，保证两次调用间隔不小于 min_interval 秒""" def __init__(self, min_interval: float = 0.3): self.min_interval = min_interval self.last_call_time = 0 def wait(self): elapsed = time.time() - self.last_call_time if elapsed < self.min_interval: time.sleep(self.min_interval - elapsed) self.last_call_time = time.time() limiter = RateLimiter(0.3)

限速器本身很简单，但注意它要设计成全局单例，不要在每个调用函数里 new 一个，否则各函数之间的调用间隔不受控。我最初就是把限速器写在了转链接口内部，结果查券和解析接口同时调用，限速完全失效。把它提升到 JdClient 层面或者作为参数传入每个 call 方法，才能从全局控制调用节奏。

6. 把工具做成服务：多群分发、定时任务和跟单核对

6.1 多群分发：用一个 webhook 通道做分发，而不是每个群一个机器人

推广者通常管理不止一个群，有的是宝妈群，有的是数码群，有的只是朋友圈好友拉起来的杂货群。不同群适合不同的商品，如果每个群配一个机器人，webhook 地址管理起来很快会乱。我的做法是建一个「分发配置表」，用文本文件记录每个群的 webhook 地址和商品类目偏好，程序按类目自动匹配目标群。

# dispatch.py import json, config def load_dispatch_config(path="dispatch.json"): """读取分发配置，按群名区分webhook地址和偏爱类目""" with open(path, "r", encoding="utf-8") as f: return json.load(f) # dispatch.json 示例 # {"数码群": {"webhook": "https://...", "categories": ["手机", "电脑"]}}

分发逻辑很简单：转链成功后有商品类目信息，去配置表里找对应类目的群，把文案推送过去；找不到类目匹配的群，回退到默认群。这个文件用纯文本管理，不用数据库，因为配置格式稳定、数据量小，改一个群地址只需要改一行。

6.2 定时任务：把转链集中到几个人少的时间窗口执行

实时响应每条消息虽然体验好，但 QPS 和链路稳定性都难以把控。更稳妥的运营节奏是定时批量处理：把一天内收集到的商品链接攒起来，在凌晨或午后固定时间统一转链、统一分发，既避开接口限流，也让群里的推送时间固定下来，群成员形成习惯。

我用系统自带的计划任务来调度，不引入 Celery 这类重工具。核心变化是 main.py 增加一个 batch 模式：

# main.py 批量处理入口 def batch_handle(link_list_file: str): client = JdClient(config.APP_KEY, config.SECRET_KEY) limiter = RateLimiter(0.3) with open(link_list_file, "r", encoding="utf-8") as f: raw_lines = [line.strip() for line in f if line.strip()] for line in raw_lines: try: sku_id = parse_input(line) result = transform(client, sku_id) msg = format_message(...) push_to_group(config.WEBHOOK_URL, msg) time.sleep(0.5) # 每处理一条，额外停0.5秒，进一步降低限流概率 except Exception as e: print(f"处理失败: {line}, 错误: {e}")

批量模式里我故意把每条之间加了 0.5 秒 sleep，和限速器的 0.3 秒叠加，实际调用间隔接近 0.8 秒。这个频率一天处理几百条完全够用，而且不会触发任何限流。单条处理失败的链接打印错误后继续处理下一条，不要中断整个批次——有一两条失败的链接是常态，可能是商品已下架，也可能是京口令解析临时失败，回头单独补跑就行。

6.3 验证方法：用「跟单数据」反查转链是否有效

工具上线后最核心的验证不是看推送成功没有，而是看一段时间后的跟单数据。京东联盟后台有「订单明细报表」，可以按推广位看到每一笔订单的来源和预估佣金。我的习惯是每周导一次订单明细，抽查前一周推送过的短链是否产生了对应订单。

具体做法是给每次转链留一个标识：main.py 里把 skuId、短链、推送时间、目标群写入一个 CSV 行，名字用当天日期，比如 20240218_trans.csv。到了对账日，打开后台报表，按短链和推送时间去匹配，能看到订单的就能确认工具链路没有问题，看不到订单的要排查是用户没下单，还是转链参数有问题。如果 100 条推送只有几条订单，属于正常转化率；如果连续一周一条订单都没有，基本可以断定 pid 或 positionId 配置错了，回到第 3.1 节检查。

这个核对环节是我踩过最深的坑换来的教训——早期工具推送一切正常，群里也有人回复下单了，但后台跟单数据全为零，后来发现 pid 里的推广位 ID 填成了另一个不常用推广位的，所有流量都记到了那个沉默推广位上。从那以后「转链必核对、上新必抽查」就成了我的固定习惯，每周花十分钟看报表，远好过月底发现一个月的推广都白做了。希望帮到你。

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