定时同步客户数据

2026-09-28 09:10:27 +08:00
parent 533984c77c
commit 73e240a872
9 changed files with 559 additions and 0 deletions
@@ -0,0 +1,27 @@
+package com.ruoyi.customer.customerinfo.mapper;
+
+import java.util.List;
+import com.ruoyi.customer.customerinfo.domain.BmsCustomer;
+
+/**
+ * 客户主表 ctr_customer 同步写库（MASTER 数据源）
+ *
+ * @author ruoyi
+ */
+public interface CustomerSyncMapper
+{
+    /**
+     * 库中未删除的全部客户编码，用于判定新增/更新
+     */
+    List<String> selectExistingCodes();
+
+    /**
+     * 新增客户（同步专用，字段集与老系统导入脚本一致）
+     */
+    int insertSyncCustomer(BmsCustomer customer);
+
+    /**
+     * 按客户编码更新客户（同步专用）
+     */
+    int updateSyncCustomerByCode(BmsCustomer customer);
+}
@@ -0,0 +1,17 @@
+package com.ruoyi.customer.customerinfo.mapper;
+
+import java.util.List;
+import java.util.Map;
+
+/**
+ * 老系统 Oracle TAB_CUSTOMER 只读查询（SLAVE 数据源，账号只读）
+ *
+ * @author ruoyi
+ */
+public interface OracleCustomerSyncMapper
+{
+    /**
+     * 全量读取 TAB_CUSTOMER 同步所需列（数据量约数百行，单次读取）
+     */
+    List<Map<String, Object>> selectAllTabCustomer();
+}
@@ -0,0 +1,309 @@
+package com.ruoyi.customer.customerinfo.sync;
+
+import java.math.BigDecimal;
+import java.math.RoundingMode;
+import java.util.ArrayList;
+import java.util.Date;
+import java.util.HashSet;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Service;
+import com.ruoyi.common.utils.StringUtils;
+import com.ruoyi.customer.customerinfo.domain.BmsCustomer;
+import com.ruoyi.customer.customerinfo.mapper.CustomerSyncMapper;
+
+/**
+ * 老系统 Oracle TAB_CUSTOMER -> ctr_customer 每日同步。
+ *
+ * 字段映射、字典转换与 D:\\生产客户数据\\scripts\\import_customers.py 完全一致：
+ * 按 customer_code 判定新增/更新；源为空的字段不写（走默认值/保留原值）。
+ *
+ * @author ruoyi
+ */
+@Service
+public class CustomerSyncService
+{
+    private static final Logger log = LoggerFactory.getLogger(CustomerSyncService.class);
+
+    /** 写库人标记，便于事后甄别同步数据 */
+    public static final String SYNC_MARKER = "oracle-sync";
+
+    /** 老系统 CUSTOMER_TYPE（中文）-> customer_type（0直客 1同行 2集运） */
+    private static final Map<String, String> CUSTOMER_TYPE_MAP = new LinkedHashMap<>();
+    /** 老系统 CUSTOMER_BALANCE_MODE（中文）-> settlement_type（0月结 1周结 2日结 3现结 4票结 5半月结） */
+    private static final Map<String, String> SETTLEMENT_MAP = new LinkedHashMap<>();
+    static
+    {
+        CUSTOMER_TYPE_MAP.put("直客", "0");
+        CUSTOMER_TYPE_MAP.put("同行", "1");
+        CUSTOMER_TYPE_MAP.put("集运", "2");
+
+        SETTLEMENT_MAP.put("月结", "0");
+        SETTLEMENT_MAP.put("周结", "1");
+        SETTLEMENT_MAP.put("日结", "2");
+        SETTLEMENT_MAP.put("现结", "3");
+        SETTLEMENT_MAP.put("现金", "3");
+        SETTLEMENT_MAP.put("票结", "4");
+        SETTLEMENT_MAP.put("支票", "4");
+        SETTLEMENT_MAP.put("半月结", "5");
+    }
+
+    /** ctr_customer 字符串列长度（超长截断，避免夜间任务因单行异常整批失败） */
+    private static final int LEN_CODE = 50;
+    private static final int LEN_NAME = 100;
+    private static final int LEN_CODE_DW = 50;
+    private static final int LEN_USER_ID = 64;
+    private static final int LEN_USER = 50;
+    private static final int LEN_REMARK = 500;
+    /** 告警明细最多打印条数 */
+    private static final int WARN_LOG_LIMIT = 50;
+
+    @Autowired
+    private OracleCustomerReader oracleCustomerReader;
+
+    @Autowired
+    private CustomerSyncWriter customerSyncWriter;
+
+    @Autowired
+    private CustomerSyncMapper customerSyncMapper;
+
+    /**
+     * 执行一次全量同步。
+     *
+     * @return 同步结果摘要
+     */
+    public String syncFromOracle()
+    {
+        long t0 = System.currentTimeMillis();
+        List<Map<String, Object>> rows = oracleCustomerReader.readAll();
+
+        Set<String> existingCodes = new HashSet<>(customerSyncMapper.selectExistingCodes());
+
+        List<BmsCustomer> inserts = new ArrayList<>();
+        List<BmsCustomer> updates = new ArrayList<>();
+        Set<String> seenCodes = new HashSet<>();
+        int blankRows = 0;
+        List<String> errors = new ArrayList<>();
+        List<String> warnings = new ArrayList<>();
+
+        for (int i = 0; i < rows.size(); i++)
+        {
+            int rowNo = i + 1;
+            Map<String, Object> row = rows.get(i);
+            String code = clean(getStr(row, "CUSTOMER_CODE"));
+            String name = clean(getStr(row, "CUSTOMER_NAME"));
+            if (code == null && name == null)
+            {
+                blankRows++;
+                continue;
+            }
+            if (code == null)
+            {
+                errors.add("行" + rowNo + ": CUSTOMER_CODE 为空，跳过");
+                continue;
+            }
+            if (name == null)
+            {
+                errors.add("行" + rowNo + " " + code + ": CUSTOMER_NAME 为空，跳过");
+                continue;
+            }
+            if (!seenCodes.add(code))
+            {
+                errors.add("行" + rowNo + " " + code + ": Oracle 源内编码重复，跳过");
+                continue;
+            }
+
+            BmsCustomer c = mapRow(row, code, name, warnings);
+            if (existingCodes.contains(code))
+            {
+                c.setUpdateBy(SYNC_MARKER);
+                c.setUpdateTime(new Date());
+                updates.add(c);
+            }
+            else
+            {
+                c.setCreateBy(SYNC_MARKER);
+                inserts.add(c);
+            }
+        }
+
+        int[] written = customerSyncWriter.writeBatch(inserts, updates);
+        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);
+        log.info(summary);
+        errors.forEach(e -> log.warn("  同步跳过: {}", e));
+        warnings.stream().limit(WARN_LOG_LIMIT).forEach(w -> log.warn("  同步告警: {}", w));
+        if (warnings.size() > WARN_LOG_LIMIT)
+        {
+            log.warn("  其余 {} 条告警略", warnings.size() - WARN_LOG_LIMIT);
+        }
+        if (!errors.isEmpty())
+        {
+            log.warn("同步存在 {} 条跳过记录，请核对 TAB_CUSTOMER 源数据", errors.size());
+        }
+        return summary;
+    }
+
+    /**
+     * 单行映射（逻辑对齐 import_customers.py 的 build_row）
+     */
+    private BmsCustomer mapRow(Map<String, Object> row, String code, String name, List<String> warnings)
+    {
+        BmsCustomer c = new BmsCustomer();
+        c.setCustomerCode(truncate(code, LEN_CODE, code, "customer_code", warnings));
+        c.setCustomerName(truncate(name, LEN_NAME, code, "customer_name", warnings));
+        c.setDelFlag("0");
+
+        c.setCustomerCodeDw(truncate(getStr(row, "CUSTOMER_OUTER_CODE"), LEN_CODE_DW, code, "customer_code_dw", warnings));
+        c.setSalesUserId(truncate(getStr(row, "BELONG_SALER"), LEN_USER_ID, code, "sales_user_id", warnings));
+        c.setSalesUser(truncate(getStr(row, "BELONG_SALER_NAME"), LEN_USER, code, "sales_user", warnings));
+        c.setSettlementUser(truncate(getStr(row, "SETTLEMENT_MAN"), LEN_USER, code, "settlement_user", warnings));
+        c.setServiceUser(truncate(getStr(row, "SERVICE_MAN"), LEN_USER, code, "service_user", warnings));
+        c.setRemark(truncate(getStr(row, "REMARK"), LEN_REMARK, code, "remark", warnings));
+
+        c.setWeightThreshold(toDecimal(row, "WEIGHT_LIMIT", code, warnings));
+        c.setCreditEd(toDecimal(row, "CREDIT_LIMIT", code, warnings));
+        c.setTempEd(toDecimal(row, "TEMP_AMOUNT", code, warnings));
+        c.setAccountBalance(toDecimal(row, "ACCOUNT_AMOUNT", code, warnings));
+        c.setForeignCurrencyBalance(toDecimal(row, "EUR_AMOUNT", code, warnings));
+
+        String customerType = clean(getStr(row, "CUSTOMER_TYPE"));
+        if (customerType != null)
+        {
+            String mapped = CUSTOMER_TYPE_MAP.get(customerType);
+            if (mapped != null)
+            {
+                c.setCustomerType(mapped);
+            }
+            else
+            {
+                warnings.add(code + ": 未知业务类型 " + customerType + "，按默认直客");
+            }
+        }
+
+        String balanceMode = clean(getStr(row, "CUSTOMER_BALANCE_MODE"));
+        if (balanceMode != null)
+        {
+            String mapped = SETTLEMENT_MAP.get(balanceMode);
+            if (mapped != null)
+            {
+                c.setSettlementType(mapped);
+            }
+            else
+            {
+                warnings.add(code + ": 未知结算方式 " + balanceMode + "，按默认月结");
+            }
+        }
+
+        c.setIsEnable(mapEnable(getStr(row, "BL_OPEN")));
+
+        Date createTime = getDate(row, "CREATE_DATE");
+        if (createTime == null)
+        {
+            createTime = getDate(row, "GMT_CREATE");
+        }
+        c.setCreateTime(createTime);
+        return c;
+    }
+
+    /** BL_OPEN：1启用 -> is_enable=0(正常)；0停用 -> 1；空按正常 */
+    private static String mapEnable(String v)
+    {
+        String s = clean(v);
+        if (s == null)
+        {
+            return "0";
+        }
+        return ("1".equals(s) || "1.0".equals(s)) ? "0" : "1";
+    }
+
+    private static String clean(String v)
+    {
+        if (v == null)
+        {
+            return null;
+        }
+        String s = v.trim();
+        return s.isEmpty() ? null : s;
+    }
+
+    /**
+     * 超长截断并记告警（Python 脚本对超长行直接报错跳过；夜间无人值守改为截断更稳）
+     */
+    private static String truncate(String value, int maxLen, String code, String column, List<String> warnings)
+    {
+        String s = clean(value);
+        if (s != null && s.length() > maxLen)
+        {
+            warnings.add(String.format("%s: %s 超长(%d>%d)已截断: %s...", code, column, s.length(), maxLen, s.substring(0, 20)));
+            s = s.substring(0, maxLen);
+        }
+        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 Date getDate(Map<String, Object> row, String key)
+    {
+        Object v = getIgnoreCase(row, key);
+        if (v == null)
+        {
+            return null;
+        }
+        if (v instanceof Date)
+        {
+            return (Date) v;
+        }
+        return null;
+    }
+
+    private static String getStr(Map<String, Object> row, String key)
+    {
+        Object v = getIgnoreCase(row, key);
+        return v == null ? null : String.valueOf(v);
+    }
+
+    /** Oracle HashMap 列名按大写返回，这里做兼容查找 */
+    private static Object getIgnoreCase(Map<String, Object> row, String key)
+    {
+        if (row.containsKey(key))
+        {
+            return row.get(key);
+        }
+        for (Map.Entry<String, Object> e : row.entrySet())
+        {
+            if (e.getKey() != null && e.getKey().equalsIgnoreCase(key))
+            {
+                return e.getValue();
+            }
+        }
+        return null;
+    }
+}
@@ -0,0 +1,40 @@
+package com.ruoyi.customer.customerinfo.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.ruoyi.customer.customerinfo.domain.BmsCustomer;
+import com.ruoyi.customer.customerinfo.mapper.CustomerSyncMapper;
+
+/**
+ * 同步写库器（MASTER 数据源，整批一个事务）。
+ * 独立 Bean：保证 @Transactional 经代理生效，且与 SLAVE 读连接隔离。
+ *
+ * @author ruoyi
+ */
+@Service
+public class CustomerSyncWriter
+{
+    @Autowired
+    private CustomerSyncMapper customerSyncMapper;
+
+    /**
+     * @return 实际写入行数（新增+更新）；任一行 SQL 失败整批回滚
+     */
+    @Transactional(rollbackFor = Exception.class)
+    public int[] writeBatch(List<BmsCustomer> inserts, List<BmsCustomer> updates)
+    {
+        int insertCount = 0;
+        int updateCount = 0;
+        for (BmsCustomer c : inserts)
+        {
+            insertCount += customerSyncMapper.insertSyncCustomer(c);
+        }
+        for (BmsCustomer c : updates)
+        {
+            updateCount += customerSyncMapper.updateSyncCustomerByCode(c);
+        }
+        return new int[] {insertCount, updateCount};
+    }
+}
@@ -0,0 +1,39 @@
+package com.ruoyi.customer.customerinfo.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.ruoyi.common.annotation.DataSource;
+import com.ruoyi.common.enums.DataSourceType;
+import com.ruoyi.customer.customerinfo.mapper.OracleCustomerSyncMapper;
+
+/**
+ * Oracle TAB_CUSTOMER 读取器。
+ * 走 SLAVE（Oracle 只读账号）数据源；必须独立 Bean，
+ * 否则 @DataSource 切面在同类自调用时不生效。
+ *
+ * @author ruoyi
+ */
+@Service
+public class OracleCustomerReader
+{
+    @Autowired
+    private OracleCustomerSyncMapper oracleCustomerSyncMapper;
+
+    @DataSource(DataSourceType.SLAVE)
+    public List<Map<String, Object>> readAll()
+    {
+        try
+        {
+            // 避免 PageHelper 把 Oracle 查询包成分页语句
+            PageHelper.clearPage();
+            return oracleCustomerSyncMapper.selectAllTabCustomer();
+        }
+        finally
+        {
+            PageHelper.clearPage();
+        }
+    }
+}
@@ -0,0 +1,25 @@
+package com.ruoyi.customer.customerinfo.task;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Component;
+import com.ruoyi.customer.customerinfo.sync.CustomerSyncService;
+
+/**
+ * 老系统客户资料定时同步任务（sys_job 调用目标：customerSyncTask.syncFromOracle）
+ *
+ * @author ruoyi
+ */
+@Component("customerSyncTask")
+public class CustomerSyncTask
+{
+    @Autowired
+    private CustomerSyncService customerSyncService;
+
+    /**
+     * 每天 00:00 从 Oracle TAB_CUSTOMER 全量同步客户主数据到 ctr_customer
+     */
+    public void syncFromOracle()
+    {
+        customerSyncService.syncFromOracle();
+    }
+}
@@ -0,0 +1,72 @@
+<?xml version="1.0" encoding="UTF-8" ?>
+<!DOCTYPE mapper
+PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN"
+"http://mybatis.org/dtd/mybatis-3-mapper.dtd">
+<mapper namespace="com.ruoyi.customer.customerinfo.mapper.CustomerSyncMapper">
+
+    <select id="selectExistingCodes" resultType="java.lang.String">
+        SELECT customer_code FROM ctr_customer WHERE del_flag = '0'
+    </select>
+
+    <!-- 字段集与 scripts/import_customers.py 的 MAP_COLUMNS 保持一致；
+         源行为空的字段不写，走数据库默认值 -->
+    <insert id="insertSyncCustomer" parameterType="com.ruoyi.customer.customerinfo.domain.BmsCustomer">
+        INSERT INTO ctr_customer (
+            customer_code, customer_name, is_enable,
+            del_flag, create_by
+            <if test="customerCodeDw != null">, customer_code_dw</if>
+            <if test="customerType != null">, customer_type</if>
+            <if test="salesUserId != null">, sales_user_id</if>
+            <if test="salesUser != null">, sales_user</if>
+            <if test="settlementUser != null">, settlement_user</if>
+            <if test="serviceUser != null">, service_user</if>
+            <if test="settlementType != null">, settlement_type</if>
+            <if test="tempEd != null">, temp_ed</if>
+            <if test="accountBalance != null">, account_balance</if>
+            <if test="foreignCurrencyBalance != null">, foreign_currency_balance</if>
+            <if test="weightThreshold != null">, weight_threshold</if>
+            <if test="creditEd != null">, credit_ed</if>
+            <if test="remark != null">, remark</if>
+            <if test="createTime != null">, create_time</if>
+        ) VALUES (
+            #{customerCode}, #{customerName}, #{isEnable},
+            '0', #{createBy}
+            <if test="customerCodeDw != null">, #{customerCodeDw}</if>
+            <if test="customerType != null">, #{customerType}</if>
+            <if test="salesUserId != null">, #{salesUserId}</if>
+            <if test="salesUser != null">, #{salesUser}</if>
+            <if test="settlementUser != null">, #{settlementUser}</if>
+            <if test="serviceUser != null">, #{serviceUser}</if>
+            <if test="settlementType != null">, #{settlementType}</if>
+            <if test="tempEd != null">, #{tempEd}</if>
+            <if test="accountBalance != null">, #{accountBalance}</if>
+            <if test="foreignCurrencyBalance != null">, #{foreignCurrencyBalance}</if>
+            <if test="weightThreshold != null">, #{weightThreshold}</if>
+            <if test="creditEd != null">, #{creditEd}</if>
+            <if test="remark != null">, #{remark}</if>
+            <if test="createTime != null">, #{createTime}</if>
+        )
+    </insert>
+
+    <update id="updateSyncCustomerByCode" parameterType="com.ruoyi.customer.customerinfo.domain.BmsCustomer">
+        UPDATE ctr_customer
+        SET is_enable = #{isEnable}
+            <if test="customerName != null">, customer_name = #{customerName}</if>
+            <if test="customerCodeDw != null">, customer_code_dw = #{customerCodeDw}</if>
+            <if test="customerType != null">, customer_type = #{customerType}</if>
+            <if test="salesUserId != null">, sales_user_id = #{salesUserId}</if>
+            <if test="salesUser != null">, sales_user = #{salesUser}</if>
+            <if test="settlementUser != null">, settlement_user = #{settlementUser}</if>
+            <if test="serviceUser != null">, service_user = #{serviceUser}</if>
+            <if test="settlementType != null">, settlement_type = #{settlementType}</if>
+            <if test="tempEd != null">, temp_ed = #{tempEd}</if>
+            <if test="accountBalance != null">, account_balance = #{accountBalance}</if>
+            <if test="foreignCurrencyBalance != null">, foreign_currency_balance = #{foreignCurrencyBalance}</if>
+            <if test="weightThreshold != null">, weight_threshold = #{weightThreshold}</if>
+            <if test="creditEd != null">, credit_ed = #{creditEd}</if>
+            <if test="remark != null">, remark = #{remark}</if>
+            , update_by = #{updateBy}
+            , update_time = #{updateTime}
+        WHERE customer_code = #{customerCode} AND del_flag = '0'
+    </update>
+</mapper>
@@ -0,0 +1,17 @@
+<?xml version="1.0" encoding="UTF-8" ?>
+<!DOCTYPE mapper
+PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN"
+"http://mybatis.org/dtd/mybatis-3-mapper.dtd">
+<mapper namespace="com.ruoyi.customer.customerinfo.mapper.OracleCustomerSyncMapper">
+
+    <!-- 老系统客户主表全量读取；列集与 import_customers.py 的映射保持一致 -->
+    <select id="selectAllTabCustomer" resultType="java.util.HashMap" fetchSize="500">
+        SELECT CUSTOMER_CODE, CUSTOMER_NAME, CUSTOMER_OUTER_CODE,
+               BELONG_SALER, BELONG_SALER_NAME, SETTLEMENT_MAN, SERVICE_MAN,
+               REMARK, WEIGHT_LIMIT, CREDIT_LIMIT, TEMP_AMOUNT,
+               ACCOUNT_AMOUNT, EUR_AMOUNT,
+               CUSTOMER_TYPE, CUSTOMER_BALANCE_MODE, BL_OPEN,
+               CREATE_DATE, GMT_CREATE
+        FROM WEISHI_USER.TAB_CUSTOMER
+    </select>
+</mapper>
@@ -0,0 +1,13 @@
+-- ----------------------------
+-- 定时任务：老系统 Oracle TAB_CUSTOMER -> ctr_customer 客户资料每日同步
+-- 调用目标对应 Bean：com.ruoyi.customer.customerinfo.task.CustomerSyncTask#syncFromOracle
+-- cron 0 0 0 * * ? = 每天 00:00:00；concurrent=1 禁止并发；misfire=3 放弃错过的执行
+-- 幂等：按 customer_code 新增/更新，可重复执行
+-- ----------------------------
+INSERT INTO sys_job (job_name, job_group, invoke_target, cron_expression,
+                     misfire_policy, concurrent, status, create_by, create_time, remark)
+SELECT '老系统客户资料每日同步', 'DEFAULT', 'customerSyncTask.syncFromOracle()', '0 0 0 * * ?',
+       '3', '1', '0', 'admin', NOW(),
+       '每天00:00从Oracle(WSDB.TAB_CUSTOMER，只读账号)全量同步客户主数据到ctr_customer；写库标记 create_by/update_by=oracle-sync'
+FROM DUAL
+WHERE NOT EXISTS (SELECT 1 FROM sys_job WHERE invoke_target = 'customerSyncTask.syncFromOracle()');