dataservice-yy/src/main/java/com/threecloud/dataserviceyy/service/FtpSyncService.java

package com.threecloud.dataserviceyy.service;

import com.threecloud.dataserviceyy.config.FtpSyncProperties;
import com.threecloud.dataserviceyy.config.FtpSyncProperties.FtpCityConfig;
import com.threecloud.dataserviceyy.entity.MidVoiceCallRecord;
import com.threecloud.dataserviceyy.mapper.MidVoiceCallRecordMapper;
import com.threecloud.dataserviceyy.util.FilePathUtil;
import com.threecloud.dataserviceyy.util.FileUploadUtil;
import com.threecloud.dataserviceyy.util.FtpUtil;
import org.apache.commons.net.ftp.FTPClient;
import org.apache.commons.net.ftp.FTPFile;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;
import org.springframework.util.StringUtils;

import java.io.BufferedReader;
import java.io.ByteArrayInputStream;
import java.io.File;
import java.io.InputStreamReader;
import java.text.SimpleDateFormat;
import java.util.*;

/**
 * FTP 录音同步服务（参考淮南 yydc-tb-server 的 MineService 重写）
 *
 * 【功能】从FTP读取录音数据txt → 解析 → 下载录音文件 → 上传OSS → 入库
 * 【策略】增量同步、无重试、数据库批量防重、txt处理完归档
 * 【部署】每个地市单独部署一份，每个服务只处理一个地市的FTP
 *
 * txt文件格式（GBK编码，※分隔）：
 *   kssj※jssj※hjls※hjzls※zjhm※bjhm※thsc※...※thfx
 *
 *   kssj  : 开始时间 yyyyMMddHHmmss
 *   jssj  : 结束时间 yyyyMMddHHmmss
 *   hjls  : 呼叫流水（录音文件主名）
 *   hjzls : 呼叫主/被叫流水
 *   zjhm  : 主叫号码
 *   bjhm  : 被叫号码
 *   thsc  : 通话时长（秒）
 *   thfx  : 通话方向 1=呼入 0=呼出
 *
 * 录音文件命名：{hjls}_{hjzls}.mp3
 * 录音文件路径：{ftpRecordDir}/{yyyyMMdd}/{hjls}_{hjzls}.mp3
 */
@Service
public class FtpSyncService {

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

    @Autowired
    private FtpSyncProperties properties;
    @Autowired
    private MidVoiceCallRecordMapper callRecordMapper;
    @Autowired
    private FileUploadUtil fileUploadUtil;

    /**
     * 定时同步任务
     */
    @Scheduled(cron = "${ftp-sync.sync-interval-cron:0 0 0/2 * * ?}")
    public void scheduledSync() {
        logger.info("【FTP定时】========== FTP录音同步开始 ==========");
        long startTime = System.currentTimeMillis();
        try {
            executeSync();
        } catch (Exception e) {
            logger.error("【FTP定时】同步任务异常: {}", e.getMessage(), e);
        }
        long costTime = System.currentTimeMillis() - startTime;
        logger.info("【FTP定时】========== FTP录音同步结束，耗时 {} 秒 ==========", costTime / 1000);
    }

    /**
     * 执行同步（主入口）
     */
    public void executeSync() {
        // 清理过期临时文件
        cleanOldFiles();

        List<FtpCityConfig> cities = properties.getCities();
        if (cities == null || cities.isEmpty()) {
            logger.warn("【FTP主流程】未配置任何地市FTP，跳过同步");
            return;
        }
        logger.info("【FTP主流程】配置了 {} 个地市", cities.size());

        for (FtpCityConfig city : cities) {
            try {
                syncCity(city);
            } catch (Exception e) {
                logger.error("【FTP主流程】地市 {} 同步失败: {}", city.getCityName(), e.getMessage(), e);
            }
        }
    }

    /**
     * 同步单个地市
     */
    private void syncCity(FtpCityConfig city) throws Exception {
        logger.info("【FTP地市】═══════════════════════════════════════");
        logger.info("【FTP地市】开始同步: {}({}), FTP={}:{}",
                city.getCityName(), city.getCityCode(), city.getFtpHost(), city.getFtpPort());

        // 参数校验
        if (!StringUtils.hasText(city.getFtpHost()) || !StringUtils.hasText(city.getFtpUsername())) {
            logger.warn("【FTP地市】{} FTP未配置完整，跳过", city.getCityName());
            return;
        }

        // 批量查询数据库已存在的call_record_id（防重）
        String prefix = city.getCityCode() + "_FTP_";
        Set<String> existingIds = loadExistingCallRecordIds(prefix);

        FTPClient ftp = null;
        try {
            // 连接FTP
            ftp = FtpUtil.connect(city.getFtpHost(), city.getFtpPort(),
                    city.getFtpUsername(), city.getFtpPassword());
            logger.info("【FTP地市】FTP连接成功");

            // 列出txt源目录所有文件
            FTPFile[] files = FtpUtil.listFiles(ftp, city.getFtpSourceDir());
            int txtCount = countTxtFiles(files);
            logger.info("【FTP地市】目录 {} 下有 {} 个txt文件", city.getFtpSourceDir(), txtCount);

            int fileSuccess = 0;
            int fileFail = 0;

            for (FTPFile file : files) {
                if (!file.isFile() || !file.getName().toLowerCase().endsWith(".txt")) {
                    continue;
                }
                try {
                    boolean ok = processTxtFile(ftp, file, city, existingIds);
                    if (ok) fileSuccess++;
                    else fileFail++;
                } catch (Exception e) {
                    fileFail++;
                    logger.error("【FTP地市】处理文件失败: {}, 原因={}", file.getName(), e.getMessage());
                }
            }
            logger.info("【FTP地市】{} 同步完成: 成功{}个文件, 失败{}个文件",
                    city.getCityName(), fileSuccess, fileFail);
        } finally {
            FtpUtil.disconnect(ftp);
        }
    }

