feat : B1 캘린더용 정책상품 회차·온통청년 청년정책 수집

src/main/java/apptive/fin/apicollector/Source.java

@@ -4,5 +4,6 @@ public enum Source {
     ALL,
     ONTONG, // 온통청년 API
     FSS, // 금감원 API
-    KFB // 은행연합회 소비자포털 비교공시(입출금자유예금)
+    KFB, // 은행연합회 소비자포털 비교공시(입출금자유예금)
+    CALENDAR // B1 캘린더: 정책상품 회차(manual-product-rounds.json → policy_product_round)와 온통청년 청년정책(youth_policy)
 }

src/main/java/apptive/fin/apicollector/batch/SourceDecider.java

@@ -23,6 +23,7 @@ public FlowExecutionStatus decide(
             case FSS -> new FlowExecutionStatus("FSS");
             case ONTONG ->  new FlowExecutionStatus("ONTONG_YOUTH");
             case KFB -> new FlowExecutionStatus("KFB");
+            case CALENDAR -> new FlowExecutionStatus("CALENDAR");
         };
     }
 }

src/main/java/apptive/fin/apicollector/client/OntongYouthClient.java

@@ -1,17 +1,22 @@
 package apptive.fin.apicollector.client;
 
 import apptive.fin.apicollector.config.CollectorProperties;
+import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Qualifier;
 import tools.jackson.databind.JsonNode;
 import org.springframework.stereotype.Component;
 import org.springframework.web.client.RestClient;
+import org.springframework.web.client.RestClientException;
 
 import java.util.ArrayList;
 import java.util.List;
 
