"""reporter.py - autoresearch 완료 보고서 생성 모듈""" from pathlib import Path from typing import Any # Sonnet 단가: input $3/M, output $15/M _SONNET_INPUT_RATE = 3.0 _SONNET_OUTPUT_RATE = 15.0 # Haiku 단가: input $0.80/M, output $4/M _HAIKU_INPUT_RATE = 0.80 _HAIKU_OUTPUT_RATE = 4.0 def estimate_cost( input_tokens: int, output_tokens: int, model: str = "sonnet", ) -> float: """입출력 토큰 수를 달러 비용으로 환산. Args: input_tokens: 입력 토큰 수 output_tokens: 출력 토큰 수 model: 모델 종류 ("sonnet" 또는 "haiku"). 기본값 "sonnet" Returns: 예상 비용 (USD) """ if model.lower() == "haiku": input_rate = _HAIKU_INPUT_RATE output_rate = _HAIKU_OUTPUT_RATE else: # 기본값: sonnet input_rate = _SONNET_INPUT_RATE output_rate = _SONNET_OUTPUT_RATE cost = (input_tokens / 1_000_000) * input_rate + (output_tokens / 1_000_000) * output_rate return cost def _fmt_score(score: float) -> str: return f"{score:.2f}" def _fmt_score_delta(before: float, after: float) -> str: delta = after - before sign = "+" if delta >= 0 else "" return f"{_fmt_score(before)} → {_fmt_score(after)} ({sign}{delta:.2f})" def _calc_change_rate(initial: float, final: float) -> str: """초기→최종 변화율 문자열. 초기값이 0이면 N/A 반환.""" if initial == 0.0: return "N/A" rate = (final - initial) / initial * 100 sign = "+" if rate >= 0 else "" return f"{sign}{rate:.1f}%" def _fmt_number(n: int) -> str: """숫자를 천 단위 쉼표 포맷으로 변환.""" return f"{n:,}" def generate_report(log_data: dict[str, Any], output_path: str) -> str: """changelog JSON → 마크다운 보고서 변환 후 파일로 저장. Args: log_data: changelog.py의 create_log/add_round/finalize_log로 만들어진 dict output_path: 보고서 저장 경로 Returns: 저장된 보고서 파일의 절대 경로 """ skill: str = log_data.get("skill", "unknown") started_at: str = log_data.get("started_at", "") ended_at: str = log_data.get("ended_at", "") rounds: list[dict[str, Any]] = log_data.get("rounds", []) final_score: float = float(log_data.get("final_score", 0.0)) total_rounds: int = int(log_data.get("total_rounds", 0)) # 시작/종료 시간 포맷팅 (ISO → 사람이 읽기 쉬운 형식) started_display = _format_datetime(started_at) ended_display = _format_datetime(ended_at) if ended_at else "-" # 초기 점수: 첫 번째 라운드의 score_before initial_score: float = 0.0 if rounds: initial_score = float(rounds[0].get("score_before", 0.0)) change_rate = _calc_change_rate(initial_score, final_score) # kept / reverted 분리 kept_rounds = [r for r in rounds if r.get("decision") == "kept"] reverted_rounds = [r for r in rounds if r.get("decision") == "reverted"] # 토큰 합산 total_input_tokens = sum(int(r.get("input_tokens", 0)) for r in rounds) total_output_tokens = sum(int(r.get("output_tokens", 0)) for r in rounds) # Mutation + Execution = Sonnet, Judge = Haiku (세분화 비용 계산) sonnet_input = sum(int(r.get("mutation_input_tokens", 0)) + int(r.get("execution_input_tokens", 0)) for r in rounds) sonnet_output = sum( int(r.get("mutation_output_tokens", 0)) + int(r.get("execution_output_tokens", 0)) for r in rounds ) haiku_input = sum(int(r.get("judge_input_tokens", 0)) for r in rounds) haiku_output = sum(int(r.get("judge_output_tokens", 0)) for r in rounds) # 하위호환: 새 필드가 없으면 기존 방식 (전부 Sonnet) if sonnet_input == 0 and haiku_input == 0: total_cost = estimate_cost(total_input_tokens, total_output_tokens, model="sonnet") cost_breakdown = "(Sonnet 기준, 추정치)" else: sonnet_cost = estimate_cost(sonnet_input, sonnet_output, model="sonnet") haiku_cost = estimate_cost(haiku_input, haiku_output, model="haiku") total_cost = sonnet_cost + haiku_cost cost_breakdown = f"(Sonnet: ${sonnet_cost:.4f} + Haiku: ${haiku_cost:.4f})" lines: list[str] = [] # 제목 lines.append(f"# Autoresearch 완료 보고서: {skill}") lines.append("") # 요약 섹션 lines.append("## 요약") lines.append(f"- 시작: {started_display}") lines.append(f"- 종료: {ended_display}") lines.append(f"- 총 라운드: {total_rounds}") lines.append(f"- 초기 점수: {_fmt_score(initial_score)} → 최종 점수: {_fmt_score(final_score)} ({change_rate})") lines.append("") # 유지된 변경 섹션 lines.append(f"## 유지된 변경 ({len(kept_rounds)}건)") if kept_rounds: lines.append("| # | 유형 | 설명 | 점수 변화 |") lines.append("|---|------|------|----------|") for idx, r in enumerate(kept_rounds, start=1): rtype = r.get("mutation_type", "") rdesc = r.get("mutation_description", "") sbefore = float(r.get("score_before", 0.0)) safter = float(r.get("score_after", 0.0)) lines.append(f"| {idx} | {rtype} | {rdesc} | {_fmt_score_delta(sbefore, safter)} |") else: lines.append("없음") lines.append("") # 롤백된 변경 섹션 lines.append(f"## 롤백된 변경 ({len(reverted_rounds)}건)") if reverted_rounds: lines.append("| # | 유형 | 설명 | 점수 변화 |") lines.append("|---|------|------|----------|") for idx, r in enumerate(reverted_rounds, start=1): rtype = r.get("mutation_type", "") rdesc = r.get("mutation_description", "") sbefore = float(r.get("score_before", 0.0)) safter = float(r.get("score_after", 0.0)) lines.append(f"| {idx} | {rtype} | {rdesc} | {_fmt_score_delta(sbefore, safter)} |") else: lines.append("없음") lines.append("") # 비용 섹션 lines.append("## 비용") lines.append(f"- 입력 토큰: {_fmt_number(total_input_tokens)}") lines.append(f"- 출력 토큰: {_fmt_number(total_output_tokens)}") lines.append(f"- 예상 비용: ${total_cost:.4f} {cost_breakdown}") lines.append("") report_content = "\n".join(lines) # 부모 디렉토리 자동 생성 후 저장 output = Path(output_path) output.parent.mkdir(parents=True, exist_ok=True) output.write_text(report_content, encoding="utf-8") return str(output.resolve()) def _format_datetime(iso_str: str) -> str: """ISO 8601 타임스탬프 → 사람이 읽기 쉬운 포맷 변환. 예) "2026-03-26T00:00:00+00:00" → "2026-03-26 00:00 UTC" 파싱 실패 시 원본 문자열 반환. """ if not iso_str: return "-" try: from datetime import datetime, timezone dt = datetime.fromisoformat(iso_str) dt_utc = dt.astimezone(timezone.utc) return dt_utc.strftime("%Y-%m-%d %H:%M UTC") except (ValueError, AttributeError): return iso_str