gold-java/src/main/java/com/example/demo/Mysql/MysqlServiceImpl.java

package com.example.demo.Mysql;


import com.example.demo.Util.BaseDES;
import com.example.demo.domain.entity.User;
import com.example.demo.service.coin.AdminService;
import com.example.demo.service.coin.MarketService;
import com.example.demo.service.coin.UserService;
import org.apache.commons.lang3.ObjectUtils;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.http.*;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.web.client.RestTemplate;

import javax.sql.DataSource;
import java.math.BigDecimal;
import java.sql.*;
import java.time.LocalDateTime;
import java.time.Month;
import java.time.format.DateTimeFormatter;
import java.util.*;


@Service

public class MysqlServiceImpl implements MysqlService {
    @Autowired
    private RestTemplate restTemplate;

    Set<Integer> validZeroTypes = new HashSet<>(Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 11, 12, 13, 14, 18, 19, 20, 21, 22, 23, 24, 26, 28, 29, 35, 36, 40, 45, 46, 47, 48, 49, 53, 54, 60));
   Set<Integer> validOneTypes = new HashSet<>(Arrays.asList(9, 15, 17, 25, 27, 37, 41, 42, 43, 50, 51, 62));
   Set<Integer> validTwoTypes = new HashSet<>(Arrays.asList(52,61));
   Set<Integer> validThreeTypes = new HashSet<>(Arrays.asList(10, 16, 30, 31, 32, 33, 34, 39, 44));
   Set<Integer> validFourTypes = new HashSet<>(Arrays.asList(55, 56, 57, 58, 59, 63, 64, 65));
    LocalDateTime now = LocalDateTime.now();
    Month currentMonth = now.getMonth();
    @Autowired
    private AdminService adminService;
    @Autowired
    private UserService userService;
    @Autowired
    private MarketService marketService;


    @Autowired
    @Qualifier("sqlserver1DataSource")
    private DataSource sqlserver1DataSource;

    @Autowired
    @Qualifier("mysql1DataSource")
    private DataSource mysql1DataSource;

    private static final Logger logger = LoggerFactory.getLogger(MysqlServiceImpl.class);





    @Override
    @Transactional(transactionManager = "mysqlTransactionManager") // 👈 保证插入和用户更新在一个事务
    public void getSqlserverData() throws Exception {
        logger.info("开始从 SQL Server 同步数据到 MySQL");

        try (Connection sqlServerConn = sqlserver1DataSource.getConnection();
             Connection mysqlConn = mysql1DataSource.getConnection()) {

            logger.info("开始查询数据...");

            int pageSize = 100;
            int offset = 0;
            boolean hasMoreData = true;

            // 👇 恢复动态时间查询（原注释掉的硬编码时间已移除）
            String querySql = """
            SELECT
                id, gtype, jwcode, free, core_jb, buy_jb, cz_time, cz_user, cz_bz, operation_platform, goods_name
            FROM
                hwhcGold.dbo.user_gold_records
            WHERE cz_time >= ?
            ORDER BY
                cz_time ASC
            OFFSET ? ROWS FETCH NEXT ? ROWS ONLY;
            """;

            String insertSql = """
            INSERT IGNORE INTO user_gold_record 
            (order_code, jwcode, sum_gold, permanent_gold, free_june, free_december,
             task_gold, pay_platform, goods_name, refund_type, refund_model, remark, type, admin_id,
             audit_status, create_time, flag, update_time, audit_time, is_refund, uid)
            VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
            """;

            while (hasMoreData) {
                try (PreparedStatement sqlServerStmt = sqlServerConn.prepareStatement(querySql)) {
                    // 👇 恢复动态时间参数
                    sqlServerStmt.setTimestamp(1, Timestamp.valueOf(LocalDateTime.now().minusHours(1)));
                    sqlServerStmt.setInt(2, offset);
                    sqlServerStmt.setInt(3, pageSize);

                    ResultSet resultSet = sqlServerStmt.executeQuery();

                    if (!resultSet.next()) {
                        logger.info("无更多数据，offset={}", offset);
                        break;
                    }

                    // 👇 步骤1：收集本批次所有记录
                    List<RecordData> batchRecords = new ArrayList<>();
                    List<String> batchUids = new ArrayList<>();

                    do {
                        RecordData data = extractRecordData(resultSet); // 👈 抽取数据
                        batchRecords.add(data);
                        batchUids.add(data.uid);
                    } while (resultSet.next());

                    logger.info("本批次共 {} 条记录", batchRecords.size());

                    // 👇 步骤2：批量查询哪些 uid 已存在（性能优化）
                    Set<String> existingUids = getExistingUids(mysqlConn, batchUids);
                    logger.info("已存在记录数: {}", existingUids.size());

                    // 👇 步骤3：准备批量插入
                    try (PreparedStatement mysqlStmt = mysqlConn.prepareStatement(insertSql)) {

                        for (RecordData data : batchRecords) {
                            if (validFourTypes.contains(data.gtype)) {
                                logger.debug("跳过 validFourTypes 类型记录，gtype={}, uid={}", data.gtype, data.uid);
                                continue;
                            }
                            if ("4".equals(data.operation_platform)) {
                                logger.debug("跳过 operation_platform=4 的记录，uid={}", data.uid);
                                continue;
                            }
                            // 👇 跳过已存在的记录（避免重复更新用户余额）
                            if (existingUids.contains(data.uid)) {
                                logger.debug("跳过重复记录，uid={}", data.uid);
                                continue;
                            }

                            // 👇 设置插入参数
                            setStatementParams(mysqlStmt, data);

                            mysqlStmt.addBatch();

                            // 👇 只有新记录才更新用户余额（关键修复！）
                            updateUserBalance(mysqlConn, data);
                        }

                        // 👇 执行批量插入
                        int[] results = mysqlStmt.executeBatch();
                        logger.info("成功插入新记录 {} 条", results.length);
                    }

                    offset += pageSize;
                }
            }

            logger.info("✅ 数据同步完成");
        } catch (SQLException e) {
            logger.error("数据库操作失败", e);
            throw new RuntimeException("同步失败", e);
        }
    }

    static class RecordData {
        int gtype, jwcode, free, core_jb, buy_jb;
        Timestamp cz_time;
        String cz_user, cz_bz, operation_platform, goods_name, uid;
        String orderNumber; // 预生成，避免重复计算
    }

    private RecordData extractRecordData(ResultSet rs) throws SQLException {
        RecordData data = new RecordData();
        data.gtype = rs.getInt("gtype");
        data.jwcode = rs.getInt("jwcode");
        data.free = rs.getInt("free");
        data.core_jb = rs.getInt("core_jb");
        data.buy_jb = rs.getInt("buy_jb");
        data.cz_time = rs.getTimestamp("cz_time");
        data.cz_user = rs.getString("cz_user");
        data.cz_bz = rs.getString("cz_bz");
        data.operation_platform = rs.getString("operation_platform");
        data.goods_name = rs.getString("goods_name");
        data.uid = rs.getString("id");

        // 预生成订单号（避免在循环中重复生成）
        String timestampPart = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyyMMddHHmmssSSS"));
        String uuidPart = UUID.randomUUID().toString().replaceAll("-", "");
        data.orderNumber = timestampPart + "_" + uuidPart;

        return data;
    }

    private Set<String> getExistingUids(Connection conn, List<String> uids) throws SQLException {
        if (uids.isEmpty()) return Collections.emptySet();

        String placeholders = String.join(",", Collections.nCopies(uids.size(), "?"));
        String sql = "SELECT uid FROM user_gold_record WHERE uid IN (" + placeholders + ")";

        Set<String> existing = new HashSet<>();
        try (PreparedStatement stmt = conn.prepareStatement(sql)) {
            for (int i = 0; i < uids.size(); i++) {
                stmt.setString(i + 1, uids.get(i));
            }
            try (ResultSet rs = stmt.executeQuery()) {
                while (rs.next()) {
                    existing.add(rs.getString("uid"));
                }
            }
        }
        return existing;
    }
    private void setStatementParams(PreparedStatement stmt, RecordData data) throws SQLException {
        String name = data.cz_user;

        // 设置 admin_id
        if (StringUtils.isNumeric(name)) {
            try {
                String adminIdStr = adminService.getId(name);
                if (adminIdStr != null && StringUtils.isNumeric(adminIdStr)) {
                    stmt.setInt(14, Integer.parseInt(adminIdStr));
                } else {
                    stmt.setInt(14, 99999);
                }
            } catch (Exception e) {
                logger.warn("解析 admin_id 失败，cz_user={}", name, e);
                stmt.setInt(14, 99999);
            }
        } else {
            stmt.setInt(14, 99999);
        }

        // refund_type, refund_model
        stmt.setString(10, null);
        stmt.setNull(11, Types.INTEGER);

        // 根据 gtype 设置 type 和 order_code
        if (validFourTypes.contains(data.gtype)) {
            throw new IllegalArgumentException("不应处理 validFourTypes 类型，应在上层过滤"); // 安全兜底
        }

        if (validZeroTypes.contains(data.gtype)) {
            stmt.setInt(13, 0);
            stmt.setNull(20, Types.INTEGER);
            stmt.setString(1, "ERPCZ_" + data.orderNumber);
        } else if (validOneTypes.contains(data.gtype)) {
            stmt.setInt(13, 1);
            stmt.setInt(20, 0);
            stmt.setString(1, "ERPXF_" + data.orderNumber);
        } else if (validTwoTypes.contains(data.gtype)) {
            stmt.setInt(13, 2);
            stmt.setInt(20, 0);
            stmt.setString(1, "ERPTK_" + data.orderNumber);
            stmt.setString(10, "退款商品");
            stmt.setInt(11, 0);
        } else if (validThreeTypes.contains(data.gtype)) {
            stmt.setInt(13, 3);
            stmt.setNull(20, Types.INTEGER);
            stmt.setString(1, "ERPQT_" + data.orderNumber);
        }

        stmt.setInt(2, data.jwcode);
        stmt.setInt(3, data.free + data.core_jb + data.buy_jb);
        stmt.setInt(4, data.buy_jb);

        if (currentMonth.getValue() >= 7) {
            stmt.setInt(5, data.free);
            stmt.setInt(6, 0);
        } else {
            stmt.setInt(5, 0);
            stmt.setInt(6, data.free);
        }

        stmt.setInt(7, data.core_jb);

        // pay_platform
        String platform = data.operation_platform;
        if ("1".equals(platform)) {
            stmt.setString(8, "ERP");
        } else if ("2".equals(platform)) {
            stmt.setString(8, "HomilyLink");
        } else if ("3".equals(platform)) {
            stmt.setString(8, "HomilyChart");
        } else if ("4".equals(platform)) {
            throw new IllegalArgumentException("不应处理 platform=4，应在上层过滤");
        } else if ("0".equals(platform)) {
            stmt.setString(8, "初始化金币");
        } else {
            stmt.setString(8, "其他");
        }

        stmt.setString(9, data.goods_name);
        stmt.setString(12, data.cz_bz);
        stmt.setInt(15, 3);
        stmt.setTimestamp(16, data.cz_time);

        if (data.cz_bz != null && data.cz_bz.contains("测试") && data.cz_bz.contains("员工")) {
            stmt.setInt(17, 0);
        } else {
            stmt.setInt(17, 1);
        }

        stmt.setTimestamp(18, data.cz_time);
        stmt.setTimestamp(19, data.cz_time);
        stmt.setString(21, data.uid);
    }

    private void updateUserBalance(Connection conn, RecordData data) throws Exception {
        logger.info("处理用户余额更新，jwcode={}", data.jwcode);

        User user = userService.selectAllUser(String.valueOf(data.jwcode));
        BigDecimal freeBD = BigDecimal.valueOf(data.free);
        BigDecimal buyJbBD = BigDecimal.valueOf(data.buy_jb);
        BigDecimal coreJbBD = BigDecimal.valueOf(data.core_jb);

        if (ObjectUtils.isEmpty(user)) {
            logger.info("用户不存在，jwcode={}", data.jwcode);
            user = new User();
            String country = "未知";
            String name = "未知";

            try {
                BaseDES des = new BaseDES();
                String desjwcode = des.encrypt(String.valueOf(data.jwcode));

                Map<String, String> requestBody = new HashMap<>();
                requestBody.put("jwcode", desjwcode);

                HttpHeaders headers = new HttpHeaders();
                headers.setContentType(MediaType.APPLICATION_JSON);
                HttpEntity<Map<String, String>> entity = new HttpEntity<>(requestBody, headers);

                ResponseEntity<Map> response = restTemplate.exchange(
                        "http://hwapi.rzfwq.com/hwjnApp/hwhc-login/hwhclogin/hc/login/clent/info",
                        HttpMethod.POST, entity, Map.class);

                if (response.getStatusCode().is2xxSuccessful()) {
                    Map<String, Object> responseBody = response.getBody();
                    if (responseBody != null) {
                        Map<String, Object> dataMap = (Map<String, Object>) responseBody.get("data");
                        if (dataMap != null) {
                            name = (String) dataMap.get("name");
                            Object countryObj = dataMap.get("country");
                            if (countryObj instanceof List) {
                                country = (String) countryObj;
                            }
                            logger.info("获取用户信息成功: name={}, country={}", name, country);
                        }
                    }
                }
            } catch (Exception e) {
                logger.warn("获取用户信息失败，jwcode={}", data.jwcode, e);
                country = "未知";
            }

            String market = marketService.getMarketIdsDao(country);
            user.setJwcode(data.jwcode);
            user.setName(name);
            user.setMarket(market);
            userService.addUser(user);
            logger.info("用户创建成功，jwcode={}", data.jwcode);

            // 重新查询确保数据完整
            user = userService.selectAllUser(String.valueOf(data.jwcode));
        }

        // 更新当前金币
        if (currentMonth.getValue() >= 7) {
            BigDecimal newFreeJune = user.getCurrentFreeJune().add(freeBD);
            if (newFreeJune.compareTo(BigDecimal.ZERO) >= 0) {
                user.setCurrentFreeJune(newFreeJune);
            } else {
                BigDecimal remaining = newFreeJune;
                user.setCurrentFreeJune(BigDecimal.ZERO);
                user.setCurrentFreeDecember(user.getCurrentFreeDecember().add(remaining));
            }
        } else {
            BigDecimal newFreeDec = user.getCurrentFreeDecember().add(freeBD);
            if (newFreeDec.compareTo(BigDecimal.ZERO) >= 0) {
                user.setCurrentFreeDecember(newFreeDec);
            } else {
                BigDecimal remaining = newFreeDec;
                user.setCurrentFreeDecember(BigDecimal.ZERO);
                user.setCurrentFreeJune(remaining);
            }
        }

        user.setCurrentPermanentGold(user.getCurrentPermanentGold().add(buyJbBD));
        user.setCurrentTaskGold(user.getCurrentTaskGold().add(coreJbBD));

        // 更新统计字段
        if (validZeroTypes.contains(data.gtype)) {
            user.setRechargeNum(user.getRechargeNum() + 1);
            user.setSumPermanentGold(user.getSumPermanentGold().add(buyJbBD));
            user.setSumTaskGold(user.getSumTaskGold().add(coreJbBD));
            if (currentMonth.getValue() >= 7) {
                user.setSumFreeJune(user.getSumFreeJune().add(freeBD));
            } else {
                user.setSumFreeDecember(user.getSumFreeDecember().add(freeBD));
            }
        } else if (validOneTypes.contains(data.gtype)) {
            user.setConsumeNum(user.getConsumeNum() + 1);
            user.setSumConsumePermanent(user.getSumConsumePermanent().add(buyJbBD));
            user.setSumConsumeTask(user.getSumConsumeTask().add(coreJbBD));
            user.setSumConsumeFree(user.getSumConsumeFree().add(freeBD));
        }

        userService.updateAllGold(user);
        logger.info("用户余额更新成功，jwcode={}", data.jwcode);
    }
    private boolean checkRecordExists(Connection conn, String uid) throws SQLException {
        String sql = "SELECT 1 FROM user_gold_record WHERE uid = ? LIMIT 1";
        try (PreparedStatement stmt = conn.prepareStatement(sql)) {
            stmt.setString(1, uid);
            try (ResultSet rs = stmt.executeQuery()) {
                return rs.next();
            }
        }
    }

}