|
|
|
@ -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<String> existingIds = loadExistingCallRecordIds(prefix); |
|
|
|
Map<String, FtpOrgMapping> 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<String> existingIds) throws Exception { |
|
|
|
Set<String> existingIds, |
|
|
|
Map<String, FtpOrgMapping> 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<String> existingIds, |
|
|
|
Map<String, FtpOrgMapping> 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<String, FtpOrgMapping> 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); // 通话结束时间
|
|
|
|
|
wang
3 weeks ago