供应商管理数据同步, tms系统产品,服务渠道,产品渠道关系数据同步

This commit is contained in:
wuweihua
2026-10-12 10:28:06 +08:00
parent d08485d738
commit 2a9825c498
29 changed files with 1646 additions and 127 deletions
@@ -4,14 +4,22 @@ import java.util.List;
import java.util.Map;
/**
* 老系统 Oracle TAB_SUPPLIER 只读查询(SLAVE 数据源,账号只读)
* 老系统 Oracle 供应商相关表只读查询(SLAVE 数据源,账号只读)
* 数据源(2026-10-10 修正,原 TAB_SUPPLIER 取数有误):
* TAB_SERVICE_INFO 服务商(即供应商主数据)
* TAB_SERVICE_CHANNEL 服务渠道(一个服务商下多个渠道,按 SERVICE_CODE 关联)
*
* @author bms
*/
public interface OracleSupplierSyncMapper
{
/**
* 全量读取 TAB_SUPPLIER 同步所需列(数据量约数百行,单次读取)
* 全量读取服务商(供应商)主数据 TAB_SERVICE_INFO
*/
List<Map<String, Object>> selectAllTabSupplier();
List<Map<String, Object>> selectAllServiceInfo();
/**
* 全量读取服务渠道 TAB_SERVICE_CHANNEL(仅需服务商编码+渠道编码,用于聚合 channel_ids)
*/
List<Map<String, Object>> selectAllServiceChannel();
}
@@ -10,7 +10,7 @@ import com.bms.common.enums.DataSourceType;
import com.bms.customer.supplierinfo.mapper.OracleSupplierSyncMapper;
/**
* Oracle TAB_SUPPLIER 读取器。
* Oracle 服务商/服务渠道读取器。
* 走 SLAVE(Oracle 只读账号)数据源;必须独立 Bean,
* 否则 @DataSource 切面在同类自调用时不生效。
*
@@ -22,14 +22,30 @@ public class OracleSupplierReader
@Autowired
private OracleSupplierSyncMapper oracleSupplierSyncMapper;
/** 读取服务商(供应商)主数据 TAB_SERVICE_INFO */
@DataSource(DataSourceType.SLAVE)
public List<Map<String, Object>> readAll()
public List<Map<String, Object>> readServiceInfo()
{
try
{
// 避免 PageHelper 把 Oracle 查询包成分页语句
PageHelper.clearPage();
return oracleSupplierSyncMapper.selectAllTabSupplier();
return oracleSupplierSyncMapper.selectAllServiceInfo();
}
finally
{
PageHelper.clearPage();
}
}
/** 读取服务渠道 TAB_SERVICE_CHANNEL */
@DataSource(DataSourceType.SLAVE)
public List<Map<String, Object>> readServiceChannels()
{
try
{
PageHelper.clearPage();
return oracleSupplierSyncMapper.selectAllServiceChannel();
}
finally
{
@@ -1,11 +1,10 @@
package com.bms.customer.supplierinfo.sync;
import java.math.BigDecimal;
import java.math.RoundingMode;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Date;
import java.util.HashSet;
import java.util.LinkedHashMap;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
@@ -13,16 +12,18 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import com.bms.common.utils.StringUtils;
import com.bms.customer.supplierinfo.domain.BmsSupplier;
import com.bms.customer.supplierinfo.mapper.SupplierSyncMapper;
/**
* 老系统 Oracle TAB_SUPPLIER -> sup_supplier 每日同步。
* 老系统 Oracle 服务商数据 -> sup_supplier 每日同步。
*
* 范围:只同步 sup_supplier 现有列能承接的字段;三个类型字段口径完全以 TAB_SUPPLIER 为准
* (供应商类型 SUP_TYPE、开票类型 INVOICE_TYPE、卡车行类型 KACHE_TYPE)。
* 按 supplier_code 判定新增/更新;源为空的非类型字段不写(新增走默认值/更新保留原值)。
* 数据源(2026-10-10 修正,原 TAB_SUPPLIER 取数有误):
* TAB_SERVICE_INFO 服务商(即供应商)主数据
* TAB_SERVICE_CHANNEL 服务渠道,按 SERVICE_CODE 归属服务商,
* 其全部渠道编码(含停用)聚合写入 channel_ids
*
* 按 supplier_code(SERVICE_CODE) 判定新增/更新;源为空的字段不写(新增走默认值/更新保留原值)。
*
* @author bms
*/
@@ -34,15 +35,18 @@ public class SupplierSyncService
/** 写库人标记,便于事后甄别同步数据 */
public static final String SYNC_MARKER = "oracle-sync";
/** 三个类型字段的合法字典值(以老系统 TAB_SUPPLIER 为准) */
private static final Set<String> SUP_TYPE_CODES = new HashSet<>(Arrays.asList("1", "2", "3", "4", "5", "6"));
private static final Set<String> INVOICE_TYPE_CODES = new HashSet<>(Arrays.asList("1", "2", "3", "4"));
private static final Set<String> KACHE_TYPE_CODES = new HashSet<>(Arrays.asList("0", "1"));
/** 源中文服务商类型 -> sup_supplier.supplier_type 字典码(4清关行 5尾派服务商) */
private static final String TYPE_QINGGUAN = "清关";
private static final String TYPE_WEICHENG = "尾程渠道";
private static final String CODE_QINGGUAN_HANG = "4";
private static final String CODE_WEIPAI = "5";
/** sup_supplier 字符串列长度 */
private static final int LEN_CODE = 50;
private static final int LEN_NAME = 100;
private static final int LEN_REMARK = 500;
/** channel_ids varchar(1000) */
private static final int LEN_CHANNEL_IDS = 1000;
/** 告警明细最多打印条数 */
private static final int WARN_LOG_LIMIT = 50;
@@ -63,7 +67,21 @@ public class SupplierSyncService
public String syncFromOracle()
{
long t0 = System.currentTimeMillis();
List<Map<String, Object>> rows = oracleSupplierReader.readAll();
List<Map<String, Object>> rows = oracleSupplierReader.readServiceInfo();
List<Map<String, Object>> channelRows = oracleSupplierReader.readServiceChannels();
// 按服务商编码聚合渠道编码(去重、保序)
Map<String, LinkedHashSet<String>> channelMap = new LinkedHashMap<>();
for (Map<String, Object> ch : channelRows)
{
String serviceCode = clean(getStr(ch, "SERVICE_CODE"));
String channelCode = clean(getStr(ch, "CHANNEL_CODE"));
if (serviceCode == null || channelCode == null)
{
continue;
}
channelMap.computeIfAbsent(serviceCode, k -> new LinkedHashSet<>()).add(channelCode);
}
Set<String> existingCodes = new HashSet<>(supplierSyncMapper.selectExistingCodes());
@@ -78,8 +96,8 @@ public class SupplierSyncService
{
int rowNo = i + 1;
Map<String, Object> row = rows.get(i);
String code = clean(getStr(row, "SUP_CODE"));
String name = clean(getStr(row, "SUP_NAME"));
String code = clean(getStr(row, "SERVICE_CODE"));
String name = clean(getStr(row, "SERVICE_NAME"));
if (code == null && name == null)
{
blankRows++;
@@ -87,12 +105,12 @@ public class SupplierSyncService
}
if (code == null)
{
errors.add("行" + rowNo + ": SUP_CODE 为空,跳过");
errors.add("行" + rowNo + ": SERVICE_CODE 为空,跳过");
continue;
}
if (name == null)
{
errors.add("行" + rowNo + " " + code + ": SUP_NAME 为空,跳过");
errors.add("行" + rowNo + " " + code + ": SERVICE_NAME 为空,跳过");
continue;
}
if (!seenCodes.add(code))
@@ -101,7 +119,8 @@ public class SupplierSyncService
continue;
}
BmsSupplier s = mapRow(row, code, name, warnings);
LinkedHashSet<String> channels = channelMap.getOrDefault(code, new LinkedHashSet<>());
BmsSupplier s = mapRow(row, code, name, channels, warnings);
if (existingCodes.contains(code))
{
s.setUpdateBy(SYNC_MARKER);
@@ -119,8 +138,9 @@ public class SupplierSyncService
long costMs = System.currentTimeMillis() - t0;
String summary = String.format(
"Oracle供应商同步完成:源行 %d,空行 %d,新增 %d,更新 %d,跳过错误 %d,告警 %d,耗时 %dms",
rows.size(), blankRows, written[0], written[1], errors.size(), warnings.size(), costMs);
"Oracle服务商同步完成:服务商 %d 家(渠道 %d 个),空行 %d,新增 %d,更新 %d,跳过错误 %d,告警 %d,耗时 %dms",
rows.size(), channelRows.size(), blankRows, written[0], written[1],
errors.size(), warnings.size(), costMs);
log.info(summary);
errors.forEach(e -> log.warn(" 同步跳过: {}", e));
warnings.stream().limit(WARN_LOG_LIMIT).forEach(w -> log.warn(" 同步告警: {}", w));
@@ -130,7 +150,7 @@ public class SupplierSyncService
}
if (!errors.isEmpty())
{
log.warn("同步存在 {} 条跳过记录,请核对 TAB_SUPPLIER 源数据", errors.size());
log.warn("同步存在 {} 条跳过记录,请核对 TAB_SERVICE_INFO 源数据", errors.size());
}
return summary;
}
@@ -138,30 +158,48 @@ public class SupplierSyncService
/**
* 单行映射
*/
private BmsSupplier mapRow(Map<String, Object> row, String code, String name, List<String> warnings)
private BmsSupplier mapRow(Map<String, Object> row, String code, String name,
LinkedHashSet<String> channels, List<String> warnings)
{
BmsSupplier s = new BmsSupplier();
s.setSupplierCode(truncate(code, LEN_CODE, code, "supplier_code", warnings));
s.setSupplierName(truncate(name, LEN_NAME, code, "supplier_name", warnings));
s.setDelFlag("0");
s.setSupplierType(toCode(row, "SUP_TYPE", SUP_TYPE_CODES, code, "供应商类型", warnings));
s.setInvoiceType(toCode(row, "INVOICE_TYPE", INVOICE_TYPE_CODES, code, "开票类型", warnings));
s.setKacheType(toCode(row, "KACHE_TYPE", KACHE_TYPE_CODES, code, "卡车行类型", warnings));
s.setSupplierType(mapServiceType(getStr(row, "SERVICE_TYPE"), code, warnings));
s.setIsEnable(mapEnable(getStr(row, "BL_OPEN")));
s.setEd(toDecimal(row, "QUOTA", code, warnings));
Long period = toLong(row, "SETTLEMENT_PERIOD", code, warnings);
s.setZq(period);
s.setRemark(truncate(getStr(row, "REMARK"), LEN_REMARK, code, "remark", warnings));
Date createTime = getDate(row, "CREATE_TIME");
Date createTime = getDate(row, "CREATE_DATE");
s.setCreateTime(createTime);
s.setChannelIds(joinChannelCodes(channels, code, warnings));
return s;
}
/**
* 源 SERVICE_TYPE 为中文:尾程渠道 -> 5(尾派服务商),清关 -> 4(清关行);
* 其他/空值置空并告警,由业务人员手工选择。
*/
private static String mapServiceType(String raw, String supplierCode, List<String> warnings)
{
String v = clean(raw);
if (v == null)
{
return null;
}
if (TYPE_WEICHENG.equals(v))
{
return CODE_WEIPAI;
}
if (TYPE_QINGGUAN.equals(v))
{
return CODE_QINGGUAN_HANG;
}
warnings.add(supplierCode + ": 未知服务商类型 " + v + ",类型置空");
return null;
}
/** BL_OPEN:1启用 -> is_enable=0(正常);0停用 -> 1;空按正常 */
private static String mapEnable(String v)
{
@@ -174,37 +212,34 @@ public class SupplierSyncService
}
/**
* Oracle NUMBER(1) 经 JDBC 多为 BigDecimal(String.valueOf 可能带 ".0"),
* 统一按整数取值后校验字典合法性;源为 null 返回 null(更新时以源为准置空)。
* 渠道编码英文逗号拼接;channel_ids 长度上限 1000,
* 超长时按完整编码边界截断并告警(保证不产生半个编码)。
*/
private static String toCode(Map<String, Object> row, String key, Set<String> validCodes,
String supplierCode, String label, List<String> warnings)
private static String joinChannelCodes(LinkedHashSet<String> channels, String supplierCode, List<String> warnings)
{
Object v = getIgnoreCase(row, key);
if (v == null || StringUtils.isEmpty(String.valueOf(v).trim()))
if (channels == null || channels.isEmpty())
{
return null;
}
String code;
if (v instanceof Number)
StringBuilder sb = new StringBuilder();
int included = 0;
for (String ch : channels)
{
code = String.valueOf(((Number) v).longValue());
}
else
{
code = String.valueOf(v).trim();
int dot = code.indexOf('.');
if (dot >= 0)
int add = ch.length() + (sb.length() > 0 ? 1 : 0);
if (sb.length() + add > LEN_CHANNEL_IDS)
{
code = code.substring(0, dot);
warnings.add(String.format("%s: 渠道数量过多,channel_ids 超 %d 字符,仅保留前 %d 个渠道(共 %d 个)",
supplierCode, LEN_CHANNEL_IDS, included, channels.size()));
break;
}
if (sb.length() > 0)
{
sb.append(',');
}
sb.append(ch);
included++;
}
if (!validCodes.contains(code))
{
warnings.add(supplierCode + ": 未知" + label + " " + v + ",该类型置空");
return null;
}
return code;
return sb.toString();
}
private static String clean(String v)
@@ -231,45 +266,6 @@ public class SupplierSyncService
return s;
}
private static BigDecimal toDecimal(Map<String, Object> row, String key, String code, List<String> warnings)
{
Object v = getIgnoreCase(row, key);
if (v == null || StringUtils.isEmpty(String.valueOf(v).trim()))
{
return null;
}
try
{
BigDecimal d = (v instanceof BigDecimal) ? (BigDecimal) v
: (v instanceof Number ? BigDecimal.valueOf(((Number) v).doubleValue())
: new BigDecimal(String.valueOf(v).trim()));
return d.setScale(2, RoundingMode.HALF_UP);
}
catch (Exception e)
{
warnings.add(code + ": 字段 " + key + " 非法数值 " + v + ",该字段不写入");
return null;
}
}
private static Long toLong(Map<String, Object> row, String key, String code, List<String> warnings)
{
Object v = getIgnoreCase(row, key);
if (v == null || StringUtils.isEmpty(String.valueOf(v).trim()))
{
return null;
}
try
{
return (v instanceof Number) ? ((Number) v).longValue() : Long.parseLong(String.valueOf(v).trim());
}
catch (Exception e)
{
warnings.add(code + ": 字段 " + key + " 非法整数 " + v + ",该字段不写入");
return null;
}
}
private static Date getDate(Map<String, Object> row, String key)
{
Object v = getIgnoreCase(row, key);
@@ -16,7 +16,7 @@ public class SupplierSyncTask
private SupplierSyncService supplierSyncService;
/**
* 每天 00:00 从 Oracle TAB_SUPPLIER 全量同步供应商主数据到 sup_supplier
* 每天 00:00 从 Oracle TAB_SERVICE_INFO/TAB_SERVICE_CHANNEL 全量同步服务商(供应商)数据到 sup_supplier
*/
public void syncFromOracle()
{
@@ -0,0 +1,85 @@
package com.bms.masterdata.domain;
import java.math.BigDecimal;
import java.util.Date;
import com.fasterxml.jackson.annotation.JsonFormat;
import lombok.Data;
/**
* 线路产品镜像对象 srv_line_product(Oracle WEISHI_USER.TAB_LINE_PRODUCT)
*
* @author bms
* @date 2026-10-12
*/
@Data
public class LineProduct
{
/** 源主键 ID */
private String id;
/** 产品(线路)编码 */
private String lineCode;
/** 产品(线路)名称 */
private String lineName;
private String productType;
private String lineType;
/** 是否可收货(1是 0否) */
private String blCargo;
private String cargoName;
private String productStorage;
private String productFirst;
private String productClearance;
private String productStorageCode;
private String productFirstCode;
private String productClearanceCode;
/** 是否启用(1启用 0停用) */
private String blOpen;
/** 是否分段(1是 0否) */
private String blSection;
private String dispCode;
private String dispName;
private String feeClass;
private String shipperId;
/** 发货规则 */
private String shipmentRule;
/** 限重(kg) */
private BigDecimal weight;
private String cargoType;
private String productCategory;
/** 体积基数(抛货系数) */
private BigDecimal volumeBasis;
private String weightType;
private String serviceType;
private String shipmentType;
private BigDecimal taxAmount;
private String taxCurrency;
private String batteryType;
private String remark;
private String createName;
private String createManCode;
@JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss", timezone = "GMT+8")
private Date createDate;
private String modifier;
private String modifierCode;
@JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss", timezone = "GMT+8")
private Date modifyDate;
@JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss", timezone = "GMT+8")
private Date gmtCreate;
@JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss", timezone = "GMT+8")
private Date gmtModified;
}
@@ -0,0 +1,45 @@
package com.bms.masterdata.domain;
import java.math.BigDecimal;
import lombok.Data;
/**
* 产品-渠道关系镜像对象 srv_product_channel(Oracle WEISHI_USER.TAB_PRODUCT_CHANNEL)
* 注意:源端 (productId, channelCode) 不唯一(同产品同渠道按国家/重量区间多行),主键为源 GUID。
*
* @author bms
* @date 2026-10-12
*/
@Data
public class ProductChannel
{
/** 源主键 GUID */
private String guid;
/** 产品ID(srv_line_product.id) */
private String productId;
/** 渠道编码(srv_service_channel.channel_code) */
private String channelCode;
private String channelName;
/** 优先级(数字越小越优先) */
private Integer priority;
/** 是否启用(1启用 0停用) */
private String isEnabled;
private String lineCode;
private String lineName;
private String countryCode;
private BigDecimal startWeight;
private BigDecimal endWeight;
/** 下单量 */
private Long orderCount;
/** 下单量占比 */
private BigDecimal orderCountRatio;
}
@@ -0,0 +1,69 @@
package com.bms.masterdata.domain;
import java.util.Date;
import com.fasterxml.jackson.annotation.JsonFormat;
import lombok.Data;
/**
* 服务渠道镜像对象 srv_service_channel(Oracle WEISHI_USER.TAB_SERVICE_CHANNEL)
*
* @author bms
* @date 2026-10-12
*/
@Data
public class ServiceChannel
{
/** 源主键 ID */
private String id;
/** 渠道编码 */
private String channelCode;
/** 渠道名称 */
private String channelName;
/** 所属服务商编码 */
private String serviceCode;
/** 所属服务商名称 */
private String serviceName;
/** 是否启用(1启用 0停用) */
private String blOpen;
/** 是否支持多票(1是 0否) */
private String blMultiple;
private String cargoType;
private String payType;
private String shipperName;
private String shipperId;
private String queryType;
private String trajectoryCode;
private String finalServiceName;
private String desPort;
/** 重量类型 */
private String weightType;
/** 体积除数(抛货系数) */
private Long dimDivisor;
private String weightRule;
private String shipmentType;
private String dataFrom;
private String remarks;
private String createName;
private String createManCode;
@JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss", timezone = "GMT+8")
private Date createDate;
private String modifier;
private String modifierCode;
@JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss", timezone = "GMT+8")
private Date modifyDate;
@JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss", timezone = "GMT+8")
private Date gmtCreate;
@JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss", timezone = "GMT+8")
private Date gmtModified;
}
@@ -0,0 +1,25 @@
package com.bms.masterdata.mapper;
import java.util.List;
import org.apache.ibatis.annotations.Param;
import com.bms.masterdata.domain.LineProduct;
import com.bms.masterdata.domain.ProductChannel;
import com.bms.masterdata.domain.ServiceChannel;
/**
* 渠道/产品 镜像表写库(MASTER 数据源,按源主键 upsert)
*
* @author bms
* @date 2026-10-12
*/
public interface ChannelMasterSyncMapper
{
/** 按源 ID 批量 upsert 服务渠道 */
int upsertServiceChannel(@Param("list") List<ServiceChannel> list);
/** 按源 ID 批量 upsert 线路产品 */
int upsertLineProduct(@Param("list") List<LineProduct> list);
/** 按源 GUID 批量 upsert 产品-渠道关系 */
int upsertProductChannel(@Param("list") List<ProductChannel> list);
}
@@ -0,0 +1,22 @@
package com.bms.masterdata.mapper;
import java.util.List;
import java.util.Map;
/**
* Oracle 渠道/产品 主数据只读查询(SLAVE 数据源,账号只读)
*
* @author bms
* @date 2026-10-12
*/
public interface OracleChannelMasterMapper
{
/** 全量读取服务渠道 TAB_SERVICE_CHANNEL */
List<Map<String, Object>> selectAllServiceChannel();
/** 全量读取线路产品 TAB_LINE_PRODUCT */
List<Map<String, Object>> selectAllLineProduct();
/** 全量读取产品-渠道关系 TAB_PRODUCT_CHANNEL */
List<Map<String, Object>> selectAllProductChannel();
}
@@ -0,0 +1,18 @@
package com.bms.masterdata.mapper;
import java.util.List;
import com.bms.masterdata.domain.ServiceChannel;
/**
* 服务渠道镜像表查询(MASTER 数据源)
*
* @author bms
* @date 2026-10-12
*/
public interface ServiceChannelMapper
{
/**
* 查询服务渠道列表(支持渠道编码/名称/服务商编码/启用状态过滤,非分页)
*/
List<ServiceChannel> selectServiceChannelList(ServiceChannel query);
}
@@ -0,0 +1,18 @@
package com.bms.masterdata.service;
import java.util.List;
import com.bms.masterdata.domain.ServiceChannel;
/**
* 服务渠道镜像 服务层
*
* @author bms
* @date 2026-10-12
*/
public interface IServiceChannelService
{
/**
* 查询服务渠道列表(非分页,供供应商绑定渠道等下拉选择场景)
*/
List<ServiceChannel> selectServiceChannelList(ServiceChannel query);
}
@@ -0,0 +1,27 @@
package com.bms.masterdata.service.impl;
import java.util.List;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import com.bms.masterdata.domain.ServiceChannel;
import com.bms.masterdata.mapper.ServiceChannelMapper;
import com.bms.masterdata.service.IServiceChannelService;
/**
* 服务渠道镜像 服务实现
*
* @author bms
* @date 2026-10-12
*/
@Service
public class ServiceChannelServiceImpl implements IServiceChannelService
{
@Autowired
private ServiceChannelMapper serviceChannelMapper;
@Override
public List<ServiceChannel> selectServiceChannelList(ServiceChannel query)
{
return serviceChannelMapper.selectServiceChannelList(query);
}
}
@@ -0,0 +1,264 @@
package com.bms.masterdata.sync;
import java.math.BigDecimal;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.Map;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import com.bms.masterdata.domain.LineProduct;
import com.bms.masterdata.domain.ProductChannel;
import com.bms.masterdata.domain.ServiceChannel;
/**
* Oracle 渠道/产品 主数据同步编排:
* 顺序 渠道 → 产品 → 产品渠道关系,按源主键幂等 upsert,可重复执行。
*
* @author bms
* @date 2026-10-12
*/
@Service
public class ChannelMasterSyncService
{
private static final Logger log = LoggerFactory.getLogger(ChannelMasterSyncService.class);
@Autowired
private OracleChannelMasterReader reader;
@Autowired
private ChannelMasterSyncWriter writer;
/**
* 全量同步三表,返回简要对账信息
*/
public String syncFromOracle()
{
long start = System.currentTimeMillis();
List<Map<String, Object>> channelRows = reader.readServiceChannels();
List<ServiceChannel> channels = new ArrayList<>(channelRows.size());
int skipChannel = 0;
for (Map<String, Object> row : channelRows)
{
String id = str(row, "ID");
if (id == null) { skipChannel++; continue; }
ServiceChannel c = new ServiceChannel();
c.setId(id);
c.setChannelCode(str(row, "CHANNEL_CODE"));
c.setChannelName(str(row, "CHANNEL_NAME"));
c.setServiceCode(str(row, "SERVICE_CODE"));
c.setServiceName(str(row, "SERVICE_NAME"));
c.setBlOpen(flag(row, "BL_OPEN"));
c.setBlMultiple(flag(row, "BL_MULTIPLE"));
c.setCargoType(str(row, "CARGO_TYPE"));
c.setPayType(str(row, "PAY_TYPE"));
c.setShipperName(str(row, "SHIPPER_NAME"));
c.setShipperId(str(row, "SHIPPER_ID"));
c.setQueryType(str(row, "QUERY_TYPE"));
c.setTrajectoryCode(str(row, "TRAJECTORY_CODE"));
c.setFinalServiceName(str(row, "FINAL_SERVICE_NAME"));
c.setDesPort(str(row, "DES_PORT"));
c.setWeightType(flag(row, "WEIGHT_TYPE"));
c.setDimDivisor(lo(row, "DIM_DIVISOR"));
c.setWeightRule(str(row, "WEIGHT_RULE"));
c.setShipmentType(str(row, "SHIPMENT_TYPE"));
c.setDataFrom(str(row, "DATA_FROM"));
c.setRemarks(str(row, "REMARKS"));
c.setCreateName(str(row, "CREATE_NAME"));
c.setCreateManCode(str(row, "CREATE_MAN_CODE"));
c.setCreateDate(date(row, "CREATE_DATE"));
c.setModifier(str(row, "MODIFIER"));
c.setModifierCode(str(row, "MODIFIER_CODE"));
c.setModifyDate(date(row, "MODIFY_DATE"));
c.setGmtCreate(date(row, "GMT_CREATE"));
c.setGmtModified(date(row, "GMT_MODIFIED"));
channels.add(c);
}
int channelAffected = writer.writeServiceChannels(channels);
List<Map<String, Object>> productRows = reader.readLineProducts();
List<LineProduct> products = new ArrayList<>(productRows.size());
int skipProduct = 0;
for (Map<String, Object> row : productRows)
{
String id = str(row, "ID");
if (id == null) { skipProduct++; continue; }
LineProduct p = new LineProduct();
p.setId(id);
p.setLineCode(str(row, "LINE_CODE"));
p.setLineName(str(row, "LINE_NAME"));
p.setProductType(str(row, "PRODUCT_TYPE"));
p.setLineType(str(row, "LINE_TYPE"));
p.setBlCargo(flag(row, "BL_CARGO"));
p.setCargoName(str(row, "CARGO_NAME"));
p.setProductStorage(str(row, "PRODUCT_STORAGE"));
p.setProductFirst(str(row, "PRODUCT_FIRST"));
p.setProductClearance(str(row, "PRODUCT_CLEARANCE"));
p.setProductStorageCode(str(row, "PRODUCT_STORAGE_CODE"));
p.setProductFirstCode(str(row, "PRODUCT_FIRST_CODE"));
p.setProductClearanceCode(str(row, "PRODUCT_CLEARANCE_CODE"));
p.setBlOpen(flag(row, "BL_OPEN"));
p.setBlSection(flag(row, "BL_SECTION"));
p.setDispCode(str(row, "DISP_CODE"));
p.setDispName(str(row, "DISP_NAME"));
p.setFeeClass(str(row, "FEE_CLASS"));
p.setShipperId(str(row, "SHIPPER_ID"));
p.setShipmentRule(flag(row, "SHIPMENT_RULE"));
p.setWeight(decimal(row, "WEIGHT"));
p.setCargoType(str(row, "CARGO_TYPE"));
p.setProductCategory(str(row, "PRODUCT_CATEGORY"));
p.setVolumeBasis(decimal(row, "VOLUME_BASIS"));
p.setWeightType(flag(row, "WEIGHT_TYPE"));
p.setServiceType(flag(row, "SERVICE_TYPE"));
p.setShipmentType(str(row, "SHIPMENT_TYPE"));
p.setTaxAmount(decimal(row, "TAX_AMOUNT"));
p.setTaxCurrency(str(row, "TAX_CURRENCY"));
p.setBatteryType(str(row, "BATTERY_TYPE"));
p.setRemark(str(row, "REMARK"));
p.setCreateName(str(row, "CREATE_NAME"));
p.setCreateManCode(str(row, "CREATE_MAN_CODE"));
p.setCreateDate(date(row, "CREATE_DATE"));
p.setModifier(str(row, "MODIFIER"));
p.setModifierCode(str(row, "MODIFIER_CODE"));
p.setModifyDate(date(row, "MODIFY_DATE"));
p.setGmtCreate(date(row, "GMT_CREATE"));
p.setGmtModified(date(row, "GMT_MODIFIED"));
products.add(p);
}
int productAffected = writer.writeLineProducts(products);
List<Map<String, Object>> relRows = reader.readProductChannels();
List<ProductChannel> relations = new ArrayList<>(relRows.size());
int skipRel = 0;
for (Map<String, Object> row : relRows)
{
String guid = str(row, "GUID");
if (guid == null) { skipRel++; continue; }
ProductChannel pc = new ProductChannel();
pc.setGuid(guid);
pc.setProductId(str(row, "PRODUCT_ID"));
pc.setChannelCode(str(row, "CHANNEL_CODE"));
pc.setChannelName(str(row, "CHANNEL_NAME"));
pc.setPriority(integer(row, "PRIORITY"));
pc.setIsEnabled(flag(row, "IS_ENABLED"));
pc.setLineName(str(row, "LINE_NAME"));
pc.setLineCode(str(row, "LINE_CODE"));
pc.setCountryCode(str(row, "COUNTRY_CODE"));
pc.setStartWeight(decimal(row, "START_WEIGHT"));
pc.setEndWeight(decimal(row, "END_WEIGHT"));
pc.setOrderCount(lo(row, "ORDER_COUNT"));
pc.setOrderCountRatio(decimal(row, "ORDER_COUNT_RATIO"));
relations.add(pc);
}
int relAffected = writer.writeProductChannels(relations);
long cost = System.currentTimeMillis() - start;
String summary = String.format(
"渠道产品同步完成,耗时%dms。服务渠道:源%d行(跳过%d),写入影响%d行;"
+ "线路产品:源%d行(跳过%d),写入影响%d行;产品渠道关系:源%d行(跳过%d),写入影响%d行",
cost, channels.size(), skipChannel, channelAffected,
products.size(), skipProduct, productAffected,
relations.size(), skipRel, relAffected);
log.info(summary);
return summary;
}
// ==================== Oracle Map 取值工具(列名忽略大小写) ====================
private static Object getIgnoreCase(Map<String, Object> row, String key)
{
Object v = row.get(key);
if (v != null)
{
return v;
}
for (Map.Entry<String, Object> e : row.entrySet())
{
if (e.getKey() != null && e.getKey().equalsIgnoreCase(key))
{
return e.getValue();
}
}
return null;
}
private static String str(Map<String, Object> row, String key)
{
Object v = getIgnoreCase(row, key);
if (v == null)
{
return null;
}
String s = v.toString().trim();
return s.isEmpty() ? null : s;
}
/** NUMBER(1) 等 0/1 标志位原样存为 "1"/"0" */
private static String flag(Map<String, Object> row, String key)
{
Object v = getIgnoreCase(row, key);
if (v == null)
{
return null;
}
if (v instanceof BigDecimal)
{
return ((BigDecimal) v).signum() == 0 ? "0" : "1";
}
if (v instanceof Number)
{
return ((Number) v).intValue() == 0 ? "0" : "1";
}
String s = v.toString().trim();
return s.isEmpty() ? null : s;
}
private static BigDecimal decimal(Map<String, Object> row, String key)
{
Object v = getIgnoreCase(row, key);
if (v == null)
{
return null;
}
if (v instanceof BigDecimal)
{
return (BigDecimal) v;
}
if (v instanceof Number)
{
return new BigDecimal(v.toString());
}
String s = v.toString().trim();
return s.isEmpty() ? null : new BigDecimal(s);
}
private static Integer integer(Map<String, Object> row, String key)
{
BigDecimal d = decimal(row, key);
return d == null ? null : d.intValue();
}
private static Long lo(Map<String, Object> row, String key)
{
BigDecimal d = decimal(row, key);
return d == null ? null : d.longValue();
}
private static Date date(Map<String, Object> row, String key)
{
Object v = getIgnoreCase(row, key);
if (v == null)
{
return null;
}
if (v instanceof Date)
{
return (Date) v;
}
String s = v.toString().trim();
return s.isEmpty() ? null : new Date(s);
}
}
@@ -0,0 +1,60 @@
package com.bms.masterdata.sync;
import java.util.List;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import com.bms.masterdata.domain.LineProduct;
import com.bms.masterdata.domain.ProductChannel;
import com.bms.masterdata.domain.ServiceChannel;
import com.bms.masterdata.mapper.ChannelMasterSyncMapper;
/**
* 渠道/产品 镜像表写库器(MASTER 数据源,分批 upsert,单表一个事务)。
* 独立 Bean:保证 @Transactional 经代理生效,且与 SLAVE 读连接隔离。
*
* @author bms
* @date 2026-10-12
*/
@Service
public class ChannelMasterSyncWriter
{
/** 单批行数,控制 SQL 包大小 */
private static final int BATCH_SIZE = 200;
@Autowired
private ChannelMasterSyncMapper mapper;
@Transactional(rollbackFor = Exception.class)
public int writeServiceChannels(List<ServiceChannel> list)
{
int rows = 0;
for (int from = 0; from < list.size(); from += BATCH_SIZE)
{
rows += mapper.upsertServiceChannel(list.subList(from, Math.min(from + BATCH_SIZE, list.size())));
}
return rows;
}
@Transactional(rollbackFor = Exception.class)
public int writeLineProducts(List<LineProduct> list)
{
int rows = 0;
for (int from = 0; from < list.size(); from += BATCH_SIZE)
{
rows += mapper.upsertLineProduct(list.subList(from, Math.min(from + BATCH_SIZE, list.size())));
}
return rows;
}
@Transactional(rollbackFor = Exception.class)
public int writeProductChannels(List<ProductChannel> list)
{
int rows = 0;
for (int from = 0; from < list.size(); from += BATCH_SIZE)
{
rows += mapper.upsertProductChannel(list.subList(from, Math.min(from + BATCH_SIZE, list.size())));
}
return rows;
}
}
@@ -0,0 +1,66 @@
package com.bms.masterdata.sync;
import java.util.List;
import java.util.Map;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import com.github.pagehelper.PageHelper;
import com.bms.common.annotation.DataSource;
import com.bms.common.enums.DataSourceType;
import com.bms.masterdata.mapper.OracleChannelMasterMapper;
/**
* Oracle 渠道/产品 主数据读取器(SLAVE 只读账号数据源)。
* 必须独立 Bean,否则 @DataSource 切面在同类自调用时不生效。
*
* @author bms
* @date 2026-10-12
*/
@Service
public class OracleChannelMasterReader
{
@Autowired
private OracleChannelMasterMapper oracleChannelMasterMapper;
@DataSource(DataSourceType.SLAVE)
public List<Map<String, Object>> readServiceChannels()
{
try
{
PageHelper.clearPage();
return oracleChannelMasterMapper.selectAllServiceChannel();
}
finally
{
PageHelper.clearPage();
}
}
@DataSource(DataSourceType.SLAVE)
public List<Map<String, Object>> readLineProducts()
{
try
{
PageHelper.clearPage();
return oracleChannelMasterMapper.selectAllLineProduct();
}
finally
{
PageHelper.clearPage();
}
}
@DataSource(DataSourceType.SLAVE)
public List<Map<String, Object>> readProductChannels()
{
try
{
PageHelper.clearPage();
return oracleChannelMasterMapper.selectAllProductChannel();
}
finally
{
PageHelper.clearPage();
}
}
}
@@ -0,0 +1,32 @@
package com.bms.masterdata.task;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import com.bms.common.utils.StringUtils;
import com.bms.masterdata.sync.ChannelMasterSyncService;
/**
* 老系统渠道/产品 主数据每日同步任务。
* sys_job.invoke_target = channelMasterSyncTask.syncFromOracle()
*
* @author bms
* @date 2026-10-12
*/
@Component("channelMasterSyncTask")
public class ChannelMasterSyncTask
{
@Autowired
private ChannelMasterSyncService channelMasterSyncService;
/**
* 全量同步 Oracle TAB_SERVICE_CHANNEL / TAB_LINE_PRODUCT / TAB_PRODUCT_CHANNEL
*/
public void syncFromOracle()
{
String summary = channelMasterSyncService.syncFromOracle();
if (StringUtils.isEmpty(summary))
{
throw new RuntimeException("老系统渠道产品同步未返回结果");
}
}
}