+@Slf4j
 @Component
 public class OntongYouthClient {
 
+    private static final int MAX_ATTEMPTS_PER_PAGE = 3;
+
     private final CollectorProperties properties;
     private final RestClient ontongYouthRestClient;
 
@@ -49,7 +54,24 @@ public List<JsonNode> fetchAll() {
         return result;
     }
 
+    // 온통청년은 가끔 JSON 대신 HTML 오류 페이지를 돌려준다(2026-11 확인, 바로 다시 요청하면 정상).
+    // I/O 오류 메시지에는 API 키가 든 요청 URL이 들어가므로 예외 종류만 남긴다.
     private JsonNode fetchPage(RestClient restClient, int page, int pageSize) {
+        String lastFailure = null;
+        for (int attempt = 1; attempt <= MAX_ATTEMPTS_PER_PAGE; attempt++) {
+            try {
+                return requestPage(restClient, page, pageSize);
+            }
+            catch (RestClientException e) {
+                lastFailure = e.getClass().getSimpleName();
+                log.warn("OntongYouthClient: page {} attempt {}/{} failed. {}", page, attempt, MAX_ATTEMPTS_PER_PAGE, lastFailure);
+            }
+        }
+        throw new IllegalStateException(
+                "Ontong youth page " + page + " failed " + MAX_ATTEMPTS_PER_PAGE + " times. last=" + lastFailure);
+    }
+
+    private JsonNode requestPage(RestClient restClient, int page, int pageSize) {
         return restClient.get()
                 .uri(uriBuilder -> uriBuilder
                         .path("/go/ythip/getPlcy")

src/main/java/apptive/fin/apicollector/config/FinancialProductSyncJobConfig.java

@@ -9,6 +9,8 @@
 import apptive.fin.apicollector.tasklet.FetchKfbRawTasklet;
 import apptive.fin.apicollector.tasklet.FetchManualRawTasklet;
 import apptive.fin.apicollector.tasklet.ResolveProductDisplayNameTasklet;
+import apptive.fin.apicollector.tasklet.SyncPolicyProductRoundTasklet;
+import apptive.fin.apicollector.tasklet.SyncYouthPolicyTasklet;
 import lombok.RequiredArgsConstructor;
 import org.springframework.batch.core.job.Job;
 import org.springframework.batch.core.job.builder.FlowBuilder;
@@ -44,6 +46,7 @@ public Job financialProductSyncJob(
         Flow fssSyncFlow,
         Flow ontongYouthSyncFlow,
         Flow kfbSyncFlow,
+        Flow calendarSyncFlow,
         Flow allSyncFlow
     ) {
         return new JobBuilder("financialProductSyncJob", jobRepository)
@@ -54,6 +57,8 @@ public Job financialProductSyncJob(
                     .on("ONTONG_YOUTH").to(ontongYouthSyncFlow)
                 .from(sourceDecider)
                     .on("KFB").to(kfbSyncFlow)
+                .from(sourceDecider)
+                    .on("CALENDAR").to(calendarSyncFlow)
                 .from(sourceDecider)
                     .on("ALL").to(allSyncFlow)
                 .end()
@@ -93,6 +98,38 @@ public Step fetchKfbRawStep(
                 .build();
     }
 
+    // 정책 목록 조회와 저장을 한 트랜잭션에 묶지 않는다. 저장은 YouthPolicySyncService가 트랜잭션을 연다.
+    @Bean
+    public Step syncYouthPolicyStep(
+            JobRepository jobRepository,
+            SyncYouthPolicyTasklet syncYouthPolicyTasklet
+    ) {
+        return new StepBuilder(SyncYouthPolicyTasklet.STEP_NAME, jobRepository)
+                .tasklet(syncYouthPolicyTasklet, new ResourcelessTransactionManager())
+                .build();
+    }
+
+    @Bean
+    public Step syncPolicyProductRoundStep(
+            JobRepository jobRepository,
+            PlatformTransactionManager transactionManager,
+            SyncPolicyProductRoundTasklet syncPolicyProductRoundTasklet
+    ) {
+        return new StepBuilder(SyncPolicyProductRoundTasklet.STEP_NAME, jobRepository)
+                .tasklet(syncPolicyProductRoundTasklet, transactionManager)
+                .build();
+    }
+
+    // B1 캘린더는 하루 1회 이상 갱신해야 해서 상품 수집과 따로 돌릴 수 있게 둔다.
+    // 정책상품 회차는 로컬 파일이라 온통청년 호출보다 먼저 둔다.
+    @Bean
+    public Flow calendarSyncFlow(Step syncPolicyProductRoundStep, Step syncYouthPolicyStep) {
+        return new FlowBuilder<Flow>("calendarSyncFlow")
+                .start(syncPolicyProductRoundStep)
+                .next(syncYouthPolicyStep)
+                .build();
+    }
+
     @Bean
     public Flow fssSyncFlow(
             Step fetchFssRawStep,
@@ -154,8 +191,12 @@ public Flow allSyncFlow(
             Step deactivateMissingProductStep,
             Step bankProductUrlStep,
             Step resolveProductDisplayNameStep,
+            Step syncPolicyProductRoundStep,
+            Step syncYouthPolicyStep,
             Step failIfAnyStepFailedStep
     ) {
+        // 캘린더는 맨 뒤에 둔다. 정책상품 회차는 정부상품 정규화 뒤라야 상품을 찾을 수 있고,
+        // 온통청년이 실패해도 상품 수집은 이미 끝나 있다.
         // KFB는 외부 HTML 스크래핑이라 깨지기 쉽다. 실패해도 다른 소스는 끝까지 돌리고(KFB 비활성화는
         // DeactivateMissingProductTasklet이 건너뜀), 마지막 step에서 잡을 FAILED로 끝내 실패를 드러낸다.
         return new FlowBuilder<Flow>("allSyncFlow")
@@ -170,6 +211,8 @@ public Flow allSyncFlow(
                 .next(deactivateMissingProductStep)
                 .next(bankProductUrlStep)
                 .next(resolveProductDisplayNameStep)
+                .next(syncPolicyProductRoundStep)
+                .next(syncYouthPolicyStep)
                 .next(failIfAnyStepFailedStep)
                 .build();
     }

src/main/java/apptive/fin/apicollector/policyround/PolicyProductRound.java

@@ -0,0 +1,68 @@
+package apptive.fin.apicollector.policyround;
+
+import apptive.fin.apicollector.global.entity.BaseTimeEntity;
+import jakarta.persistence.Column;
+import jakarta.persistence.Entity;
+import jakarta.persistence.GeneratedValue;
+import jakarta.persistence.GenerationType;
+import jakarta.persistence.Id;
+import jakarta.persistence.Table;
+import lombok.AccessLevel;
+import lombok.Getter;
+import lombok.NoArgsConstructor;
+
+import java.time.Instant;
+import java.time.LocalDate;
+import java.util.Objects;
+
+/** 1계층(Y-Fin 정책상품) 신청 회차. 스키마는 Fin-BE V10__create_youth_policy_schema.sql */
+@Entity
+@Table(name = "policy_product_round")
+@Getter
+@NoArgsConstructor(access = AccessLevel.PROTECTED)
+public class PolicyProductRound extends BaseTimeEntity {
+
+    @Id
+    @GeneratedValue(strategy = GenerationType.IDENTITY)
+    private Long id;
+
+    @Column(name = "product_id", nullable = false)
+    private Long productId;
+
+    @Column(name = "round_no", nullable = false)
+    private Integer roundNo;
+
+    @Column(name = "apply_start_date", nullable = false)
+    private LocalDate applyStartDate;
+
+    @Column(name = "apply_end_date", nullable = false)
+    private LocalDate applyEndDate;
+
+    @Column(name = "source_url", columnDefinition = "TEXT")
+    private String sourceUrl;
+
+    // 날짜가 바뀐 시각. Fin-BE 캘린더가 7일간 '일정 변경'으로 표시한다.
+    @Column(name = "schedule_changed_at")
+    private Instant scheduleChangedAt;
+
+    PolicyProductRound(Long productId, int roundNo, ProductRounds.Round round) {
+        this.productId = productId;
+        this.roundNo = roundNo;
+        this.applyStartDate = round.start();
+        this.applyEndDate = round.end();
+        this.sourceUrl = round.source();
+    }
+
+    /** @return 바뀐 값이 있으면 true */
+    boolean apply(ProductRounds.Round round, Instant changedAt) {
+        boolean datesChanged = !round.start().equals(applyStartDate) || !round.end().equals(applyEndDate);
+        boolean sourceChanged = !Objects.equals(round.source(), sourceUrl);
+        if (datesChanged) {
+            this.applyStartDate = round.start();
+            this.applyEndDate = round.end();
+            this.scheduleChangedAt = changedAt;
+        }
+        this.sourceUrl = round.source();
+        return datesChanged || sourceChanged;
+    }
+}

src/main/java/apptive/fin/apicollector/policyround/PolicyProductRoundRepository.java

@@ -0,0 +1,6 @@
+package apptive.fin.apicollector.policyround;
+
+import org.springframework.data.jpa.repository.JpaRepository;
+
+public interface PolicyProductRoundRepository extends JpaRepository<PolicyProductRound, Long> {
+}

src/main/java/apptive/fin/apicollector/policyround/PolicyProductRoundSyncService.java

@@ -0,0 +1,85 @@
+package apptive.fin.apicollector.policyround;
+
+import apptive.fin.apicollector.Source;
+import apptive.fin.apicollector.product.entity.Product;
+import apptive.fin.apicollector.product.entity.ProductSource;
+import apptive.fin.apicollector.product.repository.ProductRepository;
+import apptive.fin.apicollector.product.repository.ProductSourceRepository;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Service;
+import org.springframework.transaction.annotation.Transactional;
+
+import java.time.Instant;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+/**
+ * manual-product-rounds.json으로 policy_product_round를 맞춘다. 파일이 원천이라 파일에 없는 회차는 지운다.
+ * 회차는 상품별 파일 안 순서(round_no)로 맞춰, 날짜가 바뀐 회차만 변경 시각을 남긴다.
+ */
+@Slf4j
+@Service
+@RequiredArgsConstructor
+public class PolicyProductRoundSyncService {
+
+    // 수동 정부상품은 product_source ONTONG으로 저장된다(ManualProductNormalizer).
+    private static final String MANUAL_PRODUCT_SOURCE = Source.ONTONG.name();
+
+    private final PolicyProductRoundRepository roundRepository;
+    private final ProductSourceRepository productSourceRepository;
+    private final ProductRepository productRepository;
+
+    /** @param missingProducts DB에 아직 없는 상품 코드(정부상품 수집 전이면 생긴다) */
+    public record SyncResult(int inserted, int updated, int deleted, List<String> missingProducts) {
+    }
+
+    @Transactional
+    public SyncResult sync(List<ProductRounds> entries, Instant syncedAt) {
+        Optional<ProductSource> source = productSourceRepository.findByCode(MANUAL_PRODUCT_SOURCE);
+        Map<Long, List<PolicyProductRound>> existingByProduct = roundRepository.findAll().stream()
+                .collect(Collectors.groupingBy(PolicyProductRound::getProductId));
+
+        int inserted = 0;
+        int updated = 0;
+        int deleted = 0;
+        List<String> missingProducts = new ArrayList<>();
+        for (ProductRounds entry : entries) {
+            Optional<Product> product = source.flatMap(s -> productRepository.findBySourceAndProductCode(s, entry.productCode()));
+            if (product.isEmpty()) {
+                missingProducts.add(entry.productCode());
+                continue;
+            }
+            Long productId = product.get().getId();
+            Map<Integer, PolicyProductRound> existing = existingByProduct.getOrDefault(productId, List.of()).stream()
+                    .collect(Collectors.toMap(PolicyProductRound::getRoundNo, Function.identity()));
+            existingByProduct.remove(productId);
+
+            for (int i = 0; i < entry.rounds().size(); i++) {
+                int roundNo = i + 1;
+                ProductRounds.Round next = entry.rounds().get(i);
+                PolicyProductRound round = existing.remove(roundNo);
+                if (round == null) {
+                    roundRepository.save(new PolicyProductRound(productId, roundNo, next));
+                    inserted++;
+                }
+                else if (round.apply(next, syncedAt)) {
+                    updated++;
+                }
+            }
+            roundRepository.deleteAll(existing.values());
+            deleted += existing.size();
+        }
+
+        // 파일에서 상품 항목이 통째로 빠졌으면 그 상품의 회차도 지운다.
+        List<PolicyProductRound> orphaned = existingByProduct.values().stream().flatMap(List::stream).toList();
+        roundRepository.deleteAll(orphaned);
+        deleted += orphaned.size();
+
+        return new SyncResult(inserted, updated, deleted, missingProducts);
+    }
+}

src/main/java/apptive/fin/apicollector/policyround/ProductRounds.java

@@ -0,0 +1,29 @@
+package apptive.fin.apicollector.policyround;
+
+import java.time.LocalDate;
+import java.util.List;
+
+/**
+ * manual-product-rounds.json의 한 항목. 상품 코드(manual-products.json의 productCode)별 신청 회차.
+ * 회차가 없으면(그해 모집이 없거나 공고 전) rounds를 비우고 note에 사유를 적는다.
+ */
+public record ProductRounds(String productCode, List<Round> rounds, String note) {
+
+    /** @param source 신청기간 근거(공고문, 보도자료 등) */
+    public record Round(LocalDate start, LocalDate end, String source, String note) {
+    }
+
+    /** 사람이 고치는 파일이라 저장 전에 형식을 확인한다. 틀리면 파일을 고쳐야 하므로 실패시킨다. */
+    public void validate() {
+        if (productCode == null || productCode.isBlank() || rounds == null) {
+            throw new IllegalStateException("productCode와 rounds는 필수입니다. productCode=" + productCode);
+        }
+        for (Round round : rounds) {
+            if (round.start() == null || round.end() == null || round.start().isAfter(round.end())) {
+                throw new IllegalStateException(
+                        "회차의 시작일과 종료일이 없거나 순서가 틀렸습니다. productCode=%s, start=%s, end=%s"
+                                .formatted(productCode, round.start(), round.end()));
+            }
+        }
+    }
+}

src/main/java/apptive/fin/apicollector/tasklet/BankProductUrlTasklet.java

@@ -86,7 +86,7 @@ private List<String> targetSourceCodes() {
         return switch (collectorProperties.source()) {
             case ALL -> List.of(Source.FSS.name(), Source.KFB.name());
             case FSS, KFB -> List.of(collectorProperties.source().name());
-            case ONTONG -> List.of();
+            case ONTONG, CALENDAR -> List.of();
         };
     }