msNewPayment2.0/src/main/java/zs/payment/messages/KafkaConsumer.java

package zs.payment.messages;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;



import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cloud.context.config.annotation.RefreshScope;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import org.springframework.transaction.annotation.Transactional;
import zs.payment.dto.OrderDetailDTO;
import zs.payment.dto.SkuDTO;
import zs.payment.entity.mysql.*;
import zs.payment.enums.ProductSupplyRedis;
import zs.payment.req.ProductDetailReq;
import zs.payment.resp.Result;
import zs.payment.service.Imseweishopgoods.ImsEweiShopGoodsService;
import zs.payment.service.imseweishopcategory.ImsEweiShopCategoryService;
import zs.payment.service.imseweishopgoodsoption.ImsEweiShopGoodsOptionService;
import zs.payment.service.imseweishopgoodsspec.ImsEweiShopGoodsSpecService;
import zs.payment.service.imseweishopgoodsspecitem.ImsEweiShopGoodsSpecItemService;
import zs.payment.service.supply.paysupply.PaySupplyService;
import zs.payment.service.supply.product.ProductSupplyService;
import zs.payment.utils.RedisUtils;
import zs.payment.utils.TimeUtils;

import java.math.BigDecimal;
import java.util.*;


@Slf4j
@Component
@RefreshScope
public class KafkaConsumer {

    @Autowired
    private PaySupplyService paySupplyService;
    @Autowired
    private ProductSupplyService productSupplyService;

    @Autowired
    private RedisUtils redisUtils;

    @Autowired
    private ImsEweiShopCategoryService imsEweiShopCategoryService;
    //商品表
    @Autowired
    private ImsEweiShopGoodsService imsEweiShopGoodsService;
    //规格
    @Autowired
    private ImsEweiShopGoodsSpecService imsEweiShopGoodsSpecService;
    //规格项目
    @Autowired
    private ImsEweiShopGoodsSpecItemService imsEweiShopGoodsSpecItemService;
    @Autowired
    private ImsEweiShopGoodsOptionService imsEweiShopGoodsOptionService;


    /**
     * 商品下架
     *
     * @param messsage
     */
    @KafkaListener(
        topics = "goods.undercarriage",
        groupId = "pay2-group"
    )
    public void goodsUndercarriage(String messsage) {
        JSONObject jsonObject = JSONObject.parseObject(messsage);
        int productId = jsonObject.getIntValue("product_id");

        // 更新商品状态为下架   update ims_ewei_shop_goods set  status = 0 where catch_source=yunxin and catch_id=productId
        Boolean b = redisUtils.sismember(ProductSupplyRedis.SYNC_SUPPLY_ID, productId + "");
        if (b) {
            b = imsEweiShopGoodsService.updateStatusBySupplyId(String.valueOf(productId), "yunxin", 0);
            if (!b) {
                log.error("--------商品{}下架状态修改失败--------", productId);
            }
        }
    }


    /**
     * 商品上架
     *
     * @param messsage
     */
    @KafkaListener(
        topics = "goods.on.sale",
        groupId = "pay2-group"
    )
    public void goodsOnSale(String messsage) {
        JSONObject jsonObject = JSONObject.parseObject(messsage);
        int productId = jsonObject.getIntValue("product_id");

        // 更新商品状态为上架   update ims_ewei_shop_goods set  status = 1 where catch_source=yunxin and catch_id=productId
        Boolean b = redisUtils.sismember(ProductSupplyRedis.SYNC_SUPPLY_ID, productId + "");
        if (b) {
            b = imsEweiShopGoodsService.updateStatusBySupplyId(String.valueOf(productId), "yunxin", 1);
            if (!b) {
                log.error("--------商品{}上架状态修改失败--------", productId);
            }
        }
    }


    /**
     * 商品删除
     *
     * @param messsage
     */
    @KafkaListener(
        topics = "goods.delete",
        groupId = "pay2-group"
    )
    public void goodsDelete(String messsage) {
        JSONObject jsonObject = JSONObject.parseObject(messsage);
        int productId = jsonObject.getIntValue("product_id");

        // 更新商品状态为下架   update ims_ewei_shop_goods set  status = 0 where catch_source=yunxin and catch_id=productId
        Boolean b = redisUtils.sismember(ProductSupplyRedis.SYNC_SUPPLY_ID, productId + "");
        if (b) {
            b = imsEweiShopGoodsService.deleteBySupplyId(String.valueOf(productId), "yunxin");
            if (!b) {
                log.error("--------商品{}删除状态修改失败--------", productId);
            } else {
                //删除key
                redisUtils.srem(ProductSupplyRedis.SYNC_SUPPLY_ID, productId + "");
            }
        }

    }


