[HCR-226] analysis-server-v0.0.5

.dockerignore

@@ -11,3 +11,8 @@ __pycache__
 .env
 tests
 README.md
+node_modules
+package-lock.json
+pnpm-lock.yaml
+e2e_test_efs
+e2e_test_efs_agg

.env.example

@@ -0,0 +1,36 @@
+# Runtime
+APP_ENV=local
+APP_NAME=recommendation-server
+APP_MODE=realtime
+APP_PORT=8000
+LOG_LEVEL=INFO
+
+# Database (Java 레포와 동일 DB)
+# 1) 명시 URL (권장: 로컬 개발)
+DATABASE_URL=postgresql+asyncpg://postgres:postgres@localhost:5432/holliverse
+# DB_URL도 동일 의미로 사용 가능
+
+# 2) ECS/Secrets Manager 분해 주입 (POSTGRES_DSN 미설정 시 자동 조합)
+# - Secret JSON key(username/password)를 POSTGRES_USER/POSTGRES_PASSWORD로 매핑해 주입
+# - host/port/db는 일반 env로 주입
+# POSTGRES_HOST=your-instance.xxxxx.ap-northeast-2.rds.amazonaws.com
+# POSTGRES_PORT=5432
+# POSTGRES_DB=holliverse
+# POSTGRES_USER=holliverse
+# POSTGRES_PASSWORD=*****
+# POSTGRES_SSLMODE=require
+
+# 3) 배치용 raw DSN (asyncpg)
+# POSTGRES_DSN=postgresql://USER:PASSWORD@HOST:5432/DBNAME?sslmode=require
+
+# RDS 연결 시 SSL 필요하면 true (기본 false)
+DATABASE_SSL=false
+
+# OpenAI
+OPENAI_API_KEY=your_openai_api_key
+OPENAI_CHAT_MODEL=gpt-4o-mini
+OPENAI_EMBEDDING_MODEL=text-embedding-3-small
+
+# Recommendation
+RECOMMEND_TOP_K=3
+CACHE_TTL_DAYS=7

.gitignore

@@ -1,5 +1,29 @@
+# IDE 
 .idea
-.venv
-.env
+
+# Python
 __pycache__/
 *.pyc
