Browse Source

ftp代码

ftpmain
wang 3 weeks ago
parent
commit
58392eddcc
  1. 2
      config/application-external.yml
  2. 7
      src/main/java/com/threecloud/dataserviceyy/config/FtpSyncProperties.java
  3. 63
      src/main/java/com/threecloud/dataserviceyy/service/FtpSyncService.java
  4. 76
      src/main/java/com/threecloud/dataserviceyy/util/FtpUtil.java

2
config/application-external.yml

@ -52,6 +52,7 @@ ftp-sync:
ftp-record-dir: /rec/ ftp-record-dir: /rec/
ftp-archive-dir: /voice_record/processed/ ftp-archive-dir: /voice_record/processed/
phone-area-code: "0554" phone-area-code: "0554"
passive-mode: true # 是否被动模式:true=被动模式(PASV,默认),false=主动模式(PORT)
# ===== 蚌埠 ===== # ===== 蚌埠 =====
- city-code: "340300" - city-code: "340300"
@ -64,6 +65,7 @@ ftp-sync:
ftp-record-dir: /rec/ ftp-record-dir: /rec/
ftp-archive-dir: /voice_record/processed/ ftp-archive-dir: /voice_record/processed/
phone-area-code: "0552" phone-area-code: "0552"
passive-mode: true # 是否被动模式:true=被动模式(PASV,默认),false=主动模式(PORT)
# ==================== 服务端口 ==================== # ==================== 服务端口 ====================
server: server:

7
src/main/java/com/threecloud/dataserviceyy/config/FtpSyncProperties.java

@ -95,6 +95,10 @@ public class FtpSyncProperties {
* - 留空保留原始号码 * - 留空保留原始号码
*/ */
private String phoneAreaCode; private String phoneAreaCode;
/**
* FTP传输模式true=被动模式(PASV默认)false=主动模式(PORT)
*/
private boolean passiveMode = true;
public String getCityCode() { return cityCode; } public String getCityCode() { return cityCode; }
public void setCityCode(String cityCode) { this.cityCode = cityCode; } public void setCityCode(String cityCode) { this.cityCode = cityCode; }
@ -125,5 +129,8 @@ public class FtpSyncProperties {
public String getPhoneAreaCode() { return phoneAreaCode; } public String getPhoneAreaCode() { return phoneAreaCode; }
public void setPhoneAreaCode(String phoneAreaCode) { this.phoneAreaCode = phoneAreaCode; } public void setPhoneAreaCode(String phoneAreaCode) { this.phoneAreaCode = phoneAreaCode; }
public boolean isPassiveMode() { return passiveMode; }
public void setPassiveMode(boolean passiveMode) { this.passiveMode = passiveMode; }
} }
} }

63
src/main/java/com/threecloud/dataserviceyy/service/FtpSyncService.java

