汉高 POS 与商品 Mapping 推送
维护与实施指南

开始之前 · 零基础先读

这一节不讲代码。看完你会明白：这个系统为谁解决什么问题、两份数据分别是什么、你每天会在哪几个页面操作。

先讲一个每天都在发生的场景

假设周一有位顾客在京东「汉高家清京东自营旗舰店」买了 2 瓶施华蔻洗发水。接下来会发生这些事：

如果没有这个系统

两份数据，用大白话解释

先记住这 13 个词

你每天会碰到的 3 个界面 + 1 个后台

本指南的图索引

0读前必看 & 30 分钟上手

如果你今天刚接手，按下面 5 步走一遍，就能把整条链路跑通并定位问题。

0.1 你需要拿到的权限/账号

0.2 30 分钟上手路径

0.3 一分钟版本（背下来）

1业务背景

先搞清楚“为什么要推两份数据”，后面的字段和过滤规则才有判断依据。

1.1 客户与场景

汉高（Henkel）是鲲驰代运营的品牌方之一，涉及施华蔻（Schwarzkopf）、丝蕴（Syoss）、资生堂专业美发、汉高家清等品牌，在猫超、京东自营、京东 POP、唯品会等多个渠道销售。汉高侧使用 DME 平台（域名 dmeapi-prod.schwarzkopfclub.com.cn）接收鲲驰推送的数据，用于：

• POS 日销：每天按「店铺 × SKU × 日期」报送销量、销售额、客户数、订单数等，供汉高做终端销售分析；
• 商品 Mapping：把平台侧的商品编码（京东 SKU、猫超 sku_id、唯品会 goodsCode 等）对照到汉高的商品主数据（鲲驰码 / 69 条形码 / 子品项），供汉高识别“卖的是哪个货”。

数据来源：爬虫从各平台抓取原始订单/退款/商品数据并存入数仓；本系统只从数仓取数、清洗、推送，不负责采集。

1.2 两份数据的区别

1.3 汉高接口与鉴权

2系统全景

六个系统各司其职，任何一段出问题都会表现为“汉高没收到”，所以必须能画出这张图。

2.1 系统职责

2.2 POS 日销链路

2.3 商品 Mapping 链路

3核心数据模型

这一章是维护的地基。改配置、查问题、写 SQL，都围绕这些表和集合。

3.1 mongo report_job_task（任务配置，最重要）

集合：thirdhub.report_job_task，_id 就是 jobName。共约 30 条配置。

3.2 esb-broker 路由配置

一条 POS 配置的 targetPath 样例：

http://isb-gateway.ikunchi.com/esb-broker-provider/message/send
  ?sourceService=report
  &sourceMethod=jd-order
  &targetService=hangao_report
  &targetMethod=jd_report_order复制

Mapping 的 targetPath：

http://isb-gateway.ikunchi.com/esb-broker-provider/message/send
  ?sourceService=pipeline
  &sourceMethod=mapping_publish
  &targetService=hangao_product
  &targetMethod=product_mapping复制

3.3 esb-broker 消息与日志

状态字典：

• messageStatus：PENDING(0) 待投递 → SENDING(3) 投递中 → ENDING(2) 已结束。ENDING 不代表成功。
• resultStatus：OK(0) 成功、ERROR(1) 业务码不对、FAILURE(2) 网络/超时等系统失败。

3.4 data-flow 模板配置

3.5 fms_fnd 业务表

brand_platform_sku_mapping 关键字段

3.6 上游数仓表

这些表都在数仓里，数据来源是爬虫从各平台抓取的原始订单/退款/商品数据。我们作为使用方，只负责按日期取数、清洗、推送，不负责采集。

4POS 推送实现

核心类：mall-pipeline-provider/src/main/java/com/kunchi/mall/pipeline/job/report/CommReportPublishJobHandler.java

4.1 执行流程（逐行对应代码）

