diff --git a/analytics-service/src/main/java/kt/aivle/analytics/adapter/in/web/HistoricalAnalyticsController.java b/analytics-service/src/main/java/kt/aivle/analytics/adapter/in/web/HistoricalAnalyticsController.java index 6681d28..be4d499 100644 --- a/analytics-service/src/main/java/kt/aivle/analytics/adapter/in/web/HistoricalAnalyticsController.java +++ b/analytics-service/src/main/java/kt/aivle/analytics/adapter/in/web/HistoricalAnalyticsController.java @@ -1,7 +1,5 @@ package kt.aivle.analytics.adapter.in.web; -import java.util.List; - import org.springframework.http.ResponseEntity; import org.springframework.validation.annotation.Validated; import org.springframework.web.bind.annotation.GetMapping; @@ -14,7 +12,6 @@ import io.swagger.v3.oas.annotations.tags.Tag; import kt.aivle.analytics.adapter.in.web.dto.response.AccountMetricsResponse; import kt.aivle.analytics.adapter.in.web.dto.response.EmotionAnalysisResponse; -import kt.aivle.analytics.adapter.in.web.dto.response.PostCommentsResponse; import kt.aivle.analytics.adapter.in.web.dto.response.PostMetricsResponse; import kt.aivle.analytics.application.port.in.AnalyticsQueryUseCase; import kt.aivle.common.code.CommonResponseCode; @@ -55,20 +52,6 @@ public ResponseEntity> getHistoricalAccountM return ResponseEntity.ok(ApiResponse.of(CommonResponseCode.OK, response)); } - @Operation(summary = "히스토리 게시물 댓글 조회", description = "게시물 댓글 히스토리를 페이지네이션으로 조회합니다.") - @GetMapping("/posts/comments") - public ResponseEntity>> getHistoricalPostComments( - @RequestParam("date") String dateStr, - @RequestParam("accountId") Long accountId, - @RequestParam(value = "postId", required = false) Long postId, - @RequestParam(value = "page", defaultValue = "0") Integer page, - @RequestParam(value = "size", defaultValue = "20") Integer size, - @RequestHeader("X-USER-ID") Long userId) { - - List response = analyticsQueryUseCase.getHistoricalPostComments(userId, dateStr, accountId, postId, page, size); - - return ResponseEntity.ok(ApiResponse.of(CommonResponseCode.OK, response)); - } @Operation(summary = "히스토리 게시물 감정분석 조회", description = "특정 날짜의 게시물 댓글 감정분석 결과와 키워드를 조회합니다.") @GetMapping("/posts/emotion-analysis") diff --git a/analytics-service/src/main/java/kt/aivle/analytics/adapter/in/web/RealtimeAnalyticsController.java b/analytics-service/src/main/java/kt/aivle/analytics/adapter/in/web/RealtimeAnalyticsController.java index 782778d..5674bdc 100644 --- a/analytics-service/src/main/java/kt/aivle/analytics/adapter/in/web/RealtimeAnalyticsController.java +++ b/analytics-service/src/main/java/kt/aivle/analytics/adapter/in/web/RealtimeAnalyticsController.java @@ -1,7 +1,5 @@ package kt.aivle.analytics.adapter.in.web; -import java.util.List; - import org.springframework.http.ResponseEntity; import org.springframework.validation.annotation.Validated; import org.springframework.web.bind.annotation.GetMapping; @@ -13,7 +11,7 @@ import io.swagger.v3.oas.annotations.Operation; import io.swagger.v3.oas.annotations.tags.Tag; import kt.aivle.analytics.adapter.in.web.dto.response.AccountMetricsResponse; -import kt.aivle.analytics.adapter.in.web.dto.response.PostCommentsResponse; +import kt.aivle.analytics.adapter.in.web.dto.response.PostCommentsPageResponse; import kt.aivle.analytics.adapter.in.web.dto.response.PostMetricsResponse; import kt.aivle.analytics.adapter.in.web.dto.response.ReportResponse; import kt.aivle.analytics.application.port.in.AnalyticsQueryUseCase; @@ -56,16 +54,16 @@ public ResponseEntity> getRealtimeAccountMet return ResponseEntity.ok(ApiResponse.of(CommonResponseCode.OK, response)); } - @Operation(summary = "실시간 게시물 댓글 조회", description = "특정 게시물의 실시간 댓글을 페이지네이션으로 조회합니다.") + @Operation(summary = "실시간 게시물 댓글 조회", description = "특정 게시물의 실시간 댓글을 YouTube API 네이티브 페이지네이션으로 조회합니다.") @GetMapping("/posts/comments") - public ResponseEntity>> getRealtimePostComments( + public ResponseEntity> getRealtimePostComments( @RequestParam("accountId") Long accountId, @RequestParam(value = "postId", required = false) Long postId, - @RequestParam(value = "page", defaultValue = "0") Integer page, - @RequestParam(value = "size", defaultValue = "20") Integer size, + @RequestParam(value = "pageToken", required = false) String pageToken, + @RequestParam(value = "size", defaultValue = "5") Integer size, @RequestHeader("X-USER-ID") Long userId) { - List response = analyticsQueryUseCase.getRealtimePostComments(userId, accountId, postId, page, size); + PostCommentsPageResponse response = analyticsQueryUseCase.getRealtimePostComments(userId, accountId, postId, pageToken, size); return ResponseEntity.ok(ApiResponse.of(CommonResponseCode.OK, response)); } diff --git a/analytics-service/src/main/java/kt/aivle/analytics/adapter/in/web/dto/response/PostCommentsPageResponse.java b/analytics-service/src/main/java/kt/aivle/analytics/adapter/in/web/dto/response/PostCommentsPageResponse.java new file mode 100644 index 0000000..8d03a34 --- /dev/null +++ b/analytics-service/src/main/java/kt/aivle/analytics/adapter/in/web/dto/response/PostCommentsPageResponse.java @@ -0,0 +1,46 @@ +package kt.aivle.analytics.adapter.in.web.dto.response; + +import java.util.List; + +import lombok.Builder; +import lombok.Getter; + +/** + * YouTube API 페이지네이션을 지원하는 댓글 응답 DTO + */ +@Getter +@Builder +public class PostCommentsPageResponse { + + /** + * 댓글 데이터 목록 + */ + private List data; + + /** + * 다음 페이지 요청용 토큰 (YouTube API) + */ + private String nextPageToken; + + /** + * 다음 페이지 존재 여부 + */ + private boolean hasNextPage; + + /** + * 현재 페이지의 댓글 수 + */ + private int currentPageSize; + + /** + * 빈 페이지 응답 생성 + */ + public static PostCommentsPageResponse empty() { + return PostCommentsPageResponse.builder() + .data(List.of()) + .nextPageToken(null) + .hasNextPage(false) + .currentPageSize(0) + .build(); + } +} diff --git a/analytics-service/src/main/java/kt/aivle/analytics/adapter/in/websocket/ReportWebSocketHandler.java b/analytics-service/src/main/java/kt/aivle/analytics/adapter/in/websocket/ReportWebSocketHandler.java index dfc13e7..7f3b2b3 100644 --- a/analytics-service/src/main/java/kt/aivle/analytics/adapter/in/websocket/ReportWebSocketHandler.java +++ b/analytics-service/src/main/java/kt/aivle/analytics/adapter/in/websocket/ReportWebSocketHandler.java @@ -13,6 +13,7 @@ import org.springframework.web.socket.WebSocketSession; import org.springframework.web.socket.handler.TextWebSocketHandler; +import kt.aivle.analytics.adapter.in.web.dto.response.ReportResponse; import kt.aivle.analytics.adapter.in.websocket.dto.MessageType; import kt.aivle.analytics.adapter.in.websocket.dto.ReportRequestMessage; import kt.aivle.analytics.adapter.in.websocket.dto.WebSocketResponseMessage; @@ -29,7 +30,7 @@ public class ReportWebSocketHandler extends TextWebSocketHandler { private final ObjectMapper objectMapper; private final AtomicLong taskCounter = new AtomicLong(0); - private final ConcurrentHashMap> activeTasks = new ConcurrentHashMap<>(); + private final ConcurrentHashMap> activeTasks = new ConcurrentHashMap<>(); @Override public void afterConnectionEstablished(WebSocketSession session) throws Exception { @@ -79,31 +80,39 @@ private void processReportAsync(WebSocketSession session, ReportRequestMessage r log.info("[WebSocket] AI 분석 보고서 생성 요청 - postId: {}, accountId: {}, storeId: {}", request.getPostId(), request.getAccountId(), request.getStoreId()); + // 중복 요청 방지: 이미 진행 중인 작업이 있는지 확인 + if (!activeTasks.isEmpty()) { + log.info("[WebSocket] 이미 진행 중인 작업이 있습니다. 요청을 무시합니다. - sessionId: {}", session.getId()); + return; // 조용히 무시 + } + // 진행률 전송 시작 - sendMessage(session, WebSocketResponseMessage.progress(0, "AI 분석 보고서 생성을 시작합니다...")); + sendMessage(session, WebSocketResponseMessage.progress(10, "AI 분석 보고서 생성을 시작합니다...")); // accountId로 userId 조회 후 AI 분석 진행 // userId는 sns_account 테이블에서 조회하여 사용 - analyticsQueryUseCase.generateReportAsync( + CompletableFuture reportFuture = analyticsQueryUseCase.generateReportAsync( request.getAccountId(), request.getPostId(), - request.getStoreId() + request.getStoreId(), + (progress, message) -> { + // 진행률 콜백으로 WebSocket 메시지 전송 + sendMessage(session, WebSocketResponseMessage.progress(progress, message)); + } ) - .thenAccept(wsMessage -> { - // WebSocket 메시지를 직접 전송 - sendMessage(session, wsMessage); - }) - .exceptionally(throwable -> { - log.error("[WebSocket] AI 분석 처리 오류: {}", throwable.getMessage(), throwable); - sendMessage(session, WebSocketResponseMessage.error("AI 분석 처리 오류: " + throwable.getMessage())); - return null; - }) - .whenComplete((result, throwable) -> { - activeTasks.remove(taskId); + .whenComplete((reportResponse, throwable) -> { if (throwable != null) { - log.error("[WebSocket] 작업 완료 오류: {}", throwable.getMessage()); + // 에러 발생 시 + log.error("[WebSocket] AI 분석 처리 오류: {}", throwable.getMessage(), throwable); + sendMessage(session, WebSocketResponseMessage.error("AI 분석 처리 오류: " + throwable.getMessage())); + } else { + // 성공 시 최종 완료 메시지 전송 + sendMessage(session, WebSocketResponseMessage.complete(reportResponse, "AI 분석 보고서가 완성되었습니다!")); } + // 작업 정리 + activeTasks.remove(taskId); + // 완료 후 연결 유지를 위한 지연 (캐싱된 데이터 처리 시 중요) if (isSessionActive(session)) { try { @@ -114,8 +123,8 @@ private void processReportAsync(WebSocketSession session, ReportRequestMessage r } }); - // 작업 추적 - activeTasks.put(taskId, CompletableFuture.completedFuture(null)); + // 작업 추적 - 실제 CompletableFuture를 저장 + activeTasks.put(taskId, reportFuture); } private void sendMessage(WebSocketSession session, WebSocketResponseMessage message) { @@ -127,9 +136,8 @@ private void sendMessage(WebSocketSession session, WebSocketResponseMessage m } String jsonMessage = objectMapper.writeValueAsString(message); - log.info("[WebSocket] 메시지 전송: {}", jsonMessage); session.sendMessage(new TextMessage(jsonMessage)); - log.debug("[WebSocket] 메시지 전송: {}", jsonMessage); + log.info("[WebSocket] 메시지 전송: {}", jsonMessage); } catch (IOException e) { log.error("[WebSocket] 메시지 전송 실패: {}", e.getMessage(), e); } diff --git a/analytics-service/src/main/java/kt/aivle/analytics/adapter/out/infrastructure/YouTubeApiAdapter.java b/analytics-service/src/main/java/kt/aivle/analytics/adapter/out/infrastructure/YouTubeApiAdapter.java index 2c386a1..607becd 100644 --- a/analytics-service/src/main/java/kt/aivle/analytics/adapter/out/infrastructure/YouTubeApiAdapter.java +++ b/analytics-service/src/main/java/kt/aivle/analytics/adapter/out/infrastructure/YouTubeApiAdapter.java @@ -19,6 +19,7 @@ import org.springframework.stereotype.Component; import kt.aivle.analytics.adapter.in.web.dto.response.AccountMetricsResponse; +import kt.aivle.analytics.adapter.in.web.dto.response.PostCommentsPageResponse; import kt.aivle.analytics.adapter.in.web.dto.response.PostCommentsResponse; import kt.aivle.analytics.adapter.in.web.dto.response.PostMetricsResponse; import kt.aivle.analytics.application.port.out.infrastructure.ExternalApiPort; @@ -102,44 +103,13 @@ public VideoStatistics getVideoStatistics(String videoId) { } @Override - public List getVideoComments(String videoId) { - return getVideoCommentsWithLimit(videoId, 20); // 기본 20개로 제한 - } - - @Override - public List getVideoCommentsWithLimit(String videoId, int limit) { + public PostCommentsPageResponse getVideoCommentsWithPagination(String videoId, String pageToken, int maxResults) { try { YouTube.CommentThreads.List request = getYouTubeClient().commentThreads() .list(List.of("snippet")) .setKey(apiKey) .setVideoId(videoId) - .setMaxResults(Math.min(limit, 100L)); // 최대 100개, 요청된 개수와 비교하여 작은 값 사용 - - CommentThreadListResponse response = request.execute(); - - if (response.getItems() == null) { - return List.of(); - } - - return response.getItems().stream() - .limit(limit) // 추가로 limit 적용 - .map(this::toPostCommentsResponse) - .collect(Collectors.toList()); - - } catch (IOException e) { - handleYouTubeApiError(e, "video comments"); - return List.of(); - } - } - - @Override - public List getVideoCommentsWithPagination(String videoId, String pageToken) { - try { - YouTube.CommentThreads.List request = getYouTubeClient().commentThreads() - .list(List.of("snippet")) - .setKey(apiKey) - .setVideoId(videoId) - .setMaxResults(100L); + .setMaxResults(Math.min(maxResults, 100L)); if (pageToken != null) { request.setPageToken(pageToken); @@ -147,41 +117,25 @@ public List getVideoCommentsWithPagination(String videoId, CommentThreadListResponse response = request.execute(); - if (response.getItems() == null) { - return List.of(); - } - - return response.getItems().stream() - .map(this::toPostCommentsResponse) - .collect(Collectors.toList()); - - } catch (IOException e) { - handleYouTubeApiError(e, "video comments with pagination"); - return List.of(); - } - } - - @Override - public String getNextPageToken(String videoId, String currentPageToken) { - try { - YouTube.CommentThreads.List request = getYouTubeClient().commentThreads() - .list(List.of("snippet")) - .setKey(apiKey) - .setVideoId(videoId) - .setMaxResults(100L); - - if (currentPageToken != null) { - request.setPageToken(currentPageToken); - } - - CommentThreadListResponse response = request.execute(); - return response.getNextPageToken(); + List comments = response.getItems() != null + ? response.getItems().stream() + .map(this::toPostCommentsResponse) + .collect(Collectors.toList()) + : List.of(); + + return PostCommentsPageResponse.builder() + .data(comments) + .nextPageToken(response.getNextPageToken()) + .hasNextPage(response.getNextPageToken() != null) + .currentPageSize(comments.size()) + .build(); } catch (IOException e) { - handleYouTubeApiError(e, "get next page token"); - return null; + handleYouTubeApiError(e, "video comments with pagination response"); + return PostCommentsPageResponse.empty(); } } + // AI 분석은 별도 AiAnalysisAdapter에서 처리 diff --git a/analytics-service/src/main/java/kt/aivle/analytics/application/port/in/AnalyticsQueryUseCase.java b/analytics-service/src/main/java/kt/aivle/analytics/application/port/in/AnalyticsQueryUseCase.java index 68c8d8a..91b4692 100644 --- a/analytics-service/src/main/java/kt/aivle/analytics/application/port/in/AnalyticsQueryUseCase.java +++ b/analytics-service/src/main/java/kt/aivle/analytics/application/port/in/AnalyticsQueryUseCase.java @@ -1,14 +1,12 @@ package kt.aivle.analytics.application.port.in; -import java.util.List; import java.util.concurrent.CompletableFuture; import kt.aivle.analytics.adapter.in.web.dto.response.AccountMetricsResponse; import kt.aivle.analytics.adapter.in.web.dto.response.EmotionAnalysisResponse; -import kt.aivle.analytics.adapter.in.web.dto.response.PostCommentsResponse; +import kt.aivle.analytics.adapter.in.web.dto.response.PostCommentsPageResponse; import kt.aivle.analytics.adapter.in.web.dto.response.PostMetricsResponse; import kt.aivle.analytics.adapter.in.web.dto.response.ReportResponse; -import kt.aivle.analytics.adapter.in.websocket.dto.WebSocketResponseMessage; public interface AnalyticsQueryUseCase { @@ -17,20 +15,18 @@ public interface AnalyticsQueryUseCase { AccountMetricsResponse getRealtimeAccountMetrics(Long userId, Long accountId); - List getRealtimePostComments(Long userId, Long accountId, Long postId, Integer page, Integer size); + PostCommentsPageResponse getRealtimePostComments(Long userId, Long accountId, Long postId, String pageToken, Integer size); // 히스토리 데이터 조회 (date 파라미터 필수) PostMetricsResponse getHistoricalPostMetrics(Long userId, String dateStr, Long accountId, Long postId); AccountMetricsResponse getHistoricalAccountMetrics(Long userId, String dateStr, Long accountId); - List getHistoricalPostComments(Long userId, String dateStr, Long accountId, Long postId, Integer page, Integer size); - EmotionAnalysisResponse getHistoricalEmotionAnalysis(Long userId, String dateStr, Long accountId, Long postId); // AI 보고서 생성 (캐시 포함) ReportResponse generateReport(Long userId, Long accountId, Long postId, Long storeId); // 통합된 비동기 AI 보고서 생성 (WebSocket용) - 캐시 확인 포함 - CompletableFuture> generateReportAsync(Long accountId, Long postId, Long storeId); + CompletableFuture generateReportAsync(Long accountId, Long postId, Long storeId, ProgressCallback callback); } diff --git a/analytics-service/src/main/java/kt/aivle/analytics/application/port/in/ProgressCallback.java b/analytics-service/src/main/java/kt/aivle/analytics/application/port/in/ProgressCallback.java new file mode 100644 index 0000000..c8ed74a --- /dev/null +++ b/analytics-service/src/main/java/kt/aivle/analytics/application/port/in/ProgressCallback.java @@ -0,0 +1,6 @@ +package kt.aivle.analytics.application.port.in; + +@FunctionalInterface +public interface ProgressCallback { + void onProgress(int percentage, String message); +} diff --git a/analytics-service/src/main/java/kt/aivle/analytics/application/port/in/dto/PostCommentsQueryRequest.java b/analytics-service/src/main/java/kt/aivle/analytics/application/port/in/dto/PostCommentsQueryRequest.java index 1c43a36..3512602 100644 --- a/analytics-service/src/main/java/kt/aivle/analytics/application/port/in/dto/PostCommentsQueryRequest.java +++ b/analytics-service/src/main/java/kt/aivle/analytics/application/port/in/dto/PostCommentsQueryRequest.java @@ -1,7 +1,5 @@ package kt.aivle.analytics.application.port.in.dto; -import java.time.LocalDate; - import jakarta.validation.constraints.Max; import jakarta.validation.constraints.Min; import lombok.Builder; @@ -12,13 +10,16 @@ @Getter @NoArgsConstructor @SuperBuilder -public class PostCommentsQueryRequest extends BaseQueryRequest { +public class PostCommentsQueryRequest { + + private Long userId; // 사용자 ID private Long postId; // null이면 모든 게시물 - @Min(value = 0, message = "Page must be 0 or greater") - @Builder.Default - private Integer page = 0; // 기본값 0 + /** + * YouTube API 페이지네이션 토큰 (첫 페이지는 null) + */ + private String pageToken; @Min(value = 1, message = "Size must be 1 or greater") @Max(value = 100, message = "Size must be 100 or less") @@ -27,22 +28,18 @@ public class PostCommentsQueryRequest extends BaseQueryRequest { private Long accountId; // 계정 ID - public static PostCommentsQueryRequest forCurrentDate(Long postId, Integer page, Integer size, Long accountId) { - return new PostCommentsQueryRequest(null, postId, page, size, accountId); - } - public static PostCommentsQueryRequest forDate(LocalDate date, Long postId, Integer page, Integer size, Long accountId) { - return new PostCommentsQueryRequest(date, postId, page, size, accountId); + public static PostCommentsQueryRequest forPost(Long userId, Long postId, String pageToken, Integer size, Long accountId) { + return new PostCommentsQueryRequest(userId, postId, pageToken, size, accountId); } - public static PostCommentsQueryRequest forLatestPostByAccountId(Long accountId, Integer page, Integer size) { - return new PostCommentsQueryRequest(null, null, page, size, accountId); + public static PostCommentsQueryRequest forLatestPostByAccountId(Long userId, Long accountId, String pageToken, Integer size) { + return new PostCommentsQueryRequest(userId, null, pageToken, size, accountId); } - - public PostCommentsQueryRequest(LocalDate date, Long postId, Integer page, Integer size, Long accountId) { - super(date); + public PostCommentsQueryRequest(Long userId, Long postId, String pageToken, Integer size, Long accountId) { + this.userId = userId; this.postId = postId; - this.page = page != null ? page : 0; + this.pageToken = pageToken; this.size = size != null ? size : 20; this.accountId = accountId; } diff --git a/analytics-service/src/main/java/kt/aivle/analytics/application/port/out/infrastructure/ExternalApiPort.java b/analytics-service/src/main/java/kt/aivle/analytics/application/port/out/infrastructure/ExternalApiPort.java index c7dd3c4..f40fc06 100644 --- a/analytics-service/src/main/java/kt/aivle/analytics/application/port/out/infrastructure/ExternalApiPort.java +++ b/analytics-service/src/main/java/kt/aivle/analytics/application/port/out/infrastructure/ExternalApiPort.java @@ -3,7 +3,7 @@ import java.util.List; import kt.aivle.analytics.adapter.in.web.dto.response.AccountMetricsResponse; -import kt.aivle.analytics.adapter.in.web.dto.response.PostCommentsResponse; +import kt.aivle.analytics.adapter.in.web.dto.response.PostCommentsPageResponse; import kt.aivle.analytics.adapter.in.web.dto.response.PostMetricsResponse; /** @@ -22,25 +22,10 @@ public interface ExternalApiPort { */ VideoStatistics getVideoStatistics(String videoId); - /** - * YouTube 비디오 댓글 조회 - */ - List getVideoComments(String videoId); - - /** - * YouTube 비디오 댓글 조회 (개수 제한) - */ - List getVideoCommentsWithLimit(String videoId, int limit); - /** * YouTube 비디오 댓글 조회 (페이지네이션 지원) */ - List getVideoCommentsWithPagination(String videoId, String pageToken); - - /** - * 다음 페이지 토큰 조회 - */ - String getNextPageToken(String videoId, String currentPageToken); + PostCommentsPageResponse getVideoCommentsWithPagination(String videoId, String pageToken, int maxResults); /** * 실시간 게시물 메트릭 조회 diff --git a/analytics-service/src/main/java/kt/aivle/analytics/application/service/AnalyticsQueryService.java b/analytics-service/src/main/java/kt/aivle/analytics/application/service/AnalyticsQueryService.java index 6c6513f..fdcbcd5 100644 --- a/analytics-service/src/main/java/kt/aivle/analytics/application/service/AnalyticsQueryService.java +++ b/analytics-service/src/main/java/kt/aivle/analytics/application/service/AnalyticsQueryService.java @@ -2,7 +2,6 @@ import java.time.LocalDate; import java.time.LocalDateTime; -import java.util.ArrayList; import java.util.List; import java.util.Map; import java.util.Optional; @@ -17,13 +16,13 @@ import kt.aivle.analytics.adapter.in.event.dto.PostInfoResponseMessage; import kt.aivle.analytics.adapter.in.web.dto.response.AccountMetricsResponse; import kt.aivle.analytics.adapter.in.web.dto.response.EmotionAnalysisResponse; -import kt.aivle.analytics.adapter.in.web.dto.response.PostCommentsResponse; +import kt.aivle.analytics.adapter.in.web.dto.response.PostCommentsPageResponse; import kt.aivle.analytics.adapter.in.web.dto.response.PostMetricsResponse; import kt.aivle.analytics.adapter.in.web.dto.response.ReportResponse; -import kt.aivle.analytics.adapter.in.websocket.dto.WebSocketResponseMessage; import kt.aivle.analytics.adapter.out.infrastructure.dto.AiReportRequest; import kt.aivle.analytics.adapter.out.infrastructure.dto.AiReportResponse; import kt.aivle.analytics.application.port.in.AnalyticsQueryUseCase; +import kt.aivle.analytics.application.port.in.ProgressCallback; import kt.aivle.analytics.application.port.in.dto.AccountMetricsQueryRequest; import kt.aivle.analytics.application.port.in.dto.PostCommentsQueryRequest; import kt.aivle.analytics.application.port.in.dto.PostMetricsQueryRequest; @@ -94,19 +93,19 @@ public AccountMetricsResponse getRealtimeAccountMetrics(Long userId, Long accoun } @Override - @Cacheable(value = "realtime-comments", key = "'comments-' + #userId + ',' + #accountId + ',' + #postId + ',' + #page + ',' + #size") - public List getRealtimePostComments(Long userId, Long accountId, Long postId, Integer page, Integer size) { + @Cacheable(value = "realtime-comments", key = "'comments-' + #userId + ',' + #accountId + ',' + #postId + ',' + #pageToken + ',' + #size") + public PostCommentsPageResponse getRealtimePostComments(Long userId, Long accountId, Long postId, String pageToken, Integer size) { validationPort.validateAccountId(accountId); PostCommentsQueryRequest queryRequest; if (postId != null) { - queryRequest = PostCommentsQueryRequest.forCurrentDate(postId, page, size, accountId); + queryRequest = PostCommentsQueryRequest.forPost(userId, postId, pageToken, size, accountId); } else { - queryRequest = PostCommentsQueryRequest.forLatestPostByAccountId(accountId, page, size); + queryRequest = PostCommentsQueryRequest.forLatestPostByAccountId(userId, accountId, pageToken, size); } - return getRealtimePostCommentsInternal(userId, queryRequest); + return getRealtimePostCommentsInternal(queryRequest); } // 히스토리 데이터 조회 메서드들 @@ -155,21 +154,6 @@ public AccountMetricsResponse getHistoricalAccountMetrics(Long userId, String da return getAccountMetricsInternal(userId, queryRequest); } - @Override - @Cacheable(value = "history-comments", key = "'history-comments-' + #userId + ',' + #dateStr + ',' + #accountId + ',' + #postId + ',' + #page + ',' + #size") - public List getHistoricalPostComments(Long userId, String dateStr, Long accountId, Long postId, Integer page, Integer size) { - LocalDate date = validationPort.validateAndParseDate(dateStr); - validationPort.validateAccountId(accountId); - - PostCommentsQueryRequest queryRequest; - if (postId != null) { - queryRequest = PostCommentsQueryRequest.forDate(date, postId, page, size, accountId); - } else { - queryRequest = PostCommentsQueryRequest.forLatestPostByAccountId(accountId, page, size); - } - - return getHistoricalPostCommentsInternal(userId, queryRequest, date); - } @Override @Cacheable(value = "history-emotion-analysis", key = "'history-emotion-' + #userId + ',' + #dateStr + ',' + #accountId + ',' + #postId") @@ -200,7 +184,7 @@ public ReportResponse generateReport(Long userId, Long accountId, Long postId, L // 통합된 비동기 AI 보고서 생성 (WebSocket용) - 캐시 확인 포함 @Override - public CompletableFuture> generateReportAsync(Long accountId, Long postId, Long storeId) { + public CompletableFuture generateReportAsync(Long accountId, Long postId, Long storeId, ProgressCallback callback) { log.info("[WebSocket] 비동기 AI 보고서 생성 시작 - postId: {}", postId); // accountId로 userId 조회 @@ -223,45 +207,43 @@ public CompletableFuture> generateRepor }) .thenCompose(cachedReport -> { if (cachedReport != null) { - // 캐시된 보고서가 있으면 즉시 완료 - return CompletableFuture.supplyAsync(() -> WebSocketResponseMessage.complete(cachedReport, "캐시된 보고서를 찾았습니다!")); + // 캐시된 보고서가 있으면 콜백으로 알림 + callback.onProgress(100, "캐시된 보고서를 찾았습니다!"); + return CompletableFuture.completedFuture(cachedReport); } // 2. 캐시가 없으면 단계별로 처리 return CompletableFuture.supplyAsync(() -> { log.info("[WebSocket] 1단계: SNS 서비스에서 post 정보 가져오기 - postId: {}", postId); postInfoRef.set(getPostInfo(userId, accountId, postId, storeId)); - - // 1단계 완료 시 progress 메시지 반환 - return WebSocketResponseMessage.progress(25, "SNS 서비스에서 게시물 정보를 가져왔습니다."); + callback.onProgress(25, "SNS 서비스에서 게시물 정보를 가져왔습니다."); + return postInfoRef.get(); }) - .thenCompose(progressMessage1 -> + .thenCompose(postInfo -> CompletableFuture.supplyAsync(() -> { log.info("[WebSocket] 2단계: 게시물 메트릭 조회 - postId: {}", postId); postMetricsRef.set(getPostMetrics(userId, accountId, postId)); - - // 2단계 완료 시 progress 메시지 반환 - return WebSocketResponseMessage.progress(50, "게시물 메트릭을 조회했습니다."); + callback.onProgress(50, "게시물 메트릭을 조회했습니다."); + return postMetricsRef.get(); }) - .thenCompose(progressMessage2 -> - CompletableFuture.supplyAsync(() -> { - log.info("[WebSocket] 3단계: AI 보고서 생성 - postId: {}", postId); - ReportResponse reportResponse = generateAiReport(userId, accountId, postId, storeId, postInfoRef.get(), postMetricsRef.get()); + ) + .thenCompose(metrics -> + CompletableFuture.supplyAsync(() -> { + log.info("[WebSocket] 3단계: AI 보고서 생성 - postId: {}", postId); + ReportResponse reportResponse = generateAiReport(userId, accountId, postId, storeId, postInfoRef.get(), postMetricsRef.get()); - // 새로 생성된 보고서를 캐시에 저장 (cachedReport가 null일 때만) - try { - String cacheKey = postId + "_" + userId + "_" + accountId + "_" + storeId; - cacheManager.getCache("report").put(cacheKey, reportResponse); - log.info("[WebSocket] 새로 생성된 보고서를 캐시에 저장 - postId: {}, cacheKey: {}", postId, cacheKey); - } catch (Exception e) { - log.warn("[WebSocket] 캐시 저장 중 에러 발생 - postId: {}, error: {}", postId, e.getMessage()); - } - - // 3단계 완료 시 바로 complete 메시지 반환 - return WebSocketResponseMessage.complete(reportResponse, "AI 분석 보고서가 완성되었습니다!"); - }) - ) + // 새로 생성된 보고서를 캐시에 저장 + try { + String cacheKey = postId + "_" + userId + "_" + accountId + "_" + storeId; + cacheManager.getCache("report").put(cacheKey, reportResponse); + log.info("[WebSocket] 새로 생성된 보고서를 캐시에 저장 - postId: {}, cacheKey: {}", postId, cacheKey); + } catch (Exception e) { + log.warn("[WebSocket] 캐시 저장 중 에러 발생 - postId: {}, error: {}", postId, e.getMessage()); + } + return reportResponse; + }) ); + }); } // ===== PRIVATE METHODS ===== @@ -318,33 +300,6 @@ private AccountMetricsResponse getAccountMetricsInternal(Long userId, AccountMet .build(); } - /** - * 히스토리 댓글 조회 내부 로직 (날짜 기준 필터링) - */ - private List getHistoricalPostCommentsInternal(Long userId, PostCommentsQueryRequest request, LocalDate date) { - log.info("Getting historical post comments for userId: {}, postId: {}, accountId: {}, date: {}, page: {}, size: {}", - userId, request.getPostId(), request.getAccountId(), date, request.getPage(), request.getSize()); - - Long targetPostId; - if (request.getPostId() != null) { - targetPostId = request.getPostId(); - // postId가 제공된 경우 계정 ID 검증 - validatePostAccountId(targetPostId, request.getAccountId()); - } else { - targetPostId = getLatestPostIdByAccountId(request.getAccountId()); - } - - // 날짜 기준으로 publishedAt 이전의 댓글을 최신순으로 페이지네이션하여 조회 - List comments = snsPostCommentMetricRepositoryPort.findByPostIdAndPublishedAtBeforeWithPagination( - targetPostId, date, request.getPage(), request.getSize()); - - log.info("Retrieved historical comments from DB for postId: {}, date: {}, page: {}, size: {}, result count: {}", - targetPostId, date, request.getPage(), request.getSize(), comments.size()); - - return comments.stream() - .map(this::toSnsPostCommentsResponse) - .collect(Collectors.toList()); - } /** * 내부 히스토리 감정분석 로직 (캐시 적용) @@ -377,7 +332,9 @@ private EmotionAnalysisResponse getHistoricalEmotionAnalysisInternal(Long userId private PostMetricsResponse getRealtimePostMetricsInternal(Long userId, PostMetricsQueryRequest request) { log.info("Getting realtime post metrics for userId: {}, postId: {}, accountId: {}", userId, request.getPostId(), request.getAccountId()); - Long targetPostId = getTargetPostIdForRealtime(userId, request); + Long targetPostId = request.getPostId() != null + ? request.getPostId() + : getLatestPostIdByAccountId(request.getAccountId()); // postId가 제공된 경우 계정 ID 검증 if (request.getPostId() != null) { @@ -405,10 +362,13 @@ private AccountMetricsResponse getRealtimeAccountMetricsInternal(Long userId, Ac responses.get(0); } - private List getRealtimePostCommentsInternal(Long userId, PostCommentsQueryRequest request) { - log.info("Getting realtime post comments for userId: {}, postId: {}, accountId: {}", userId, request.getPostId(), request.getAccountId()); + private PostCommentsPageResponse getRealtimePostCommentsInternal(PostCommentsQueryRequest request) { + log.info("Getting realtime post comments for userId: {}, postId: {}, accountId: {}, pageToken: {}", + request.getUserId(), request.getPostId(), request.getAccountId(), request.getPageToken()); - Long targetPostId = getTargetPostIdForRealtime(userId, request); + Long targetPostId = request.getPostId() != null + ? request.getPostId() + : getLatestPostIdByAccountId(request.getAccountId()); // postId가 제공된 경우 계정 ID 검증 if (request.getPostId() != null) { @@ -421,33 +381,21 @@ private List getRealtimePostCommentsInternal(Long userId, SnsPost post = snsPostRepositoryPort.findById(targetPostId) .orElseThrow(() -> new BusinessException(AnalyticsErrorCode.POST_NOT_FOUND)); - // 외부 API에서 댓글 조회 (개수 제한 적용) - - List comments = externalApiPort.getVideoCommentsWithLimit(post.getSnsPostId(), request.getSize()); - log.info("Retrieved comments from external API for postId: {}, comment count: {}", targetPostId, comments.size()); - - // 페이지네이션 적용 - int start = request.getPage() * request.getSize(); - int end = Math.min(start + request.getSize(), comments.size()); - - if (start >= comments.size()) { - return List.of(); // 빈 목록 반환 - } + // 외부 API에서 댓글 조회 (YouTube API 네이티브 페이지네이션 사용) + PostCommentsPageResponse response = externalApiPort.getVideoCommentsWithPagination( + post.getSnsPostId(), + request.getPageToken(), + request.getSize() + ); - List paginatedComments = new ArrayList<>(comments.subList(start, end)); - log.info("Applied pagination for postId: {}, page: {}, size: {}, result count: {}", - targetPostId, request.getPage(), request.getSize(), paginatedComments.size()); + log.info("Retrieved comments from external API for postId: {}, comment count: {}, hasNextPage: {}, nextPageToken: {}", + targetPostId, response.getCurrentPageSize(), response.isHasNextPage(), response.getNextPageToken()); - return paginatedComments; + return response; } // 헬퍼 메서드들 - - - - - /** * 계정 ID로 최근 게시물 ID 조회 @@ -539,27 +487,6 @@ private EmotionAnalysisResponse buildEmotionAnalysisResponse(Long postId, List toSnsAccountMetricsResponseFromJoin(List fetchCommentsFromAPI(SnsPost post, Long postI postId, pageCount, pageToken != null ? "있음" : "없음"); // ExternalApiPort를 통해 댓글 조회 (페이지네이션 지원) - List pageComments = externalApiPort.getVideoCommentsWithPagination(post.getSnsPostId(), pageToken); + PostCommentsPageResponse pageResponse = externalApiPort.getVideoCommentsWithPagination(post.getSnsPostId(), pageToken, 100); + List pageComments = pageResponse.getData(); if (pageComments.isEmpty()) { log.info("📄 빈 페이지 - postId: {}, 페이지: {}", postId, pageCount); @@ -347,7 +349,7 @@ private List fetchCommentsFromAPI(SnsPost post, Long postI } // 다음 페이지 토큰 가져오기 - pageToken = externalApiPort.getNextPageToken(post.getSnsPostId(), pageToken); + pageToken = pageResponse.getNextPageToken(); } while (pageToken != null);