+*.pyo
+*.pyd
+dist/
+build/
+*.egg-info/
+.eggs/
+
+# Environment
+.venv
+.env
+
+# Logs 
+*.log
+logs/
+
+# Node
+node_modules/
+
+# OS
+.DS_Store
+
+# E2E test data (keep out of git)
+e2e_test_efs/**

.husky/pre-commit

@@ -0,0 +1 @@
+pnpm test

Dockerfile

@@ -1,14 +1,49 @@
-FROM python:3.11-slim
+FROM python:3.11-slim AS builder
 
-WORKDIR /app
+WORKDIR /build
 
 ENV PYTHONDONTWRITEBYTECODE=1 \
-    PYTHONUNBUFFERED=1
+    PYTHONUNBUFFERED=1 \
+    PIP_DISABLE_PIP_VERSION_CHECK=1 \
+    PIP_NO_CACHE_DIR=1
 
 COPY requirements.txt ./
-RUN pip install --no-cache-dir -r requirements.txt
+
+RUN apt-get update \
+    && apt-get install -y --no-install-recommends build-essential python3-dev \
+    && pip wheel --wheel-dir /wheels -r requirements.txt \
+    && rm -rf /var/lib/apt/lists/*
+
+
+FROM python:3.11-slim AS runtime
+
+WORKDIR /app
+
+ENV PYTHONDONTWRITEBYTECODE=1 \
+    PYTHONUNBUFFERED=1 \
+    PIP_DISABLE_PIP_VERSION_CHECK=1 \
+    PIP_NO_CACHE_DIR=1 \
+    APP_MODE=realtime \
+    APP_HOST=0.0.0.0 \
+    APP_PORT=8000
+
+RUN addgroup --system appgroup \
+    && adduser --system --ingroup appgroup --home /app appuser
+
+COPY --from=builder /wheels /wheels
+RUN pip install --no-cache-dir /wheels/* \
+    && python -m spacy download ko_core_news_sm \
+    && rm -rf /wheels
 
 COPY app ./app
+COPY scripts ./scripts
+COPY docker-entrypoint.sh ./docker-entrypoint.sh
+
+RUN chmod +x /app/docker-entrypoint.sh \
+    && chown -R appuser:appgroup /app
+
+USER appuser
 
 EXPOSE 8000
-CMD ["python", "-m", "uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]
+
+ENTRYPOINT ["/app/docker-entrypoint.sh"]

README.md

@@ -1,5 +1,5 @@
 # counseling-analytics
-
+하이
 FastAPI 개발 서버 실행 가이드입니다.
 
 ## Prerequisites
@@ -56,7 +56,13 @@ pip install -r requirements.txt
 ## 5) FastAPI 서버 실행
 
 ```bash
-python -m uvicorn app.main:app --reload --host 0.0.0.0 --port 8000
+python -m uvicorn app.realtime.main:app --reload --host 0.0.0.0 --port 8000
+```
+
+## 5-1) 배치 1회 실행
+
+```bash
+python -m app.batch.main
 ```
 
 ## 6) 접속 확인

app/batch/main.py

@@ -0,0 +1,88 @@
+"""One-off Kafka batch entrypoint for keyword mapping/extraction."""
+
+from __future__ import annotations
+
+import asyncio
+import json
+import logging
+
+from aiokafka import AIOKafkaConsumer
+
+from app.core.config import get_settings
+from app.core.logging import configure_logging
+from app.infra.postgres.analysis_repository import AnalysisRepository
+from app.infra.postgres.client import create_postgres_pool
+from app.infra.postgres.dispatch_outbox_repository import DispatchOutboxRepository
+from app.schemas.analysis_request_message import AnalysisRequestMessage
+from app.services.kafka_analysis_consumer_service import KafkaAnalysisConsumerService
+from app.services.sql_keyword_analysis_service import SqlKeywordAnalysisService
+
+logger = logging.getLogger(__name__)
+
+
+async def run_once() -> int:
+    settings = get_settings()
+    configure_logging(settings.debug)
+
+    service = KafkaAnalysisConsumerService(settings)
+    service._db_pool = await create_postgres_pool(settings)  # noqa: SLF001
+    service._analysis_repository = AnalysisRepository(service._db_pool)  # noqa: SLF001
+    service._outbox_repository = DispatchOutboxRepository(service._db_pool)  # noqa: SLF001
+    service._analysis_service = SqlKeywordAnalysisService()  # noqa: SLF001
+
+    consumer = AIOKafkaConsumer(
+        settings.kafka_analysis_request_topic,
+        bootstrap_servers=[s.strip() for s in settings.kafka_bootstrap_servers.split(",") if s.strip()],
+        group_id=settings.kafka_consumer_group_id,
+        auto_offset_reset=settings.kafka_auto_offset_reset,
+        enable_auto_commit=False,
+        value_deserializer=lambda value: json.loads(value.decode("utf-8")),
+    )
+
+    processed_count = 0
+    received_count = 0
+    dropped_count = 0
+    try:
+        await consumer.start()
+        polled = await consumer.getmany(
+            timeout_ms=settings.kafka_poll_timeout_ms,
+            max_records=settings.kafka_batch_size,
+        )
+
+        messages: list[AnalysisRequestMessage] = []
+        for _, records in polled.items():
+            for record in records:
+                received_count += 1
+                parsed = service._parse_message(record.value)  # noqa: SLF001
+                if parsed is None:
+                    dropped_count += 1
+                    continue
+                messages.append(parsed)
+
+        if messages:
+            for chunk in service._chunk(messages, settings.kafka_batch_size):  # noqa: SLF001
+                await service._process_batch(chunk)  # noqa: SLF001
+                processed_count += len(chunk)
+            await consumer.commit()
+        elif received_count > 0 and dropped_count == received_count:
+            await consumer.commit()
+
+        logger.info(
+            "Keyword batch run finished once. received=%d dropped=%d processed=%d",
+            received_count,
+            dropped_count,
+            processed_count,
+        )
+        return processed_count
+    finally:
+        await consumer.stop()
+        if service._db_pool is not None:  # noqa: SLF001
+            await service._db_pool.close()  # noqa: SLF001
+
+
+def main() -> None:
+    asyncio.run(run_once())
+
+
+if __name__ == "__main__":
+    main()