Loading...
Loading...
02-reusable-code-python/news/reporter.py
"""
뉴스 인텔리전스 마크다운 보고서 생성기.
AnalysisReport를 기반으로 트렌드·기회·요약이 포함된
마크다운 보고서를 생성하고 파일로 저장한다.
@source: 00-general-pro
@extracted: 2026-03-08
@version: 1.0.0
의존성:
- pydantic >= 2.0
사용법:
```python
from news.reporter import ReportGenerator
from news.models import AnalysisReport
generator = ReportGenerator(report_dir="docs/intel")
report_path = generator.save_report(analysis_report)
card_paths = generator.save_opportunity_cards(
analysis_report
)
print(f"보고서: {report_path}")
```
"""
import logging
import re
from datetime import datetime
from pathlib import Path
from .models import (
AnalysisReport,
AppOpportunity,
TrendSignal,
)
logger = logging.getLogger(__name__)
# ============================================
# 보고서 생성기
# ============================================
class ReportGenerator:
"""뉴스 인텔리전스 마크다운 보고서 생성기.
AnalysisReport를 마크다운 형식으로 변환하여
전체 보고서 및 개별 기회 카드를 파일로 저장한다.
Attributes:
report_dir: 보고서 저장 디렉토리 경로.
"""
def __init__(self, report_dir: str = "docs/intel") -> None:
"""초기화.
Args:
report_dir: 보고서 저장 디렉토리. 기본 docs/intel.
"""
self.report_dir = Path(report_dir)
def generate_full_report(
self, report: AnalysisReport
) -> str:
"""전체 마크다운 보고서를 생성한다.
Args:
report: 종합 분석 보고서.
Returns:
마크다운 형식의 전체 보고서 문자열.
"""
sections = [
self._format_summary(report),
self._format_trend_section(report.trends),
self._format_opportunity_section(
report.opportunities
),
self._format_next_steps(report),
]
return "\n\n".join(sections)
def save_report(self, report: AnalysisReport) -> str:
"""보고서를 파일로 저장한다.
{report_dir}/reports/YYYY-MM-DD-intel-report.md 형식.
Args:
report: 종합 분석 보고서.
Returns:
저장된 파일 경로 문자열.
"""
reports_dir = self.report_dir / "reports"
reports_dir.mkdir(parents=True, exist_ok=True)
date_str = datetime.now().strftime("%Y-%m-%d")
filename = f"{date_str}-intel-report.md"
filepath = reports_dir / filename
content = self.generate_full_report(report)
filepath.write_text(content, encoding="utf-8")
logger.info("보고서 저장 완료: %s", filepath)
return str(filepath)
def save_opportunity_cards(
self, report: AnalysisReport
) -> list[str]:
"""각 기회를 개별 마크다운 카드로 저장한다.
{report_dir}/opportunities/YYYY-MM-DD-{kebab-title}.md
Args:
report: 종합 분석 보고서.
Returns:
저장된 카드 파일 경로 목록.
"""
opps_dir = self.report_dir / "opportunities"
opps_dir.mkdir(parents=True, exist_ok=True)
date_str = datetime.now().strftime("%Y-%m-%d")
saved_paths: list[str] = []
for opp in report.opportunities:
kebab_title = self._to_kebab_case(opp.title)
filename = f"{date_str}-{kebab_title}.md"
filepath = opps_dir / filename
content = self._format_opportunity_card(opp)
filepath.write_text(content, encoding="utf-8")
saved_paths.append(str(filepath))
logger.info("기회 카드 저장: %s", filepath)
return saved_paths
# ============================================
# 섹션 포매터
# ============================================
def _format_summary(
self, report: AnalysisReport
) -> str:
"""보고서 헤더와 핵심 지표 요약을 생성한다.
Args:
report: 종합 분석 보고서.
Returns:
요약 섹션 마크다운 문자열.
"""
domains_str = (
", ".join(report.domains) if report.domains
else "general"
)
generated_str = report.generated_at.strftime(
"%Y-%m-%d %H:%M"
)
return (
"# 뉴스 인텔리전스 보고서\n\n"
f"> 생성일: {generated_str}"
f" | 도메인: {domains_str}"
f" | 기간: {report.period_days}일"
f" | 수집 항목: {report.raw_item_count}개\n\n"
"## 핵심 요약\n\n"
"| 지표 | 값 |\n"
"|------|-----|\n"
f"| 수집 항목 수 | {report.raw_item_count} |\n"
f"| 식별된 트렌드 | {len(report.trends)} |\n"
f"| 도출된 기회 | {len(report.opportunities)} |\n"
f"| 분석 소요 시간"
f" | {report.analysis_duration_seconds}초 |"
)
def _format_trend_section(
self, trends: list[TrendSignal]
) -> str:
"""트렌드 테이블과 상세 설명을 생성한다.
Args:
trends: 트렌드 시그널 목록.
Returns:
트렌드 섹션 마크다운 문자열.
"""
if not trends:
return "## 주요 트렌드\n\n_트렌드가 감지되지 않았습니다._"
lines = [
"## 주요 트렌드\n",
"| # | 키워드 | 빈도 | 소스 | 카테고리 | 단계 |",
"|---|--------|------|------|---------|------|",
]
for i, trend in enumerate(trends, 1):
sources_str = ", ".join(trend.sources)
category = trend.category or "-"
stage = self._translate_stage(trend.stage)
lines.append(
f"| {i} | {trend.keyword}"
f" | {trend.frequency}"
f" | {sources_str}"
f" | {category}"
f" | {stage} |"
)
# 성장률 상위 5개 하이라이트
growing = sorted(
[t for t in trends if t.growth_rate > 0],
key=lambda t: t.growth_rate,
reverse=True,
)[:5]
if growing:
lines.append("")
lines.append("### 급성장 키워드")
lines.append("")
for t in growing:
lines.append(
f"- **{t.keyword}**:"
f" 성장률 {t.growth_rate:+.0%}"
f" (빈도 {t.frequency},"
f" {self._translate_stage(t.stage)})"
)
return "\n".join(lines)
def _format_opportunity_section(
self, opps: list[AppOpportunity]
) -> str:
"""기회 카드와 점수 비교 테이블을 생성한다.
Args:
opps: 앱 기회 목록.
Returns:
기회 섹션 마크다운 문자열.
"""
if not opps:
return (
"## 앱 기회\n\n"
"_도출된 기회가 없습니다._"
)
lines = ["## 앱 기회\n"]
for i, opp in enumerate(opps, 1):
lines.append(
f"### {i}."
f" {opp.title}"
f" ({opp.total_score:.0f}점)\n"
)
lines.append(
"| 차원 | 점수 |"
)
lines.append("|------|------|")
lines.append(
f"| 기술 실현성"
f" | {opp.tech_feasibility:.1f}/10 |"
)
lines.append(
f"| 시장 수요"
f" | {opp.market_demand:.1f}/10 |"
)
lines.append(
f"| 차별화"
f" | {opp.differentiation:.1f}/10 |"
)
lines.append(
f"| 타이밍"
f" | {opp.timing:.1f}/10 |"
)
lines.append(
f"| 확장성"
f" | {opp.scalability:.1f}/10 |"
)
lines.append("")
lines.append(opp.description)
lines.append("")
lines.append(
f"**수요 근거:**"
f" {'; '.join(opp.demand_signals)}"
)
lines.append(
f"**추천 스택:**"
f" {', '.join(opp.suggested_stack)}"
)
lines.append(
f"**관련 트렌드:**"
f" {', '.join(opp.related_trends)}"
)
lines.append("")
lines.append("---")
lines.append("")
return "\n".join(lines)
def _format_opportunity_card(
self, opp: AppOpportunity
) -> str:
"""개별 기회 카드 마크다운을 생성한다.
Args:
opp: 앱 기회.
Returns:
기회 카드 마크다운 문자열.
"""
date_str = datetime.now().strftime("%Y-%m-%d")
return (
f"# {opp.title}\n\n"
f"> 생성일: {date_str}"
f" | 종합 점수: {opp.total_score:.0f}점\n\n"
"## 점수 매트릭스\n\n"
"| 차원 | 점수 |\n"
"|------|------|\n"
f"| 기술 실현성"
f" | {opp.tech_feasibility:.1f}/10 |\n"
f"| 시장 수요"
f" | {opp.market_demand:.1f}/10 |\n"
f"| 차별화"
f" | {opp.differentiation:.1f}/10 |\n"
f"| 타이밍"
f" | {opp.timing:.1f}/10 |\n"
f"| 확장성"
f" | {opp.scalability:.1f}/10 |\n\n"
"## 설명\n\n"
f"{opp.description}\n\n"
"## 수요 근거\n\n"
+ "\n".join(
f"- {s}" for s in opp.demand_signals
)
+ "\n\n"
"## 추천 스택\n\n"
+ ", ".join(opp.suggested_stack)
+ "\n\n"
"## 관련 트렌드\n\n"
+ ", ".join(opp.related_trends)
+ "\n\n"
"## 경쟁 제품\n\n"
+ (
"\n".join(
f"- {p}"
for p in opp.competing_products
)
if opp.competing_products
else "_아직 식별된 경쟁 제품 없음_"
)
+ "\n"
)
def _format_next_steps(
self, report: AnalysisReport
) -> str:
"""다음 단계 섹션을 생성한다.
Args:
report: 종합 분석 보고서.
Returns:
다음 단계 섹션 마크다운 문자열.
"""
domain_str = (
report.domains[0] if report.domains
else "ai"
)
return (
"## 다음 단계\n\n"
"- `/kdyintel --market` — 마케팅 전략 분석\n"
f"- `/kdyidea --domain {domain_str}`"
" — 아이디어 정제\n"
"- `/kdygenesis` — 프로젝트 생성"
)
# ============================================
# 유틸리티
# ============================================
def _to_kebab_case(self, text: str) -> str:
"""문자열을 kebab-case로 변환한다.
Args:
text: 원본 문자열.
Returns:
kebab-case 문자열 (최대 50자).
"""
# 특수문자 → 하이픈, 연속 하이픈 정리
cleaned = re.sub(r"[^a-zA-Z0-9가-힣\s-]", "", text)
kebab = re.sub(r"[\s_]+", "-", cleaned.strip())
kebab = re.sub(r"-+", "-", kebab)
kebab = kebab.strip("-").lower()
return kebab[:50] if kebab else "untitled"
def _translate_stage(self, stage: str) -> str:
"""트렌드 단계 영문을 한국어로 변환한다.
Args:
stage: 영문 트렌드 단계.
Returns:
한국어 트렌드 단계.
"""
stage_map = {
"emerging": "신흥",
"growing": "성장",
"mainstream": "주류",
"declining": "하락",
"unknown": "미분류",
}
return stage_map.get(stage, stage)
# ============================================
# CLI 엔트리포인트
# ============================================
def _parse_args() -> "argparse.Namespace":
"""CLI 인수를 파싱한다."""
import argparse
parser = argparse.ArgumentParser(
description="뉴스 인텔리전스 보고서 생성기",
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog=(
"예시:\n"
" uv run -m news.reporter"
" --input analysis.json"
" --output-dir docs/intel\n"
" cat analysis.json | uv run -m news.reporter"
" --stdout"
),
)
parser.add_argument(
"--input",
type=str,
default=None,
help="분석 결과 JSON 파일 (기본: stdin)",
)
parser.add_argument(
"--format",
type=str,
default="markdown",
choices=["markdown"],
help="출력 형식 (기본: markdown)",
)
parser.add_argument(
"--output-dir",
type=str,
default="docs/intel",
help="보고서 저장 디렉토리 (기본: docs/intel)",
)
parser.add_argument(
"--stdout",
action="store_true",
default=False,
help="파일 저장 대신 stdout 출력",
)
return parser.parse_args()
def _main() -> None:
"""동기 메인 함수."""
import json
import sys
from pathlib import Path
args = _parse_args()
# 입력 읽기
if args.input:
input_path = Path(args.input)
raw_json = input_path.read_text(encoding="utf-8")
else:
raw_json = sys.stdin.read()
data = json.loads(raw_json)
analysis_report = AnalysisReport.model_validate(data)
# 보고서 생성
generator = ReportGenerator(report_dir=args.output_dir)
if args.stdout:
# stdout 출력
content = generator.generate_full_report(
analysis_report
)
sys.stdout.write(content + "\n")
else:
# 파일 저장
report_path = generator.save_report(
analysis_report
)
card_paths = generator.save_opportunity_cards(
analysis_report
)
print(f"보고서 저장: {report_path}")
for path in card_paths:
print(f"기회 카드: {path}")
if __name__ == "__main__":
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
)
_main()