    /**
     * 处理单个txt文件
     *
     * @return true=成功，false=失败
     */
    private boolean processTxtFile(FTPClient ftp, FTPFile file, FtpCityConfig city,
                                    Set<String> existingIds) throws Exception {
        String fileName = file.getName();
        String sourcePath = city.getFtpSourceDir() + "/" + fileName;
        logger.info("【FTP文件】开始处理: {}", fileName);

        // 下载txt到内存
        byte[] txtBytes = FtpUtil.downloadFile(ftp, sourcePath);
        int lineCount = 0;
        int successCount = 0;
        int skipCount = 0;
        int failCount = 0;

        try (BufferedReader reader = new BufferedReader(
                new InputStreamReader(new ByteArrayInputStream(txtBytes), properties.getTxtEncoding()))) {
            String line;
            while ((line = reader.readLine()) != null) {
                if (line.indexOf(properties.getFieldSeparator()) <= 0) {
                    continue;
                }
                lineCount++;
                try {
                    boolean processed = processLine(line, city, existingIds, ftp);
                    if (processed) successCount++;
                    else skipCount++;
                } catch (Exception e) {
                    failCount++;
                    logger.error("【FTP文件】解析单行失败: {}, 原因={}", line, e.getMessage());
                }
            }
        }

        logger.info("【FTP文件】{} 解析完成: 共{}行, 新增{}条, 跳过{}条, 失败{}条",
                fileName, lineCount, successCount, skipCount, failCount);

        // 移动到归档目录
        try {
            String archivePath = city.getFtpArchiveDir() + "/" + fileName;
            FtpUtil.ensureDir(ftp, city.getFtpArchiveDir());
            FtpUtil.rename(ftp, sourcePath, archivePath);
            logger.info("【FTP文件】已归档: {} -> {}", sourcePath, archivePath);
        } catch (Exception e) {
            logger.error("【FTP文件】归档失败: {}, 原因={}", fileName, e.getMessage());
        }

        return true;
    }

