✨ Company Compliance 수집 JOB 추가
This commit is contained in:
hyojin kim
부모
f4421fa455
커밋
e63607a69d
@ -0,0 +1,228 @@
|
||||
package com.snp.batch.jobs.compliance.batch.config;
|
||||
|
||||
import com.snp.batch.common.batch.config.BaseMultiStepJobConfig;
|
||||
import com.snp.batch.jobs.compliance.batch.dto.CompanyComplianceDto;
|
||||
import com.snp.batch.jobs.compliance.batch.dto.ComplianceDto;
|
||||
import com.snp.batch.jobs.compliance.batch.entity.CompanyComplianceEntity;
|
||||
import com.snp.batch.jobs.compliance.batch.entity.ComplianceEntity;
|
||||
import com.snp.batch.jobs.compliance.batch.processor.CompanyComplianceDataProcessor;
|
||||
import com.snp.batch.jobs.compliance.batch.reader.CompanyComplianceDataRangeReader;
|
||||
import com.snp.batch.jobs.compliance.batch.reader.ComplianceDataRangeReader;
|
||||
import com.snp.batch.jobs.compliance.batch.writer.CompanyComplianceDataWriter;
|
||||
import com.snp.batch.service.BatchApiLogService;
|
||||
import com.snp.batch.service.BatchDateService;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.springframework.batch.core.Job;
|
||||
import org.springframework.batch.core.Step;
|
||||
import org.springframework.batch.core.configuration.annotation.StepScope;
|
||||
import org.springframework.batch.core.job.builder.JobBuilder;
|
||||
import org.springframework.batch.core.repository.JobRepository;
|
||||
import org.springframework.batch.core.step.builder.StepBuilder;
|
||||
import org.springframework.batch.core.step.tasklet.Tasklet;
|
||||
import org.springframework.batch.item.ItemProcessor;
|
||||
import org.springframework.batch.item.ItemReader;
|
||||
import org.springframework.batch.item.ItemWriter;
|
||||
import org.springframework.batch.repeat.RepeatStatus;
|
||||
import org.springframework.beans.factory.annotation.Qualifier;
|
||||
import org.springframework.beans.factory.annotation.Value;
|
||||
import org.springframework.context.annotation.Bean;
|
||||
import org.springframework.context.annotation.Configuration;
|
||||
import org.springframework.jdbc.core.JdbcTemplate;
|
||||
import org.springframework.transaction.PlatformTransactionManager;
|
||||
import org.springframework.web.reactive.function.client.WebClient;
|
||||
|
||||
import java.time.OffsetDateTime;
|
||||
import java.time.ZoneId;
|
||||
import java.time.format.DateTimeFormatter;
|
||||
import java.util.Map;
|
||||
|
||||
@Slf4j
|
||||
@Configuration
|
||||
public class CompanyComplianceImportRangeJobConfig extends BaseMultiStepJobConfig<CompanyComplianceDto, CompanyComplianceEntity> {
|
||||
|
||||
private final JdbcTemplate jdbcTemplate;
|
||||
private final WebClient maritimeServiceApiWebClient;
|
||||
private final CompanyComplianceDataRangeReader companyComplianceDataRangeReader;
|
||||
private final CompanyComplianceDataProcessor companyComplianceDataProcessor;
|
||||
private final CompanyComplianceDataWriter companyComplianceDataWriter;
|
||||
private final BatchDateService batchDateService;
|
||||
private final BatchApiLogService batchApiLogService;
|
||||
|
||||
@Value("${app.batch.webservice-api.url}")
|
||||
private String maritimeServiceApiUrl;
|
||||
protected String getApiKey() {return "COMPANY_COMPLIANCE_IMPORT_API";}
|
||||
protected String getBatchUpdateSql() {
|
||||
return String.format("UPDATE SNP_DATA.BATCH_LAST_EXECUTION SET LAST_SUCCESS_DATE = NOW(), UPDATED_AT = NOW() WHERE API_KEY = '%s'", getApiKey());}
|
||||
|
||||
@Override
|
||||
protected int getChunkSize() {
|
||||
return 5000;
|
||||
}
|
||||
public CompanyComplianceImportRangeJobConfig(
|
||||
JobRepository jobRepository,
|
||||
PlatformTransactionManager transactionManager,
|
||||
CompanyComplianceDataRangeReader companyComplianceDataRangeReader,
|
||||
CompanyComplianceDataProcessor companyComplianceDataProcessor,
|
||||
CompanyComplianceDataWriter companyComplianceDataWriter,
|
||||
JdbcTemplate jdbcTemplate,
|
||||
@Qualifier("maritimeServiceApiWebClient")WebClient maritimeServiceApiWebClient,
|
||||
BatchDateService batchDateService,
|
||||
BatchApiLogService batchApiLogService) {
|
||||
super(jobRepository, transactionManager);
|
||||
this.jdbcTemplate = jdbcTemplate;
|
||||
this.companyComplianceDataRangeReader = companyComplianceDataRangeReader;
|
||||
this.maritimeServiceApiWebClient = maritimeServiceApiWebClient;
|
||||
this.companyComplianceDataProcessor = companyComplianceDataProcessor;
|
||||
this.companyComplianceDataWriter = companyComplianceDataWriter;
|
||||
this.batchDateService = batchDateService;
|
||||
this.batchApiLogService = batchApiLogService;
|
||||
}
|
||||
|
||||
@Override
|
||||
protected String getJobName() {
|
||||
return "CompanyComplianceImportRangeJob";
|
||||
}
|
||||
@Override
|
||||
protected String getStepName() {
|
||||
return "CompanyComplianceImportRangeStep";
|
||||
}
|
||||
|
||||
@Override
|
||||
protected Job createJobFlow(JobBuilder jobBuilder) {
|
||||
return jobBuilder
|
||||
.start(companyComplianceImportRangeStep()) // 1단계 실행
|
||||
.next(currentCompanyComplianceUpdateStep()) // 2단계 실행 (1단계 실패 시 실행 안 됨)
|
||||
.next(companyComplianceHistoryValueChangeManageStep()) // 3단계 실행 (2단계 실패 시 실행 안 됨)
|
||||
.next(companyComplianceLastExecutionUpdateStep()) // 4단계: 모두 완료 시, BATCH_LAST_EXECUTION 마지막 성공일자 업데이트
|
||||
.build();
|
||||
}
|
||||
|
||||
@Override
|
||||
protected ItemReader<CompanyComplianceDto> createReader() {
|
||||
return companyComplianceDataRangeReader;
|
||||
}
|
||||
|
||||
@Bean
|
||||
@StepScope
|
||||
public CompanyComplianceDataRangeReader companyComplianceDataRangeReader(
|
||||
@Value("#{stepExecution.jobExecution.id}") Long jobExecutionId, // SpEL로 ID 추출
|
||||
@Value("#{stepExecution.id}") Long stepExecutionId
|
||||
) {
|
||||
CompanyComplianceDataRangeReader reader = new CompanyComplianceDataRangeReader(maritimeServiceApiWebClient, batchDateService, batchApiLogService, maritimeServiceApiUrl);
|
||||
reader.setExecutionIds(jobExecutionId, stepExecutionId); // ID 세팅
|
||||
return reader;
|
||||
}
|
||||
@Override
|
||||
protected ItemProcessor<CompanyComplianceDto, CompanyComplianceEntity> createProcessor() {
|
||||
return companyComplianceDataProcessor;
|
||||
}
|
||||
|
||||
@Override
|
||||
protected ItemWriter<CompanyComplianceEntity> createWriter() {
|
||||
return companyComplianceDataWriter;
|
||||
}
|
||||
|
||||
@Bean(name = "CompanyComplianceImportRangeJob")
|
||||
public Job companyComplianceImportRangeJob() {
|
||||
return job();
|
||||
}
|
||||
|
||||
@Bean(name = "CompanyComplianceImportRangeStep")
|
||||
public Step companyComplianceImportRangeStep() {
|
||||
return step();
|
||||
}
|
||||
/**
|
||||
* 2단계: Current Company Compliance 업데이트
|
||||
*/
|
||||
@Bean
|
||||
public Tasklet currentCompanyComplianceUpsertTasklet() {
|
||||
return (contribution, chunkContext) -> {
|
||||
log.info(">>>>> Company Compliance Upsert 프로시저 호출 시작");
|
||||
|
||||
// PostgreSQL 기준 프로시저 호출 (CALL)
|
||||
jdbcTemplate.execute("CALL new_snp.upsert_current_company_compliance()");
|
||||
|
||||
log.info(">>>>> Company Compliance Upsert 프로시저 호출 완료");
|
||||
return RepeatStatus.FINISHED;
|
||||
};
|
||||
}
|
||||
@Bean(name = "CurrentCompanyComplianceUpdateStep")
|
||||
public Step currentCompanyComplianceUpdateStep() {
|
||||
return new StepBuilder("CurrentCompanyComplianceUpdateStep", jobRepository)
|
||||
.tasklet(currentCompanyComplianceUpsertTasklet(), transactionManager)
|
||||
.build();
|
||||
}
|
||||
|
||||
/**
|
||||
* 3단계: Compliance History Value Change 관리
|
||||
*/
|
||||
@Bean
|
||||
public Tasklet companyComplianceHistoryValueChangeManageTasklet() {
|
||||
return (contribution, chunkContext) -> {
|
||||
log.info(">>>>> Company Compliance History Value Change Manage 프로시저 호출 시작");
|
||||
|
||||
// 1. 입력 포맷(UTC 'Z' 포함) 및 프로시저용 타겟 포맷 정의
|
||||
DateTimeFormatter inputFormatter = DateTimeFormatter.ofPattern("yyyy-MM-dd'T'HH:mm:ss.SSX");
|
||||
DateTimeFormatter targetFormatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss.SSS");
|
||||
|
||||
Map<String, String> params = batchDateService.getDateRangeWithTimezoneParams(getApiKey());
|
||||
|
||||
String rawFromDate = params.get("fromDate");
|
||||
String rawToDate = params.get("toDate");
|
||||
|
||||
// 2. UTC 문자열 -> OffsetDateTime -> Asia/Seoul 변환 -> LocalDateTime 추출
|
||||
String startDt = convertToKstString(rawFromDate, inputFormatter, targetFormatter);
|
||||
String endDt = convertToKstString(rawToDate, inputFormatter, targetFormatter);
|
||||
|
||||
log.info("Company Compliance History Value Change Manage 프로시저 변수 (KST 변환): 시작일: {}, 종료일: {}", startDt, endDt);
|
||||
|
||||
// 3. 프로시저 호출 (안전한 파라미터 바인딩 권장)
|
||||
jdbcTemplate.update("CALL new_snp.company_compliance_history_value_change_manage(CAST(? AS TIMESTAMP), CAST(? AS TIMESTAMP))", startDt, endDt);
|
||||
|
||||
log.info(">>>>> Company Compliance History Value Change Manage 프로시저 호출 완료");
|
||||
return RepeatStatus.FINISHED;
|
||||
};
|
||||
}
|
||||
/**
|
||||
* UTC 문자열을 한국 시간(KST) 문자열로 변환하는 헬퍼 메소드
|
||||
*/
|
||||
private String convertToKstString(String rawDate, DateTimeFormatter input, DateTimeFormatter target) {
|
||||
if (rawDate == null) return null;
|
||||
|
||||
// 1. 문자열을 OffsetDateTime으로 파싱 (Z를 인식하여 UTC 시간으로 인지함)
|
||||
return OffsetDateTime.parse(rawDate, input)
|
||||
// 2. 시간대를 서울(+09:00)로 변경 (값이 9시간 더해짐)
|
||||
.atZoneSameInstant(ZoneId.of("Asia/Seoul"))
|
||||
// 3. 프로시저 형식에 맞게 포맷팅
|
||||
.format(target);
|
||||
}
|
||||
@Bean(name = "CompanyComplianceHistoryValueChangeManageStep")
|
||||
public Step companyComplianceHistoryValueChangeManageStep() {
|
||||
return new StepBuilder("CompanyComplianceHistoryValueChangeManageStep", jobRepository)
|
||||
.tasklet(companyComplianceHistoryValueChangeManageTasklet(), transactionManager)
|
||||
.build();
|
||||
}
|
||||
|
||||
/**
|
||||
* 4단계: 모든 스텝 성공 시 배치 실행 로그(날짜) 업데이트
|
||||
*/
|
||||
@Bean
|
||||
public Tasklet companyComplianceLastExecutionUpdateTasklet() {
|
||||
return (contribution, chunkContext) -> {
|
||||
log.info(">>>>> 모든 스텝 성공: BATCH_LAST_EXECUTION 업데이트 시작");
|
||||
|
||||
jdbcTemplate.execute(getBatchUpdateSql());
|
||||
|
||||
log.info(">>>>> BATCH_LAST_EXECUTION 업데이트 완료");
|
||||
return RepeatStatus.FINISHED;
|
||||
};
|
||||
}
|
||||
@Bean(name = "CompanyComplianceLastExecutionUpdateStep")
|
||||
public Step companyComplianceLastExecutionUpdateStep() {
|
||||
return new StepBuilder("CompanyComplianceLastExecutionUpdateStep", jobRepository)
|
||||
.tasklet(companyComplianceLastExecutionUpdateTasklet(), transactionManager)
|
||||
.build();
|
||||
}
|
||||
|
||||
|
||||
}
|
||||
@ -0,0 +1,61 @@
|
||||
package com.snp.batch.jobs.compliance.batch.dto;
|
||||
|
||||
import com.fasterxml.jackson.annotation.JsonProperty;
|
||||
import lombok.AllArgsConstructor;
|
||||
import lombok.Builder;
|
||||
import lombok.Data;
|
||||
import lombok.NoArgsConstructor;
|
||||
|
||||
@Data
|
||||
@Builder
|
||||
@NoArgsConstructor
|
||||
@AllArgsConstructor
|
||||
public class CompanyComplianceDto {
|
||||
@JsonProperty("owcode")
|
||||
private String owcode;
|
||||
|
||||
@JsonProperty("lastUpdated")
|
||||
private String lastUpdated;
|
||||
|
||||
@JsonProperty("companyOverallComplianceStatus")
|
||||
private Integer companyOverallComplianceStatus;
|
||||
|
||||
@JsonProperty("companyOnAustralianSanctionList")
|
||||
private Integer companyOnAustralianSanctionList;
|
||||
|
||||
@JsonProperty("companyOnBESSanctionList")
|
||||
private Integer companyOnBESSanctionList;
|
||||
|
||||
@JsonProperty("companyOnCanadianSanctionList")
|
||||
private Integer companyOnCanadianSanctionList;
|
||||
|
||||
@JsonProperty("companyInOFACSanctionedCountry")
|
||||
private Integer companyInOFACSanctionedCountry;
|
||||
|
||||
@JsonProperty("companyInFATFJurisdiction")
|
||||
private Integer companyInFATFJurisdiction;
|
||||
|
||||
@JsonProperty("companyOnEUSanctionList")
|
||||
private Integer companyOnEUSanctionList;
|
||||
|
||||
@JsonProperty("companyOnOFACSanctionList")
|
||||
private Integer companyOnOFACSanctionList;
|
||||
|
||||
@JsonProperty("companyOnOFACNONSDNSanctionList")
|
||||
private Integer companyOnOFACNONSDNSanctionList;
|
||||
|
||||
@JsonProperty("companyOnOFACSSISanctionList")
|
||||
private Integer companyOnOFACSSISanctionList;
|
||||
|
||||
@JsonProperty("parentCompanyNonCompliance")
|
||||
private Integer parentCompanyNonCompliance;
|
||||
|
||||
@JsonProperty("companyOnSwissSanctionList")
|
||||
private Integer companyOnSwissSanctionList;
|
||||
|
||||
@JsonProperty("companyOnUAESanctionList")
|
||||
private Integer companyOnUAESanctionList;
|
||||
|
||||
@JsonProperty("companyOnUNSanctionList")
|
||||
private Integer companyOnUNSanctionList;
|
||||
}
|
||||
@ -1,17 +0,0 @@
|
||||
package com.snp.batch.jobs.compliance.batch.dto;
|
||||
|
||||
import lombok.AllArgsConstructor;
|
||||
import lombok.Builder;
|
||||
import lombok.Data;
|
||||
import lombok.NoArgsConstructor;
|
||||
|
||||
import java.util.List;
|
||||
|
||||
@Data
|
||||
@Builder
|
||||
@NoArgsConstructor
|
||||
@AllArgsConstructor
|
||||
public class ComplianceResponse {
|
||||
|
||||
private List<ComplianceDto> complianceDtoList;
|
||||
}
|
||||
@ -0,0 +1,47 @@
|
||||
package com.snp.batch.jobs.compliance.batch.entity;
|
||||
|
||||
import com.snp.batch.common.batch.entity.BaseEntity;
|
||||
import lombok.AllArgsConstructor;
|
||||
import lombok.Data;
|
||||
import lombok.EqualsAndHashCode;
|
||||
import lombok.NoArgsConstructor;
|
||||
import lombok.experimental.SuperBuilder;
|
||||
|
||||
@Data
|
||||
@SuperBuilder
|
||||
@NoArgsConstructor
|
||||
@AllArgsConstructor
|
||||
@EqualsAndHashCode(callSuper = true)
|
||||
public class CompanyComplianceEntity extends BaseEntity {
|
||||
private String owcode;
|
||||
|
||||
private String lastUpdated;
|
||||
|
||||
private Integer companyOverallComplianceStatus;
|
||||
|
||||
private Integer companyOnAustralianSanctionList;
|
||||
|
||||
private Integer companyOnBESSanctionList;
|
||||
|
||||
private Integer companyOnCanadianSanctionList;
|
||||
|
||||
private Integer companyInOFACSanctionedCountry;
|
||||
|
||||
private Integer companyInFATFJurisdiction;
|
||||
|
||||
private Integer companyOnEUSanctionList;
|
||||
|
||||
private Integer companyOnOFACSanctionList;
|
||||
|
||||
private Integer companyOnOFACNONSDNSanctionList;
|
||||
|
||||
private Integer companyOnOFACSSISanctionList;
|
||||
|
||||
private Integer parentCompanyNonCompliance;
|
||||
|
||||
private Integer companyOnSwissSanctionList;
|
||||
|
||||
private Integer companyOnUAESanctionList;
|
||||
|
||||
private Integer companyOnUNSanctionList;
|
||||
}
|
||||
@ -1,6 +1,5 @@
|
||||
package com.snp.batch.jobs.compliance.batch.entity;
|
||||
|
||||
import com.fasterxml.jackson.annotation.JsonProperty;
|
||||
import com.snp.batch.common.batch.entity.BaseEntity;
|
||||
import lombok.AllArgsConstructor;
|
||||
import lombok.Data;
|
||||
|
||||
@ -0,0 +1,36 @@
|
||||
package com.snp.batch.jobs.compliance.batch.processor;
|
||||
|
||||
import com.snp.batch.common.batch.processor.BaseProcessor;
|
||||
import com.snp.batch.jobs.compliance.batch.dto.CompanyComplianceDto;
|
||||
import com.snp.batch.jobs.compliance.batch.entity.CompanyComplianceEntity;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
@Slf4j
|
||||
@Component
|
||||
public class CompanyComplianceDataProcessor extends BaseProcessor<CompanyComplianceDto, CompanyComplianceEntity> {
|
||||
|
||||
@Override
|
||||
protected CompanyComplianceEntity processItem(CompanyComplianceDto dto) throws Exception {
|
||||
|
||||
CompanyComplianceEntity entity = CompanyComplianceEntity.builder()
|
||||
.owcode(dto.getOwcode())
|
||||
.lastUpdated(dto.getLastUpdated())
|
||||
.companyOverallComplianceStatus(dto.getCompanyOverallComplianceStatus())
|
||||
.companyOnAustralianSanctionList(dto.getCompanyOnAustralianSanctionList())
|
||||
.companyOnBESSanctionList(dto.getCompanyOnBESSanctionList())
|
||||
.companyOnCanadianSanctionList(dto.getCompanyOnCanadianSanctionList())
|
||||
.companyInOFACSanctionedCountry(dto.getCompanyInOFACSanctionedCountry())
|
||||
.companyInFATFJurisdiction(dto.getCompanyInFATFJurisdiction())
|
||||
.companyOnEUSanctionList(dto.getCompanyOnEUSanctionList())
|
||||
.companyOnOFACSanctionList(dto.getCompanyOnOFACSanctionList())
|
||||
.companyOnOFACNONSDNSanctionList(dto.getCompanyOnOFACNONSDNSanctionList())
|
||||
.companyOnOFACSSISanctionList(dto.getCompanyOnOFACSSISanctionList())
|
||||
.parentCompanyNonCompliance(dto.getParentCompanyNonCompliance())
|
||||
.companyOnSwissSanctionList(dto.getCompanyOnSwissSanctionList())
|
||||
.companyOnUAESanctionList(dto.getCompanyOnUAESanctionList())
|
||||
.companyOnUNSanctionList(dto.getCompanyOnUNSanctionList())
|
||||
.build();
|
||||
return entity;
|
||||
}
|
||||
}
|
||||
@ -11,7 +11,6 @@ import org.springframework.stereotype.Component;
|
||||
public class ComplianceDataProcessor extends BaseProcessor<ComplianceDto, ComplianceEntity> {
|
||||
@Override
|
||||
protected ComplianceEntity processItem(ComplianceDto dto) throws Exception {
|
||||
log.debug("AIS 최신 항적 데이터 처리 시작: imoNumber={}", dto.getLrimoShipNo());
|
||||
|
||||
ComplianceEntity entity = ComplianceEntity.builder()
|
||||
// 1. Primary Keys
|
||||
@ -53,8 +52,6 @@ public class ComplianceDataProcessor extends BaseProcessor<ComplianceDto, Compli
|
||||
.shipUNSanctionList(dto.getShipUNSanctionList())
|
||||
.build();
|
||||
|
||||
log.debug("AIS 최신 항적 데이터 처리 완료: imoNumber={}", dto.getLrimoShipNo());
|
||||
|
||||
return entity;
|
||||
}
|
||||
}
|
||||
|
||||
@ -0,0 +1,108 @@
|
||||
package com.snp.batch.jobs.compliance.batch.reader;
|
||||
|
||||
import com.snp.batch.common.batch.reader.BaseApiReader;
|
||||
import com.snp.batch.jobs.compliance.batch.dto.CompanyComplianceDto;
|
||||
import com.snp.batch.service.BatchApiLogService;
|
||||
import com.snp.batch.service.BatchDateService;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.springframework.core.ParameterizedTypeReference;
|
||||
import org.springframework.web.reactive.function.client.WebClient;
|
||||
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
|
||||
@Slf4j
|
||||
public class CompanyComplianceDataRangeReader extends BaseApiReader<CompanyComplianceDto> {
|
||||
private final BatchDateService batchDateService; // ✨ BatchDateService 필드 추가
|
||||
private final BatchApiLogService batchApiLogService;
|
||||
String maritimeServiceApiUrl;
|
||||
private List<CompanyComplianceDto> allData;
|
||||
private int currentBatchIndex = 0;
|
||||
private final int batchSize = 5000;
|
||||
|
||||
public CompanyComplianceDataRangeReader(WebClient webClient, BatchDateService batchDateService, BatchApiLogService batchApiLogService, String maritimeServiceApiUrl) {
|
||||
super(webClient);
|
||||
this.batchDateService = batchDateService;
|
||||
this.batchApiLogService = batchApiLogService;
|
||||
this.maritimeServiceApiUrl = maritimeServiceApiUrl;
|
||||
enableChunkMode();
|
||||
}
|
||||
@Override
|
||||
protected String getReaderName() {
|
||||
return "CompanyComplianceDataRangeReader";
|
||||
}
|
||||
@Override
|
||||
protected String getApiPath() {
|
||||
return "/RiskAndCompliance/UpdatedCompanyComplianceList";
|
||||
}
|
||||
protected String getApiKey() {
|
||||
return "COMPANY_COMPLIANCE_IMPORT_API";
|
||||
}
|
||||
@Override
|
||||
protected void resetCustomState() {
|
||||
this.currentBatchIndex = 0;
|
||||
this.allData = null;
|
||||
}
|
||||
|
||||
@Override
|
||||
protected List<CompanyComplianceDto> fetchNextBatch() throws Exception{
|
||||
// 모든 배치 처리 완료 확인
|
||||
if (allData == null) {
|
||||
allData = callApiWithBatch();
|
||||
|
||||
if (allData == null || allData.isEmpty()) {
|
||||
log.warn("[{}] 조회된 데이터 없음 → 종료", getReaderName());
|
||||
return null;
|
||||
}
|
||||
|
||||
log.info("[{}] 총 {}건 데이터 조회됨. batchSize = {}", getReaderName(), allData.size(), batchSize);
|
||||
}
|
||||
|
||||
// 2) 이미 끝까지 읽었으면 종료
|
||||
if (currentBatchIndex >= allData.size()) {
|
||||
log.info("[{}] 모든 배치 처리 완료", getReaderName());
|
||||
return null;
|
||||
}
|
||||
|
||||
// 3) 이번 배치의 end 계산
|
||||
int end = Math.min(currentBatchIndex + batchSize, allData.size());
|
||||
|
||||
// 4) 현재 batch 리스트 잘라서 반환
|
||||
List<CompanyComplianceDto> batch = allData.subList(currentBatchIndex, end);
|
||||
|
||||
int batchNum = (currentBatchIndex / batchSize) + 1;
|
||||
int totalBatches = (int) Math.ceil((double) allData.size() / batchSize);
|
||||
|
||||
log.info("[{}] 배치 {}/{} 처리 중: {}건", getReaderName(), batchNum, totalBatches, batch.size());
|
||||
|
||||
// 다음 batch 인덱스 이동
|
||||
currentBatchIndex = end;
|
||||
updateApiCallStats(totalBatches, batchNum);
|
||||
|
||||
return batch;
|
||||
}
|
||||
|
||||
@Override
|
||||
protected void afterFetch(List<CompanyComplianceDto> data){
|
||||
try{
|
||||
if (data == null) {
|
||||
log.info("[{}] 배치 처리 성공", getReaderName());
|
||||
}
|
||||
}catch (Exception e){
|
||||
log.info("[{}] 배치 처리 실패", getReaderName());
|
||||
log.info("[{}] API 호출 종료", getReaderName());
|
||||
}
|
||||
}
|
||||
private List<CompanyComplianceDto> callApiWithBatch() {
|
||||
Map<String, String> params = batchDateService.getDateRangeWithTimezoneParams(getApiKey());
|
||||
// 부모 클래스의 공통 모듈 호출 (단 한 줄로 처리 가능)
|
||||
return executeListApiCall(
|
||||
maritimeServiceApiUrl,
|
||||
getApiPath(),
|
||||
params,
|
||||
new ParameterizedTypeReference<List<CompanyComplianceDto>>() {},
|
||||
batchApiLogService
|
||||
);
|
||||
}
|
||||
|
||||
}
|
||||
@ -0,0 +1,9 @@
|
||||
package com.snp.batch.jobs.compliance.batch.repository;
|
||||
|
||||
import com.snp.batch.jobs.compliance.batch.entity.CompanyComplianceEntity;
|
||||
|
||||
import java.util.List;
|
||||
|
||||
public interface CompanyComplianceRepository {
|
||||
void saveCompanyComplianceAll(List<CompanyComplianceEntity> items);
|
||||
}
|
||||
@ -0,0 +1,116 @@
|
||||
package com.snp.batch.jobs.compliance.batch.repository;
|
||||
|
||||
import com.snp.batch.common.batch.repository.BaseJdbcRepository;
|
||||
import com.snp.batch.jobs.compliance.batch.entity.CompanyComplianceEntity;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.springframework.jdbc.core.JdbcTemplate;
|
||||
import org.springframework.jdbc.core.RowMapper;
|
||||
import org.springframework.stereotype.Repository;
|
||||
|
||||
import java.sql.PreparedStatement;
|
||||
import java.sql.Types;
|
||||
import java.util.List;
|
||||
|
||||
@Slf4j
|
||||
@Repository("CompanyComplianceRepository")
|
||||
public class CompanyComplianceRepositoryImpl extends BaseJdbcRepository<CompanyComplianceEntity, Long> implements CompanyComplianceRepository{
|
||||
public CompanyComplianceRepositoryImpl(JdbcTemplate jdbcTemplate) {
|
||||
super(jdbcTemplate);
|
||||
}
|
||||
|
||||
@Override
|
||||
protected String getTableName() {
|
||||
return null;
|
||||
}
|
||||
|
||||
@Override
|
||||
protected RowMapper<CompanyComplianceEntity> getRowMapper() {
|
||||
return null;
|
||||
}
|
||||
|
||||
@Override
|
||||
protected Long extractId(CompanyComplianceEntity entity) {
|
||||
return null;
|
||||
}
|
||||
|
||||
@Override
|
||||
protected String getInsertSql() {
|
||||
return null;
|
||||
}
|
||||
|
||||
@Override
|
||||
protected String getUpdateSql() {
|
||||
return """
|
||||
INSERT INTO new_snp.tb_company_compliance_hstry(
|
||||
owcode, lastupdated,
|
||||
companyoverallcompliancestatus, companyonaustraliansanctionlist, companyonbessanctionlist, companyoncanadiansanctionlist, companyinofacsanctionedcountry,
|
||||
companyinfatfjurisdiction, companyoneusanctionlist, companyonofacsanctionlist, companyonofacnonsdnsanctionlist, companyonofacssilist,
|
||||
companyonswisssanctionlist, companyonuaesanctionlist, companyonunsanctionlist, parentcompanycompliancerisk
|
||||
)VALUES(
|
||||
?, ?::timestamp, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?
|
||||
)ON CONFLICT (owcode, lastupdated)
|
||||
DO UPDATE SET
|
||||
companyoverallcompliancestatus = EXCLUDED.companyoverallcompliancestatus,
|
||||
companyonaustraliansanctionlist = EXCLUDED.companyonaustraliansanctionlist,
|
||||
companyonbessanctionlist = EXCLUDED.companyonbessanctionlist,
|
||||
companyoncanadiansanctionlist = EXCLUDED.companyoncanadiansanctionlist,
|
||||
companyinofacsanctionedcountry = EXCLUDED.companyinofacsanctionedcountry,
|
||||
companyinfatfjurisdiction = EXCLUDED.companyinfatfjurisdiction,
|
||||
companyoneusanctionlist = EXCLUDED.companyoneusanctionlist,
|
||||
companyonofacsanctionlist = EXCLUDED.companyonofacsanctionlist,
|
||||
companyonofacnonsdnsanctionlist = EXCLUDED.companyonofacnonsdnsanctionlist,
|
||||
companyonofacssilist = EXCLUDED.companyonofacssilist,
|
||||
companyonswisssanctionlist = EXCLUDED.companyonswisssanctionlist,
|
||||
companyonuaesanctionlist = EXCLUDED.companyonuaesanctionlist,
|
||||
companyonunsanctionlist = EXCLUDED.companyonunsanctionlist,
|
||||
parentcompanycompliancerisk = EXCLUDED.parentcompanycompliancerisk
|
||||
""";
|
||||
}
|
||||
|
||||
@Override
|
||||
protected void setInsertParameters(PreparedStatement ps, CompanyComplianceEntity entity) throws Exception {
|
||||
}
|
||||
|
||||
@Override
|
||||
protected void setUpdateParameters(PreparedStatement ps, CompanyComplianceEntity entity) throws Exception {
|
||||
int idx = 1;
|
||||
ps.setString(idx++, entity.getOwcode());
|
||||
ps.setString(idx++, entity.getLastUpdated());
|
||||
ps.setObject(idx++, entity.getCompanyOverallComplianceStatus(), Types.INTEGER);
|
||||
ps.setObject(idx++, entity.getCompanyOnAustralianSanctionList(), Types.INTEGER);
|
||||
ps.setObject(idx++, entity.getCompanyOnBESSanctionList(), Types.INTEGER);
|
||||
ps.setObject(idx++, entity.getCompanyOnCanadianSanctionList(), Types.INTEGER);
|
||||
ps.setObject(idx++, entity.getCompanyInOFACSanctionedCountry(), Types.INTEGER);
|
||||
ps.setObject(idx++, entity.getCompanyInFATFJurisdiction(), Types.INTEGER);
|
||||
ps.setObject(idx++, entity.getCompanyOnEUSanctionList(), Types.INTEGER);
|
||||
ps.setObject(idx++, entity.getCompanyOnOFACSanctionList(), Types.INTEGER);
|
||||
ps.setObject(idx++, entity.getCompanyOnOFACNONSDNSanctionList(), Types.INTEGER);
|
||||
ps.setObject(idx++, entity.getCompanyOnOFACSSISanctionList(), Types.INTEGER);
|
||||
ps.setObject(idx++, entity.getCompanyOnSwissSanctionList(), Types.INTEGER);
|
||||
ps.setObject(idx++, entity.getCompanyOnUAESanctionList(), Types.INTEGER);
|
||||
ps.setObject(idx++, entity.getCompanyOnUNSanctionList(), Types.INTEGER);
|
||||
ps.setObject(idx++, entity.getParentCompanyNonCompliance(), Types.INTEGER);
|
||||
}
|
||||
|
||||
@Override
|
||||
protected String getEntityName() {
|
||||
return "CompanyComplianceEntity";
|
||||
}
|
||||
|
||||
@Override
|
||||
public void saveCompanyComplianceAll(List<CompanyComplianceEntity> items) {
|
||||
if (items == null || items.isEmpty()) {
|
||||
return;
|
||||
}
|
||||
jdbcTemplate.batchUpdate(getUpdateSql(), items, items.size(),
|
||||
(ps, entity) -> {
|
||||
try {
|
||||
setUpdateParameters(ps, entity);
|
||||
} catch (Exception e) {
|
||||
log.error("배치 수정 파라미터 설정 실패", e);
|
||||
throw new RuntimeException(e);
|
||||
}
|
||||
});
|
||||
log.info("{} 전체 저장 완료: 수정={} 건", getEntityName(), items.size());
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,24 @@
|
||||
package com.snp.batch.jobs.compliance.batch.writer;
|
||||
|
||||
import com.snp.batch.common.batch.writer.BaseWriter;
|
||||
import com.snp.batch.jobs.compliance.batch.entity.CompanyComplianceEntity;
|
||||
import com.snp.batch.jobs.compliance.batch.repository.CompanyComplianceRepository;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
import java.util.List;
|
||||
|
||||
@Slf4j
|
||||
@Component
|
||||
public class CompanyComplianceDataWriter extends BaseWriter<CompanyComplianceEntity> {
|
||||
private final CompanyComplianceRepository complianceRepository;
|
||||
public CompanyComplianceDataWriter(CompanyComplianceRepository complianceRepository) {
|
||||
super("CompanyComplianceRepository");
|
||||
this.complianceRepository = complianceRepository;
|
||||
}
|
||||
|
||||
@Override
|
||||
protected void writeItems(List<CompanyComplianceEntity> items) throws Exception {
|
||||
complianceRepository.saveCompanyComplianceAll(items);
|
||||
}
|
||||
}
|
||||
불러오는 중...
Reference in New Issue
Block a user