运单DTO：WaybillBaseData

订单DTO：FullOrderBaseData

乘丰OMS-OS接入文档

index:rdmp_full_order_basedata

{
  "query": {
    "exists": {
      "field": "omsCreateTime"
    }
  },
  "_source": ["channelType","srcCityName","customerReferenceCode","stateCode","lastOrderOperateCode","lastOrderOperateReasonCode","operateCodeType","region","omsCreateTime"]
}

{
  "query": {
   "bool": {
      "filter": [
        {
          "term": {
            "special_cargo_flag": 1
          }
        }
      ]
    }
  },
  "_source": ["special_cargo_flag","is_magnetic_inspection"]
}

新增字段

channelType 渠道类型

以下有优先级：

1）字段“source_zone_code”优先匹配4PL代码，显示“By 4PL”

2) src_type_name类型是“2W DC”，“MIX DC”，“4W DC”“Station ”，显示“DC”

3）src_type_name类型是“Onsite area”,"Onsite code"，显示“Onsite”

4）src_type_name类型是“Project Department”，“RHUB”，“CHUB”显示“PUP”

5）src_type_name是“KE SHOP”，“DCSP SHOP”，“KBD SHOP”，“FC SHOP”，“4PL”，“RTSP”，“PSP”，显示“shop”

6）其他对接不到，或src_type_name为空的，显示“其他”

运单source_zone_code为寄件网点，对应订单srcDeptCode
运单src_type_name为寄件网点类型，对应订单srcDeptCodeTypeName

public class FullOrderBaseDataProcessFuction...
    // 添加常量定义
    protected static final List<String> DEPT_TYPE_DC = Arrays.asList("2W DC", "MIX DC", "4W DC", "Station");
    protected static final List<String> DEPT_TYPE_PUP = Arrays.asList("Project Department", "RHUB", "CHUB");
    protected static final List<String> DEPT_TYPE_SHOP = Arrays.asList("KE SHOP", "DCSP SHOP", "KBD SHOP", "FC SHOP", "4PL", "RTSP", "PSP");
    protected static final List<String> DEPT_TYPE_ONSITE = Arrays.asList("Onsite area", "Onsite code");
    ...
    private void setChannelType(FullOrderBaseData fullOrder) {
        String srcDeptCode = fullOrder.getSrcDeptCode();
        String srcDeptCodeTypeName = fullOrder.getSrcDeptCodeTypeName();

        if ("4PL".equals(srcDeptCode)) {
            fullOrder.setChannelType("By 4PL");
            return;
        }

        if (srcDeptCodeTypeName != null) {
            if (DEPT_TYPE_DC.contains(srcDeptCodeTypeName)) {
                fullOrder.setChannelType("DC");
                return;
            }

            if (DEPT_TYPE_ONSITE.contains(srcDeptCodeTypeName)) {
                fullOrder.setChannelType("Onsite");
                return;
            }

            if (DEPT_TYPE_PUP.contains(srcDeptCodeTypeName)) {
                fullOrder.setChannelType("PUP");
                return;
            }

            if (DEPT_TYPE_SHOP.contains(srcDeptCodeTypeName)) {
                fullOrder.setChannelType("shop");
                return;
            }
        }

        fullOrder.setChannelType("其他");
    }

srcCityNam原寄地城市名称