    /**
     * 处理单行记录
     *
     * @return true=成功入库，false=跳过
     */
    private boolean processLine(String line, FtpCityConfig city, Set<String> existingIds,
                                 FTPClient ftp) throws Exception {
        String[] fields = line.split(properties.getFieldSeparator(), -1);
        if (fields.length < 10) {
            logger.debug("【FTP行】字段数不足（{}<10），跳过: {}", fields.length, line);
            return false;
        }

        String kssj = trimToNull(fields[0]);    // 开始时间 yyyyMMddHHmmss
        String jssj = trimToNull(fields[1]);    // 结束时间
        String hjls = trimToNull(fields[2]);    // 呼叫流水（录音主名）
        String hjzls = trimToNull(fields[3]);   // 呼叫主/被叫流水
        String zjhm = trimToNull(fields[4]);    // 主叫号码
        String bjhm = trimToNull(fields[5]);    // 被叫号码
        String thscStr = trimToNull(fields[6]); // 通话时长
        String thfx = trimToNull(fields[9]);    // 通话方向

        if (kssj == null || jssj == null || hjls == null || hjzls == null) {
            return false;
        }

        int thsc = 0;
        try { thsc = thscStr == null ? 0 : Integer.parseInt(thscStr); } catch (NumberFormatException ignored) {}

        // 解析时间
        Date callStartTime = parseDate(kssj);
        Date callEndTime = parseDate(jssj);
        if (callStartTime == null) {
            return false;
        }

        // 通话方向：1=呼入 0=呼出（老系统thfx字段）
        boolean isOutgoing = thfx != null && thfx.contains("0");
        String callDirection = isOutgoing ? "2" : "1"; // 1呼入 2呼出（数据库规范）

        // 录音文件名
        String recordFileName = hjls + "_" + hjzls + ".mp3";

        // callRecordId 唯一标识
        String callRecordId = city.getCityCode() + "_FTP_" + hjls;

        // 防重复检查
        if (existingIds.contains(callRecordId)) {
            logger.debug("【FTP行】已存在，跳过: {}", callRecordId);
            return false;
        }

        // 主叫/被叫处理（去除 0553 区号前缀 - 参考老代码）
        if (zjhm != null && zjhm.startsWith("0553")) {
            zjhm = zjhm.substring(4);
        }
        if (bjhm != null && bjhm.startsWith("0553")) {
            bjhm = bjhm.substring(4);
        }

        // 通话状态：1正常接通
        String callStatus = "1";

        // 构建本地和OSS路径
        SimpleDateFormat yyyyMMdd = new SimpleDateFormat("yyyyMMdd");
        String dateStr = yyyyMMdd.format(callStartTime);
        String localPath = FilePathUtil.buildLocalPath(
                properties.getTempPath(), city.getCityCode(), callStartTime, "ftp", recordFileName);
        String ossPath = FilePathUtil.buildOssPath(city.getCityCode(), callStartTime, recordFileName);

        // 下载录音文件
        String remoteRecordPath = city.getFtpRecordDir() + "/" + dateStr + "/" + recordFileName;
        byte[] recordBytes = FtpUtil.downloadFile(ftp, remoteRecordPath);
        logger.info("【FTP行】下载录音: {} ({} 字节)", recordFileName, recordBytes.length);

        // 上传OSS
        String ossUrl = fileUploadUtil.uploadWav(city.getCityName(), ossPath, recordFileName, recordBytes);
        logger.info("【FTP行】上传OSS: {}", ossUrl);

        // 构建并保存实体
        MidVoiceCallRecord rec = new MidVoiceCallRecord();
        rec.setCallRecordId(callRecordId);
        rec.setDeviceNo("FTP_" + city.getCityCode());
        rec.setCityCode(city.getCityCode());
        rec.setCityName(city.getCityName());
        rec.setOrgCode(city.getCityCode());
        rec.setRecordingFileName(recordFileName);
        rec.setCallStartTime(callStartTime);
        rec.setCallEndTime(callEndTime);
        rec.setCallDuration(thsc);
        rec.setCallDirection(callDirection);

        if (isOutgoing) {
            rec.setCallTel(zjhm != null ? zjhm : "");
            rec.setCalledTel(bjhm != null ? bjhm : "");
        } else {
            rec.setCallTel(bjhm != null ? bjhm : "");
            rec.setCalledTel(zjhm != null ? zjhm : "");
        }
        rec.setCallStatus(callStatus);
        rec.setRecordingFileSize(recordBytes.length);
        rec.setRecordingFilePath(ossUrl);

        callRecordMapper.insert(rec);
        existingIds.add(callRecordId); // 防止同批内重复
        logger.info("【FTP行】保存成功: id={}, callRecordId={}", rec.getId(), callRecordId);

        return true;
    }

    /**
     * 批量加载已存在的call_record_id（防重）
     */
    private Set<String> loadExistingCallRecordIds(String prefix) {
        try {
            List<String> ids = callRecordMapper.selectCallRecordIdsByPrefix(prefix);
            logger.info("【FTP防重】已加载 {} 条历史记录（prefix={}）", ids.size(), prefix);
            return new HashSet<>(ids);
        } catch (Exception e) {
            logger.error("【FTP防重】加载历史记录失败，将全部当作新数据: {}", e.getMessage());
            return new HashSet<>();
        }
    }

    /**
     * 清理过期临时文件
     */
    private void cleanOldFiles() {
        File tempDir = new File(properties.getTempPath());
        if (!tempDir.exists()) {
            return;
        }
        long retainMillis = (long) properties.getRetainDays() * 24 * 60 * 60 * 1000;
        long now = System.currentTimeMillis();
        deleteOldFilesRecursive(tempDir, now - retainMillis);
    }

    private void deleteOldFilesRecursive(File dir, long threshold) {
        if (!dir.isDirectory()) {
            if (dir.lastModified() < threshold) {
                dir.delete();
            }
            return;
        }
        File[] files = dir.listFiles();
        if (files == null) return;
        for (File f : files) {
            deleteOldFilesRecursive(f, threshold);
        }
        // 删除空目录
        File[] remainFiles = dir.listFiles();
        if (remainFiles == null || remainFiles.length == 0) {
            dir.delete();
        }
    }

    /**
     * 统计txt文件数
     */
    private int countTxtFiles(FTPFile[] files) {
        int count = 0;
        if (files == null) return 0;
        for (FTPFile f : files) {
            if (f.isFile() && f.getName().toLowerCase().endsWith(".txt")) {
                count++;
            }
        }
        return count;
    }

    /**
     * 解析日期 yyyyMMddHHmmss
     */
    private Date parseDate(String s) {
        if (s == null) return null;
        try {
            return new SimpleDateFormat("yyyyMMddHHmmss").parse(s);
        } catch (Exception e) {
            return null;
        }
    }

    private String trimToNull(String s) {
        if (s == null) return null;
        s = s.trim();
        return s.isEmpty() ? null : s;
    }
}