@ -129,7 +129,7 @@ public class FtpSyncService {
FTPClient ftp = null; FTPClient ftp = null;
try { try {
ftp = FtpUtil.connect(city.getFtpHost(), city.getFtpPort(), ftp = FtpUtil.connect(city.getFtpHost(), city.getFtpPort(),
city.getFtpUsername(), city.getFtpPassword()); city.getFtpUsername(), city.getFtpPassword(), city.isPassiveMode());
logger.info("【FTP地市】FTP连接成功"); logger.info("【FTP地市】FTP连接成功");
FTPFile[] files = FtpUtil.listFiles(ftp, city.getFtpSourceDir()); FTPFile[] files = FtpUtil.listFiles(ftp, city.getFtpSourceDir());
@ -140,6 +140,11 @@ public class FtpSyncService {
Set<String> processedFiles = loadProcessedFileNames(city.getCityCode()); Set<String> processedFiles = loadProcessedFileNames(city.getCityCode());
logger.info("【FTP地市】已处理文件记录: {} 个", processedFiles.size()); logger.info("【FTP地市】已处理文件记录: {} 个", processedFiles.size());
// 按文件名升序排序,确保按时间先后顺序依次处理
if (files != null && files.length > 0) {
Arrays.sort(files, Comparator.comparing(FTPFile::getName));
}
int fileSuccess = 0; int fileSuccess = 0;
int fileFail = 0; int fileFail = 0;
int fileSkipped = 0; int fileSkipped = 0;
@ -149,10 +154,9 @@ public class FtpSyncService {
continue; continue;
} }
String fileName = file.getName(); String fileName = file.getName();
// 历史目录中只处理当天已经封口的时间段文件: // 检查是否为有效的数据TXT文件(支持跨天、一天内多个切片文件;排除空文件或未封口占位文件)
// 例如 20260722123-20260722163.txt;空文件或无横杠的占位文件暂不处理。 if (!isValidDataTxtFile(file)) {
if (!isTodayCompletedIntervalTxtFile(file)) { logger.debug("【FTP文件】非有效数据文件,跳过: {} ({} 字节)",
logger.debug("【FTP文件】非当天有效时间段文件,跳过: {} ({} 字节)",
fileName, file.getSize()); fileName, file.getSize());
continue; continue;
} }
@ -173,7 +177,7 @@ public class FtpSyncService {
try { try {
FtpUtil.disconnect(ftp); // 关闭老连接 FtpUtil.disconnect(ftp); // 关闭老连接
ftp = FtpUtil.connect(city.getFtpHost(), city.getFtpPort(), ftp = FtpUtil.connect(city.getFtpHost(), city.getFtpPort(),
city.getFtpUsername(), city.getFtpPassword()); city.getFtpUsername(), city.getFtpPassword(), city.isPassiveMode());
logger.info("【FTP地市】重新连接 FTP 成功,正在重试处理刚才失败的文件: {}", fileName); logger.info("【FTP地市】重新连接 FTP 成功,正在重试处理刚才失败的文件: {}", fileName);
// 重新尝试处理文件 // 重新尝试处理文件
@ -341,15 +345,21 @@ public class FtpSyncService {
String remoteRecordPath = recordDir + "/" + dateStr + "/" + recordFileName; String remoteRecordPath = recordDir + "/" + dateStr + "/" + recordFileName;
logger.info("【FTP行】准备下载录音: {}", remoteRecordPath); logger.info("【FTP行】准备下载录音: {}", remoteRecordPath);
byte[] recordBytes = FtpUtil.downloadFile(ftp, remoteRecordPath); byte[] recordBytes = FtpUtil.downloadFile(ftp, remoteRecordPath);
logger.info("【FTP行】下载录音: {} ({} 字节)", recordFileName, recordBytes.length); String ossUrl = null;
int fileSize = 0;
// 上传到OSS(获取可访问的URL) if (recordBytes != null && recordBytes.length > 0) {
String ossUrl = fileUploadUtil.uploadWav(ossPath, recordFileName, recordBytes); fileSize = recordBytes.length;
logger.info("【FTP行】上传OSS: {}", ossUrl); 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, 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); callRecordMapper.insert(rec);
existingIds.add(callRecordId); // 防止同批内重复(本次循环内) existingIds.add(callRecordId); // 防止同批内重复(本次循环内)
logger.info("【FTP行】保存成功: id={}, callRecordId={}", rec.getId(), callRecordId); logger.info("【FTP行】保存成功: id={}, callRecordId={}", rec.getId(), callRecordId);
@ -478,19 +488,32 @@ public class FtpSyncService {
} }
/** /**
* 判断是否为当天已经生成完成的时间段TXT文件 * 判断是否为有效且已封口的TXT数据文件支持跨天多天一天多个时间切片文件
* 文件名规则以当天yyyyMMdd开头日期后同时包含起止时间横杠且文件非空 * 兼容格式
* 1. 芜湖等带横杠时间段格式 202608170700-202608170730.txt要求两端时间戳完整排除未写完的占位文件
* 2. 淮南/蚌埠等纯时间戳格式 202608170740.txtyyyyMMddHHmm.txt yyyyMMddHHmmss.txt
*/ */
private boolean isTodayCompletedIntervalTxtFile(FTPFile file) { private boolean isValidDataTxtFile(FTPFile file) {
if (!FtpUtil.isTxtFile(file) || file.getSize() <= 0) { if (!FtpUtil.isTxtFile(file) || file.getSize() <= 0) {
return false; return false;
} }
String fileName = file.getName(); String fileName = file.getName();
String today = DateUtil.formatDate(new Date(), "yyyyMMdd"); // 去掉后缀 .txt
int separatorIndex = fileName.indexOf('-', today.length()); String nameWithoutExt = fileName.substring(0, fileName.length() - ".txt".length());
return fileName.startsWith(today)
&& separatorIndex > today.length() // 基础规则:文件名开头至少是 8 位合法数字日期(yyyyMMdd,如 20260817)
&& separatorIndex < fileName.length() - ".txt".length() - 1; 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}");
}
} }
/** /**

76
src/main/java/com/threecloud/dataserviceyy/util/FtpUtil.java

@ -25,7 +25,7 @@ public class FtpUtil {
private static final int READ_TIMEOUT = 60000; private static final int READ_TIMEOUT = 60000;
/** /**
* 连接FTP服务器一次性连接处理所有操作最后调用 disconnect 关闭 * 连接FTP服务器默认使用被动模式
* *
* @param host FTP服务器IP * @param host FTP服务器IP
* @param port FTP服务器端口 * @param port FTP服务器端口
@ -34,21 +34,50 @@ public class FtpUtil {
* @return FTPClient 连接成功返回客户端失败抛异常 * @return FTPClient 连接成功返回客户端失败抛异常
*/ */
public static FTPClient connect(String host, int port, String username, String password) throws IOException { 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(); FTPClient ftp = new FTPClient();
// 设置UTF-8编码(必须在 connect 前设置)
ftp.setControlEncoding("UTF-8");
ftp.setConnectTimeout(CONNECT_TIMEOUT); ftp.setConnectTimeout(CONNECT_TIMEOUT);
ftp.setDefaultTimeout(CONNECT_TIMEOUT);
ftp.setDataTimeout(READ_TIMEOUT); ftp.setDataTimeout(READ_TIMEOUT);
logger.info("正在尝试连接 FTP: {}:{} (模式: {})", host, port, passiveMode ? "被动(PASV)" : "主动(PORT)");
ftp.connect(host, port); ftp.connect(host, port);
boolean loginOk = ftp.login(username, password); boolean loginOk = ftp.login(username, password);
if (!loginOk) { if (!loginOk) {
ftp.disconnect(); ftp.disconnect();
throw new IOException("FTP登录失败,请检查用户名密码"); throw new IOException("FTP登录失败,请检查用户名密码");
} }
// 设置被动模式(必须,避开防火墙)
ftp.enterLocalPassiveMode();
// 设置二进制传输(mp3/wav必须用二进制) // 设置二进制传输(mp3/wav必须用二进制)
ftp.setFileType(FTP.BINARY_FILE_TYPE); ftp.setFileType(FTP.BINARY_FILE_TYPE);
// 设置UTF-8编码(文件名有中文) // 关闭远程IP验证(增强在多网卡/NAT环境下的数据连接兼容性)
ftp.setControlEncoding("UTF-8"); 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); logger.info("FTP连接成功: {}:{}", host, port);
return ftp; return ftp;
} }
@ -94,12 +123,22 @@ public class FtpUtil {
* @param remotePath 远程文件路径绝对或相对于当前工作目录 * @param remotePath 远程文件路径绝对或相对于当前工作目录
* @return 文件字节内容 * @return 文件字节内容
*/ */
/**
* 下载文件到本地字节数组自动完成FTP pending命令
*
* @param ftp 已连接的FTPClient
* @param remotePath 远程文件路径绝对或相对于当前工作目录
* @return 文件字节内容若文件在FTP上不存在则返回 null
*/
public static byte[] downloadFile(FTPClient ftp, String remotePath) throws IOException { public static byte[] downloadFile(FTPClient ftp, String remotePath) throws IOException {
InputStream in = null; InputStream in = null;
try { try {
in = ftp.retrieveFileStream(remotePath); in = ftp.retrieveFileStream(remotePath);
if (in == null) { 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(); ByteArrayOutputStream out = new ByteArrayOutputStream();
byte[] buffer = new byte[8192]; byte[] buffer = new byte[8192];
@ -108,14 +147,27 @@ public class FtpUtil {
out.write(buffer, 0, len); out.write(buffer, 0, len);
} }
return out.toByteArray(); return out.toByteArray();
} catch (IOException e) {
// 下载异常时尝试终止当前传输命令
try {
ftp.abort();
} catch (Exception ignored) {
}
throw e;
} finally { } finally {
// 必须关闭流并调用 completePendingCommand,否则后续FTP操作会失败 // 关键:只有成功打开了输入流,才需要 close 并调用 completePendingCommand
if (in != null) { if (in != null) {
try { in.close(); } catch (IOException ignored) {} try {
} in.close();
boolean ok = ftp.completePendingCommand(); } catch (IOException ignored) {}
if (!ok) { try {
logger.warn("FTP completePendingCommand 返回 false: {}", remotePath); boolean ok = ftp.completePendingCommand();
if (!ok) {
logger.warn("FTP completePendingCommand 返回 false: {}", remotePath);
}
} catch (Exception e) {
logger.warn("FTP completePendingCommand 异常: {}, msg={}", remotePath, e.getMessage());
}
} }
} }
} }

Loading…
Cancel
Save