From 58392eddcc6ffff0e8acdfc7b7cbddb8268c8f1c Mon Sep 17 00:00:00 2001 From: wang Date: Mon, 17 Aug 2026 15:31:00 +0800 Subject: [PATCH] =?UTF-8?q?ftp=E4=BB=A3=E7=A0=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- config/application-external.yml | 2 + .../config/FtpSyncProperties.java | 7 ++ .../dataserviceyy/service/FtpSyncService.java | 63 ++++++++++----- .../dataserviceyy/util/FtpUtil.java | 76 ++++++++++++++++--- 4 files changed, 116 insertions(+), 32 deletions(-) diff --git a/config/application-external.yml b/config/application-external.yml index d8e28c6..50b2a4f 100644 --- a/config/application-external.yml +++ b/config/application-external.yml @@ -52,6 +52,7 @@ ftp-sync: ftp-record-dir: /rec/ ftp-archive-dir: /voice_record/processed/ phone-area-code: "0554" + passive-mode: true # 是否被动模式：true=被动模式(PASV，默认)，false=主动模式(PORT) # ===== 蚌埠 ===== - city-code: "340300" @@ -64,6 +65,7 @@ ftp-sync: ftp-record-dir: /rec/ ftp-archive-dir: /voice_record/processed/ phone-area-code: "0552" + passive-mode: true # 是否被动模式：true=被动模式(PASV，默认)，false=主动模式(PORT) # ==================== 服务端口 ==================== server: diff --git a/src/main/java/com/threecloud/dataserviceyy/config/FtpSyncProperties.java b/src/main/java/com/threecloud/dataserviceyy/config/FtpSyncProperties.java index d9fc66b..bd6f7bf 100644 --- a/src/main/java/com/threecloud/dataserviceyy/config/FtpSyncProperties.java +++ b/src/main/java/com/threecloud/dataserviceyy/config/FtpSyncProperties.java @@ -95,6 +95,10 @@ public class FtpSyncProperties { * - 留空：保留原始号码 */ private String phoneAreaCode; + /** + * FTP传输模式：true=被动模式(PASV，默认)，false=主动模式(PORT) + */ + private boolean passiveMode = true; public String getCityCode() { return cityCode; } public void setCityCode(String cityCode) { this.cityCode = cityCode; } @@ -125,5 +129,8 @@ public class FtpSyncProperties { public String getPhoneAreaCode() { return phoneAreaCode; } public void setPhoneAreaCode(String phoneAreaCode) { this.phoneAreaCode = phoneAreaCode; } + + public boolean isPassiveMode() { return passiveMode; } + public void setPassiveMode(boolean passiveMode) { this.passiveMode = passiveMode; } } } diff --git a/src/main/java/com/threecloud/dataserviceyy/service/FtpSyncService.java b/src/main/java/com/threecloud/dataserviceyy/service/FtpSyncService.java index dec03f2..d83b436 100644 --- a/src/main/java/com/threecloud/dataserviceyy/service/FtpSyncService.java +++ b/src/main/java/com/threecloud/dataserviceyy/service/FtpSyncService.java @@ -129,7 +129,7 @@ public class FtpSyncService { FTPClient ftp = null; try { ftp = FtpUtil.connect(city.getFtpHost(), city.getFtpPort(), - city.getFtpUsername(), city.getFtpPassword()); + city.getFtpUsername(), city.getFtpPassword(), city.isPassiveMode()); logger.info("【FTP地市】FTP连接成功"); FTPFile[] files = FtpUtil.listFiles(ftp, city.getFtpSourceDir()); @@ -140,6 +140,11 @@ public class FtpSyncService { Set processedFiles = loadProcessedFileNames(city.getCityCode()); logger.info("【FTP地市】已处理文件记录: {} 个", processedFiles.size()); + // 按文件名升序排序，确保按时间先后顺序依次处理 + if (files != null && files.length > 0) { + Arrays.sort(files, Comparator.comparing(FTPFile::getName)); + } + int fileSuccess = 0; int fileFail = 0; int fileSkipped = 0; @@ -149,10 +154,9 @@ public class FtpSyncService { continue; } String fileName = file.getName(); - // 历史目录中只处理当天已经封口的时间段文件： - // 例如 20260722123-20260722163.txt；空文件或无横杠的占位文件暂不处理。 - if (!isTodayCompletedIntervalTxtFile(file)) { - logger.debug("【FTP文件】非当天有效时间段文件，跳过: {} ({} 字节)", + // 检查是否为有效的数据TXT文件（支持跨天、一天内多个切片文件；排除空文件或未封口占位文件） + if (!isValidDataTxtFile(file)) { + logger.debug("【FTP文件】非有效数据文件，跳过: {} ({} 字节)", fileName, file.getSize()); continue; } @@ -173,7 +177,7 @@ public class FtpSyncService { try { FtpUtil.disconnect(ftp); // 关闭老连接 ftp = FtpUtil.connect(city.getFtpHost(), city.getFtpPort(), - city.getFtpUsername(), city.getFtpPassword()); + city.getFtpUsername(), city.getFtpPassword(), city.isPassiveMode()); logger.info("【FTP地市】重新连接 FTP 成功，正在重试处理刚才失败的文件: {}", fileName); // 重新尝试处理文件 @@ -341,15 +345,21 @@ public class FtpSyncService { String remoteRecordPath = recordDir + "/" + dateStr + "/" + recordFileName; logger.info("【FTP行】准备下载录音: {}", remoteRecordPath); byte[] recordBytes = FtpUtil.downloadFile(ftp, remoteRecordPath); - logger.info("【FTP行】下载录音: {} ({} 字节)", recordFileName, recordBytes.length); - - // 上传到OSS（获取可访问的URL） - String ossUrl = fileUploadUtil.uploadWav(ossPath, recordFileName, recordBytes); - logger.info("【FTP行】上传OSS: {}", ossUrl); + String ossUrl = null; + int fileSize = 0; + if (recordBytes != null && recordBytes.length > 0) { + fileSize = recordBytes.length; + logger.info("【FTP行】下载录音: {} ({} 字节)", recordFileName, fileSize); + // 上传到OSS（获取可访问的URL） + ossUrl = fileUploadUtil.uploadWav(ossPath, recordFileName, recordBytes); + logger.info("【FTP行】上传OSS: {}", ossUrl); + } else { + logger.warn("【FTP行】录音文件不存在或为空，跳过OSS上传: {}", remoteRecordPath); + } // 构建实体对象并入库 MidVoiceCallRecord rec = buildCallRecord(city, callRecordId, callStartTime, callEndTime, - thsc, isOutgoing, zjhm, bjhm, recordFileName, recordBytes.length, ossUrl); + thsc, isOutgoing, zjhm, bjhm, recordFileName, fileSize, ossUrl); callRecordMapper.insert(rec); existingIds.add(callRecordId); // 防止同批内重复（本次循环内） logger.info("【FTP行】保存成功: id={}, callRecordId={}", rec.getId(), callRecordId); @@ -478,19 +488,32 @@ public class FtpSyncService { } /** - * 判断是否为当天已经生成完成的时间段TXT文件。 - * 文件名规则：以当天yyyyMMdd开头，日期后同时包含起止时间横杠，且文件非空。 + * 判断是否为有效且已封口的TXT数据文件（支持跨天、多天、一天多个时间切片文件） + * 兼容格式： + * 1. 芜湖等带横杠时间段格式：如 202608170700-202608170730.txt（要求两端时间戳完整，排除未写完的占位文件） + * 2. 淮南/蚌埠等纯时间戳格式：如 202608170740.txt（yyyyMMddHHmm.txt 或 yyyyMMddHHmmss.txt） */ - private boolean isTodayCompletedIntervalTxtFile(FTPFile file) { + private boolean isValidDataTxtFile(FTPFile file) { if (!FtpUtil.isTxtFile(file) || file.getSize() <= 0) { return false; } String fileName = file.getName(); - String today = DateUtil.formatDate(new Date(), "yyyyMMdd"); - int separatorIndex = fileName.indexOf('-', today.length()); - return fileName.startsWith(today) - && separatorIndex > today.length() - && separatorIndex < fileName.length() - ".txt".length() - 1; + // 去掉后缀 .txt + String nameWithoutExt = fileName.substring(0, fileName.length() - ".txt".length()); + + // 基础规则：文件名开头至少是 8 位合法数字日期（yyyyMMdd，如 20260817） + if (nameWithoutExt.length() < 8 || !nameWithoutExt.substring(0, 8).matches("\\d{8}")) { + return false; + } + + if (nameWithoutExt.contains("-")) { + // 模式1：芜湖等时间段文件（必须包含横杠且两端时间戳完整） + int separatorIndex = nameWithoutExt.indexOf('-'); + return separatorIndex >= 8 && separatorIndex < nameWithoutExt.length() - 1; + } else { + // 模式2：淮南/蚌埠等纯时间戳文件（纯数字，如 202608170740） + return nameWithoutExt.matches("\\d{12,14}"); + } } /** diff --git a/src/main/java/com/threecloud/dataserviceyy/util/FtpUtil.java b/src/main/java/com/threecloud/dataserviceyy/util/FtpUtil.java index 0870e2d..9707045 100644 --- a/src/main/java/com/threecloud/dataserviceyy/util/FtpUtil.java +++ b/src/main/java/com/threecloud/dataserviceyy/util/FtpUtil.java @@ -25,7 +25,7 @@ public class FtpUtil { private static final int READ_TIMEOUT = 60000; /** - * 连接FTP服务器（一次性连接处理所有操作，最后调用 disconnect 关闭） + * 连接FTP服务器（默认使用被动模式） * * @param host FTP服务器IP * @param port FTP服务器端口 @@ -34,21 +34,50 @@ public class FtpUtil { * @return FTPClient 连接成功返回客户端，失败抛异常 */ public static FTPClient connect(String host, int port, String username, String password) throws IOException { + return connect(host, port, username, password, true); + } + + /** + * 连接FTP服务器（支持指定主动/被动模式） + * + * @param host FTP服务器IP + * @param port FTP服务器端口 + * @param username 用户名 + * @param password 密码 + * @param passiveMode 是否使用被动模式（true=被动模式，false=主动模式） + * @return FTPClient 连接成功返回客户端，失败抛异常 + */ + public static FTPClient connect(String host, int port, String username, String password, boolean passiveMode) throws IOException { FTPClient ftp = new FTPClient(); + // 设置UTF-8编码（必须在 connect 前设置） + ftp.setControlEncoding("UTF-8"); ftp.setConnectTimeout(CONNECT_TIMEOUT); + ftp.setDefaultTimeout(CONNECT_TIMEOUT); ftp.setDataTimeout(READ_TIMEOUT); + + logger.info("正在尝试连接 FTP: {}:{} (模式: {})", host, port, passiveMode ? "被动(PASV)" : "主动(PORT)"); ftp.connect(host, port); boolean loginOk = ftp.login(username, password); if (!loginOk) { ftp.disconnect(); throw new IOException("FTP登录失败，请检查用户名密码"); } - // 设置被动模式（必须，避开防火墙） - ftp.enterLocalPassiveMode(); + // 设置二进制传输（mp3/wav必须用二进制） ftp.setFileType(FTP.BINARY_FILE_TYPE); - // 设置UTF-8编码（文件名有中文） - ftp.setControlEncoding("UTF-8"); + // 关闭远程IP验证（增强在多网卡/NAT环境下的数据连接兼容性） + ftp.setRemoteVerificationEnabled(false); + + if (passiveMode) { + // 设置被动模式 + ftp.enterLocalPassiveMode(); + // 关键：防止 FTP 服务端在 PASV 响应中返回内部局域网 IP / 127.0.0.1 导致数据端口连接超时 + ftp.setPassiveNatWorkaroundStrategy(hostname -> host); + } else { + // 设置主动模式 + ftp.enterLocalActiveMode(); + } + logger.info("FTP连接成功: {}:{}", host, port); return ftp; } @@ -94,12 +123,22 @@ public class FtpUtil { * @param remotePath 远程文件路径（绝对或相对于当前工作目录） * @return 文件字节内容 */ + /** + * 下载文件到本地字节数组（自动完成FTP pending命令） + * + * @param ftp 已连接的FTPClient + * @param remotePath 远程文件路径（绝对或相对于当前工作目录） + * @return 文件字节内容，若文件在FTP上不存在则返回 null + */ public static byte[] downloadFile(FTPClient ftp, String remotePath) throws IOException { InputStream in = null; try { in = ftp.retrieveFileStream(remotePath); if (in == null) { - throw new IOException("下载文件失败（无输入流）: " + remotePath); + int replyCode = ftp.getReplyCode(); + String replyMsg = ftp.getReplyString(); + logger.warn("FTP文件不存在或无法打开输入流: {}, replyCode={}, msg={}", remotePath, replyCode, replyMsg != null ? replyMsg.trim() : ""); + return null; } ByteArrayOutputStream out = new ByteArrayOutputStream(); byte[] buffer = new byte[8192]; @@ -108,14 +147,27 @@ public class FtpUtil { out.write(buffer, 0, len); } return out.toByteArray(); + } catch (IOException e) { + // 下载异常时尝试终止当前传输命令 + try { + ftp.abort(); + } catch (Exception ignored) { + } + throw e; } finally { - // 必须关闭流并调用 completePendingCommand，否则后续FTP操作会失败 + // 关键：只有成功打开了输入流，才需要 close 并调用 completePendingCommand if (in != null) { - try { in.close(); } catch (IOException ignored) {} - } - boolean ok = ftp.completePendingCommand(); - if (!ok) { - logger.warn("FTP completePendingCommand 返回 false: {}", remotePath); + try { + in.close(); + } catch (IOException ignored) {} + try { + boolean ok = ftp.completePendingCommand(); + if (!ok) { + logger.warn("FTP completePendingCommand 返回 false: {}", remotePath); + } + } catch (Exception e) { + logger.warn("FTP completePendingCommand 异常: {}, msg={}", remotePath, e.getMessage()); + } } } }