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.LocalDate;
import java.time.LocalDateTime;
import java.time.Month;
import java.time.format.DateTimeFormatter;
import java.util.*;
import java.util.Date;
import java.util.stream.Collectors;


@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,67));
   Set<Integer> validOneTypes = new HashSet<>(Arrays.asList(9, 15, 17, 25, 27, 37, 41, 42, 43, 50, 51, 62,68));
   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",rollbackFor = Exception.class) // 👈 保证插入和用户更新在一个事务
    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,uid
            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,link_id)
            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);
        }
    }


    @Override
    @Transactional(transactionManager = "mysqlTransactionManager") // 👈 保证插入和用户更新在一个事务
    public void getSqlserverDataDay() 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, uid
            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,link_id)
            VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
            """;

            while (hasMoreData) {
                try (PreparedStatement sqlServerStmt = sqlServerConn.prepareStatement(querySql)) {
                    // 👇 恢复动态时间参数
                    sqlServerStmt.setTimestamp(1, Timestamp.valueOf(LocalDate.now().minusDays(1).atStartOfDay()));
                    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);

                        // 👉【新增一行调用】👈
                        // 传入当前连接 + 本批次记录，自动完成路由更新
                        updateUserRegionWallet(mysqlConn, batchRecords);
                    }

                    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;
        Integer permanent_gold;
        Timestamp cz_time;
        String cz_user, cz_bz, operation_platform, goods_name, uid;
        String orderNumber; // 预生成，避免重复计算
        String linkId;

    }

    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");
        data.linkId = rs.getString("uid");

        // 预生成订单号（避免在循环中重复生成）
        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.setInt(20, 0);
            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 if ("5".equals(platform)){
            stmt.setString(8, "DC");
        }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);
        }
        Timestamp now = new Timestamp(System.currentTimeMillis());
        stmt.setTimestamp(18, now);
        stmt.setTimestamp(19, data.cz_time);
        stmt.setString(21, data.uid);
        stmt.setString(22, data.linkId);
    }

    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");
                            country = getCountryWithDefault(dataMap, "未知");

                            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();
            }
        }
    }
    private String getCountryWithDefault(Map<String, Object> dataMap, String defaultValue) {
        Object countryObj = dataMap.get("treelist");

        if (countryObj instanceof String) {
            String countryStr = ((String) countryObj).trim();
            if (countryStr.isEmpty()) {
                return defaultValue;
            }

            String[] parts = countryStr.split("-");
            return parts.length >= 3 ? parts[2] : defaultValue;
        }

        return defaultValue;
    }

    /**
     * 查 g_order 获取 pay_style，路由更新 user_region_wallet 余额
     * @param mysqlConn 当前事务连接（⚠️必须使用此连接）
     * @param records 本批次有效记录
     */
    private void updateUserRegionWallet(Connection mysqlConn, List<RecordData> records) {
        if (records == null || records.isEmpty()) return;

        // 1. 收集 jwcode 去重
        List<Integer> jwcodes = records.stream()
                .map(r -> r.jwcode).filter(Objects::nonNull).distinct()
                .collect(Collectors.toList());
        if (jwcodes.isEmpty()) return;

        // 2. 批量查 g_order 获取 pay_style（state=1 支付完成）
        String inSql = String.join(",", Collections.nCopies(jwcodes.size(), "?"));
        String querySql = "SELECT jwcode, pay_style FROM g_order WHERE jwcode IN (" + inSql + ") AND state = 1";

        Map<Integer, Integer> jwcodePayStyleMap = new HashMap<>();
        try (PreparedStatement stmt = mysqlConn.prepareStatement(querySql)) {
            for (int i = 0; i < jwcodes.size(); i++) stmt.setInt(i + 1, jwcodes.get(i));
            try (ResultSet rs = stmt.executeQuery()) {
                while (rs.next()) jwcodePayStyleMap.put(rs.getInt("jwcode"), rs.getInt("pay_style"));
            }
        } catch (SQLException e) {
            throw new RuntimeException("查询 g_order 失败", e);
        }

        // 3. 按 pay_style 路由到 wallet_id，累加永久金币
        Map<String, Integer> userWalletGoldMap = new HashMap<>(); // key: "jwcode_walletId" -> 累加金额
        for (RecordData data : records) {
            if (!jwcodePayStyleMap.containsKey(data.jwcode)) continue;
            if (data.permanent_gold == null || data.permanent_gold <= 0) continue;

            Integer payStyle = jwcodePayStyleMap.get(data.jwcode);
            Integer walletId = mapPayStyleToWalletId(payStyle);
            if (walletId == null) continue;

            String key = data.jwcode + "_" + walletId;
            userWalletGoldMap.merge(key, data.permanent_gold, Integer::sum);
        }
        if (userWalletGoldMap.isEmpty()) return;

        // 4. UPSERT 批量更新 user_region_wallet（不存在则插入，存在则累加）
        String upsertSql = """
        INSERT INTO user_region_wallet 
        (jwcode, wallet_id, current_permanent_gold, create_time, update_time) 
        VALUES (?, ?, ?, NOW(), NOW())
        ON DUPLICATE KEY UPDATE 
            current_permanent_gold = current_permanent_gold + VALUES(current_permanent_gold),
            update_time = NOW()
        """;

        try (PreparedStatement stmt = mysqlConn.prepareStatement(upsertSql)) {
            for (Map.Entry<String, Integer> entry : userWalletGoldMap.entrySet()) {
                String[] parts = entry.getKey().split("_");
                int jwcode = Integer.parseInt(parts[0]);
                int walletId = Integer.parseInt(parts[1]);

                stmt.setInt(1, jwcode);
                stmt.setString(2, String.valueOf(walletId));
                stmt.setInt(3, entry.getValue());
                stmt.addBatch();
            }
            int[] results = stmt.executeBatch();
            logger.info("✅ 地区钱包余额更新 {} 条", results.length);
        } catch (SQLException e) {
            throw new RuntimeException("更新地区钱包失败", e);
        }
    }

    /**
     * pay_style → wallet_id 映射（⚠️ 请替换成你实际的 wallet_id）
     */
    private Integer mapPayStyleToWalletId(Integer payStyle) {
        if (payStyle == null) return null;
        return switch (payStyle) {
            case 3,4,9 -> 4;      // 微信/支付宝 → 华东钱包
            case 5,6 -> 1;      // Stripe/PaymentAsia → 华南钱包
            case 7 -> 3; // 其他 → 西南钱包
            case 10 -> 7;
            case 15 -> 2;
            default -> null;
        };
    }
}