[DP-386] 트렌드 배치 오케스트레이터 + 스케줄러 구현

CLAUDE.md

@@ -41,6 +41,11 @@
   - **퀴즈 레벨별 출제 방향 차별화 + 누락 레벨 재시도 로직 추가 (DP-265)**
   - **주간 인사이트 키워드 분석 강화: UserRepository(user_tags+content_tags) + 미탐색 태그 기반 추천 글 + Sonnet 전환 (DP-254)**
   - **영어 제목 번역 후 translated_title을 DynamoDB ai_summaries + PostgreSQL contents에 저장 (DP-328)**
+  - **태그 정규화(TagNormalizer) + 빈도 집계(FrequencyAnalyzer) + TF-IDF/tokenizer 구현 (DP-380, DP-381)**
+  - **트렌드 랭킹 TrendRanker + TrendDataLoader + TrendSnapshotRepository 구현 (DP-378, DP-379, DP-383)**
+  - **Top 5 콘텐츠 LLM 서사 요약 TopPostsSummaryGenerator 구현 (DP-404)**
+  - **수집 동향 LLM 서사 요약 CollectionSummaryGenerator + TrendSignals 구현 (DP-384)**
+  - **트렌드 배치 오케스트레이터 TrendOrchestrator + run_trend_batch.py + run_trend_scheduler.py 구현 (DP-386)**
 
 ---
 
@@ -67,6 +72,7 @@
 2. **PostgreSQL 직접 저장** — `ContentRepository` (Backend push 없음)
 3. **AI 요약 + 퀴즈 자동 생성** — `ContentPipeline`: 저장 직후 4레벨 요약 + 4레벨 퀴즈 생성 → DynamoDB. 요약 성공 시 tags·category → PostgreSQL UPDATE
 4. **AI 질문/답변/리포트** — RefineService, AnswerService, InsightService
+5. **트렌드 분석 배치** — `TrendOrchestrator`: 일/주/월 단위 태그 빈도·TF-IDF·LLM 서사 요약 → `trend_snapshots` PostgreSQL 저장
 5. **출력 JSON 스키마 검증 + 파싱 실패 대응**
 6. **캐시/저장/로그 기록**
 
@@ -179,6 +185,13 @@ DATABASE_URL=postgresql://... python scripts/reprocess_quiz.py <content_id>
 # DynamoDB → PostgreSQL tags/category 동기화
 DATABASE_URL=postgresql://... python scripts/sync_ai_metadata.py
 
+# 트렌드 분석 1회 실행 (unit: daily/weekly/monthly)
+DATABASE_URL=postgresql://... python scripts/run_trend_batch.py --unit weekly
+DATABASE_URL=postgresql://... python scripts/run_trend_batch.py --unit daily --force
+
+# 트렌드 스케줄러 (daily 00:05 / weekly 월 00:10 / monthly 1일 00:15 KST)
+DATABASE_URL=postgresql://... python scripts/run_trend_scheduler.py
+
 # 테스트
 pytest -q
 ```

README.md

@@ -112,6 +112,8 @@ ContentPipeline.process_content()
 | `scripts/run_backfill_batch.py` | 1회 수집 실행 — PostgreSQL 저장 + AI 처리 (요약+퀴즈+임베딩) |
 | `scripts/run_scheduler.py` | 6시간 간격 자동 반복 실행 |
 | `scripts/run_collect_and_save.py` | 로컬 JSONL 저장 전용 (AI 처리 없음, 개발용) |
+| `scripts/run_trend_batch.py` | 트렌드 분석 1회 실행 (`--unit daily/weekly/monthly`, `--force`) |
+| `scripts/run_trend_scheduler.py` | 트렌드 분석 자동 실행 (daily 00:05 / weekly 월 00:10 / monthly 1일 00:15 KST) |
 | `scripts/init_postgres.py` | PostgreSQL UNIQUE 인덱스 초기화 (배포 시 1회) |
 | `scripts/init_vectors.py` | FAISS 인덱스 초기화 |
 | `scripts/reindex_vectors.py` | FAISS 인덱스 재빌드 (인덱스 유실 시) |
@@ -129,13 +131,39 @@ DATABASE_URL=postgresql://... python scripts/run_backfill_batch.py
 # 스케줄러 (6시간 간격 자동 반복)
 DATABASE_URL=postgresql://... python scripts/run_scheduler.py
 
+# 트렌드 분석 1회 실행
+DATABASE_URL=postgresql://... python scripts/run_trend_batch.py --unit weekly
+DATABASE_URL=postgresql://... python scripts/run_trend_batch.py --unit daily --force
+
+# 트렌드 분석 스케줄러 (daily/weekly/monthly 자동 실행)
+DATABASE_URL=postgresql://... python scripts/run_trend_scheduler.py
+
 # 개발 서버
 uvicorn main:app --reload
 
 # FAISS 재빌드 (인덱스 유실 시)
 python scripts/reindex_vectors.py
 ```
 
