diff --git a/src/main/java/com/threecloud/dataserviceyy/service/FtpSyncService.java b/src/main/java/com/threecloud/dataserviceyy/service/FtpSyncService.java index 2ef37e8..dec03f2 100644 --- a/src/main/java/com/threecloud/dataserviceyy/service/FtpSyncService.java +++ b/src/main/java/com/threecloud/dataserviceyy/service/FtpSyncService.java @@ -149,6 +149,13 @@ public class FtpSyncService { continue; } String fileName = file.getName(); + // 历史目录中只处理当天已经封口的时间段文件: + // 例如 20260722123-20260722163.txt;空文件或无横杠的占位文件暂不处理。 + if (!isTodayCompletedIntervalTxtFile(file)) { + logger.debug("【FTP文件】非当天有效时间段文件,跳过: {} ({} 字节)", + fileName, file.getSize()); + continue; + } // 检查是否已处理过 if (processedFiles.contains(fileName)) { logger.debug("【FTP文件】已处理过,跳过: {}", fileName); @@ -162,9 +169,25 @@ public class FtpSyncService { fileFail++; } } catch (Exception e) { - fileFail++; - logger.error("【FTP地市】处理文件严重失败: {}, 原因={}。将终止本轮同步以防止连接断开导致后续文件持续报错。", fileName, e.getMessage()); - break; + logger.warn("【FTP地市】处理文件失败: {}, 原因={}。正在尝试重新连接 FTP 并重试...", fileName, e.getMessage()); + try { + FtpUtil.disconnect(ftp); // 关闭老连接 + ftp = FtpUtil.connect(city.getFtpHost(), city.getFtpPort(), + city.getFtpUsername(), city.getFtpPassword()); + logger.info("【FTP地市】重新连接 FTP 成功,正在重试处理刚才失败的文件: {}", fileName); + + // 重新尝试处理文件 + if (processTxtFile(ftp, file, city, existingIds)) { + fileSuccess++; + } else { + fileFail++; + } + } catch (Exception retryEx) { + fileFail++; + logger.error("【FTP地市】重新连接或重试处理文件仍然失败: {}, 原因={}", fileName, retryEx.getMessage()); + // 重连失败,判定为彻底断联,终止本轮同步 + break; + } } } logger.info("【FTP地市】{} 同步完成: 成功{}个文件, 跳过{}个(已处理), 失败{}个文件", @@ -209,6 +232,14 @@ public class FtpSyncService { skipCount++; } } catch (Exception e) { + // 如果是网络连接重置或其它 IO 严重异常,不在此吞掉,直接向上抛出以触发外部重连重试 + if (e instanceof java.io.IOException || (e.getMessage() != null && ( + e.getMessage().contains("Connection") || + e.getMessage().contains("Socket") || + e.getMessage().contains("reset") || + e.getMessage().contains("broken pipe")))) { + throw e; + } failCount++; errorMsg = e.getMessage(); logger.error("【FTP文件】解析单行失败: {}, 原因={}", line, e.getMessage()); @@ -446,6 +477,22 @@ public class FtpSyncService { } } + /** + * 判断是否为当天已经生成完成的时间段TXT文件。 + * 文件名规则:以当天yyyyMMdd开头,日期后同时包含起止时间横杠,且文件非空。 + */ + private boolean isTodayCompletedIntervalTxtFile(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; + } + /** * 拼接FTP路径,自动处理末尾斜杠,避免双斜杠 * diff --git a/src/main/java/com/threecloud/dataserviceyy/util/FileUploadUtil.java b/src/main/java/com/threecloud/dataserviceyy/util/FileUploadUtil.java index 809f78e..694da68 100644 --- a/src/main/java/com/threecloud/dataserviceyy/util/FileUploadUtil.java +++ b/src/main/java/com/threecloud/dataserviceyy/util/FileUploadUtil.java @@ -60,8 +60,8 @@ public class FileUploadUtil { if (fileName != null && !fileName.isEmpty() && path.endsWith(fileName)) { path = path.substring(0, path.length() - fileName.length()); } - if (path.endsWith("/")) { - path = path.substring(0, path.length() - 1); + if (!path.endsWith("/")) { + path = path + "/"; } logger.debug("上传文件到OSS, path={}, 大小={} bytes", path, fileData.length); @@ -140,41 +140,42 @@ public class FileUploadUtil { return ""; } + String rawUrl = null; if (content instanceof Map) { Map map = (Map) content; Object url = map.get("fileUrl"); if (url != null) { - return url.toString(); - } - url = map.get("url"); - if (url != null) { - return url.toString(); + rawUrl = url.toString(); + } else { + url = map.get("url"); + if (url != null) { + rawUrl = url.toString(); + } } } - String contentStr = content.toString().trim(); - if (contentStr.startsWith("{") || contentStr.startsWith("[")) { - try { - Object parsed = JSON.parse(contentStr); - if (parsed instanceof Map) { - return parseFileUrl(parsed); + if (rawUrl == null) { + String contentStr = content.toString().trim(); + if (contentStr.startsWith("{") || contentStr.startsWith("[")) { + try { + Object parsed = JSON.parse(contentStr); + if (parsed instanceof Map) { + return parseFileUrl(parsed); + } + } catch (Exception e) { + logger.debug("解析JSON失败: {}", e.getMessage()); } - } catch (Exception e) { - logger.debug("解析JSON失败: {}", e.getMessage()); } + rawUrl = contentStr; } - if (contentStr.startsWith("http://") || contentStr.startsWith("https://")) { - return contentStr; - } - - // 相对路径,拼接 base-url - if (contentStr.startsWith("/")) { + // 相对路径,统一拼接 base-url + if (rawUrl.startsWith("/")) { String base = ossBaseUrl.endsWith("/") ? ossBaseUrl.substring(0, ossBaseUrl.length() - 1) : ossBaseUrl; - return base + contentStr; + return base + rawUrl; } - return contentStr; + return rawUrl; } /**