public class FullOrderProcessor {
    private void updateDepartmentInfo(FullOrderBaseData fullOrder, String deptCode, String flag) {
    private void updateCityInfo(FullOrderBaseData fullOrder, String cityCode, String flag) {

customerReferenceCode客户类型

public class FullOrderProcessor {
    private void updateCustomerInfo(FullOrderBaseData fullOrder, String monthlyCard) {
         fullOrder.setMonthlyCard(monthlyCard);
         // 填充客户类型信息
        Optional<CustomerProfileInfo> customerProfileInfoOpt = CacheContainer.getInstance().fetchCustomerProfileCfg(monthlyCard);
        if (customerProfileInfoOpt.isPresent()) {
            CustomerProfileInfo customerProfileInfo = customerProfileInfoOpt.get();
            fullOrder.setCustomerReferenceCode(customerProfileInfo.getCustomerReferenceCode());
        }

stateCode订单状态原始值

public class FullOrderProcessor {
    public void updateOmsStatusOrder(FullOrderBaseData fullOrder, OmsStatusOrderDto omsStatusOrderDto) {
        String orderCode = omsStatusOrderDto.getOrderCode();
        fullOrder.setStateCode(orderCode);

lastOrderOperateCode订单任务

public class FullOrderProcessor {
    public void updateOmsStatusOrder(FullOrderBaseData fullOrder, OmsStatusOrderDto omsStatusOrderDto) {
         if (omsStatusOrderDto.getOrderOperates() != null && omsStatusOrderDto.getOrderOperates().size() > 0) {
            int size = omsStatusOrderDto.getOrderOperates().size();
            OrderOperate orderOperate = omsStatusOrderDto.getOrderOperates().get(size - 1);
            String operateCode = orderOperate.getOperateCode();
            fullOrder.setLastOrderOperateCode(operateCode);

lastOrderOperateReasonCode异常原因码

①当orderoperates中operateCode=4002时，取对应reasonCode的值，

②当orderoperates中operateCode！=4002时，直接置空

public class FullOrderProcessor {
    public void updateOmsStatusOrder(FullOrderBaseData fullOrder, OmsStatusOrderDto omsStatusOrderDto) {
         if (omsStatusOrderDto.getOrderOperates() != null && omsStatusOrderDto.getOrderOperates().size() > 0) {
            int size = omsStatusOrderDto.getOrderOperates().size();
            OrderOperate orderOperate = omsStatusOrderDto.getOrderOperates().get(size - 1);
            String operateCode = orderOperate.getOperateCode();
            fullOrder.setLastOrderOperateCode(operateCode);
            // 根据条件设置lastOrderOperateReasonCode
            if ("4002".equals(operateCode)) {
                fullOrder.setLastOrderOperateReasonCode(reasonCode);
            } else {
                fullOrder.setLastOrderOperateReasonCode(null);
            }

operateCodeType操作码类型

1，operateCode是40000,40037，显示DISP；

2，operateCode是40001，显示PIN；

3，operateCode是40003,40072，显示PUP；表示已揽收成功

4，operateCode是当40002，reasonCode匹配kex的异常表，如40014对应的是PUX07

public class FullOrderProcessor {
    public void updateOmsStatusOrder(FullOrderBaseData fullOrder, OmsStatusOrderDto omsStatusOrderDto) {
         if (omsStatusOrderDto.getOrderOperates() != null && omsStatusOrderDto.getOrderOperates().size() > 0) {
            int size = omsStatusOrderDto.getOrderOperates().size();
            OrderOperate orderOperate = omsStatusOrderDto.getOrderOperates().get(size - 1);
            String operateCode = orderOperate.getOperateCode();
            fullOrder.setLastOrderOperateCode(operateCode);
            // 根据条件设置lastOrderOperateReasonCode
            if ("4002".equals(operateCode)) {
                fullOrder.setLastOrderOperateReasonCode(reasonCode);
            } else {
                fullOrder.setLastOrderOperateReasonCode(null);
            }
            
             // 设置operateCodeType
            if ("40000".equals(operateCode) || "40037".equals(operateCode)) {
                fullOrder.setOperateCodeType("DISP");
            } else if ("40001".equals(operateCode)) {
                fullOrder.setOperateCodeType("PIN");
            } else if ("40003".equals(operateCode) || "40072".equals(operateCode)) {
                fullOrder.setOperateCodeType("PUP");
            } else if ("40002".equals(operateCode)) {
                // 从异常表中获取对应的PUX编码
                if ("40021".equals(reasonCode)) {
                    fullOrder.setOperateCodeType("PUX03");
                } else if ("40004".equals(reasonCode)) {
                    fullOrder.setOperateCodeType("PUX06");
                } else if ("40005".equals(reasonCode)) {
                    fullOrder.setOperateCodeType("PUX17");
                } else if ("40075".equals(reasonCode)) {
                    fullOrder.setOperateCodeType("PUX08");
                } else if ("40074".equals(reasonCode)) {
                    fullOrder.setOperateCodeType("PUX35");
                } else if ("40014".equals(reasonCode)) {
                    fullOrder.setOperateCodeType("PUX07");
                } else if ("40036".equals(reasonCode)) {
                    fullOrder.setOperateCodeType("PUX20");
                } else if ("40035".equals(reasonCode)) {
                    fullOrder.setOperateCodeType("PUX01");
                }
            }
        }

region国家

omsCreateTime传入OMS时间

public class FullOrderProcessor {    
    public void updateByOrderInfoFvp(FullOrderBaseData fullOrder, OrderInfoToFvpDto orderInfoToFvpDto) {
        TimeZone timeZone = getTimeZoneByRegion(orderInfoToFvpDto.getRegion());
        // 设置 region 和 omsCreateTime
        fullOrder.setRegion(orderInfoToFvpDto.getRegion());
        fullOrder.setOmsCreateTime(TimeZoneUtil.dateToTimeStr(orderInfoToFvpDto.getCreateTime(),
                TimeUtil.DATE_TIME_SECOND_PATTERN, timeZone));

月结卡号（客户编码）

代码主流程

FullOrderDataToKafka

SingleOutputStreamOperator<Tuple2<String, String>> orderDetail
        = fvpOmsSource.union(cxOmsSource,omsStatusSource,omsToCXSource,operationSource)
        .keyBy(0)
        .process(new FullOrderBaseDataProcessFuction(waybillBaseDataProperty))
        .name(RDMP_BASEDATA_PROCESS).uid(RDMP_BASEDATA_PROCESS)
        .setParallelism(fullOrderProcParrel);

FullOrderBaseDataProcessFuction

@Override
public void processElement(Tuple3<String, String, byte[]> input, Context context, Collector<Tuple2<String, String>> output) {
    //先从缓存中获取，假如缓存中有，直接执行
    FullOrderBaseData fetchFullOrder = FullOrderCacheContainer.getInstance().fetchFullOrder(orderId);
}

private void process(FullOrderBaseData fullOrder, String kafkaSource, byte[] msg, Collector<Tuple2<String, String>> output) throws Exception {

    try {
        //订单数据源
        KafkaSourceEnum inputSourceEnum = KafkaSourceEnum.valueOf(kafkaSource);
        if (KafkaSourceEnum.OMS_FVP_ORDERL == inputSourceEnum) {
            OrderInfoToFvpDto orderInfoToFvpDto = JSONObject.parseObject(msg, OrderInfoToFvpDto.class);
            fullOrderProcessor.updateByOrderInfoFvp(fullOrder, orderInfoToFvpDto);
        } else if (KafkaSourceEnum.OMS_CORE_CX == inputSourceEnum) {
            OmsCoreCXDto omsCoreCXDto = JSONObject.parseObject(msg, OmsCoreCXDto.class);
            fullOrderProcessor.updateByOmsCoreCX(fullOrder, omsCoreCXDto);
        } else if (KafkaSourceEnum.OMS_TO_CX == inputSourceEnum) {
            OmsToCxDto omsToCxDto = JSONObject.parseObject(msg, OmsToCxDto.class);
            fullOrderProcessor.updateByOmsToCX(fullOrder, omsToCxDto);
        } else if (KafkaSourceEnum.OMS_STATUS_ORDER == inputSourceEnum) {
            OmsStatusOrderDto omsStatusOrderDto = JSONObject.parseObject(msg, OmsStatusOrderDto.class);

            fullOrderProcessor.updateOmsStatusOrder(fullOrder, omsStatusOrderDto);
        } else if (KafkaSourceEnum.OMS_OPT_WAYBILL_LINK_ORDER == inputSourceEnum) {
            OperationWaybillLinkOrderDto opLinkOrder = JSONObject.parseObject(msg, OperationWaybillLinkOrderDto.class);
            fullOrderProcessor.updateOpWaybill(fullOrder, opLinkOrder);

        }
        TimeZone timeZone = TimeZoneUtil.getTimeZone(CacheContainer.getInstance().fetchTimeZone(fullOrder.getSrcDeptCodeRegion()), FlinkTaskPropertiesUtil.getTimeZone());
        // 非下call的数据,最晚揽收时间的兜底<订单时间+1H>
        // 预约单  非预约单设置最晚揽收时间  设置最晚收件时间
        if(StringUtils.isBlank(fullOrder.getLastTime()) && StringUtils.isNotBlank(fullOrder.getOrderTm())){
            String lastTime= TimeZoneUtil.addTm(fullOrder.getOrderTm(), Calendar.HOUR_OF_DAY,1,timeZone);
            fullOrder.setLastTime(lastTime);
        }

        // 设置订单流向 orderFlowEnum
        String srcHqCode = fullOrder.getSrcHqCode();
        String destHqCode = fullOrder.getDestHqCode();
        if (StringUtils.isBlank(srcHqCode) || StringUtils.isBlank(destHqCode)) {
            fullOrder.setOrderFlowEnum(0);
        } else if ( !OmpsOrder.mainlandArea.contains(srcHqCode) && !OmpsOrder.mainlandArea.contains(destHqCode)){
            fullOrder.setOrderFlowEnum(1);
        }else if ( OmpsOrder.mainlandArea.contains(srcHqCode) && OmpsOrder.mainlandArea.contains(destHqCode)){
            fullOrder.setOrderFlowEnum(2);
        }else if ( !OmpsOrder.mainlandArea.contains(srcHqCode) && OmpsOrder.mainlandArea.contains(destHqCode)){
            fullOrder.setOrderFlowEnum(3);
        }else if ( OmpsOrder.mainlandArea.contains(srcHqCode) && !OmpsOrder.mainlandArea.contains(destHqCode)){
            fullOrder.setOrderFlowEnum(4);
        }
        if ( OmpsOrder.ska.contains(fullOrder.getCustomerAbbr())){
            fullOrder.setOrderChannelEnum(1);
        }else if (StringUtils.isNotBlank(fullOrder.getClientCode())){
            fullOrder.setOrderChannelEnum(2);
        }
        if (fullOrder.getOrderChannelEnum()==0){
            fullOrder.setOrderChannelEnum(5);
        }
        if (StringUtils.isBlank(fullOrder.getPartitionTm()) && StringUtils.isNotBlank(fullOrder.getOrderTm())) {
            fullOrder.setPartitionTm(fullOrder.getOrderTm());
        }
        if (StringUtils.isBlank(fullOrder.getPartitionTm())) {
            fullOrder.setPartitionTm(TimeZoneUtil.format(System.currentTimeMillis(),"yyyy-MM-dd HH:mm:ss",TimeZoneUtil.getDefaultTimeZone(FlinkTaskPropertiesUtil.getTimeZone())));
        }
        fullOrder.setProcessVersionTm(System.currentTimeMillis());

        // 设置channelType
        setChannelType(fullOrder);

        output.collect(new Tuple2<>(fullOrder.getOrderId(), JSONObject.toJSONString(fullOrder)));

        FullOrderCacheContainer.getInstance().saveFullOrder(fullOrder);

        saveToHbase(fullOrder, false);

    } catch (Exception e) {
        logger.error("throw exception: {}", e);
        throw e;
    }
}