+## 트렌드 분석 배치 파이프라인
+
+AI 서버가 일/주/월 단위로 태그 빈도·TF-IDF·LLM 서사 요약을 생성하고 PostgreSQL `trend_snapshots`에 저장한다. Backend는 이 테이블을 읽어 Frontend에 전달한다.
+
+```
+TrendOrchestrator.run(unit, period_start, period_end)
+    ├─ TrendDataLoader         — cur/prev 콘텐츠 + 조회수 4개 병렬 쿼리
+    ├─ TagNormalizer            — rapidfuzz 동의어 정규화
+    ├─ FrequencyAnalyzer        — 태그 빈도 집계 + 증감 상태 (new/up/down/same)
+    ├─ KoreanTokenizer + TfidfAnalyzer — TF-IDF 키워드 추출 (제목 기반)
+    ├─ TrendRanker              — 조회수 Top 5 콘텐츠 선정
+    ├─ TopPostsSummaryGenerator — Top 5 콘텐츠 LLM 서사 요약 (Bedrock)
+    ├─ CollectionSummaryGenerator — 수집 동향 LLM 서사 요약 (Bedrock, weekly/monthly만)
+    └─ TrendSnapshotRepository  — trend_snapshots upsert
+```
+
+- `force=False`이고 동일 기간 스냅샷이 이미 있으면 재생성 없이 즉시 반환
+- LLM 실패 시 해당 요약 필드만 `null`로 저장, 스냅샷 저장은 계속 진행
+
 ## DynamoDB 테이블 목록
 
 | 테이블 | 내용 |

app/core/CLAUDE.md

@@ -52,6 +52,8 @@ Anthropic Tool Use 스키마를 Bedrock Converse API `toolConfig` 형식으로 
 | `refine.py` | `SYSTEM_PROMPT`, `REFINE_TOOL`, `build_user_prompt()` — 레벨별 질문 개선 + 컨텍스트 청크 |
 | `answer.py` | `SYSTEM_PROMPT`, `ANSWER_TOOL`, `build_user_prompt()` — 아티클+RAG 컨텍스트 기반 답변 |
 | `insight.py` | `SYSTEM_PROMPT`, `INSIGHT_TOOL`, `build_user_prompt()` — 주간 활동/읽은글/스크랩/질문 기반 인사이트 |
+| `trend_top_posts.py` | `SYSTEM_PROMPT`, `TOOL_SAVE_TOP_POSTS_SUMMARY`, `build_user_prompt()` — Top 5 콘텐츠 주제 흐름 서사 요약 (DP-404) |
+| `trend_collection.py` | `SYSTEM_PROMPT`, `TOOL_SAVE_COLLECTION_SUMMARY`, `build_user_prompt()` — 수집 동향 서사 요약 (DP-384) |
 
 ### quiz.py 구성 요소 (DP-265)
 

app/repositories/CLAUDE.md

@@ -18,6 +18,7 @@ DynamoDB + PostgreSQL 접근 레이어. 각 도메인별 저장/조회 로직을
 | `question_vector_repository.py` | `QuestionVectorRepository` | DynamoDB | `rag_questions` | 질문 임베딩 upsert 저장 (DP-234) |
 | `event_repository.py` | `EventRepository` | DynamoDB | `event_logs` | AI 처리 이벤트 로그 + 일별 dedup (DP-252) |
 | `insight_repository.py` | `InsightRepository` | DynamoDB | `weekly_report_insights` | 주간 인사이트 upsert 저장 (DP-259) |
+| `trend_repository.py` | `TrendSnapshotRepository` | PostgreSQL | `trend_snapshots` | 트렌드 분석 결과 upsert/조회 (DP-378, DP-386) |
 
 ---
 

app/repositories/content_repository.py

@@ -221,12 +221,13 @@ def find_by_published_range(self, start: datetime, end: datetime) -> list[dict]:
         with self._engine.begin() as conn:
             result = conn.execute(
                 text("""
-                    SELECT id, title, translated_title, category, tags,
-                           source_id, published_at
-                    FROM contents
-                    WHERE published_at >= :start AND published_at < :end
-                      AND is_available = true
-                    ORDER BY published_at DESC
+                    SELECT c.id, c.title, c.translated_title, c.category, c.tags,
+                           cs.name AS source_name, c.thumbnail_url, c.published_at
+                    FROM contents c
+                    LEFT JOIN content_sources cs ON cs.id = c.source_id
+                    WHERE c.published_at >= :start AND c.published_at < :end
+                      AND c.is_available = true
+                    ORDER BY c.published_at DESC
                 """),
                 {"start": start, "end": end},
             )