1. 读配置：task(param) 用 {"jobName":"..."} 查 report_job_task；查不到返回 false。
2. 生成批次号：yyyyMMddHHmmssSSS-8位随机，贯穿本次所有推送流水。
3. 拼上游 URL：startTime = nextStartTime - compensationMinutes，按 timePattern 格式化，然后给 timeFieldList 每个字段都塞这个值，再拼 param 静态参数和 limit/offset。
4. 取总数：GET 一次，用 successField/successValue（status=200）判断成功，读 countField=count。
5. 无数据分支：totalCount=0 → 记告警、建 no_data=true 的流水、抛异常。异常导致水位不推进，下次重试同一天。
6. 逐页取数：读 dataListField=results，先累计 mapping 去重数据，再逐条 sendMessage。
7. 汇总 mapping：把本批次所有页的 (shopName, sku) 去重后批量调 fms-fnd addMapping（每 1000 条一批）。只对含 sku 字段的行生效（京东系）。
8. 推进水位：nextStartTime += intervalMinutes，写回 mongo。

4.2 日期窗口与分页

分页参数固定写死：limit=pageSize(50)、offset=(pageIndex-1)*pageSize。第 1 页先取总数，再循环 1..totalPage。

4.3 单条推送与状态回写

createTask   → fms-fnd /brandDataPushTask/v2/create   (task_status=0)
markProcessing → /brandDataPushTask/v2/update        (task_status=1)
HttpUtil.doPost(esb-broker /message/send)
   └─ 响应 data 非空 → markSuccess (task_status=2)
   └─ 否则          → markFail    (task_status=3)
复制

4.4 现有 POS 任务清单

以 mongo 实际配置为准（aiops 页面可查）。下表按渠道分组，帮助新人快速定位。

4.5 period 语义与京东回流（day_p1）

为什么京东要单独做“回流”

京东的 POS 数据有一个特点：历史数据会回溯变化。同一张订单/同一个 SKU，今天抓到的“10 月 1 日销量”过几天可能被平台修正。比如 10 月 1 日某个品当天看到总共卖了 10,000，到 10 月 5 日再抓同一天，可能已经变成 15,000。所以每天按 by-day 抓到的数据只是“当时的快照”，不是最终值。

和汉高约定：次月 7 号，把上个月 1 号到月底的 by-day 数据完整重抓一遍，数仓里标记为 period=day_p1，再完整推一次给汉高。汉高会同时保留“每天常规”和“次月回流”两份数据，两份都会看；回流用于修正/复核上个月的最终数。

技术上怎么实现

• 配置：mongo report_job_task 里每组京东店铺都有一条 _huiliu 任务（jobInfo 带“(回流)”），param.period=day_p1，其余字段与常规任务一致。
• 取数：还是同一个 CommReportPublishJobHandler，只是把 period=day_p1 传给 spider_data，数仓返回回流版本的数据。
• 标记：data-flow 的京东模板判断 period == "day_p1"，输出 "dataType":"回流"；否则输出“常规”。
• 节奏：handler 一次只处理一天（nextStartTime 每次 +1 天）。所以“重推一个月”= 把水位拨到上月 1 号，然后连续触发约 28~31 次，每次都推一天。xxl-job 上看到的是一批密集执行，不是一次请求。
• 实测：2026 年 10 月这批，资生堂/汉高家清在 10-08 09:07~09:25 跑完 9 月全月；施华蔻两家在 10-08 13:02~13:52 跑完，最后一天在 10-09 14:21 补上。实际日期以 xxl-job 调度为准。

执行/核对步骤

1. 确认目标月份：例如 10 月要推 9 月，水位应设为 2026-09-01 00:00:00。
2. 在 xxl-job 找到对应店铺的 *_huiliu 任务，连续触发，直到水位走过 2026-09-30（约 30 次）。
3. 查 brand_data_push_task：job_info 带“(回流)”的记录数应约等于“上月天数 × 每天常规条数”。
4. 抽查 broker_message：subscriberId=hangao_report-jd_report_order，reqBody 里应包含 "period":"day_p1"，time/start_date 覆盖 09-01~09-30；respBody 外层 201、内层 code=0。
5. 如果某天漏了：把水位拨回那一天，单独触发一次；不要重跑整月（会重复推送）。

4.6 新增一个 POS 任务（两种方式）

方式 A：aiops-web 页面（推荐）