    /**
     * 商品添加
     *
     * @param messsage
     */
    @KafkaListener(
        topics = "goods.sync",
        groupId = "pay2-group"
    )
    @Transactional(rollbackFor = Exception.class)
    public void goodsSync(String messsage) {

        log.info("---准备开始商品入库{}---",messsage);
        List<Integer> ids = JSONArray.parseArray(messsage, Integer.class);
        ProductDetailReq req = new ProductDetailReq();
        req.setIds(ids);
        Result result = productSupplyService.detailList(req);

        if (result.isSuccess()) {

            List<OrderDetailDTO> list = JSONObject.parseArray(
                result.getBody().get("data").toString(),
                OrderDetailDTO.class
            );
            //todo  导入到企云商城
            list.forEach(dto -> {
                //todo 先查看该商品编号是否已同步企云商城。若没同步，则判断商品分类是否存在。不存在则创建

                Boolean b = redisUtils.sismember(ProductSupplyRedis.SYNC_SUPPLY_ID, dto.getId().toString());
                if (!b) {
                    //不存在，则需要添加

                    Integer uniacid = 465;

                    //第一级
                    Long pcate = 0L;
                    //第二级
                    Long ccate = 0L;
                    //第三级
                    Long tcate = 0L;

                    //未同步，则判断商品分类
                    ImsEweiShopCategory category1 = imsEweiShopCategoryService.selectByName(dto.getCategory_1().getName(), dto.getCategory_1().getLevel(), 0L, uniacid);
                    if (!Optional.ofNullable(category1).isPresent()) {
                        //创建新的商品分类
                        category1 = new ImsEweiShopCategory();
                        category1.setLevel(1);
                        category1.setName(dto.getCategory_1().getName());
                        category1.setParentId(0);
                        category1.setUniacId(uniacid);
                        imsEweiShopCategoryService.save(category1);
                    }
                    pcate = category1.getId();
                    // 存在，再查询商品分类第二级
                    ImsEweiShopCategory category2 = imsEweiShopCategoryService.selectByName(dto.getCategory_2().getName(), dto.getCategory_2().getLevel(), pcate, uniacid);
                    if (!Optional.ofNullable(category2).isPresent()) {
                        category2 = new ImsEweiShopCategory();
                        category2.setLevel(2);
                        category2.setName(dto.getCategory_2().getName());
                        category2.setParentId(pcate.intValue());
                        category2.setUniacId(uniacid);
                        imsEweiShopCategoryService.save(category2);
                    }
                    ccate = category2.getId();

                    // 再查询商品分类第三级
                    ImsEweiShopCategory category3 = imsEweiShopCategoryService.selectByName(dto.getCategory_3().getName(), dto.getCategory_3().getLevel(), ccate, uniacid);
                    if (!Optional.ofNullable(category3).isPresent()) {
                        category3 = new ImsEweiShopCategory();
                        category3.setLevel(3);
                        category3.setName(dto.getCategory_3().getName());
                        category3.setParentId(ccate.intValue());
                        category3.setUniacId(uniacid);
                        imsEweiShopCategoryService.save(category3);
                    }
                    tcate = category3.getId();

                    //todo 创建商品
                    ImsEweiShopGoods goods = new ImsEweiShopGoods();
                    //直接作为推荐商品展示
                    goods.setIsrecommand(1);
                    //商品分类
                    goods.setPcate(pcate);
                    goods.setCcate(ccate);
                    goods.setTcate(tcate);
                    goods.setPcates("");
                    goods.setCcates("");
                    goods.setTcates("");
                    goods.setCates(pcate+","+ccate+","+tcate);
                    goods.setType(1);
                    //启用商品规格
                    goods.setHasoption(1);
                    //状态  1=上架
                    goods.setStatus(1);
                    goods.setDisplayorder(0);
                    goods.setTitle(dto.getTitle());
                    //商品图
                    goods.setThumb(dto.getImage_url());
                    //商品单位： 件
                    goods.setUnit(dto.getUnit());
                    goods.setContent(dto.getDetail_images());
                    //================以下价格对应企业商城商品价格====================
                    // 商品售价-用户需要支付的价格 ，要求售价不能大于原价【原需求如此，这边处理方案为原价=商品售价】
                    goods.setMarketprice(new BigDecimal(dto.getMarket_price() ).divide(new BigDecimal(100)));
                    // 原价
                    goods.setProductprice(new BigDecimal(dto.getMarket_price() ).divide(new BigDecimal(100)));
                    // 生态供应价
                    goods.setCostprice(new BigDecimal(dto.getCost_price() ).divide(new BigDecimal(100)));
                    // 成本价
                    goods.setCost_of_price(new BigDecimal(dto.getCost_price() ).divide(new BigDecimal(100)));


                    goods.setTotal(dto.getStock());
                    // 0 拍下减库存 1 付款减库存 2 永久不减
                    goods.setTotalcnf(0);
                    goods.setSales(dto.getSales());
                    goods.setSalesreal(dto.getSales());
                    goods.setIsnew(0);
                    goods.setIshot(0);
                    goods.setIsdiscount(0);
                    //默认值，避免金额为0
                    goods.setWeight(new BigDecimal(1));
                    //商品来源
                    goods.setSupply_source("yunxin");
                    //芸信商城的商品主键
                    goods.setSupply_id(dto.getId().toString());
//                    goods.setCates("");
                    //数据库里的时间戳是秒级
                    goods.setCreatetime(TimeUtils.getLongFromCurrentTime() / 1000);
                    goods.setUpdatetime(TimeUtils.getLongFromCurrentTime() / 1000);
                    goods.setDeleted(0);
                    //多商户id
                    goods.setMerchid(0);
                    // merchid==0?0:1
                    goods.setMerchsale(0);
//                    goods.setShorttitle();
                    //商品编号
                    goods.setGoodssn(dto.getSn());
                    //商品条码
                    goods.setProductsn(dto.getBarcode());
                    //默认发布给云享生态
                    goods.setUniacid(uniacid);

                    //创建商品表
                    boolean b2 = imsEweiShopGoodsService.save(goods);

                    if (!b2) {
                        log.error("商品{}创建失败", dto.getId().toString());
                        return;
                    }

                    //创建规格
                    ImsEweiShopGoodsSpec goodsSpec = new ImsEweiShopGoodsSpec();
                    goodsSpec.setTitle("规格");
                    goodsSpec.setUniacid(uniacid);
                    goodsSpec.setGoodsid(goods.getId().intValue());

                    boolean b3 = imsEweiShopGoodsSpecService.save(goodsSpec);


                    for (int i = 0; i < dto.getSkus().size(); i++) {
                        SkuDTO sku = dto.getSkus().get(i);
                        ImsEweiShopGoodsSpecItem goodsSpecItem = new ImsEweiShopGoodsSpecItem();
                        goodsSpecItem.setShowVar(1);
                        goodsSpecItem.setUniacid(uniacid);
                        goodsSpecItem.setSpecid(goodsSpec.getId().intValue());
                        goodsSpecItem.setTitle(sku.getTitle());
                        goodsSpecItem.setDisplayorder(i);

                        //创建规格项目
                        boolean b4 = imsEweiShopGoodsSpecItemService.save(goodsSpecItem);

                        //创建规格项目的价格
                        ImsEweiShopGoodsOption goodsOption = new ImsEweiShopGoodsOption();
                        goodsOption.setTitle(sku.getTitle());
                        goodsOption.setUniacid(uniacid);
                        goodsOption.setGoodsid(goods.getId().intValue());
                        goodsOption.setSpecs(goodsSpecItem.getId().toString());
                        //原价，可用于打折，最大价格
                        goodsOption.setProductprice(new BigDecimal(sku.getGuide_price() ).divide(new BigDecimal(100)));
                        //现价 不能大于原价
                        goodsOption.setMarketprice(new BigDecimal(sku.getGuide_price() ).divide(new BigDecimal(100)));
                        //成本价
                        goodsOption.setCostprice(new BigDecimal(sku.getCost_price() ).divide(new BigDecimal(100)));
                        //预售价 = 售价
                        goodsOption.setPresellprice(new BigDecimal(sku.getPrice() ).divide(new BigDecimal(100)));

                        goodsOption.setStock(sku.getStock());
                        goodsOption.setWeight(new BigDecimal(sku.getWeight()));
                        //下单与获取运费验证是否可售等使用这个
                        goodsOption.setSkuId(sku.getId().toString());
                        goodsOption.setGoodssn(sku.getSn());
                        goodsOption.setDay(0);
                        goodsOption.setAllfullbackprice(new BigDecimal(0));
                        goodsOption.setFullbackprice(new BigDecimal(0));
                        goodsOption.setExchange_postage(new BigDecimal(0));
                        goodsOption.setIsfullback(0);
                        goodsOption.setIslive(0);
                        goodsOption.setLiveprice(new BigDecimal(0));
                        //企云商城的商品重量为0时，金额为0
                        goodsOption.setWeight(sku.getWeight() == 0 ? new BigDecimal(1) : new BigDecimal(sku.getWeight()));

                        boolean b5 = imsEweiShopGoodsOptionService.save(goodsOption);
                    }
                    redisUtils.sadd(ProductSupplyRedis.SYNC_SUPPLY_ID, dto.getId().toString());
                } else {
                    log.error("====商品{}已同步过=====", dto.getId().toString());

                }

            });
            log.info("=======productId:{} goodsAlter end=======", ids);
        } else {
            log.error("kafka consumer goods.sync 商品{}详情查询失败", messsage);
        }


    }


    //------------订单----------------

    /**
     * 物流订单发货
     *
     * @param messsage
     */
    @KafkaListener(
        topics = "order.delivery",
        groupId = "pay2-group"
    )
    public void goodsDelivery(String orderSn) {
        //通过这个第三方订单编号获取物流信息
        paySupplyService.updateOrderLogistic(orderSn);
        log.info("=======订单发货-第三方订单编号:{} =======", orderSn);
    }


}