app/services/CLAUDE.md

@@ -20,6 +20,15 @@
 | `question_embedding_service.py` | `QuestionEmbeddingOrchestrator` | 질문 임베딩 → DynamoDB rag_questions + FAISS questions 저장 (DP-234) |
 | `similar_question_service.py` | `SimilarQuestionService` | FAISS questions 인덱스 유사 질문 검색 (DP-235) |
 | `insight_service.py` | `InsightService` | Bedrock Tool Use 기반 주간 학습 인사이트 생성 (DP-259) |
+| `trend/normalize.py` | `TagNormalizer` | rapidfuzz 기반 태그 동의어 정규화 (DP-380) |
+| `trend/frequency.py` | `FrequencyAnalyzer` | 태그 빈도 집계 + 증감 상태 판정 (DP-380) |
+| `trend/tokenizer.py` | `KoreanTokenizer` | kiwipiepy 형태소 분석 기반 토크나이저 (DP-381) |
+| `trend/tfidf.py` | `TfidfAnalyzer` | scikit-learn TF-IDF 키워드 추출 (DP-381) |
+| `trend/ranking.py` | `TrendRanker` | 조회수 기반 Top 5 콘텐츠 + 복합 점수 Top 10 태그 선정 (DP-383) |
+| `trend/data_loader.py` | `TrendDataLoader` | cur/prev 기간 데이터 병렬 로드 (DP-379) |
+| `trend/top_posts_summary.py` | `TopPostsSummaryGenerator` | Top 5 콘텐츠 주제 흐름 LLM 서사 요약 (DP-404) |
+| `trend/collection_summary.py` | `CollectionSummaryGenerator` | 수집 동향 LLM 서사 요약 + `TrendSignals` 데이터클래스 (DP-384) |
+| `trend/orchestrator.py` | `TrendOrchestrator` | 일/주/월 트렌드 배치 오케스트레이터 + `compute_period()` (DP-386) |
 
 ---
 

app/services/trend/CLAUDE.md

@@ -0,0 +1,97 @@
+# CLAUDE.md — app/services/trend/
+
+트렌드 분석 파이프라인 서비스 모음. 태그 정규화 → 빈도 집계 → TF-IDF → 랭킹 → LLM 서사 요약 → 스냅샷 저장까지의 전 흐름을 담당한다.
+
+---
+
+## 파일 구성
+
+| 파일 | 클래스/함수 | 역할 |
+|------|------------|------|
+| `normalize.py` | `TagNormalizer` | rapidfuzz ratio 기반 태그 동의어 정규화 (DP-380) |
+| `frequency.py` | `FrequencyAnalyzer`, `TagFrequency` | cur/prev 태그 빈도 집계 + 증감 상태 판정 (DP-380) |
+| `tokenizer.py` | `KoreanTokenizer` | kiwipiepy NNG/NNP/SL POS 필터 기반 한국어 토크나이저 (DP-381) |
+| `tfidf.py` | `TfidfAnalyzer` | scikit-learn TfidfVectorizer 기반 상위 키워드 추출 (DP-381) |
+| `ranking.py` | `TrendRanker`, `RankedTag` | 기간 조회수 Top 5 콘텐츠 + 복합 점수 Top 10 태그 선정 (DP-383) |
+| `data_loader.py` | `TrendDataLoader`, `TrendRawData` | cur/prev 기간 4개 쿼리 병렬 로드 (DP-379, DP-386) |
+| `top_posts_summary.py` | `TopPostsSummaryGenerator` | Top 5 콘텐츠 주제 흐름 LLM 서사 요약 (DP-404) |
+| `collection_summary.py` | `CollectionSummaryGenerator`, `TrendSignals` | 수집 동향 LLM 서사 요약 (DP-384) |
+| `orchestrator.py` | `TrendOrchestrator`, `compute_period` | 배치 오케스트레이터 — 위 컴포넌트를 순서대로 조합 (DP-386) |
+
+---
+
+## 트렌드 배치 전체 흐름
+
+```
+compute_period(unit) → (period_start, period_end)
+    ↓
+TrendDataLoader.load(start, end)
+    ├─ find_by_published_range(start, end)       → cur_contents (source_name, thumbnail_url 포함)
+    ├─ find_by_published_range(prev_start, start) → prev_contents
+    ├─ find_view_counts_by_period(start, end)     → cur_view_counts
+    └─ find_view_counts_by_period(prev_start, start) → prev_view_counts
+    ↓
+TagNormalizer.normalize(cur_tags, prev_tags)
+    ↓
+FrequencyAnalyzer.analyze(cur_tags, prev_tags) → list[TagFrequency]
+    ↓
+KoreanTokenizer.tokenize(titles) → TfidfAnalyzer.extract() → tfidf_keywords[:15]
+    ↓
+TrendRanker.rank_contents(cur_view_counts, cur_contents) → top_contents_raw
+    ↓
+TopPostsSummaryGenerator.generate(top_contents_raw, unit, ..., prev_summary) → top_posts_summary
+CollectionSummaryGenerator.generate(TrendSignals(...)) → collection_summary
+    ↓
+TrendResponse 조립 → TrendSnapshotRepository.upsert()
+```
+
+---
+
+## TagFrequency / TrendSignals 데이터 계약
+
+### TagFrequency (frequency.py)
+```python
+@dataclass
+class TagFrequency:
+    keyword: str
+    cur_count: int
+    prev_count: int
+    delta: int
+    growth_rate: float | None  # None = state="new"
+    state: str  # "new" | "up" | "down" | "same"
+```
+
+### TrendSignals (collection_summary.py)
+```python
+@dataclass
+class TrendSignals:
+    unit: str                       # "daily" | "weekly" | "monthly"
+    period_start: str               # ISO 날짜 문자열
+    period_end: str
+    cur_content_count: int
+    prev_content_count: int = 0
+    top_tags: list[TagFrequency] = field(default_factory=list)
+    tfidf_keywords: list[str] = field(default_factory=list)
+    prev_summary: str | None = None  # 이전 기간 collection_summary
+```
+
+---
+
+## LLM 실패 처리 정책
+
+| 서비스 | 실패 시 동작 |
+|--------|------------|
+| `TopPostsSummaryGenerator` | `AIUpstreamError` / `AITimeoutError` 전파 → 오케스트레이터에서 catch → `top_posts_summary=None` |
+| `CollectionSummaryGenerator` | 내부 catch → `None` 반환 (daily 포함) |
+
+두 경우 모두 스냅샷 저장은 계속 진행된다.
+
+---
+
+## compute_period 단위별 로직
+
+| unit | 기준 | period_start | period_end |
+|------|------|-------------|------------|
+| daily | 어제 자정~오늘 자정 | `today - 1일` | `today` |
+| weekly | 이전 주 월~월 | `이번 주 월요일 - 7일` | `이번 주 월요일` |
+| monthly | 이전 달 1일~이번 달 1일 | `지난달 1일` | `이번달 1일` |