1. 打开「汉高 → Report任务配置」→ 点「新增任务」。
2. 按 7.4 的字段说明填写。注意 任务名称(ID) 必须唯一，建议沿用 {渠道}{店铺}ReportToHangao 命名。
3. 保存后到 xxl-job 新增一个调度：Handler=CommReportPublishJobHandler，参数={"jobName":"你填的ID"}，路由策略选「第一个」或「一致性哈希」，阻塞策略选「单机串行」。
4. 手动触发一次（建议把 nextStartTime 设为目标日期 00:00:00），观察日志与 brand_data_push_task。

方式 B：直接插 mongo（批量/脚本）

db.report_job_task.insertOne({
  "_id": "Jd110ReportToHangao",
  "jobInfo": "xxx店==>hangao",
  "nextStartTime": "2026-10-20 00:00:00",
  "reportPath": "http://10.0.1.216:9008/openapi/spider_data",
  "targetPath": "http://isb-gateway.ikunchi.com/esb-broker-provider/message/send?sourceService=report&sourceMethod=jd-order&targetService=hangao_report&targetMethod=jd_report_order",
  "intervalMinutes": 1440,
  "compensationMinutes": 0,
  "retryCount": 3,
  "retrySecond": 10,
  "pageSize": 50,
  "pageSleep": 300,
  "param": {
    "table": "src_ppzh_report_manage_pro",
    "db_type": "pg",
    "fields": "record_load_time,username,start_date,end_date,period,brand,time,pro_name,sku,first_category_name,second_category_name,third_category_name,shop_name,shop_type,pv,uv,avg_visit_num,avg_stay_time,deal_user,deal_rate,deal_num,deal_pro_num,deal_amt,deal_price_avg,to_cart_num,to_cart_rate",
    "shop_name": "xxx店",
    "period": "day"
  },
  "timeFieldList": ["time", "start_date"],
  "timePattern": "yyyy-MM-dd",
  "successField": "status",
  "successValue": "200",
  "countField": "count",
  "dataListField": "results",
  "createTime": new Date(),
  "updateTime": new Date(),
  "version": NumberLong(0)
})复制

4.7 补数（重跑某一天）

1. 在 aiops 页面找到任务，点「改时间」，把 下次执行时间 改为要补的日期（页面会强制成 YYYY-MM-DD 00:00:00）。
2. 到 xxl-job 手动触发一次该任务。
3. 观察：brand_data_push_task 新增记录 → broker_message 的 resultStatus。
4. 要补多天就重复 2~3 步；任务每次只前进一天。

5Mapping 推送实现

核心类：mall-pipeline-provider/src/main/java/com/kunchi/mall/pipeline/job/report/CommSkuMappingPublishJobHandler.java

5.1 数据从哪来

5.2 全量推送流程

1. 读 MappingReportToHangao 配置（pageSize=50）。
2. 调 fms-fnd POST /brandPlatformSkuMapping/v2/searchMapping，请求体只有 pageNo/pageSize（param 为空），不传 status，即不过滤状态。
3. 默认排序 updateTime desc, status desc, id desc。
4. 逐页把 DB 字段转成汉高字段（见 5.3），productCode 为空的记录直接丢弃。
5. 每页组装 {"size":N,"data":[...]}，调 pushOnce 投 esb-broker。
6. 页间休眠 pageSleep；全部完成后 nextStartTime += intervalMinutes（这个值不参与取数）。

5.3 字段转换表

5.4 过滤、版本与幂等

• 空条码过滤：productCode 为空 → 不推送。当前约 36% 的映射因此不会到达汉高（这是设计上的过滤，不是 bug）。
• 空页：整页都被过滤时，data-flow 返回 201 + "无效数据"，broker 记 OK，监控上看不出来。
• 版本机制：编辑时如果 subSkuMapping 有变化，后端会：
  • 把原记录置为 status=0、expire_time=昨天 23:59:59；
  • 新增一条 version=0、effective_time=今天 00:00:00、expire_time=9999 的记录。
  因为全量推送不过滤状态，所以旧记录（enableStates=0）和新记录都会推给汉高，由汉高按 id 去启用/禁用。
• 自动同步幂等：addMapping 按 (shopId, platformCode) 判重，任意状态已存在就跳过，不会更新名称。

5.5 新增一个店铺的 Mapping

