From a42b0f8f6f63db324fbacf0fd1ac8ecfb93ea0d8 Mon Sep 17 00:00:00 2001 From: wang Date: Wed, 30 Sep 2026 11:46:48 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BC=98=E5=8C=96ftp?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- config/application-external.yml | 15 +++- .../config/FtpSyncProperties.java | 20 +++++ .../dataserviceyy/service/FtpSyncService.java | 90 +++++++++++++++++-- 3 files changed, 115 insertions(+), 10 deletions(-) diff --git a/config/application-external.yml b/config/application-external.yml index 269b0ec..c5f911f 100644 --- a/config/application-external.yml +++ b/config/application-external.yml @@ -66,6 +66,12 @@ ftp-sync: ftp-record-dir: /rec/ ftp-archive-dir: /voice_record/processed/ phone-area-code: "0554" + # 淮南索引第8列=来源单位名称，第9列=来源单位ID（位置从1开始） + org-mapping-enabled: true + source-org-name-position: 8 + source-org-id-position: 9 + # 未完成人工机构映射时使用的org_code，请填写系统要求的完整数字编码；禁止填340400 + default-org-code: "" # 示例："730364575384027136"，请保留引号 passive-mode: true # 是否被动模式：true=被动模式(PASV，默认)，false=主动模式(PORT) # ===== 蚌埠 ===== @@ -76,9 +82,16 @@ ftp-sync: ftp-username: ftpadmin ftp-password: A3833508q ftp-source-dir: /voice_record/ - ftp-record-dir: /rec/ + # 蚌埠录音实际位于FTP根目录的 /yyyyMMdd/ 下，不是 /rec/yyyyMMdd/ + ftp-record-dir: / ftp-archive-dir: /voice_record/processed/ phone-area-code: "0552" + # 蚌埠索引格式已确认与淮南一致：第8列=来源单位名称，第9列=来源单位ID + org-mapping-enabled: true + source-org-name-position: 8 + source-org-id-position: 9 + # 未完成人工机构映射时使用的org_code，请填写系统要求的完整数字编码；禁止填340300 + default-org-code: "" # 请填写完整数字编码并保留引号 passive-mode: true # 是否被动模式：true=被动模式(PASV，默认)，false=主动模式(PORT) # ==================== 服务端口 ==================== diff --git a/src/main/java/com/threecloud/dataserviceyy/config/FtpSyncProperties.java b/src/main/java/com/threecloud/dataserviceyy/config/FtpSyncProperties.java index f43ec43..4cf3737 100644 --- a/src/main/java/com/threecloud/dataserviceyy/config/FtpSyncProperties.java +++ b/src/main/java/com/threecloud/dataserviceyy/config/FtpSyncProperties.java @@ -101,6 +101,14 @@ public class FtpSyncProperties { * - 留空：保留原始号码 */ private String phoneAreaCode; + /** 是否从TXT解析来源单位并维护FTP单位映射 */ + private boolean orgMappingEnabled = false; + /** 来源单位名称在TXT中的位置（从1开始） */ + private int sourceOrgNamePosition = -1; + /** 来源单位ID在TXT中的位置（从1开始） */ + private int sourceOrgIdPosition = -1; + /** 未完成人工机构映射时写入通话记录的默认机构编码；为空则不写 */ + private String defaultOrgCode; /** * FTP传输模式：true=被动模式(PASV，默认)，false=主动模式(PORT) */ @@ -136,6 +144,18 @@ public class FtpSyncProperties { public String getPhoneAreaCode() { return phoneAreaCode; } public void setPhoneAreaCode(String phoneAreaCode) { this.phoneAreaCode = phoneAreaCode; } + public boolean isOrgMappingEnabled() { return orgMappingEnabled; } + public void setOrgMappingEnabled(boolean orgMappingEnabled) { this.orgMappingEnabled = orgMappingEnabled; } + + public int getSourceOrgNamePosition() { return sourceOrgNamePosition; } + public void setSourceOrgNamePosition(int sourceOrgNamePosition) { this.sourceOrgNamePosition = sourceOrgNamePosition; } + + public int getSourceOrgIdPosition() { return sourceOrgIdPosition; } + public void setSourceOrgIdPosition(int sourceOrgIdPosition) { this.sourceOrgIdPosition = sourceOrgIdPosition; } + + public String getDefaultOrgCode() { return defaultOrgCode; } + public void setDefaultOrgCode(String defaultOrgCode) { this.defaultOrgCode = defaultOrgCode; } + 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 12c7953..95b3a03 100644 --- a/src/main/java/com/threecloud/dataserviceyy/service/FtpSyncService.java +++ b/src/main/java/com/threecloud/dataserviceyy/service/FtpSyncService.java @@ -2,8 +2,10 @@ package com.threecloud.dataserviceyy.service; import com.threecloud.dataserviceyy.config.FtpSyncProperties; import com.threecloud.dataserviceyy.config.FtpSyncProperties.FtpCityConfig; +import com.threecloud.dataserviceyy.entity.FtpOrgMapping; import com.threecloud.dataserviceyy.entity.FtpProcessedFile; import com.threecloud.dataserviceyy.entity.MidVoiceCallRecord; +import com.threecloud.dataserviceyy.mapper.FtpOrgMappingMapper; import com.threecloud.dataserviceyy.mapper.FtpProcessedFileMapper; import com.threecloud.dataserviceyy.mapper.MidVoiceCallRecordMapper; import com.threecloud.dataserviceyy.util.DateUtil; @@ -33,7 +35,7 @@ import java.util.*; * 【部署】每个地市单独部署一份，每个服务只处理一个地市的FTP * * txt文件格式（GBK编码，※分隔）： - * kssj※jssj※hjls※hjzls※zjhm※bjhm※thsc※...※thfx + * kssj※jssj※hjls※hjzls※zjhm※bjhm※thsc※sourceOrgName※sourceOrgId※thfx * * kssj : 开始时间 yyyyMMddHHmmss * jssj : 结束时间 yyyyMMddHHmmss @@ -42,6 +44,8 @@ import java.util.*; * zjhm : 主叫号码 * bjhm : 被叫号码 * thsc : 通话时长（秒） + * sourceOrgName : 来源单位名称（淮南格式第8列，如孔店派出所） + * sourceOrgId : 来源单位ID（淮南格式第9列，不是设备编号） * thfx : 通话方向 1=呼入 0=呼出 * * 录音文件命名：{hjls}_{hjzls}.mp3 @@ -67,6 +71,8 @@ public class FtpSyncService { @Autowired private FtpProcessedFileMapper processedFileMapper; @Autowired + private FtpOrgMappingMapper orgMappingMapper; + @Autowired private FileUploadUtil fileUploadUtil; /** @@ -125,6 +131,7 @@ public class FtpSyncService { // 批量查询数据库已存在的call_record_id（防重） String prefix = city.getCityCode() + "_FTP_"; Set existingIds = loadExistingCallRecordIds(prefix); + Map orgMappingCache = new HashMap<>(); FTPClient ftp = null; try { @@ -180,7 +187,7 @@ public class FtpSyncService { continue; } try { - int txtResult = processTxtFile(ftp, candidate.directory, file, city, existingIds); + int txtResult = processTxtFile(ftp, candidate.directory, file, city, existingIds, orgMappingCache); if (txtResult == 1) { fileSuccess++; processedFiles.add(fileName); @@ -199,7 +206,7 @@ public class FtpSyncService { logger.info("【FTP地市】重新连接 FTP 成功，正在重试处理刚才失败的文件: {}", fileName); // 重新尝试处理文件 - int txtResult = processTxtFile(ftp, candidate.directory, file, city, existingIds); + int txtResult = processTxtFile(ftp, candidate.directory, file, city, existingIds, orgMappingCache); if (txtResult == 1) { fileSuccess++; processedFiles.add(fileName); @@ -230,7 +237,8 @@ public class FtpSyncService { * @return 1=全部成功处理完毕, 2=有录音暂未就绪(下次重试), 0=解析失败 */ private int processTxtFile(FTPClient ftp, String txtDirectory, FTPFile file, FtpCityConfig city, - Set existingIds) throws Exception { + Set existingIds, + Map orgMappingCache) throws Exception { String fileName = file.getName(); String sourcePath = joinPath(txtDirectory, fileName); long fileSize = file.getSize(); @@ -255,7 +263,7 @@ public class FtpSyncService { } lineCount++; try { - int result = processLine(line, city, existingIds, ftp); + int result = processLine(line, city, existingIds, orgMappingCache, ftp); switch (result) { case 1: successCount++; @@ -314,6 +322,7 @@ public class FtpSyncService { * @return 1=成功入库, 2=已存在跳过, 3=录音未就绪跳过, 0=无效行 */ private int processLine(String line, FtpCityConfig city, Set existingIds, + Map orgMappingCache, FTPClient ftp) throws Exception { // 按分隔符拆分字段（*分隔，需转义正则特殊字符） String separator = java.util.regex.Pattern.quote(properties.getFieldSeparator()); @@ -331,8 +340,12 @@ public class FtpSyncService { String zjhm = fields[4].trim(); // zjhm : 主叫号码（包含区号，如：055312345678） String bjhm = fields[5].trim(); // bjhm : 被叫号码 String thscStr = fields[6].trim();// thsc : 通话时长（秒） + // fields[7]、fields[8] 由地市配置决定；淮南分别为来源单位名称、来源单位ID。 String thfx = fields[9].trim(); // thfx : 通话方向（1=呼入，0=呼出 - 老系统特殊编码） + // 淮南等配置启用的地市，从索引列中提取厂家单位信息并维护映射表。 + FtpOrgMapping orgMapping = resolveOrgMapping(city, fields, orgMappingCache); + // 必填字段校验 if (kssj.isEmpty() || jssj.isEmpty() || hjls.isEmpty() || hjzls.isEmpty()) { return 0; @@ -390,7 +403,7 @@ public class FtpSyncService { logger.info("【FTP行】上传OSS成功: {}", ossUrl); // 构建实体对象并入库 - MidVoiceCallRecord rec = buildCallRecord(city, callRecordId, callStartTime, callEndTime, + MidVoiceCallRecord rec = buildCallRecord(city, orgMapping, callRecordId, callStartTime, callEndTime, thsc, isOutgoing, zjhm, bjhm, recordFileName, fileSize, ossUrl); callRecordMapper.insert(rec); existingIds.add(callRecordId); // 防止同批内重复 @@ -399,6 +412,58 @@ public class FtpSyncService { return 1; } + /** + * 解析并登记FTP索引中的来源单位。字段位置使用从1开始的配置，避免不同地市格式差异写死在代码中。 + * 人工维护的 mapped_org_code / mapped_org_name 不会被同步任务覆盖。 + */ + private FtpOrgMapping resolveOrgMapping(FtpCityConfig city, String[] fields, + Map cache) { + if (!city.isOrgMappingEnabled()) { + return null; + } + + String sourceOrgName = getFieldByPosition(fields, city.getSourceOrgNamePosition()); + String sourceOrgId = getFieldByPosition(fields, city.getSourceOrgIdPosition()); + if (!StringUtils.hasText(sourceOrgId)) { + logger.warn("【FTP单位】{} 已启用单位映射，但索引第{}列单位ID为空，保留录音但不生成映射", + city.getCityName(), city.getSourceOrgIdPosition()); + return null; + } + + String cacheKey = city.getCityCode() + "|" + sourceOrgId; + FtpOrgMapping cached = cache.get(cacheKey); + if (cached != null) { + return cached; + } + + FtpOrgMapping source = new FtpOrgMapping(); + source.setCityCode(city.getCityCode()); + source.setCityName(city.getCityName()); + source.setSourceOrgId(sourceOrgId); + source.setSourceOrgName(sourceOrgName); + source.setFtpDeviceNo("FTP_" + city.getCityCode() + "_" + sourceOrgId); + orgMappingMapper.upsert(source); + + FtpOrgMapping saved = orgMappingMapper.selectBySourceOrg(city.getCityCode(), sourceOrgId); + if (saved == null) { + throw new IllegalStateException("FTP单位映射保存后未能查询到: " + cacheKey); + } + cache.put(cacheKey, saved); + logger.info("【FTP单位】识别来源单位: city={}, sourceOrgId={}, sourceOrgName={}, ftpDeviceNo={}, mappedOrgCode={}", + city.getCityCode(), sourceOrgId, sourceOrgName, saved.getFtpDeviceNo(), saved.getMappedOrgCode()); + return saved; + } + + /** 按从1开始的位置读取TXT字段；配置越界时返回空串并记录告警。 */ + private String getFieldByPosition(String[] fields, int position) { + int index = position - 1; + if (position <= 0 || index >= fields.length) { + logger.warn("【FTP单位】索引字段位置配置无效: position={}, fieldCount={}", position, fields.length); + return ""; + } + return fields[index] == null ? "" : fields[index].trim(); + } + /** * 构建通话记录实体 * @@ -414,17 +479,24 @@ public class FtpSyncService { * @param fileSize 录音文件大小（字节） * @param ossUrl OSS访问URL */ - private MidVoiceCallRecord buildCallRecord(FtpCityConfig city, String callRecordId, + private MidVoiceCallRecord buildCallRecord(FtpCityConfig city, FtpOrgMapping orgMapping, + String callRecordId, Date callStartTime, Date callEndTime, int duration, boolean isOutgoing, String zjhm, String bjhm, String fileName, int fileSize, String ossUrl) { MidVoiceCallRecord rec = new MidVoiceCallRecord(); rec.setCallRecordId(callRecordId); // 唯一标识 - rec.setDeviceNo("FTP_" + city.getCityCode()); // 设备编码（FTP_地市编码） + rec.setDeviceNo(orgMapping != null + ? orgMapping.getFtpDeviceNo() + : "FTP_" + city.getCityCode()); // 逻辑设备编码 rec.setCityCode(city.getCityCode()); // 地市编码（用于数据库分区） rec.setCityName(city.getCityName()); // 地市名称 - rec.setOrgCode(city.getCityCode()); // 单位代码（用地市编码代替） + // 人工映射的标准机构编码优先；未映射时使用地市配置值，禁止用6位city_code冒充机构编码。 + String mappedOrgCode = orgMapping != null ? orgMapping.getMappedOrgCode() : null; + rec.setOrgCode(StringUtils.hasText(mappedOrgCode) + ? mappedOrgCode + : (StringUtils.hasText(city.getDefaultOrgCode()) ? city.getDefaultOrgCode().trim() : null)); rec.setRecordingFileName(fileName); // 录音文件名 rec.setCallStartTime(callStartTime); // 通话开始时间 rec.setCallEndTime(callEndTime); // 通话结束时间