app/services/trend/data_loader.py

@@ -18,6 +18,7 @@ class TrendRawData:
     cur_contents: list[dict] = field(default_factory=list)
     cur_view_counts: dict[str, int] = field(default_factory=dict)
     prev_view_counts: dict[str, int] = field(default_factory=dict)
+    prev_contents: list[dict] = field(default_factory=list)
     summary_meta: dict[str, dict] = field(default_factory=dict)
 
 
@@ -45,10 +46,13 @@ def load(self, start: datetime, end: datetime) -> TrendRawData:
         delta = end - start
         prev_start = start - delta
 
-        with ThreadPoolExecutor(max_workers=3) as executor:
+        with ThreadPoolExecutor(max_workers=4) as executor:
             f_contents = executor.submit(
                 self._content_repo.find_by_published_range, start, end
             )
+            f_prev_contents = executor.submit(
+                self._content_repo.find_by_published_range, prev_start, start
+            )
             f_cur_views = executor.submit(
                 self._content_repo.find_view_counts_by_period, start, end
             )
@@ -57,6 +61,7 @@ def load(self, start: datetime, end: datetime) -> TrendRawData:
             )
 
         cur_contents = f_contents.result()
+        prev_contents = f_prev_contents.result()
         cur_view_counts = f_cur_views.result()
         prev_view_counts = f_prev_views.result()
 
@@ -68,8 +73,10 @@ def load(self, start: datetime, end: datetime) -> TrendRawData:
         )
 
         logger.debug(
-            "TrendDataLoader.load 완료: contents=%d cur_views=%d prev_views=%d meta=%d",
+            "TrendDataLoader.load 완료: contents=%d prev_contents=%d"
+            " cur_views=%d prev_views=%d meta=%d",
             len(cur_contents),
+            len(prev_contents),
             len(cur_view_counts),
             len(prev_view_counts),
             len(summary_meta),
@@ -79,5 +86,6 @@ def load(self, start: datetime, end: datetime) -> TrendRawData:
             cur_contents=cur_contents,
             cur_view_counts=cur_view_counts,
             prev_view_counts=prev_view_counts,
+            prev_contents=prev_contents,
             summary_meta=summary_meta,
         )