1. 确认 shop_config 已有该店铺（shop_id、shop_name、current_shop_name、is_deleted=0）。没有就找 fms-fnd 负责人加。
2. 在 ffc-web「商品对照」用 Excel 导入初始映射（见 6.9）。
3. 确认汉高模板里已有该 shopId 分支（JD 模板是按 shop_name/username 映射的，新店铺可能要加分支）。
4. 跑一次 Mapping 任务，检查 broker_message 里 shopId 是否正确。

6前端操作：ffc-web 商品对照

入口：基础数据 → 商品对照，路由 /goods-comparison，面包屑「基础数据 / 商品码对照维护列表」。对应代码 ffc-web/src/views/baseData/goods-comparison.vue。

6.1 页面结构

6.2 查询

输入条件后点「查询」，实际请求：

POST /fms-fnd-provider/brandPlatformSkuMapping/v2/findByQoWithPage
{
  "pageNo": 1,
  "pageSize": 20,
  "platformCode": "100142031852",
  "shopName": "汉高家清京东自营旗舰店",
  "productCode": "6920177948308",
  "operator": "geyongqiang",
  "status": 1
}复制

所有条件都是模糊匹配（Containing）。「清空」会重置表单并重新查询。

6.3 列表字段含义

展开行可以看到子品项：商品名称、商品条形码、数量、是否告警、标签。

6.4 新增映射

1. 点「新增」，弹窗标题「新增商品码对照」。
2. 填写：
   • 店铺名称（必填）：必须与 shop_config 完全一致；
   • 平台侧商品编码（必填）：平台 SKU；
   • 平台商品名称（必填）；
   • 条形码：新增时该字段不参与提交，由子表自动推导；
   • TP名称默认“鲲驰”，状态默认启用，操作人自动取登录账号（只读）。
3. 在子表点「增行」，填写子品项：
   • 商品名称、商品条形码、数量；
   • 是否告警：默认“是”，用于完整性告警；
   • 标签：正装 / 赠品 / 促销品。
4. 点「确定」，请求 POST /v2/save。

6.5 编辑映射（版本机制，重点）

1. 点行内「编辑」，弹窗回填当前记录与子表。
2. 可改：店铺名称、平台侧商品编码、平台商品名称、子表（名称/条码/数量/告警/标签）。主表「条形码」即使改了，也会被子表推导值覆盖。
3. 点「确定」，请求 POST /v2/update（提交完整行对象 + 最新子表）。

6.6 启用 / 禁用

• 行内「禁用」：POST /v2/update，{"ids":[id],"status":0}。
• 行内「启用」：{"ids":[id],"status":1}；已失效（expire_time ≤ 现在）的记录不能启用，会被跳过。
• 批量禁用/启用：勾选多行后点顶部按钮，逻辑同上。

禁用只影响 enableStates，不影响是否推送；要停止推送只能删除或让 product_code 为空。

6.7 批量删除

勾选行 → 「批量删除」→ 请求 POST /v2/deleteBatch，物理删除，不可恢复。删除后汉高侧不会自动删除，只会停止后续推送。

6.8 导出

点「导出」，按当前查询条件请求 GET /v2/exportMapping，下载 Excel。导出不受分页限制（后端按 Qo 查询全量）。

6.9 导入（最常用，必须掌握）

步骤

1. 点「下载模板」，得到 商品码对照.xlsx。
2. 按模板填写，一次最多传 3 个文件（可多选）。
3. 点「导入」，上传后返回形如 导入成功：X 条；重复/跳过：Y 条。
4. 刷新列表核对。

模板列（必须与表头一致）

导入规则

• 按 (店铺ID, 商家产品编码) 分组聚合；同一组多行会合并子品项。
• 已存在（任意状态）的分组整组跳过，不更新。
• product_code 推导：套装 = 所有子条码按 quantity 重复拼接；单品 = 69 条形码或首个子条码。
• 落库 source_type=1、version=0、status 取模板值。
• 店铺 ID 不在允许列表 → 整行跳过。

6.10 页面操作的副作用（务必知道）

7前端操作：aiops-web Report任务配置

入口：汉高 → Report任务配置，路由 /reportJobTask。对应代码 aiops-web/src/views/reportJobTask/。

7.1 页面结构

接口前缀：/mall-pipeline-provider/mongo/job/report/task，后端是 JobTaskController.java。

7.2 查询

POST /page，请求 {"jobName":"","jobInfo":"","pageNo":1,"pageSize":10}。

• jobName：精确匹配（Mongo _id）。
• jobInfo：模糊匹配（正则，已转义）。
• 列表只返回 5 个字段：jobName、jobInfo、nextStartTime、targetPath、param。

7.3 列表字段

7.4 新增任务（逐字段）

点「新增任务」，弹窗分四块：

基础信息

推送配置

执行参数

查询与解析

7.5 改时间（补数最常用）

行内「改时间」→ 选日期 → 确认。请求 POST /updateStartTime，参数 jobName、nextStartTime=YYYY-MM-DD 00:00:00。

7.6 改参数（只能改 4 个字段）

行内「改参数」→ 可修改：任务描述、下次执行时间、目标推送地址、查询参数(param)。其余字段保持原值。

请求 POST /updateV2，后端白名单：

jobInfo、nextStartTime、targetPath、param复制

7.7 删除

行内「删除」→ 二次确认 → POST /delete?jobName=xxx，物理删除，不可恢复。删除后 xxl-job 再触发会报「未配置任务」。

7.8 新增后还要做什么

8后端配置（不走页面）

页面覆盖不到的地方：直接改 mongo、加模板、加路由。改之前先在测试环境验证。

8.1 新增一个汉高店铺（完整清单）

8.2 新增一个汉高接口模板

假设新增「退货」推送。需要两件事：

① 新增 API 路径映射

db.data_flow_api.insertOne({
  "_id": "/message/template/receive/ffc/hangao/productReturn",
  "templateId": "ffc-hangao_product_return",
  "info": "汉高退货",
  "createTime": new Date(),
  "updateTime": new Date(),
  "version": 1
})复制

② 新增模板

db.data_flow_template.insertOne({
  "_id": "ffc-hangao_product_return",
  "info": "汉高退货",
  "targetHost": "https://dmeapi-prod.schwarzkopfclub.com.cn",
  "targetPath": "/commodity/return",
  "serviceId": "hangao",
  "expectField": "code",
  "expectValue": "0",
  "template": "[{\"refundNum\":\"${requestBody.refundId}\",\"skuId\":\"${requestBody.skuId}\"}]",
  "createTime": new Date(),
  "updateTime": new Date(),
  "version": 0
})复制

8.3 新增一个 esb-broker 路由

db.broker_subscriber.insertOne({
  "_id": "hangao_product-product_return",
  "service": "hangao_product",
  "method": "product_return",
  "binding": "broker_topic@broker.basic@basic",
  "targetHost": "http://data-flow-provider",
  "targetPath": "/message/template/receive/ffc/hangao/productReturn",
  "subscriberType": "HTTP",
  "expectCode": "code",
  "expectValue": "201",
  "retryWhenError": true,
  "retryWhenFailure": true,
  "retryTtl": 300000,
  "maxRetry": 3,
  "description": "汉高退货"
})复制

绑定关系 broker_topic@broker.basic@basic 已存在（destination=broker.basic，routingKey=basic），不用新建 binding/queue/listener。

8.4 xxl-job 配置约定

8.5 配置修改检查清单

• 新增 POS 任务：_id 唯一、timeFieldList 非空、pageSize > 0、intervalMinutes > 0、targetPath 的 targetMethod 与订阅者一致。
• 京东店铺要成对配置：常规 JdXXXReportToHangao（day）+ 回流 JdXXXReportToHangao_huiliu（day_p1），回流任务按月集中触发。
• 新增店铺：shop_config 有记录、模板有 shopId 分支、mapping 已导入。
• 新增平台接口：data_flow_api → data_flow_template → broker_subscriber 三件套齐全。
• 改模板：先用「模板测试」接口 POST /template/test 验证输出是合法 JSON。
• 改完必须手动触发一次并查 broker_message。

9日常巡检与排障

这一章是可以照着执行的操作手册。建议每天上午跑一遍 9.1。

9.1 每日巡检（5 分钟）

① 推送流水成功率

SELECT job_info,
       COUNT(*) AS total,
       SUM(task_status = 2) AS ok,
       SUM(task_status = 3) AS fail,
       MIN(create_time) AS first_time,
       MAX(create_time) AS last_time
FROM fms_fnd.brand_data_push_task
WHERE create_time >= CURDATE() - INTERVAL 1 DAY
GROUP BY job_info
ORDER BY fail DESC, job_info;复制

关注：fail > 0；某任务 24 小时无记录（可能没调度或一直失败）。

② 水位是否停滞/滞后

db.report_job_task.find({},
  {jobInfo:1, nextStartTime:1, updateTime:1})
  .sort({updateTime:-1})复制

关注：nextStartTime 是否落后业务日期 > 2 天；updateTime 是否超过 24 小时没变。

③ 下游失败与汉高业务码

db.broker_message.aggregate([
  {$match:{
    createTime:{$gte:new Date(Date.now()-24*3600*1000)},
    subscriberId:{$in:[
      "hangao_report-tm_report_order",
      "hangao_report-jd_report_order",
      "hangao_report-wph_report_order",
      "hangao_report-jd_pop_report_order",
      "hangao_product-product_mapping"
    ]}
  }},
  {$group:{_id:{sub:"$subscriberId", st:"$resultStatus"}, cnt:{$sum:1}}}
])复制

关注：ERROR（汉高业务码不对/模板异常）、FAILURE（网络/超时）。

④ Mapping 缺口

SELECT shop_name,
       COUNT(*) AS total,
       SUM(product_code IS NULL OR product_code = '') AS blank_code
FROM fms_fnd.brand_platform_sku_mapping
GROUP BY shop_name
ORDER BY blank_code DESC;复制

关注：blank_code 增长（新增 SKU 未补条码，不会推给汉高）。

⑤ Mapping 是否每天跑了

SELECT DATE(create_time) AS d, COUNT(*) AS pages
FROM fms_fnd.brand_data_push_task
WHERE job_info = '全量Mapping==>hangao'
  AND create_time >= CURDATE() - INTERVAL 7 DAY
GROUP BY DATE(create_time)
ORDER BY d DESC;复制

正常每天约 100 页（5,000 条 ÷ 50）。为 0 说明任务没跑或没数据。

⑥ 每月 7 号后：京东回流是否覆盖上月

SELECT job_info, COUNT(*) AS rows_cnt,
       MIN(create_time) AS first_time, MAX(create_time) AS last_time
FROM fms_fnd.brand_data_push_task
WHERE job_info LIKE '%回流%'
  AND create_time >= DATE_FORMAT(CURDATE(), '%Y-%m-01')
GROUP BY job_info;复制

记录数应约等于“上月天数 × 每天常规条数”。再抽查 mongo broker_message：reqBody 含 "period":"day_p1"，日期覆盖上月 1 号到月底。

9.2 症状 → 排查路径

9.3 全链路追踪（一次推送的完整证据链）

db.broker_message.findOne({_id:"你的-esb-id"})
db.broker_logging.find({messageId:"你的-esb-id"}).sort({reqTime:1})

9.4 补数 / 重推

POS 补数

1. aiops 页面「改时间」→ 目标日期。
2. xxl-job 手动触发该任务。
3. 查 brand_data_push_task 与 broker_message。
4. 多天则重复；每次只补一天。

Mapping 重推

1. mapping 是每天全量，通常不需要单独补；改完数据等次日 03:00 即可。
2. 急用：xxl-job 手动触发 MappingReportToHangao（不改任何参数）。

京东回流补推（次月）

1. 把对应店铺的 *_huiliu 任务水位改到目标月 1 号（如 2026-09-01 00:00:00）。
2. 连续触发，直到水位走过月底；漏哪天就单独把水位拨到那天再触发一次。
3. 核对 broker_message 的 period=day_p1 与日期覆盖。

单条消息重推（esb-broker）

POST /esb-broker-provider/message/sendById?id={messageId}复制

9.5 已知问题与规避（接手必读）

10变更与发布

改哪里、要不要发版、发布顺序，一张表说清。

10.1 变更类型与发布方式

10.2 代码入口索引

10.3 发布顺序建议

1. 先发布 fms-fnd（如果改了映射规则/接口）。
2. 再发布 pipeline（依赖 fms-fnd 接口）。
3. 再发布 data-flow（如果改了模板加载/鉴权逻辑；只改 mongo 模板不用发版）。
4. 最后发布前端页面。

11附录

11.1 接口清单

11.2 表/集合清单

11.3 枚举字典

11.4 店铺 ID 对照

12FAQ 20 问