diff --git a/docs/lotto-premium-roadmap.md b/docs/lotto-premium-roadmap.md new file mode 100644 index 0000000..fa0f9d0 --- /dev/null +++ b/docs/lotto-premium-roadmap.md @@ -0,0 +1,252 @@ +# 로또랩 프리미엄 서비스 고도화 로드맵 + +> 작성일: 2026-03-19 +> 목표: 번호 생성 도구 → 데이터 기반 로또 전략 코치 + +--- + +## 1. 현재 서비스 한계 + +현재 구조는 **"번호 생성 도구"** 수준으로 수익화에 한계가 있음. + +| 문제 | 내용 | +|------|------| +| 차별점 부재 | 무료 로또 번호 생성기와 구분되지 않음 | +| 신뢰 근거 부족 | 사용자가 결과를 믿을 데이터 시각화 없음 | +| 리텐션 약함 | 지속적으로 돌아올 이유가 없음 | + +--- + +## 2. 포지셔닝 전환 + +> **"번호 생성"이 아니라 "데이터 기반 로또 전략 코치"** + +사람들이 구독료를 지불하는 심리적 동기: + +- **확신**: 내가 선택한 번호가 좋은 선택이라는 데이터 근거 +- **FOMO**: 이번 주 리포트를 못 받으면 놓치는 느낌 +- **소유감**: 내 데이터와 이력이 축적된다는 느낌 + +--- + +## 3. 고도화 방향 (5가지) + +### 3-1. 당첨 근접도 추적 — 신뢰 기반 구축 + +**목표**: 기존 채점 데이터(`check_results_for_draw`)를 신뢰 지표로 전환 + +**구현 내용**: +- 추천 번호의 회차별 일치 개수 통계 집계 +- 전국 평균 대비 성과 비교 지표 노출 +- 매주 "지난 주 내 번호 성과" 이메일/푸시 발송 + +**예시 UI 문구**: +``` +"지난 52주간 우리 추천번호의 평균 일치 개수: 2.7개 (전국 평균 1.9개)" +"3개 일치율이 일반 무작위 대비 43% 높습니다" +``` + +**활용 데이터**: 기존 `recommendations` + `draws` 테이블 채점 결과 + +**우선순위**: ⭐⭐⭐ (데이터 이미 존재, 즉시 구현 가능) + +--- + +### 3-2. 개인화 분석 리포트 — 프리미엄 핵심 기능 + +**목표**: 모든 사용자에게 동일한 번호 → 개인 패턴 기반 맞춤 추천 + +**구현 내용**: +- 사용자 번호 선택 이력 패턴 분석 +- 홀짝 비율, 번호대 분포, 연속번호 포함률 등 개인 성향 분석 +- 약점을 보완한 AI 보정 추천번호 생성 + +**예시 분석 항목**: +``` +"당신은 홀수를 선호하는 경향 (67%)" +"당신이 자주 피하는 번호대: 30번대" +"당신 번호의 약점: 연속번호 포함률 낮음" +→ "이를 보완한 AI 보정 추천번호 제공" +``` + +**신규 테이블**: `user_preferences` + +**우선순위**: ⭐⭐ (신규 테이블 및 분석 로직 필요) + +--- + +### 3-3. 회차별 공략 리포트 — 킬러 콘텐츠 + +**목표**: 매주 추첨 전 발행하는 주간 분석 레포트 → 구독 유지 동기 + +**구현 내용**: +- 매주 자동 생성되는 회차별 공략 리포트 +- 과출현/냉각 번호 분석 +- 패턴 기반 번호군 추천 +- AI 신뢰도 점수 표시 + +**예시 리포트 구조**: +``` +[1180회 공략 리포트] +- 최근 10회 과출현 번호 제외 추천 +- 이번 주 "냉각 구간" 번호 (오랫동안 미출현) +- 패턴 분석: 직전 3회 연속 출현한 번호군 +- AI 신뢰도 점수: 87/100 +``` + +**스케줄러**: 매주 토요일 추첨 전 자동 생성 (APScheduler) + +**우선순위**: ⭐⭐⭐ (주간 구독 모델의 핵심 훅) + +--- + +### 3-4. 번호 포트폴리오 관리 — 차별화 UX + +**목표**: 로또를 투자처럼 관리하는 경험 제공 + +**구현 내용**: +- 세트 분류: 고위험/안정형/균형형 +- 구매 금액 직접 입력 → 수익률 자동 계산 +- 누적 투자 대비 당첨금 통계 + +**예시 화면**: +``` +내 번호 포트폴리오 +├── 고위험/고수익 세트 (출현 빈도 낮은 번호 조합) +├── 안정형 세트 (평균 출현 패턴) +└── 균형형 세트 (시뮬레이션 최적화) + +이번 주 매입: 3세트 (₩3,000) +누적 투자: ₩240,000 / 누적 당첨: ₩45,000 +수익률: -81.2% (전국 평균 대비 +12.1%) +``` + +**활용 데이터**: `best_picks`, `recommendations` 확장 + +**우선순위**: ⭐⭐ (UX 임팩트 큼, 중기 구현) + +--- + +### 3-5. 커뮤니티 + 소셜 증거 — 바이럴 유도 + +**목표**: 사용자 참여 및 구전 마케팅 + +**구현 내용**: +- 이번 주 가장 많이 선택된 번호 TOP 10 공개 +- "나와 같은 번호 선택한 회원 수" 표시 +- AI 추천으로 X개 일치 달성한 회원 수 표시 + +**예시**: +``` +"이번 주 가장 많이 선택된 번호 TOP 10" +"AI 추천 번호로 3개 일치 달성한 회원: 1,247명" +"나와 같은 번호를 선택한 회원: 34명" +``` + +**전략**: 무료 티어에 일부 공개 → 상세 분석은 유료 전환 + +**우선순위**: ⭐ (회원 시스템 구축 후 가능) + +--- + +## 4. 구독 티어 설계 + +| 기능 | 무료 | 스탠다드 (₩2,900/월) | 프리미엄 (₩5,900/월) | +|------|:----:|:----:|:----:| +| 기본 추천 번호 | 1세트 | 5세트 | 무제한 | +| 통계 분석 | 기본 | 심화 | 전체 | +| 회차 공략 리포트 | - | 주간 요약 | 풀 리포트 | +| 개인 패턴 분석 | - | - | ✓ | +| 번호 포트폴리오 | - | ✓ | ✓ | +| 당첨 근접도 통계 | - | ✓ | ✓ | +| 당첨 알림 | - | 이메일 | 이메일 + 앱 | + +--- + +## 5. 기술 구현 로드맵 + +### Phase 1 — 즉시 가능 (데이터 이미 존재) + +- [ ] 추천 이력 채점 통계 API (`GET /api/lotto/stats/performance`) +- [ ] 신뢰도 지표 UI (평균 일치 개수, 전국 평균 비교) +- [ ] 회차별 공략 리포트 API (`GET /api/lotto/report/{drw_no}`) +- [ ] 개인 추천 이력 성과 대시보드 + +### Phase 2 — 단기 (1-2주) + +- [ ] `user_preferences` 테이블 설계 및 구현 +- [ ] 개인 패턴 분석 API (`GET /api/lotto/analysis/personal`) +- [ ] 주간 리포트 자동 생성 스케줄러 (토요일 오전) +- [ ] 투자 추적 기능 (구매 금액 입력 → 수익률 계산) +- [ ] `purchase_history` 테이블 추가 + +### Phase 3 — 중기 (1개월) + +- [ ] 회원 시스템 구축 (JWT 인증, SQLite `users` 테이블) +- [ ] 구독 플랜 관리 (`subscription_plans`, `user_subscriptions` 테이블) +- [ ] 결제 연동 (Toss Payments 또는 Stripe) +- [ ] 이메일 발송 자동화 (SendGrid) +- [ ] 소셜 증거 데이터 집계 API + +--- + +## 6. DB 스키마 확장 계획 + +```sql +-- Phase 2 +CREATE TABLE purchase_history ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + draw_no INTEGER NOT NULL, + amount INTEGER NOT NULL, -- 구매 금액 (원) + sets INTEGER NOT NULL DEFAULT 1, -- 구매 세트 수 + prize INTEGER DEFAULT 0, -- 당첨금 + note TEXT, + created_at TEXT DEFAULT (datetime('now')) +); + +CREATE TABLE user_preferences ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + odd_ratio REAL, -- 홀수 선호 비율 + high_ratio REAL, -- 고번호(23+) 선호 비율 + consecutive INTEGER, -- 연속번호 포함 선호 여부 + excluded_numbers TEXT, -- JSON 배열, 기피 번호 + updated_at TEXT DEFAULT (datetime('now')) +); + +-- Phase 3 +CREATE TABLE users ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + email TEXT UNIQUE NOT NULL, + password_hash TEXT NOT NULL, + plan TEXT DEFAULT 'free', -- free | standard | premium + plan_expires_at TEXT, + created_at TEXT DEFAULT (datetime('now')) +); +``` + +--- + +## 7. API 확장 계획 + +| Phase | 메서드 | 경로 | 설명 | +|-------|--------|------|------| +| 1 | GET | `/api/lotto/stats/performance` | 추천 성과 통계 (평균 일치 수 등) | +| 1 | GET | `/api/lotto/report/latest` | 최신 회차 공략 리포트 | +| 1 | GET | `/api/lotto/report/{drw_no}` | 특정 회차 공략 리포트 | +| 2 | GET | `/api/lotto/purchase` | 구매 이력 조회 | +| 2 | POST | `/api/lotto/purchase` | 구매 이력 추가 | +| 2 | GET | `/api/lotto/purchase/stats` | 투자 수익률 통계 | +| 2 | GET | `/api/lotto/analysis/personal` | 개인 패턴 분석 | +| 3 | POST | `/api/auth/register` | 회원가입 | +| 3 | POST | `/api/auth/login` | 로그인 | +| 3 | GET | `/api/subscription/plans` | 구독 플랜 목록 | +| 3 | POST | `/api/subscription/checkout` | 결제 시작 | + +--- + +## 참고 + +- 현재 운영 중인 lotto API: `CLAUDE.md` → `lotto-lab API 목록` 섹션 참고 +- 채점 로직: `backend/app/checker.py` +- 시뮬레이션 로직: `backend/app/recommender.py` +- DB 스키마: `backend/app/db.py` `init_db()` diff --git a/docs/superpowers/plans/2026-04-05-lotto-purchase-strategy-evolution.md b/docs/superpowers/plans/2026-04-05-lotto-purchase-strategy-evolution.md deleted file mode 100644 index 045105e..0000000 --- a/docs/superpowers/plans/2026-04-05-lotto-purchase-strategy-evolution.md +++ /dev/null @@ -1,1498 +0,0 @@ -# Lotto 구매 연동 + 전략 진화 시스템 구현 계획 - -> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. - -**Goal:** 로또 추천 번호의 실제/가상 구매, 자동 결과 체크, EMA 기반 전략 진화 메타 추천 시스템 구현 - -**Architecture:** 기존 lotto-backend(backend/) 서비스에 `purchase_manager.py`, `strategy_evolver.py` 2개 모듈 추가. 기존 `purchase_history` 테이블을 ALTER TABLE로 확장하고, `strategy_performance`, `strategy_weights` 2개 테이블 신규 생성. checker.py에서 purchase 체크 연동하여 자동 순환 파이프라인 완성. - -**Tech Stack:** Python 3.12, FastAPI, SQLite, APScheduler (기존 스택 그대로) - -**Spec:** `docs/superpowers/specs/2026-04-05-lotto-purchase-strategy-evolution-design.md` - ---- - -## 파일 구조 - -### 신규 파일 -| 파일 | 역할 | -|------|------| -| `backend/app/purchase_manager.py` | 구매 이력 관리 + 결과 체크 로직 | -| `backend/app/strategy_evolver.py` | EMA 계산 + Softmax 가중치 + 스마트 추천 | -| `backend/tests/test_purchase_manager.py` | purchase_manager 단위 테스트 | -| `backend/tests/test_strategy_evolver.py` | strategy_evolver 단위 테스트 | -| `backend/tests/test_integration.py` | 체커 연동 통합 테스트 | - -### 수정 파일 -| 파일 | 변경 내용 | -|------|----------| -| `backend/app/db.py` | purchase_history ALTER + 신규 테이�� 2개 + CRUD 함수 | -| `backend/app/checker.py` | check_results_for_draw() 끝에 purchase 체크 호출 | -| `backend/app/main.py` | 신규 API 엔드포인트 + Pydantic 모델 + import | - ---- - -## Task 1: DB 스키마 확장 — purchase_history ALTER + 신규 테이블 - -**Files:** -- Modify: `backend/app/db.py:254-268` (purchase_history 생성 부분) -- Modify: `backend/app/db.py:21` (init_db 함수 내부에 신규 ��이블 추가) -- Test: `backend/tests/test_purchase_manager.py` - -- [ ] **Step 1: 테스트 파일 생성 — DB 스키마 검증 테스트** - -```python -# backend/tests/test_purchase_manager.py -import sys, os -sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "app")) - -import sqlite3 -import pytest -from unittest.mock import patch - -# 테스트용 임시 DB 경로 -TEST_DB = ":memory:" - -def _get_test_conn(): - conn = sqlite3.connect(TEST_DB) - conn.row_factory = sqlite3.Row - return conn - -def test_purchase_history_has_new_columns(): - """purchase_history 테이블에 신규 컬럼이 존재하는지 검증""" - with patch("db.DB_PATH", TEST_DB): - import db - db.DB_PATH = TEST_DB - db.init_db() - - conn = db._conn() - cols = {r["name"] for r in conn.execute("PRAGMA table_info(purchase_history)").fetchall()} - assert "numbers" in cols - assert "is_real" in cols - assert "source_strategy" in cols - assert "source_detail" in cols - assert "checked" in cols - assert "results" in cols - assert "total_prize" in cols - # 기존 컬럼도 유지 - assert "draw_no" in cols - assert "amount" in cols - assert "sets" in cols - assert "prize" in cols - assert "note" in cols - conn.close() - - -def test_strategy_performance_table_exists(): - """strategy_performance 테이블이 생성되는지 검증""" - with patch("db.DB_PATH", TEST_DB): - import db - db.DB_PATH = TEST_DB - db.init_db() - - conn = db._conn() - cols = {r["name"] for r in conn.execute("PRAGMA table_info(strategy_performance)").fetchall()} - assert "strategy" in cols - assert "draw_no" in cols - assert "sets_count" in cols - assert "total_correct" in cols - assert "avg_score" in cols - conn.close() - - -def test_strategy_weights_table_exists(): - """strategy_weights 테이블이 생성되고 초기값이 있는지 검증""" - with patch("db.DB_PATH", TEST_DB): - import db - db.DB_PATH = TEST_DB - db.init_db() - - conn = db._conn() - rows = conn.execute("SELECT * FROM strategy_weights ORDER BY strategy").fetchall() - strategies = {r["strategy"] for r in rows} - assert strategies == {"combined", "simulation", "heatmap", "manual", "custom"} - # 가중치 합이 1.0 - total_weight = sum(r["weight"] for r in rows) - assert abs(total_weight - 1.0) < 0.01 - conn.close() -``` - -- [ ] **Step 2: 테스트 실행 — 실패 확인** - -Run: `cd C:/Users/jaeoh/Desktop/workspace/web-backend && python -m pytest backend/tests/test_purchase_manager.py -v` -Expected: FAIL — 신규 컬럼/테이블 미존재 - -- [ ] **Step 3: db.py init_db()에 purchase_history ALTER 추가** - -`backend/app/db.py`의 `init_db()` 함수에서, 기존 `purchase_history` CREATE TABLE 이후(268행 근처)에 추가: - -```python - # ── purchase_history 컬럼 확장 (기존 데이터 보존) ────────────────────── - _ensure_column(conn, "purchase_history", "numbers", - "ALTER TABLE purchase_history ADD COLUMN numbers TEXT NOT NULL DEFAULT '[]'") - _ensure_column(conn, "purchase_history", "is_real", - "ALTER TABLE purchase_history ADD COLUMN is_real INTEGER NOT NULL DEFAULT 1") - _ensure_column(conn, "purchase_history", "source_strategy", - "ALTER TABLE purchase_history ADD COLUMN source_strategy TEXT NOT NULL DEFAULT 'manual'") - _ensure_column(conn, "purchase_history", "source_detail", - "ALTER TABLE purchase_history ADD COLUMN source_detail TEXT NOT NULL DEFAULT '{}'") - _ensure_column(conn, "purchase_history", "checked", - "ALTER TABLE purchase_history ADD COLUMN checked INTEGER NOT NULL DEFAULT 0") - _ensure_column(conn, "purchase_history", "results", - "ALTER TABLE purchase_history ADD COLUMN results TEXT NOT NULL DEFAULT '[]'") - _ensure_column(conn, "purchase_history", "total_prize", - "ALTER TABLE purchase_history ADD COLUMN total_prize INTEGER NOT NULL DEFAULT 0") -``` - -- [ ] **Step 4: db.py init_db()에 strategy_performance 테이블 추가** - -기존 `weekly_reports` 테이블 생성 부분(270행 근처) 이후에 추가: - -```python - # ── strategy_performance 테이블 ──────────────────────────────────────── - conn.execute( - """ - CREATE TABLE IF NOT EXISTS strategy_performance ( - id INTEGER PRIMARY KEY AUTOINCREMENT, - strategy TEXT NOT NULL, - draw_no INTEGER NOT NULL, - sets_count INTEGER NOT NULL DEFAULT 0, - total_correct INTEGER NOT NULL DEFAULT 0, - max_correct INTEGER NOT NULL DEFAULT 0, - prize_total INTEGER NOT NULL DEFAULT 0, - avg_score REAL NOT NULL DEFAULT 0.0, - updated_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ','now')), - UNIQUE(strategy, draw_no) - ); - """ - ) - - # ── strategy_weights 테이블 ──────────────────────────────────────────── - conn.execute( - """ - CREATE TABLE IF NOT EXISTS strategy_weights ( - id INTEGER PRIMARY KEY AUTOINCREMENT, - strategy TEXT NOT NULL UNIQUE, - weight REAL NOT NULL DEFAULT 0.2, - ema_score REAL NOT NULL DEFAULT 0.15, - total_sets INTEGER NOT NULL DEFAULT 0, - total_hits_3plus INTEGER NOT NULL DEFAULT 0, - updated_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ','now')) - ); - """ - ) - - # strategy_weights 초기값 시드 (이미 있으면 무시) - _INIT_WEIGHTS = [ - ("combined", 0.30, 0.15), - ("simulation", 0.25, 0.15), - ("heatmap", 0.20, 0.15), - ("manual", 0.15, 0.15), - ("custom", 0.10, 0.15), - ] - for strat, w, ema in _INIT_WEIGHTS: - conn.execute( - "INSERT OR IGNORE INTO strategy_weights (strategy, weight, ema_score) VALUES (?, ?, ?)", - (strat, w, ema), - ) -``` - -- [ ] **Step 5: 테스트 실행 — 통과 확인** - -Run: `cd C:/Users/jaeoh/Desktop/workspace/web-backend && python -m pytest backend/tests/test_purchase_manager.py -v` -Expected: 3 tests PASS - -- [ ] **Step 6: 커밋** - -```bash -cd C:/Users/jaeoh/Desktop/workspace/web-backend -git add backend/app/db.py backend/tests/test_purchase_manager.py -git commit -m "lotto-lab: DB 스키마 확장 — purchase_history ALTER + strategy 테이블 추가" -``` - ---- - -## Task 2: DB CRUD 함수 — 구매 이력 확장 + 전략 성과/가중치 - -**Files:** -- Modify: `backend/app/db.py:1140-1230` (기존 purchase CRUD 확장 + 신규 함수) -- Test: `backend/tests/test_purchase_manager.py` (추가) - -- [ ] **Step 1: 테스트 추가 — 확장된 purchase CRUD** - -`backend/tests/test_purchase_manager.py`에 추가: - -```python -def test_add_purchase_with_numbers(): - """번호 포함 구매 등록""" - with patch("db.DB_PATH", TEST_DB): - import db - db.DB_PATH = TEST_DB - db.init_db() - - result = db.add_purchase( - draw_no=1125, - amount=2000, - sets=2, - prize=0, - note="테스트", - numbers=[[3, 12, 23, 34, 38, 45], [7, 14, 21, 29, 36, 42]], - is_real=True, - source_strategy="combined", - source_detail={"recommendation_ids": [1, 2]}, - ) - assert result["draw_no"] == 1125 - assert result["is_real"] == 1 - assert result["source_strategy"] == "combined" - assert result["numbers"] == [[3, 12, 23, 34, 38, 45], [7, 14, 21, 29, 36, 42]] - assert result["checked"] == 0 - - -def test_get_purchases_filter_is_real(): - """is_real 필터 동작""" - with patch("db.DB_PATH", TEST_DB): - import db - db.DB_PATH = TEST_DB - db.init_db() - - db.add_purchase(draw_no=1125, amount=1000, sets=1, is_real=True, source_strategy="combined") - db.add_purchase(draw_no=1125, amount=1000, sets=1, is_real=False, source_strategy="simulation") - - real = db.get_purchases(is_real=True) - virtual = db.get_purchases(is_real=False) - assert all(r["is_real"] == 1 for r in real) - assert all(r["is_real"] == 0 for r in virtual) - - -def test_get_purchase_stats_by_type(): - """실제/가상 분리 통계""" - with patch("db.DB_PATH", TEST_DB): - import db - db.DB_PATH = TEST_DB - db.init_db() - - db.add_purchase(draw_no=1125, amount=2000, sets=2, prize=5000, is_real=True, source_strategy="combined") - db.add_purchase(draw_no=1125, amount=1000, sets=1, prize=0, is_real=False, source_strategy="simulation") - - stats = db.get_purchase_stats() - assert stats["total"]["invested"] == 3000 - assert stats["real"]["invested"] == 2000 - assert stats["virtual"]["invested"] == 1000 - - -def test_upsert_strategy_performance(): - """전략 성과 upsert""" - with patch("db.DB_PATH", TEST_DB): - import db - db.DB_PATH = TEST_DB - db.init_db() - - db.upsert_strategy_performance("combined", 1125, sets_count=2, total_correct=5, max_correct=3, avg_score=0.42) - rows = db.get_strategy_performance("combined") - assert len(rows) == 1 - assert rows[0]["sets_count"] == 2 - - # upsert: 같은 전략+회차 → 업데이트 - db.upsert_strategy_performance("combined", 1125, sets_count=3, total_correct=7, max_correct=4, avg_score=0.50) - rows = db.get_strategy_performance("combined") - assert len(rows) == 1 - assert rows[0]["sets_count"] == 3 - - -def test_update_strategy_weight(): - """전략 가중치 업데이트""" - with patch("db.DB_PATH", TEST_DB): - import db - db.DB_PATH = TEST_DB - db.init_db() - - db.update_strategy_weight("combined", weight=0.35, ema_score=0.28, total_sets=15, total_hits_3plus=3) - weights = db.get_strategy_weights() - combined = next(w for w in weights if w["strategy"] == "combined") - assert combined["weight"] == 0.35 - assert combined["ema_score"] == 0.28 -``` - -- [ ] **Step 2: 테스트 실행 — 실패 확인** - -Run: `cd C:/Users/jaeoh/Desktop/workspace/web-backend && python -m pytest backend/tests/test_purchase_manager.py -v` -Expected: 새 테스트들 FAIL — 함수 미존재 - -- [ ] **Step 3: db.py — add_purchase 함수 시그니처 확장** - -기존 `add_purchase` (db.py:1154) 를 확장: - -```python -def add_purchase(draw_no: int, amount: int, sets: int, prize: int = 0, note: str = "", - numbers: list = None, is_real: bool = True, - source_strategy: str = "manual", source_detail: dict = None) -> Dict[str, Any]: - import json as _json - numbers_json = _json.dumps(numbers or [], ensure_ascii=False) - detail_json = _json.dumps(source_detail or {}, ensure_ascii=False) - is_real_int = 1 if is_real else 0 - with _conn() as conn: - conn.execute( - """INSERT INTO purchase_history - (draw_no, amount, sets, prize, note, numbers, is_real, source_strategy, source_detail) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)""", - (draw_no, amount, sets, prize, note, numbers_json, is_real_int, source_strategy, detail_json), - ) - row = conn.execute("SELECT * FROM purchase_history WHERE rowid = last_insert_rowid()").fetchone() - return _purchase_row_to_dict(row) -``` - -- [ ] **Step 4: db.py — _purchase_row_to_dict 확장** - -기존 `_purchase_row_to_dict` (db.py:1142) 를 확장: - -```python -def _purchase_row_to_dict(r) -> Dict[str, Any]: - import json as _json - numbers_raw = r["numbers"] if "numbers" in r.keys() else "[]" - detail_raw = r["source_detail"] if "source_detail" in r.keys() else "{}" - results_raw = r["results"] if "results" in r.keys() else "[]" - return { - "id": r["id"], - "draw_no": r["draw_no"], - "amount": r["amount"], - "sets": r["sets"], - "prize": r["prize"], - "note": r["note"], - "created_at": r["created_at"], - "numbers": _json.loads(numbers_raw) if numbers_raw else [], - "is_real": r["is_real"] if "is_real" in r.keys() else 1, - "source_strategy": r["source_strategy"] if "source_strategy" in r.keys() else "manual", - "source_detail": _json.loads(detail_raw) if detail_raw else {}, - "checked": r["checked"] if "checked" in r.keys() else 0, - "results": _json.loads(results_raw) if results_raw else [], - "total_prize": r["total_prize"] if "total_prize" in r.keys() else 0, - } -``` - -- [ ] **Step 5: db.py — get_purchases에 is_real, strategy 필터 추가** - -기존 `get_purchases` (db.py:1164) 확장: - -```python -def get_purchases(draw_no: int = None, days: int = None, - is_real: bool = None, strategy: str = None, - checked: bool = None) -> List[Dict[str, Any]]: - conditions, params = [], [] - if draw_no is not None: - conditions.append("draw_no = ?") - params.append(draw_no) - if days: - conditions.append("created_at >= datetime('now', ? || ' days')") - params.append(f"-{days}") - if is_real is not None: - conditions.append("is_real = ?") - params.append(1 if is_real else 0) - if strategy is not None: - conditions.append("source_strategy = ?") - params.append(strategy) - if checked is not None: - conditions.append("checked = ?") - params.append(1 if checked else 0) - where = f"WHERE {' AND '.join(conditions)}" if conditions else "" - with _conn() as conn: - rows = conn.execute( - f"SELECT * FROM purchase_history {where} ORDER BY draw_no DESC, id DESC", - params, - ).fetchall() - return [_purchase_row_to_dict(r) for r in rows] -``` - -- [ ] **Step 6: db.py — get_purchase_stats 확장 (전체/실제/가상 + 전략별)** - -기존 `get_purchase_stats` (db.py:1206) 를 교체: - -```python -def get_purchase_stats() -> Dict[str, Any]: - import json as _json - - def _calc_group(rows): - if not rows: - return {"sets": 0, "invested": 0, "prize": 0, "roi": 0.0, "win_rate": 0.0} - invested = sum(r["amount"] for r in rows) - prize = sum(r["total_prize"] or r["prize"] for r in rows) - wins = sum(1 for r in rows if (r["total_prize"] or r["prize"]) > 0) - return { - "sets": sum(r["sets"] for r in rows), - "invested": invested, - "prize": prize, - "roi": round((prize / invested * 100 - 100) if invested else 0.0, 2), - "win_rate": round(wins / len(rows) * 100, 2) if rows else 0.0, - } - - with _conn() as conn: - rows = conn.execute("SELECT * FROM purchase_history").fetchall() - - all_rows = [dict(r) for r in rows] - real_rows = [r for r in all_rows if r.get("is_real", 1) == 1] - virtual_rows = [r for r in all_rows if r.get("is_real", 1) == 0] - - # 전략별 집계 - by_strategy = {} - for r in all_rows: - strat = r.get("source_strategy", "manual") - if strat not in by_strategy: - by_strategy[strat] = [] - by_strategy[strat].append(r) - - strategy_stats = {} - for strat, srows in by_strategy.items(): - s = _calc_group(srows) - # results에서 correct 수 추출 - total_correct = 0 - count_sets = 0 - hits_3plus = 0 - for r in srows: - results_raw = r.get("results", "[]") - try: - results = _json.loads(results_raw) if isinstance(results_raw, str) else results_raw - except Exception: - results = [] - for res in results: - count_sets += 1 - c = res.get("correct", 0) - total_correct += c - if c >= 3: - hits_3plus += 1 - s["avg_correct"] = round(total_correct / count_sets, 2) if count_sets else 0.0 - s["hits_3plus"] = hits_3plus - strategy_stats[strat] = s - - return { - "total": _calc_group(all_rows), - "real": _calc_group(real_rows), - "virtual": _calc_group(virtual_rows), - "by_strategy": strategy_stats, - # 하위호환: 기존 필드도 유지 - "total_records": len(all_rows), - "total_invested": sum(r["amount"] for r in all_rows), - "total_prize": sum(r.get("total_prize", 0) or r["prize"] for r in all_rows), - "net": sum(r.get("total_prize", 0) or r["prize"] for r in all_rows) - sum(r["amount"] for r in all_rows), - "return_rate": round((sum(r.get("total_prize", 0) or r["prize"] for r in all_rows) / sum(r["amount"] for r in all_rows) * 100) if sum(r["amount"] for r in all_rows) else 0.0, 2), - "prize_count": sum(1 for r in all_rows if (r.get("total_prize", 0) or r["prize"]) > 0), - "max_prize": max((r.get("total_prize", 0) or r["prize"] for r in all_rows), default=0), - } -``` - -- [ ] **Step 7: db.py — strategy_performance CRUD 함수 추가** - -db.py 파일 끝에 추가: - -```python -# ── strategy_performance CRUD ──────────────────────────────────────────────── - -def upsert_strategy_performance(strategy: str, draw_no: int, sets_count: int = 0, - total_correct: int = 0, max_correct: int = 0, - prize_total: int = 0, avg_score: float = 0.0) -> None: - with _conn() as conn: - conn.execute( - """INSERT INTO strategy_performance (strategy, draw_no, sets_count, total_correct, max_correct, prize_total, avg_score) - VALUES (?, ?, ?, ?, ?, ?, ?) - ON CONFLICT(strategy, draw_no) DO UPDATE SET - sets_count=excluded.sets_count, total_correct=excluded.total_correct, - max_correct=excluded.max_correct, prize_total=excluded.prize_total, - avg_score=excluded.avg_score, - updated_at=strftime('%Y-%m-%dT%H:%M:%fZ','now')""", - (strategy, draw_no, sets_count, total_correct, max_correct, prize_total, avg_score), - ) - - -def get_strategy_performance(strategy: str = None, days: int = None) -> List[Dict[str, Any]]: - conditions, params = [], [] - if strategy: - conditions.append("strategy = ?") - params.append(strategy) - if days: - conditions.append("updated_at >= datetime('now', ? || ' days')") - params.append(f"-{days}") - where = f"WHERE {' AND '.join(conditions)}" if conditions else "" - with _conn() as conn: - rows = conn.execute( - f"SELECT * FROM strategy_performance {where} ORDER BY draw_no ASC", - params, - ).fetchall() - return [dict(r) for r in rows] - - -# ── strategy_weights CRUD ───────────────────────────────────────────────────── - -def get_strategy_weights() -> List[Dict[str, Any]]: - with _conn() as conn: - rows = conn.execute("SELECT * FROM strategy_weights ORDER BY weight DESC").fetchall() - return [dict(r) for r in rows] - - -def update_strategy_weight(strategy: str, weight: float, ema_score: float, - total_sets: int = None, total_hits_3plus: int = None) -> None: - with _conn() as conn: - fields = "weight=?, ema_score=?, updated_at=strftime('%Y-%m-%dT%H:%M:%fZ','now')" - params = [weight, ema_score] - if total_sets is not None: - fields += ", total_sets=?" - params.append(total_sets) - if total_hits_3plus is not None: - fields += ", total_hits_3plus=?" - params.append(total_hits_3plus) - params.append(strategy) - conn.execute(f"UPDATE strategy_weights SET {fields} WHERE strategy=?", params) - - -def update_purchase_results(purchase_id: int, results: list, total_prize: int) -> None: - """구매 건의 결과를 갱신 (체커 호출 후)""" - import json as _json - with _conn() as conn: - conn.execute( - "UPDATE purchase_history SET results=?, total_prize=?, checked=1 WHERE id=?", - (_json.dumps(results, ensure_ascii=False), total_prize, purchase_id), - ) -``` - -- [ ] **Step 8: db.py — exports 업데이트 (main.py import용)** - -db.py 파일 상단이나 기존 함수들이 이미 직접 import되므로, main.py의 import 문에 새 함수를 추가할 준비만 해둠. (Task 5에서 실행) - -- [ ] **Step 9: 테스트 실행 — 통과 확인** - -Run: `cd C:/Users/jaeoh/Desktop/workspace/web-backend && python -m pytest backend/tests/test_purchase_manager.py -v` -Expected: ALL PASS - -- [ ] **Step 10: 커밋** - -```bash -cd C:/Users/jaeoh/Desktop/workspace/web-backend -git add backend/app/db.py backend/tests/test_purchase_manager.py -git commit -m "lotto-lab: 구매 CRUD 확장 + strategy_performance/weights CRUD 추가" -``` - ---- - -## Task 3: purchase_manager.py — 구매 결과 체크 로직 - -**Files:** -- Create: `backend/app/purchase_manager.py` -- Test: `backend/tests/test_purchase_manager.py` (추가) - -- [ ] **Step 1: 테스트 추가 — 구매 결과 체크** - -`backend/tests/test_purchase_manager.py`에 추가: - -```python -def test_check_purchases_for_draw(): - """특정 회차 구매 건들의 결과 체크""" - with patch("db.DB_PATH", TEST_DB): - import db - db.DB_PATH = TEST_DB - db.init_db() - - # draws 테이블에 테스트 데이터 삽입 - conn = db._conn() - conn.execute( - "INSERT OR IGNORE INTO draws (drw_no, drw_date, n1, n2, n3, n4, n5, n6, bonus) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)", - (1125, "2026-04-04", 3, 12, 23, 34, 38, 45, 7) - ) - conn.commit() - conn.close() - - # 1125회차 대상 구매 등록 (draw_no=1125) - db.add_purchase( - draw_no=1125, amount=2000, sets=2, numbers=[[3, 12, 23, 34, 38, 45], [1, 2, 3, 4, 5, 6]], - is_real=True, source_strategy="combined", - ) - - from purchase_manager import check_purchases_for_draw - checked = check_purchases_for_draw(1125) - assert checked == 1 # 1건 체크됨 - - purchases = db.get_purchases(draw_no=1125) - p = purchases[0] - assert p["checked"] == 1 - assert len(p["results"]) == 2 - # 첫 번째 세트: 6개 전부 맞음 = 1등 - assert p["results"][0]["rank"] == 1 - assert p["results"][0]["correct"] == 6 - # 두 번째 세트: 3개 맞음 (3, 34 -> 아니 3만 맞음... 확인) - # [1,2,3,4,5,6] vs [3,12,23,34,38,45] → 3 하나만 맞음 - assert p["results"][1]["correct"] == 1 - assert p["results"][1]["rank"] == 0 - - -def test_check_purchases_updates_strategy_performance(): - """결과 체크 후 strategy_performance가 갱신되는지 검증""" - with patch("db.DB_PATH", TEST_DB): - import db - db.DB_PATH = TEST_DB - db.init_db() - - conn = db._conn() - conn.execute( - "INSERT OR IGNORE INTO draws (drw_no, drw_date, n1, n2, n3, n4, n5, n6, bonus) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)", - (1126, "2026-04-11", 5, 15, 25, 35, 40, 44, 10) - ) - conn.commit() - conn.close() - - db.add_purchase( - draw_no=1126, amount=1000, sets=1, numbers=[[5, 15, 25, 10, 11, 12]], - is_real=False, source_strategy="simulation", - ) - - from purchase_manager import check_purchases_for_draw - check_purchases_for_draw(1126) - - perfs = db.get_strategy_performance("simulation") - assert len(perfs) >= 1 - p = perfs[-1] - assert p["draw_no"] == 1126 - assert p["sets_count"] == 1 -``` - -- [ ] **Step 2: 테스트 실행 — 실패 확인** - -Run: `cd C:/Users/jaeoh/Desktop/workspace/web-backend && python -m pytest backend/tests/test_purchase_manager.py::test_check_purchases_for_draw -v` -Expected: FAIL — `purchase_manager` 모듈 미존재 - -- [ ] **Step 3: purchase_manager.py 작성** - -```python -# backend/app/purchase_manager.py -""" -구매 이력 관리 + 결과 체크 모듈. - -- check_purchases_for_draw(): 특정 회차 구매 건들의 결과를 자동 체크 -- 체커의 _calc_rank 재사용 -- 결과 체크 후 strategy_performance 자동 갱신 -""" -import json -import logging -from .db import ( - _conn, get_draw, get_purchases, update_purchase_results, - upsert_strategy_performance, -) -from .checker import _calc_rank - -logger = logging.getLogger("lotto-backend") - -RANK_PRIZE = {1: 0, 2: 0, 3: 1_500_000, 4: 50_000, 5: 5_000} - - -def check_purchases_for_draw(drw_no: int) -> int: - """ - 특정 회차 결과로 해당 회차 구매 건들을 채점한다. - - Returns: 채점한 구매 건 수 - """ - win_row = get_draw(drw_no) - if not win_row: - return 0 - - win_nums = [win_row["n1"], win_row["n2"], win_row["n3"], - win_row["n4"], win_row["n5"], win_row["n6"]] - bonus = win_row["bonus"] - - # draw_no가 해당 회차이고 아직 체크 안 된 구매 건 - unchecked = get_purchases(draw_no=drw_no, checked=False) - - # 전략별 집계 - strategy_agg = {} # strategy -> {sets_count, total_correct, max_correct, prize_total, scores[]} - - count = 0 - for purchase in unchecked: - numbers_list = purchase["numbers"] - if not numbers_list: - continue - - results = [] - for nums in numbers_list: - rank, correct, has_bonus = _calc_rank(nums, win_nums, bonus) - prize = RANK_PRIZE.get(rank, 0) - results.append({ - "numbers": nums, - "rank": rank, - "correct": correct, - "has_bonus": has_bonus, - "prize": prize, - }) - - total_prize = sum(r["prize"] for r in results) - update_purchase_results(purchase["id"], results, total_prize) - - # 전략별 집계 - strat = purchase["source_strategy"] - if strat not in strategy_agg: - strategy_agg[strat] = {"sets_count": 0, "total_correct": 0, "max_correct": 0, "prize_total": 0, "scores": []} - agg = strategy_agg[strat] - for r in results: - agg["sets_count"] += 1 - agg["total_correct"] += r["correct"] - agg["max_correct"] = max(agg["max_correct"], r["correct"]) - agg["prize_total"] += r["prize"] - agg["scores"].append(r["correct"] / 6.0) - - count += 1 - - # strategy_performance 갱신 - for strat, agg in strategy_agg.items(): - avg_score = sum(agg["scores"]) / len(agg["scores"]) if agg["scores"] else 0.0 - upsert_strategy_performance( - strategy=strat, - draw_no=drw_no, - sets_count=agg["sets_count"], - total_correct=agg["total_correct"], - max_correct=agg["max_correct"], - prize_total=agg["prize_total"], - avg_score=round(avg_score, 4), - ) - - logger.info(f"[purchase_manager] {drw_no}회차 구매 {count}건 체크 완료") - return count -``` - -- [ ] **Step 4: 테스트 실행 — 통과 확인** - -Run: `cd C:/Users/jaeoh/Desktop/workspace/web-backend && python -m pytest backend/tests/test_purchase_manager.py -v` -Expected: ALL PASS - -- [ ] **Step 5: 커밋** - -```bash -cd C:/Users/jaeoh/Desktop/workspace/web-backend -git add backend/app/purchase_manager.py backend/tests/test_purchase_manager.py -git commit -m "lotto-lab: purchase_manager — 구매 결과 자동 체크 + 전략 성과 집계" -``` - ---- - -## Task 4: strategy_evolver.py — EMA + Softmax 가중치 진화 + 스마트 추천 - -**Files:** -- Create: `backend/app/strategy_evolver.py` -- Test: `backend/tests/test_strategy_evolver.py` - -- [ ] **Step 1: 테스트 파일 생성** - -```python -# backend/tests/test_strategy_evolver.py -import sys, os -sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "app")) - -import math -import pytest -from unittest.mock import patch - -TEST_DB = ":memory:" - - -def test_calc_draw_score_basic(): - """세트별 결과 → draw_score 계산""" - from strategy_evolver import calc_draw_score - - results = [ - {"correct": 3, "rank": 5}, # 3/6 + 0.1 = 0.6 - {"correct": 1, "rank": 0}, # 1/6 + 0 = 0.167 - ] - score = calc_draw_score(results) - expected = ((3/6 + 0.1) + (1/6)) / 2 - assert abs(score - expected) < 0.01 - - -def test_calc_draw_score_empty(): - """빈 결과 → 0""" - from strategy_evolver import calc_draw_score - assert calc_draw_score([]) == 0.0 - - -def test_recalculate_weights_softmax(): - """EMA → Softmax 가중치 변환""" - from strategy_evolver import _softmax_weights - - ema_scores = { - "combined": 0.30, - "simulation": 0.25, - "heatmap": 0.15, - "manual": 0.10, - "custom": 0.05, - } - weights = _softmax_weights(ema_scores) - - # 합이 1.0 - assert abs(sum(weights.values()) - 1.0) < 0.001 - # combined가 가장 높아야 함 - assert weights["combined"] > weights["simulation"] - assert weights["simulation"] > weights["heatmap"] - # 최소 가중치 5% 보장 - assert all(w >= 0.049 for w in weights.values()) - - -def test_recalculate_weights_min_weight(): - """한 전략의 EMA가 매우 낮아도 최소 5% 보장""" - from strategy_evolver import _softmax_weights - - ema_scores = { - "combined": 0.50, - "simulation": 0.01, - "heatmap": 0.01, - "manual": 0.01, - "custom": 0.01, - } - weights = _softmax_weights(ema_scores) - - assert weights["simulation"] >= 0.049 - assert weights["custom"] >= 0.049 - assert abs(sum(weights.values()) - 1.0) < 0.001 - - -def test_update_ema(): - """EMA 갱신 공식 검증""" - from strategy_evolver import ALPHA - - old_ema = 0.15 - draw_score = 0.40 - new_ema = ALPHA * draw_score + (1 - ALPHA) * old_ema - expected = 0.3 * 0.40 + 0.7 * 0.15 # = 0.12 + 0.105 = 0.225 - assert abs(new_ema - expected) < 0.001 -``` - -- [ ] **Step 2: 테스트 실행 — 실패 확인** - -Run: `cd C:/Users/jaeoh/Desktop/workspace/web-backend && python -m pytest backend/tests/test_strategy_evolver.py -v` -Expected: FAIL — `strategy_evolver` 모듈 미존재 - -- [ ] **Step 3: strategy_evolver.py 작성** - -```python -# backend/app/strategy_evolver.py -""" -전략 진화 엔진 — EMA + Softmax 기반 적응형 가중치 관리. - -- calc_draw_score(): 구매 결과 → 성과 점수 -- update_ema_for_strategy(): 특정 전략의 EMA 갱신 -- recalculate_weights(): 전 전략 EMA → Softmax → 가중치 저장 -- generate_smart_recommendation(): 가중치 기반 메타 전략 추천 -""" -import math -import json -import logging -from typing import Dict, List, Any, Tuple - -from .db import ( - get_strategy_weights, update_strategy_weight, - get_strategy_performance, get_best_picks, get_all_draw_numbers, get_latest_draw, -) -from .recommender import recommend_numbers, recommend_with_heatmap -from .analyzer import generate_combined_recommendation, score_combination, build_analysis_cache -from .db import list_recommendations_ex - -logger = logging.getLogger("lotto-backend") - -ALPHA = 0.3 # EMA 감쇠율 -TEMPERATURE = 2.0 # Softmax 온도 -MIN_WEIGHT = 0.05 # 최소 가중치 -INITIAL_EMA = 0.15 # 콜드스타트 초기값 -MIN_DATA_DRAWS = 10 # 학습 최소 회차 - -STRATEGIES = ["combined", "simulation", "heatmap", "manual", "custom"] - -RANK_BONUS = {5: 0.1, 4: 0.3, 3: 0.6, 2: 0.8, 1: 1.0} - - -def calc_draw_score(results: List[Dict]) -> float: - """구매 결과 리스트 → 평균 성과 점수""" - if not results: - return 0.0 - scores = [] - for r in results: - s = r.get("correct", 0) / 6.0 - s += RANK_BONUS.get(r.get("rank", 0), 0) - scores.append(s) - return sum(scores) / len(scores) - - -def _softmax_weights(ema_scores: Dict[str, float]) -> Dict[str, float]: - """EMA 점수 → Softmax → 최소 가중치 보장 → 정규화""" - raw = {s: math.exp(ema / TEMPERATURE) for s, ema in ema_scores.items()} - total = sum(raw.values()) - weights = {s: v / total for s, v in raw.items()} - - # 최소 가중치 보장 - clamped = {} - surplus = 0.0 - unclamped = [] - for s, w in weights.items(): - if w < MIN_WEIGHT: - clamped[s] = MIN_WEIGHT - surplus += MIN_WEIGHT - w - else: - unclamped.append(s) - clamped[s] = w - - # surplus를 unclamped에서 비례 차감 - if surplus > 0 and unclamped: - unclamped_total = sum(clamped[s] for s in unclamped) - for s in unclamped: - clamped[s] -= surplus * (clamped[s] / unclamped_total) - - # 최종 정규화 - final_total = sum(clamped.values()) - return {s: round(v / final_total, 4) for s, v in clamped.items()} - - -def update_ema_for_strategy(strategy: str, draw_score: float) -> float: - """특정 전략의 EMA 갱신 + DB 저장. 새 EMA 반환.""" - weights = get_strategy_weights() - current = next((w for w in weights if w["strategy"] == strategy), None) - old_ema = current["ema_score"] if current else INITIAL_EMA - new_ema = ALPHA * draw_score + (1 - ALPHA) * old_ema - return new_ema - - -def recalculate_weights() -> Dict[str, float]: - """전 전략 EMA → Softmax → 가중치 재계산 + DB 저장""" - weights_rows = get_strategy_weights() - ema_scores = {w["strategy"]: w["ema_score"] for w in weights_rows} - - # 누락된 전략 보충 - for s in STRATEGIES: - if s not in ema_scores: - ema_scores[s] = INITIAL_EMA - - new_weights = _softmax_weights(ema_scores) - - for s, w in new_weights.items(): - row = next((r for r in weights_rows if r["strategy"] == s), None) - update_strategy_weight( - strategy=s, - weight=w, - ema_score=ema_scores[s], - total_sets=row["total_sets"] if row else 0, - total_hits_3plus=row["total_hits_3plus"] if row else 0, - ) - - logger.info(f"[strategy_evolver] 가중치 재계산: {new_weights}") - return new_weights - - -def evolve_after_check(strategy: str, draw_no: int, results: List[Dict]) -> None: - """결과 체크 후 EMA 갱신 + 가중치 재계산 (purchase_manager에서 호출)""" - draw_score = calc_draw_score(results) - new_ema = update_ema_for_strategy(strategy, draw_score) - - weights_rows = get_strategy_weights() - current = next((w for w in weights_rows if w["strategy"] == strategy), None) - hits_3plus = sum(1 for r in results if r.get("correct", 0) >= 3) - - update_strategy_weight( - strategy=strategy, - weight=current["weight"] if current else 0.2, - ema_score=new_ema, - total_sets=(current["total_sets"] if current else 0) + len(results), - total_hits_3plus=(current["total_hits_3plus"] if current else 0) + hits_3plus, - ) - - recalculate_weights() - - -def get_weights_with_trend() -> Dict[str, Any]: - """현재 가중치 + trend 정보 반환""" - weights = get_strategy_weights() - perfs = get_strategy_performance() - - # 전략별 최근 EMA 변화 추적 (최근 5회차) - strat_perfs = {} - for p in perfs: - s = p["strategy"] - if s not in strat_perfs: - strat_perfs[s] = [] - strat_perfs[s].append(p) - - result = [] - for w in weights: - # trend 계산: 최근 5회차 avg_score 변화 - sp = strat_perfs.get(w["strategy"], []) - if len(sp) >= 5: - recent_avg = sum(p["avg_score"] for p in sp[-3:]) / 3 - older_avg = sum(p["avg_score"] for p in sp[-5:-2]) / 3 - delta = recent_avg - older_avg - trend = "up" if delta > 0.02 else ("down" if delta < -0.02 else "stable") - else: - trend = "stable" - - result.append({ - "strategy": w["strategy"], - "weight": w["weight"], - "ema_score": w["ema_score"], - "total_sets": w["total_sets"], - "hits_3plus": w["total_hits_3plus"], - "trend": trend, - }) - - # 학습 상태 - all_draws = set() - for p in perfs: - all_draws.add(p["draw_no"]) - - return { - "weights": result, - "last_evolved": weights[0]["updated_at"] if weights else None, - "min_data_draws": MIN_DATA_DRAWS, - "current_data_draws": len(all_draws), - "status": "active" if len(all_draws) >= MIN_DATA_DRAWS else "learning", - } - - -def generate_smart_recommendation(sets: int = 5) -> Dict[str, Any]: - """ - 전략 가중치 기반 메타 전략 추천. - - 1. 가중치 로드 - 2. 각 전략에서 후보 10세트 생성 - 3. meta_score = original_score × strategy_weight - 4. 상위 N세트 선출 (중복 제거) - """ - weights_data = get_strategy_weights() - weight_map = {w["strategy"]: w["weight"] for w in weights_data} - draws = get_all_draw_numbers() - if not draws: - return {"error": "No draw data"} - - latest = get_latest_draw() - cache = build_analysis_cache(draws) - past_recs = list_recommendations_ex(limit=100, sort="id_desc") - - candidates = [] # [{numbers, score, strategy, meta_score}] - seen_keys = set() - - def _add_candidate(nums: list, strategy: str, raw_score: float = None): - key = tuple(sorted(nums)) - if key in seen_keys: - return - seen_keys.add(key) - if raw_score is None: - sc = score_combination(nums, cache) - raw_score = sc["score_total"] - meta = raw_score * weight_map.get(strategy, 0.1) - candidates.append({ - "numbers": sorted(nums), - "raw_score": round(raw_score, 4), - "strategy": strategy, - "meta_score": round(meta, 4), - }) - - # combined: 10세트 - for _ in range(10): - try: - r = generate_combined_recommendation(draws) - if "final_numbers" in r: - _add_candidate(r["final_numbers"], "combined") - except Exception: - pass - - # simulation: best_picks 상위 10개 - best = get_best_picks(limit=10) - for b in best: - nums = json.loads(b["numbers"]) if isinstance(b["numbers"], str) else b["numbers"] - _add_candidate(nums, "simulation", b.get("score_total")) - - # heatmap: 10세트 - for _ in range(10): - try: - r = recommend_with_heatmap(draws, past_recs) - _add_candidate(r["numbers"], "heatmap") - except Exception: - pass - - # manual: 10세트 - for _ in range(10): - try: - r = recommend_numbers(draws) - _add_candidate(r["numbers"], "manual") - except Exception: - pass - - # meta_score 기준 정렬, 상위 N개 - candidates.sort(key=lambda c: -c["meta_score"]) - top = candidates[:sets] - - # contribution 계산 - result_sets = [] - for c in top: - # 이 번호에 기여한 전략들의 비율 - sc = score_combination(c["numbers"], cache) - contributions = {} - for strat in STRATEGIES: - contributions[strat] = round(weight_map.get(strat, 0) * sc["score_total"], 4) - contrib_total = sum(contributions.values()) or 1 - contributions = {s: round(v / contrib_total, 3) for s, v in contributions.items()} - - result_sets.append({ - "numbers": c["numbers"], - "meta_score": c["meta_score"], - "source_strategy": c["strategy"], - "contribution": contributions, - "individual_scores": {k: round(v, 4) for k, v in sc.items()}, - }) - - # 학습 상태 - perfs = get_strategy_performance() - data_draws = len(set(p["draw_no"] for p in perfs)) - status = "active" if data_draws >= MIN_DATA_DRAWS else "learning" - - return { - "sets": result_sets, - "strategy_weights_used": weight_map, - "learning_status": { - "draws_learned": data_draws, - "status": status, - "message": "" if status == "active" else f"{MIN_DATA_DRAWS}회차 이상 데이터 필요 (현재 {data_draws}회차)", - }, - "based_on_latest_draw": latest["drw_no"] if latest else None, - } -``` - -- [ ] **Step 4: 테스트 실행 — 통과 확인** - -Run: `cd C:/Users/jaeoh/Desktop/workspace/web-backend && python -m pytest backend/tests/test_strategy_evolver.py -v` -Expected: ALL PASS - -- [ ] **Step 5: 커밋** - -```bash -cd C:/Users/jaeoh/Desktop/workspace/web-backend -git add backend/app/strategy_evolver.py backend/tests/test_strategy_evolver.py -git commit -m "lotto-lab: strategy_evolver — EMA/Softmax 가중치 진화 + 스마트 추천" -``` - ---- - -## Task 5: checker.py 연동 — 자동 파이프라인 - -**Files:** -- Modify: `backend/app/checker.py:28-66` -- Test: `backend/tests/test_integration.py` - -- [ ] **Step 1: 통합 테스트 작성** - -```python -# backend/tests/test_integration.py -import sys, os -sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "app")) - -import pytest -from unittest.mock import patch - -TEST_DB = ":memory:" - - -def test_check_results_triggers_purchase_check(): - """check_results_for_draw가 purchase 체크도 트리거하는지 검증""" - with patch("db.DB_PATH", TEST_DB): - import db - db.DB_PATH = TEST_DB - db.init_db() - - # 당첨번호 삽입: 1124회차 (base), 1125회차 (결과) - conn = db._conn() - conn.execute( - "INSERT INTO draws (drw_no, drw_date, n1, n2, n3, n4, n5, n6, bonus) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)", - (1124, "2026-03-28", 1, 2, 3, 4, 5, 6, 7) - ) - conn.execute( - "INSERT INTO draws (drw_no, drw_date, n1, n2, n3, n4, n5, n6, bonus) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)", - (1125, "2026-04-04", 10, 20, 30, 35, 40, 44, 15) - ) - conn.commit() - conn.close() - - # 1125회차 대상 구매 (draw_no=1125) - db.add_purchase( - draw_no=1125, amount=1000, sets=1, - numbers=[[10, 20, 30, 1, 2, 3]], - is_real=True, source_strategy="combined", - ) - - from checker import check_results_for_draw - check_results_for_draw(1125) - - # purchase도 체크되었는지 확인 - purchases = db.get_purchases(draw_no=1125) - assert purchases[0]["checked"] == 1 - assert purchases[0]["results"][0]["correct"] == 3 # 10, 20, 30 맞음 -``` - -- [ ] **Step 2: 테스트 실행 — 실패 확인** - -Run: `cd C:/Users/jaeoh/Desktop/workspace/web-backend && python -m pytest backend/tests/test_integration.py -v` -Expected: FAIL — checker.py에 purchase 연동 없음 - -- [ ] **Step 3: checker.py 수정 — purchase 체크 연동** - -`backend/app/checker.py` 파일 끝(66행 이후)의 `check_results_for_draw` 함수에 추가: - -기존 함수의 마지막 `return count` 바로 위에 추가: - -```python - # ── 구매 이력 체크 연동 ────────────────────────────────────── - try: - from .purchase_manager import check_purchases_for_draw - purchase_count = check_purchases_for_draw(drw_no) - if purchase_count > 0: - # 전략 가중치 재계산 - from .strategy_evolver import recalculate_weights - recalculate_weights() - except ImportError: - pass # purchase_manager 미설치 시 무시 (하위호환) -``` - -- [ ] **Step 4: 테스트 실행 — 통과 확인** - -Run: `cd C:/Users/jaeoh/Desktop/workspace/web-backend && python -m pytest backend/tests/test_integration.py -v` -Expected: PASS - -- [ ] **Step 5: 커밋** - -```bash -cd C:/Users/jaeoh/Desktop/workspace/web-backend -git add backend/app/checker.py backend/tests/test_integration.py -git commit -m "lotto-lab: checker 연동 — 추첨 결과 시 purchase 자동 체크 + 가중치 재계산" -``` - ---- - -## Task 6: main.py — 신규 API 엔드포인트 - -**Files:** -- Modify: `backend/app/main.py:12-36` (import 확장) -- Modify: `backend/app/main.py:258-308` (purchase API 확장) -- Modify: `backend/app/main.py` (신규 전략/스마트 추천 API 추가) - -- [ ] **Step 1: main.py import 확장** - -`backend/app/main.py:12-36`의 import 블록에 추가: - -```python -from .db import ( - # ... 기존 import 유지 ... - # 신규: 전략 관련 - get_strategy_weights as db_get_strategy_weights, - get_strategy_performance as db_get_strategy_performance, - upsert_strategy_performance, - update_strategy_weight, - update_purchase_results, -) -from .purchase_manager import check_purchases_for_draw -from .strategy_evolver import ( - get_weights_with_trend, recalculate_weights, - generate_smart_recommendation, -) -``` - -- [ ] **Step 2: PurchaseCreate Pydantic 모델 확장** - -기존 `PurchaseCreate` (main.py:260-265) 교체: - -```python -class PurchaseCreate(BaseModel): - draw_no: int - amount: int - sets: int = 1 - prize: int = 0 - note: str = "" - # 신규 필드 - numbers: List[List[int]] = [] - is_real: bool = True - source_strategy: str = "manual" - source_detail: dict = {} -``` - -- [ ] **Step 3: PurchaseUpdate 확장** - -기존 `PurchaseUpdate` (main.py:268-273) 확장: - -```python -class PurchaseUpdate(BaseModel): - draw_no: Optional[int] = None - amount: Optional[int] = None - sets: Optional[int] = None - prize: Optional[int] = None - note: Optional[str] = None - # 신규 필드 - numbers: Optional[List[List[int]]] = None - is_real: Optional[bool] = None - source_strategy: Optional[str] = None -``` - -- [ ] **Step 4: 기존 purchase API 수정** - -`api_purchase_create` (main.py:288-291) 수정: - -```python -@app.post("/api/lotto/purchase", status_code=201) -def api_purchase_create(body: PurchaseCreate): - """구매 이력 추가 (실제/가상)""" - sets = body.sets if body.sets > 0 else max(len(body.numbers), 1) - amount = body.amount if body.amount > 0 else sets * 1000 - return add_purchase( - draw_no=body.draw_no, - amount=amount, - sets=sets, - prize=body.prize, - note=body.note, - numbers=body.numbers, - is_real=body.is_real, - source_strategy=body.source_strategy, - source_detail=body.source_detail, - ) -``` - -`api_purchase_list` (main.py:282-285) 수정: - -```python -@app.get("/api/lotto/purchase") -def api_purchase_list(draw_no: Optional[int] = None, days: Optional[int] = None, - is_real: Optional[bool] = None, strategy: Optional[str] = None): - """구매 이력 조회 (필터: draw_no, days, is_real, strategy)""" - return {"records": get_purchases(draw_no=draw_no, days=days, is_real=is_real, strategy=strategy)} -``` - -- [ ] **Step 5: 전략 API 추가** - -`main.py`의 purchase API 블록 이후에 추가: - -```python -# ── 전략 진화 API ───────────────────────────────────────��──────────────────── - -@app.get("/api/lotto/strategy/weights") -def api_strategy_weights(): - """현재 전략별 가중치 + 성과 요약 + trend""" - return get_weights_with_trend() - - -@app.get("/api/lotto/strategy/performance") -def api_strategy_performance(strategy: Optional[str] = None, days: Optional[int] = None): - """전략별 회차 성과 이력 (차트용)""" - rows = db_get_strategy_performance(strategy=strategy, days=days) - return {"records": rows} - - -@app.post("/api/lotto/strategy/evolve") -def api_strategy_evolve(): - """수동 가중치 재계산 트리거""" - new_weights = recalculate_weights() - return {"ok": True, "weights": new_weights} - - -# ── 스마트 추천 API ────────────────────────────────────────────────────────── - -@app.get("/api/lotto/recommend/smart") -def api_recommend_smart(sets: int = 5): - """전략 가중치 기반 메타 전략 추천""" - sets = max(1, min(sets, 10)) - result = generate_smart_recommendation(sets=sets) - if "error" in result: - raise HTTPException(status_code=500, detail=result["error"]) - return result -``` - -- [ ] **Step 6: 테스트 — API 스모크 테스트** - -Run: `cd C:/Users/jaeoh/Desktop/workspace/web-backend && python -c "from backend.app.main import app; print('import OK')"` 또는 docker compose up 후 curl 테스트. - -Expected: import 성공, 문법 오류 없음 - -- [ ] **Step 7: 커밋** - -```bash -cd C:/Users/jaeoh/Desktop/workspace/web-backend -git add backend/app/main.py -git commit -m "lotto-lab: 구매/전략/스마트추천 API 엔드포인트 추가" -``` - ---- - -## Task 7: 통합 검증 — Docker 빌드 + 전체 테스트 - -**Files:** (수정 없음, 검증만) - -- [ ] **Step 1: 전체 테스트 실행** - -Run: `cd C:/Users/jaeoh/Desktop/workspace/web-backend && python -m pytest backend/tests/ -v` -Expected: ALL PASS - -- [ ] **Step 2: Docker 빌드 테스트** - -Run: `cd C:/Users/jaeoh/Desktop/workspace/web-backend && docker compose build lotto-backend` -Expected: 빌드 성공 - -- [ ] **Step 3: 로컬 실행 테스트** - -Run: `cd C:/Users/jaeoh/Desktop/workspace/web-backend && docker compose up -d lotto-backend` - -API 엔드포인트 확인: -```bash -# 전략 가중치 조회 -curl http://localhost:18000/api/lotto/strategy/weights - -# 스마트 추천 -curl "http://localhost:18000/api/lotto/recommend/smart?sets=3" - -# 가상 구매 등록 -curl -X POST http://localhost:18000/api/lotto/purchase \ - -H "Content-Type: application/json" \ - -d '{"draw_no":1125,"numbers":[[3,12,23,34,38,45]],"is_real":false,"amount":1000,"sets":1,"source_strategy":"smart"}' - -# 구매 이력 조회 (가상만) -curl "http://localhost:18000/api/lotto/purchase?is_real=false" - -# 구매 통계 -curl http://localhost:18000/api/lotto/purchase/stats -``` - -Expected: 모든 API가 200 응답 - -- [ ] **Step 4: CLAUDE.md 업데이트** - -`backend/` CLAUDE.md의 lotto-lab API 목록에 신규 API 추가: - -```markdown -| GET | `/api/lotto/purchase` | 구매 이력 조회 (is_real, strategy, draw_no, days 필터) | -| POST | `/api/lotto/purchase` | 구매 등록 (실제/가상, 번호, 전략 출처 포함) | -| GET | `/api/lotto/purchase/stats` | 구매 통계 (전체/실제/가상 + 전략별) | -| GET | `/api/lotto/strategy/weights` | 전략별 가중치 + 성과 + trend | -| GET | `/api/lotto/strategy/performance` | 전략별 회차 성과 이력 (차트용) | -| POST | `/api/lotto/strategy/evolve` | 수동 가중치 재계산 | -| GET | `/api/lotto/recommend/smart` | 전략 진화 기반 메타 추천 | -``` - -lotto.db 테이블에 추가: - -```markdown -| `strategy_performance` | 전략별 회차 성과 (EMA 입력 데이터) | -| `strategy_weights` | 메타 전략 가중치 (EMA + Softmax) | -``` - -- [ ] **Step 5: 커밋** - -```bash -cd C:/Users/jaeoh/Desktop/workspace/web-backend -git add CLAUDE.md -git commit -m "lotto-lab: CLAUDE.md 신규 API + 테이블 문서 업데이트" -``` - ---- - -## 요약 - -| Task | 내용 | 파일 | -|------|------|------| -| 1 | DB 스키마 확장 | db.py (ALTER + 테이블 생성) | -| 2 | DB CRUD 함수 | db.py (purchase 확장 + strategy CRUD) | -| 3 | purchase_manager.py | 구매 결과 체크 + 전략 성과 집계 | -| 4 | strategy_evolver.py | EMA + Softmax + 스마트 추천 | -| 5 | checker.py 연동 | 자동 파이프라인 완성 | -| 6 | main.py API | 9개 엔드포인트 추가 | -| 7 | 통합 검증 | Docker 빌드 + API 테스트 + 문서 | - -**프론트엔드 작업은 별도 플랜으로 분리합니다** (web-ui 레포가 별도 Git 저장소이므로). diff --git a/docs/superpowers/plans/2026-04-05-realestate-lab.md b/docs/superpowers/plans/2026-04-05-realestate-lab.md deleted file mode 100644 index 001d5b7..0000000 --- a/docs/superpowers/plans/2026-04-05-realestate-lab.md +++ /dev/null @@ -1,1509 +0,0 @@ -# realestate-lab Implementation Plan - -> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. - -**Goal:** 부동산 청약 공고 자동 수집 + 프로필 기반 자격 매칭 독립 서비스 구축, 기존 lotto-backend 청약 코드 제거 - -**Architecture:** 공공데이터포털(한국부동산원 청약홈 API)에서 매일 공고를 수집하여 SQLite에 저장하고, 사용자 프로필 기반으로 자격 매칭 점수를 산출하는 FastAPI 독립 서비스. stock-lab/music-lab과 동일한 Docker 컨테이너 패턴. - -**Tech Stack:** Python 3.12, FastAPI, SQLite, APScheduler, requests, pydantic - ---- - -## File Structure - -### New files (realestate-lab/) - -| File | Responsibility | -|------|----------------| -| `realestate-lab/app/__init__.py` | 패키지 마커 | -| `realestate-lab/app/main.py` | FastAPI 앱, 라우트, APScheduler, startup/shutdown | -| `realestate-lab/app/db.py` | SQLite 테이블 생성, 모든 CRUD 함수 | -| `realestate-lab/app/collector.py` | 공공데이터포털 API 호출, 응답 파싱, DB 저장 | -| `realestate-lab/app/matcher.py` | 프로필 기반 매칭 점수 산출 엔진 | -| `realestate-lab/app/models.py` | Pydantic 요청/응답 모델 | -| `realestate-lab/Dockerfile` | python:3.12-alpine 기반 컨테이너 | -| `realestate-lab/requirements.txt` | 의존성 목록 | - -### Modified files - -| File | Change | -|------|--------| -| `docker-compose.yml` | realestate-lab 서비스 추가 | -| `nginx/default.conf` | `/api/realestate/` 프록시 라우팅 추가 | -| `scripts/deploy-nas.sh` | rsync 대상에 realestate-lab 추가 | -| `backend/app/main.py` | 청약 관련 모델 + 라우트 + import 제거 | -| `backend/app/db.py` | realestate_complexes, subscription_items, subscription_profile 테이블 및 CRUD 제거 | - ---- - -### Task 1: 프로젝트 스캐폴딩 (Dockerfile, requirements.txt, __init__.py) - -**Files:** -- Create: `realestate-lab/app/__init__.py` -- Create: `realestate-lab/requirements.txt` -- Create: `realestate-lab/Dockerfile` - -- [ ] **Step 1: 디렉토리 생성 및 __init__.py** - -```bash -mkdir -p realestate-lab/app -``` - -```python -# realestate-lab/app/__init__.py -``` -(빈 파일) - -- [ ] **Step 2: requirements.txt 작성** - -``` -# realestate-lab/requirements.txt -requests==2.32.3 -fastapi==0.115.6 -uvicorn[standard]==0.30.6 -apscheduler==3.10.4 -pydantic>=2.0 -``` - -- [ ] **Step 3: Dockerfile 작성** - -```dockerfile -# realestate-lab/Dockerfile -FROM python:3.12-alpine - -WORKDIR /app -COPY requirements.txt . -RUN pip install --no-cache-dir -r requirements.txt - -COPY . . - -CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"] -``` - -- [ ] **Step 4: Commit** - -```bash -git add realestate-lab/ -git commit -m "feat(realestate-lab): 프로젝트 스캐폴딩 — Dockerfile, requirements, init" -``` - ---- - -### Task 2: Pydantic 모델 정의 (models.py) - -**Files:** -- Create: `realestate-lab/app/models.py` - -- [ ] **Step 1: models.py 작성** - -```python -# realestate-lab/app/models.py -from typing import Optional, List -from pydantic import BaseModel - - -# ── 공고 ───────────────────────────────────────────────────────────────────── - -class AnnouncementCreate(BaseModel): - house_nm: str - house_secd: str = "01" - house_dtl_secd: Optional[str] = None - rent_secd: Optional[str] = None - region_code: Optional[str] = None - region_name: Optional[str] = None - address: Optional[str] = None - total_units: Optional[int] = None - rcrit_date: Optional[str] = None - receipt_start: Optional[str] = None - receipt_end: Optional[str] = None - spsply_start: Optional[str] = None - spsply_end: Optional[str] = None - gnrl_rank1_start: Optional[str] = None - gnrl_rank1_end: Optional[str] = None - winner_date: Optional[str] = None - contract_start: Optional[str] = None - contract_end: Optional[str] = None - homepage_url: Optional[str] = None - pblanc_url: Optional[str] = None - constructor: Optional[str] = None - developer: Optional[str] = None - move_in_month: Optional[str] = None - is_speculative_area: Optional[str] = None - is_price_cap: Optional[str] = None - contact: Optional[str] = None - - -class AnnouncementUpdate(BaseModel): - house_nm: Optional[str] = None - house_secd: Optional[str] = None - house_dtl_secd: Optional[str] = None - rent_secd: Optional[str] = None - region_code: Optional[str] = None - region_name: Optional[str] = None - address: Optional[str] = None - total_units: Optional[int] = None - rcrit_date: Optional[str] = None - receipt_start: Optional[str] = None - receipt_end: Optional[str] = None - spsply_start: Optional[str] = None - spsply_end: Optional[str] = None - gnrl_rank1_start: Optional[str] = None - gnrl_rank1_end: Optional[str] = None - winner_date: Optional[str] = None - contract_start: Optional[str] = None - contract_end: Optional[str] = None - homepage_url: Optional[str] = None - pblanc_url: Optional[str] = None - constructor: Optional[str] = None - developer: Optional[str] = None - move_in_month: Optional[str] = None - is_speculative_area: Optional[str] = None - is_price_cap: Optional[str] = None - contact: Optional[str] = None - - -# ── 프로필 ─────────────────────────────────────────────────────────────────── - -class ProfileUpdate(BaseModel): - name: Optional[str] = None - age: Optional[int] = None - is_homeless: Optional[bool] = None - is_householder: Optional[bool] = None - subscription_months: Optional[int] = None - subscription_amount: Optional[int] = None - family_members: Optional[int] = None - has_dependents: Optional[bool] = None - children_count: Optional[int] = None - is_newlywed: Optional[bool] = None - marriage_months: Optional[int] = None - has_newborn: Optional[bool] = None - is_first_home: Optional[bool] = None - income_level: Optional[str] = None - preferred_regions: Optional[List[str]] = None - preferred_types: Optional[List[str]] = None - min_area: Optional[float] = None - max_area: Optional[float] = None - max_price: Optional[int] = None -``` - -- [ ] **Step 2: Commit** - -```bash -git add realestate-lab/app/models.py -git commit -m "feat(realestate-lab): Pydantic 요청 모델 정의" -``` - ---- - -### Task 3: DB 레이어 (db.py) - -**Files:** -- Create: `realestate-lab/app/db.py` - -- [ ] **Step 1: db.py 작성 — 테이블 생성 + announcements CRUD** - -```python -# realestate-lab/app/db.py -import json -import sqlite3 -import logging -from typing import Dict, Any, List, Optional -from datetime import date - -logger = logging.getLogger("realestate-lab") - -DB_PATH = "/app/data/realestate.db" - - -def _conn(): - c = sqlite3.connect(DB_PATH) - c.row_factory = sqlite3.Row - c.execute("PRAGMA journal_mode=WAL;") - c.execute("PRAGMA foreign_keys=ON;") - return c - - -def init_db(): - with _conn() as conn: - # ── announcements ──────────────────────────────────────────────── - conn.execute(""" - CREATE TABLE IF NOT EXISTS announcements ( - id INTEGER PRIMARY KEY AUTOINCREMENT, - house_manage_no TEXT NOT NULL, - pblanc_no TEXT NOT NULL, - house_nm TEXT, - house_secd TEXT, - house_dtl_secd TEXT, - rent_secd TEXT, - region_code TEXT, - region_name TEXT, - address TEXT, - total_units INTEGER, - rcrit_date TEXT, - receipt_start TEXT, - receipt_end TEXT, - spsply_start TEXT, - spsply_end TEXT, - gnrl_rank1_start TEXT, - gnrl_rank1_end TEXT, - winner_date TEXT, - contract_start TEXT, - contract_end TEXT, - homepage_url TEXT, - pblanc_url TEXT, - constructor TEXT, - developer TEXT, - move_in_month TEXT, - is_speculative_area TEXT, - is_price_cap TEXT, - contact TEXT, - status TEXT NOT NULL DEFAULT '청약예정', - source TEXT NOT NULL DEFAULT 'manual', - created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ','now')), - updated_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ','now')), - UNIQUE(house_manage_no, pblanc_no) - ); - """) - conn.execute("CREATE INDEX IF NOT EXISTS idx_ann_status ON announcements(status);") - conn.execute("CREATE INDEX IF NOT EXISTS idx_ann_region ON announcements(region_name);") - - # ── announcement_models ────────────────────────────────────────── - conn.execute(""" - CREATE TABLE IF NOT EXISTS announcement_models ( - id INTEGER PRIMARY KEY AUTOINCREMENT, - house_manage_no TEXT NOT NULL, - pblanc_no TEXT NOT NULL, - model_no TEXT, - house_ty TEXT, - supply_area REAL, - general_units INTEGER DEFAULT 0, - special_units INTEGER DEFAULT 0, - multi_child_units INTEGER DEFAULT 0, - newlywed_units INTEGER DEFAULT 0, - first_life_units INTEGER DEFAULT 0, - old_parent_units INTEGER DEFAULT 0, - institution_units INTEGER DEFAULT 0, - youth_units INTEGER DEFAULT 0, - newborn_units INTEGER DEFAULT 0, - top_amount INTEGER, - UNIQUE(house_manage_no, pblanc_no, model_no) - ); - """) - - # ── user_profile ───────────────────────────────────────────────── - conn.execute(""" - CREATE TABLE IF NOT EXISTS user_profile ( - id INTEGER PRIMARY KEY DEFAULT 1, - name TEXT, - age INTEGER, - is_homeless INTEGER, - is_householder INTEGER, - subscription_months INTEGER, - subscription_amount INTEGER, - family_members INTEGER, - has_dependents INTEGER, - children_count INTEGER DEFAULT 0, - is_newlywed INTEGER, - marriage_months INTEGER, - has_newborn INTEGER, - is_first_home INTEGER, - income_level TEXT, - preferred_regions TEXT NOT NULL DEFAULT '[]', - preferred_types TEXT NOT NULL DEFAULT '[]', - min_area REAL, - max_area REAL, - max_price INTEGER, - updated_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ','now')) - ); - """) - - # ── match_results ──────────────────────────────────────────────── - conn.execute(""" - CREATE TABLE IF NOT EXISTS match_results ( - id INTEGER PRIMARY KEY AUTOINCREMENT, - announcement_id INTEGER NOT NULL, - model_id INTEGER, - match_score INTEGER NOT NULL DEFAULT 0, - match_reasons TEXT NOT NULL DEFAULT '[]', - eligible_types TEXT NOT NULL DEFAULT '[]', - is_new INTEGER NOT NULL DEFAULT 1, - created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ','now')), - UNIQUE(announcement_id, model_id) - ); - """) - - # ── collect_log ────────────────────────────────────────────────── - conn.execute(""" - CREATE TABLE IF NOT EXISTS collect_log ( - id INTEGER PRIMARY KEY AUTOINCREMENT, - collected_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ','now')), - new_count INTEGER NOT NULL DEFAULT 0, - total_count INTEGER NOT NULL DEFAULT 0, - error TEXT - ); - """) - - -# ── 상태 자동 계산 ─────────────────────────────────────────────────────────── - -def compute_status(receipt_start: str, receipt_end: str, winner_date: str) -> str: - today = date.today().isoformat() - if receipt_start and today < receipt_start: - return "청약예정" - if receipt_start and receipt_end and receipt_start <= today <= receipt_end: - return "청약중" - if receipt_end and winner_date and receipt_end < today <= winner_date: - return "결과발표" - if winner_date and today > winner_date: - return "완료" - return "청약예정" - - -# ── announcements CRUD ─────────────────────────────────────────────────────── - -def _ann_row_to_dict(r) -> Dict[str, Any]: - return {c: r[c] for c in r.keys()} - - -def upsert_announcement(data: Dict[str, Any]) -> Dict[str, Any]: - """공고 upsert — house_manage_no + pblanc_no 기준.""" - status = compute_status( - data.get("receipt_start", ""), - data.get("receipt_end", ""), - data.get("winner_date", ""), - ) - with _conn() as conn: - conn.execute(""" - INSERT INTO announcements ( - house_manage_no, pblanc_no, house_nm, house_secd, house_dtl_secd, - rent_secd, region_code, region_name, address, total_units, - rcrit_date, receipt_start, receipt_end, spsply_start, spsply_end, - gnrl_rank1_start, gnrl_rank1_end, winner_date, contract_start, - contract_end, homepage_url, pblanc_url, constructor, developer, - move_in_month, is_speculative_area, is_price_cap, contact, - status, source - ) VALUES ( - :house_manage_no, :pblanc_no, :house_nm, :house_secd, :house_dtl_secd, - :rent_secd, :region_code, :region_name, :address, :total_units, - :rcrit_date, :receipt_start, :receipt_end, :spsply_start, :spsply_end, - :gnrl_rank1_start, :gnrl_rank1_end, :winner_date, :contract_start, - :contract_end, :homepage_url, :pblanc_url, :constructor, :developer, - :move_in_month, :is_speculative_area, :is_price_cap, :contact, - :status, :source - ) - ON CONFLICT(house_manage_no, pblanc_no) DO UPDATE SET - house_nm=excluded.house_nm, - house_secd=excluded.house_secd, - house_dtl_secd=excluded.house_dtl_secd, - rent_secd=excluded.rent_secd, - region_code=excluded.region_code, - region_name=excluded.region_name, - address=excluded.address, - total_units=excluded.total_units, - rcrit_date=excluded.rcrit_date, - receipt_start=excluded.receipt_start, - receipt_end=excluded.receipt_end, - spsply_start=excluded.spsply_start, - spsply_end=excluded.spsply_end, - gnrl_rank1_start=excluded.gnrl_rank1_start, - gnrl_rank1_end=excluded.gnrl_rank1_end, - winner_date=excluded.winner_date, - contract_start=excluded.contract_start, - contract_end=excluded.contract_end, - homepage_url=excluded.homepage_url, - pblanc_url=excluded.pblanc_url, - constructor=excluded.constructor, - developer=excluded.developer, - move_in_month=excluded.move_in_month, - is_speculative_area=excluded.is_speculative_area, - is_price_cap=excluded.is_price_cap, - contact=excluded.contact, - status=excluded.status, - updated_at=strftime('%Y-%m-%dT%H:%M:%fZ','now') - """, {**data, "status": status}) - row = conn.execute( - "SELECT * FROM announcements WHERE house_manage_no = ? AND pblanc_no = ?", - (data["house_manage_no"], data["pblanc_no"]), - ).fetchone() - return _ann_row_to_dict(row) - - -def get_announcements( - region: str = None, - status: str = None, - house_type: str = None, - matched_only: bool = False, - sort: str = "date", - page: int = 1, - size: int = 20, -) -> Dict[str, Any]: - conditions, params = [], [] - if region: - conditions.append("a.region_name = ?") - params.append(region) - if status: - conditions.append("a.status = ?") - params.append(status) - if house_type: - conditions.append("a.house_secd = ?") - params.append(house_type) - - join_clause = "" - if matched_only: - join_clause = "INNER JOIN match_results m ON m.announcement_id = a.id" - - where = f"WHERE {' AND '.join(conditions)}" if conditions else "" - - order_map = {"date": "a.rcrit_date DESC", "score": "a.id DESC", "price": "a.id ASC"} - order = order_map.get(sort, "a.rcrit_date DESC") - if matched_only and sort == "score": - order = "m.match_score DESC" - - offset = (page - 1) * size - - with _conn() as conn: - total = conn.execute( - f"SELECT COUNT(*) FROM announcements a {join_clause} {where}", params - ).fetchone()[0] - rows = conn.execute( - f"SELECT a.* FROM announcements a {join_clause} {where} ORDER BY {order} LIMIT ? OFFSET ?", - params + [size, offset], - ).fetchall() - return { - "items": [_ann_row_to_dict(r) for r in rows], - "total": total, - "page": page, - "size": size, - } - - -def get_announcement(ann_id: int) -> Optional[Dict[str, Any]]: - with _conn() as conn: - row = conn.execute("SELECT * FROM announcements WHERE id = ?", (ann_id,)).fetchone() - if not row: - return None - ann = _ann_row_to_dict(row) - models = conn.execute( - "SELECT * FROM announcement_models WHERE house_manage_no = ? AND pblanc_no = ?", - (ann["house_manage_no"], ann["pblanc_no"]), - ).fetchall() - ann["models"] = [dict(m) for m in models] - return ann - - -def create_announcement(data: Dict[str, Any]) -> Dict[str, Any]: - """수동 공고 등록 (house_manage_no 자동 생성).""" - import uuid - data["house_manage_no"] = data.get("house_manage_no", f"MANUAL-{uuid.uuid4().hex[:8]}") - data["pblanc_no"] = data.get("pblanc_no", "00") - data["source"] = "manual" - return upsert_announcement(data) - - -def update_announcement(ann_id: int, data: Dict[str, Any]) -> Optional[Dict[str, Any]]: - fields = {k: v for k, v in data.items() if v is not None} - if not fields: - return get_announcement(ann_id) - - # 날짜 변경 시 status 재계산 - with _conn() as conn: - row = conn.execute("SELECT * FROM announcements WHERE id = ?", (ann_id,)).fetchone() - if not row: - return None - current = _ann_row_to_dict(row) - merged = {**current, **fields} - status = compute_status( - merged.get("receipt_start", ""), - merged.get("receipt_end", ""), - merged.get("winner_date", ""), - ) - fields["status"] = status - - set_clauses = ", ".join(f"{k} = ?" for k in fields) - set_clauses += ", updated_at = strftime('%Y-%m-%dT%H:%M:%fZ','now')" - conn.execute( - f"UPDATE announcements SET {set_clauses} WHERE id = ?", - list(fields.values()) + [ann_id], - ) - return get_announcement(ann_id) - - -def delete_announcement(ann_id: int) -> bool: - with _conn() as conn: - # 관련 매칭 결과도 삭제 - conn.execute("DELETE FROM match_results WHERE announcement_id = ?", (ann_id,)) - cur = conn.execute("DELETE FROM announcements WHERE id = ?", (ann_id,)) - return cur.rowcount > 0 - - -def update_all_statuses(): - """모든 진행중 공고의 status를 날짜 기반으로 재계산.""" - with _conn() as conn: - rows = conn.execute( - "SELECT id, receipt_start, receipt_end, winner_date FROM announcements WHERE status != '완료'" - ).fetchall() - for r in rows: - new_status = compute_status(r["receipt_start"], r["receipt_end"], r["winner_date"]) - if new_status != "완료": - conn.execute("UPDATE announcements SET status = ? WHERE id = ?", (new_status, r["id"])) - else: - conn.execute( - "UPDATE announcements SET status = ?, updated_at = strftime('%Y-%m-%dT%H:%M:%fZ','now') WHERE id = ?", - (new_status, r["id"]), - ) - - -# ── announcement_models CRUD ───────────────────────────────────────────────── - -def upsert_model(data: Dict[str, Any]): - with _conn() as conn: - conn.execute(""" - INSERT INTO announcement_models ( - house_manage_no, pblanc_no, model_no, house_ty, supply_area, - general_units, special_units, multi_child_units, newlywed_units, - first_life_units, old_parent_units, institution_units, - youth_units, newborn_units, top_amount - ) VALUES ( - :house_manage_no, :pblanc_no, :model_no, :house_ty, :supply_area, - :general_units, :special_units, :multi_child_units, :newlywed_units, - :first_life_units, :old_parent_units, :institution_units, - :youth_units, :newborn_units, :top_amount - ) - ON CONFLICT(house_manage_no, pblanc_no, model_no) DO UPDATE SET - house_ty=excluded.house_ty, - supply_area=excluded.supply_area, - general_units=excluded.general_units, - special_units=excluded.special_units, - multi_child_units=excluded.multi_child_units, - newlywed_units=excluded.newlywed_units, - first_life_units=excluded.first_life_units, - old_parent_units=excluded.old_parent_units, - institution_units=excluded.institution_units, - youth_units=excluded.youth_units, - newborn_units=excluded.newborn_units, - top_amount=excluded.top_amount - """, data) - - -# ── user_profile CRUD ──────────────────────────────────────────────────────── - -def _profile_row_to_dict(r) -> Dict[str, Any]: - d = {} - for c in r.keys(): - val = r[c] - if c in ("is_homeless", "is_householder", "has_dependents", "is_newlywed", - "has_newborn", "is_first_home"): - d[c] = bool(val) if val is not None else None - elif c in ("preferred_regions", "preferred_types"): - d[c] = json.loads(val) if val else [] - else: - d[c] = val - return d - - -def get_profile() -> Optional[Dict[str, Any]]: - with _conn() as conn: - r = conn.execute("SELECT * FROM user_profile WHERE id = 1").fetchone() - return _profile_row_to_dict(r) if r else None - - -def upsert_profile(data: Dict[str, Any]) -> Dict[str, Any]: - updates = {} - for k, v in data.items(): - if v is None: - continue - if isinstance(v, bool): - updates[k] = 1 if v else 0 - elif isinstance(v, list): - updates[k] = json.dumps(v) - else: - updates[k] = v - - with _conn() as conn: - existing = conn.execute("SELECT id FROM user_profile WHERE id = 1").fetchone() - if existing: - if updates: - set_clauses = ", ".join(f"{k} = ?" for k in updates) - set_clauses += ", updated_at = strftime('%Y-%m-%dT%H:%M:%fZ','now')" - conn.execute( - f"UPDATE user_profile SET {set_clauses} WHERE id = 1", - list(updates.values()), - ) - else: - cols = ["id"] + list(updates.keys()) - vals = [1] + list(updates.values()) - placeholders = ", ".join("?" for _ in vals) - conn.execute( - f"INSERT INTO user_profile ({', '.join(cols)}) VALUES ({placeholders})", - vals, - ) - row = conn.execute("SELECT * FROM user_profile WHERE id = 1").fetchone() - return _profile_row_to_dict(row) - - -# ── match_results CRUD ─────────────────────────────────────────────────────── - -def save_match_result(data: Dict[str, Any]): - with _conn() as conn: - conn.execute(""" - INSERT INTO match_results (announcement_id, model_id, match_score, match_reasons, eligible_types, is_new) - VALUES (:announcement_id, :model_id, :match_score, :match_reasons, :eligible_types, 1) - ON CONFLICT(announcement_id, model_id) DO UPDATE SET - match_score=excluded.match_score, - match_reasons=excluded.match_reasons, - eligible_types=excluded.eligible_types - """, { - **data, - "match_reasons": json.dumps(data.get("match_reasons", [])), - "eligible_types": json.dumps(data.get("eligible_types", [])), - }) - - -def get_matches(page: int = 1, size: int = 20) -> Dict[str, Any]: - offset = (page - 1) * size - with _conn() as conn: - total = conn.execute("SELECT COUNT(*) FROM match_results").fetchone()[0] - rows = conn.execute(""" - SELECT m.*, a.house_nm, a.region_name, a.address, a.status as ann_status, - a.receipt_start, a.receipt_end, a.winner_date, a.pblanc_url - FROM match_results m - JOIN announcements a ON a.id = m.announcement_id - ORDER BY m.is_new DESC, m.match_score DESC - LIMIT ? OFFSET ? - """, (size, offset)).fetchall() - - items = [] - for r in rows: - d = {c: r[c] for c in r.keys()} - d["match_reasons"] = json.loads(d["match_reasons"]) if d["match_reasons"] else [] - d["eligible_types"] = json.loads(d["eligible_types"]) if d["eligible_types"] else [] - items.append(d) - return {"items": items, "total": total, "page": page, "size": size} - - -def mark_match_read(match_id: int) -> bool: - with _conn() as conn: - cur = conn.execute("UPDATE match_results SET is_new = 0 WHERE id = ?", (match_id,)) - return cur.rowcount > 0 - - -def clear_match_results(): - with _conn() as conn: - conn.execute("DELETE FROM match_results") - - -# ── collect_log CRUD ───────────────────────────────────────────────────────── - -def save_collect_log(new_count: int, total_count: int, error: str = None): - with _conn() as conn: - conn.execute( - "INSERT INTO collect_log (new_count, total_count, error) VALUES (?, ?, ?)", - (new_count, total_count, error), - ) - - -def get_last_collect_log() -> Optional[Dict[str, Any]]: - with _conn() as conn: - r = conn.execute("SELECT * FROM collect_log ORDER BY id DESC LIMIT 1").fetchone() - return dict(r) if r else None - - -# ── 대시보드 ───────────────────────────────────────────────────────────────── - -def get_dashboard() -> Dict[str, Any]: - with _conn() as conn: - active = conn.execute( - "SELECT COUNT(*) FROM announcements WHERE status IN ('청약예정', '청약중')" - ).fetchone()[0] - new_matches = conn.execute( - "SELECT COUNT(*) FROM match_results WHERE is_new = 1" - ).fetchone()[0] - upcoming = conn.execute(""" - SELECT id, house_nm, receipt_start, receipt_end, status - FROM announcements - WHERE status IN ('청약예정', '청약중') - ORDER BY receipt_start ASC - LIMIT 5 - """).fetchall() - return { - "active_count": active, - "new_match_count": new_matches, - "upcoming": [dict(r) for r in upcoming], - } -``` - -- [ ] **Step 2: Commit** - -```bash -git add realestate-lab/app/db.py -git commit -m "feat(realestate-lab): DB 레이어 — 테이블 생성 + 전체 CRUD" -``` - ---- - -### Task 4: 수집기 (collector.py) - -**Files:** -- Create: `realestate-lab/app/collector.py` - -- [ ] **Step 1: collector.py 작성** - -```python -# realestate-lab/app/collector.py -import os -import logging -import requests -from typing import List, Dict, Any - -from .db import upsert_announcement, upsert_model, save_collect_log - -logger = logging.getLogger("realestate-lab") - -API_BASE = "https://api.odcloud.kr/api/ApplyhomeInfoDetailSvc/v1" -API_KEY = os.getenv("DATA_GO_KR_API_KEY", "") - -# 수집 대상 엔드포인트 (상세 + 주택형별 쌍) -DETAIL_ENDPOINTS = [ - ("getAPTLttotPblancDetail", "getAPTLttotPblancMdl"), - ("getUrbtyOfctlLttotPblancDetail", "getUrbtyOfctlLttotPblancMdl"), - ("getRemndrLttotPblancDetail", "getRemndrLttotPblancMdl"), - ("getPblPvtRentLttotPblancDetail", "getPblPvtRentLttotPblancMdl"), - ("getOPTLttotPblancDetail", "getOPTLttotPblancMdl"), -] - - -def _api_call(endpoint: str, params: dict = None) -> List[Dict[str, Any]]: - """공공데이터포털 API 호출. 페이지네이션 자동 처리.""" - if not API_KEY: - logger.warning("DATA_GO_KR_API_KEY 미설정 — API 수집 건너뜀") - return [] - - url = f"{API_BASE}/{endpoint}" - base_params = { - "serviceKey": API_KEY, - "perPage": 100, - "returnType": "JSON", - } - if params: - base_params.update(params) - - all_data = [] - page = 1 - while True: - base_params["page"] = page - try: - resp = requests.get(url, params=base_params, timeout=30) - resp.raise_for_status() - body = resp.json() - except Exception as e: - logger.error(f"API 호출 실패: {endpoint} page={page} — {e}") - break - - data = body.get("data", []) - if not data: - break - - all_data.extend(data) - total = body.get("totalCount", 0) - if len(all_data) >= total: - break - page += 1 - - return all_data - - -def _parse_apt_detail(raw: Dict[str, Any]) -> Dict[str, Any]: - """APT 상세 API 응답을 announcements 스키마로 변환.""" - return { - "house_manage_no": str(raw.get("HOUSE_MANAGE_NO", "")), - "pblanc_no": str(raw.get("PBLANC_NO", "")), - "house_nm": raw.get("HOUSE_NM"), - "house_secd": raw.get("HOUSE_SECD"), - "house_dtl_secd": raw.get("HOUSE_DTL_SECD"), - "rent_secd": raw.get("RENT_SECD"), - "region_code": raw.get("SUBSCRPT_AREA_CODE"), - "region_name": raw.get("SUBSCRPT_AREA_CODE_NM"), - "address": raw.get("HSSPLY_ADRES"), - "total_units": raw.get("TOT_SUPLY_HSHLDCO"), - "rcrit_date": raw.get("RCRIT_PBLANC_DE"), - "receipt_start": raw.get("RCEPT_BGNDE") or raw.get("SUBSCRPT_RCEPT_BGNDE"), - "receipt_end": raw.get("RCEPT_ENDDE") or raw.get("SUBSCRPT_RCEPT_ENDDE"), - "spsply_start": raw.get("SPSPLY_RCEPT_BGNDE"), - "spsply_end": raw.get("SPSPLY_RCEPT_ENDDE"), - "gnrl_rank1_start": raw.get("GNRL_RNK1_CRSPAREA_RCPTDE") or raw.get("GNRL_RCEPT_BGNDE"), - "gnrl_rank1_end": raw.get("GNRL_RNK1_CRSPAREA_ENDDE") or raw.get("GNRL_RCEPT_ENDDE"), - "winner_date": raw.get("PRZWNER_PRESNATN_DE"), - "contract_start": raw.get("CNTRCT_CNCLS_BGNDE"), - "contract_end": raw.get("CNTRCT_CNCLS_ENDDE"), - "homepage_url": raw.get("HMPG_ADRES"), - "pblanc_url": raw.get("PBLANC_URL"), - "constructor": raw.get("CNSTRCT_ENTRPS_NM"), - "developer": raw.get("BSNS_MBY_NM"), - "move_in_month": raw.get("MVN_PREARNGE_YM"), - "is_speculative_area": raw.get("SPECLT_RDN_EARTH_AT"), - "is_price_cap": raw.get("PARCPRC_ULS_AT"), - "contact": raw.get("MDHS_TELNO"), - "source": "auto", - } - - -def _parse_model(raw: Dict[str, Any]) -> Dict[str, Any]: - """주택형별 API 응답을 announcement_models 스키마로 변환.""" - top = raw.get("LTTOT_TOP_AMOUNT") - if isinstance(top, str): - top = int(top.replace(",", "")) if top.strip() else None - return { - "house_manage_no": str(raw.get("HOUSE_MANAGE_NO", "")), - "pblanc_no": str(raw.get("PBLANC_NO", "")), - "model_no": raw.get("MODEL_NO"), - "house_ty": raw.get("HOUSE_TY"), - "supply_area": float(raw["SUPLY_AR"]) if raw.get("SUPLY_AR") else None, - "general_units": raw.get("SUPLY_HSHLDCO", 0) or 0, - "special_units": raw.get("SPSPLY_HSHLDCO", 0) or 0, - "multi_child_units": raw.get("MNYCH_HSHLDCO", 0) or 0, - "newlywed_units": raw.get("NWWDS_HSHLDCO", 0) or 0, - "first_life_units": raw.get("LFE_FRST_HSHLDCO", 0) or 0, - "old_parent_units": raw.get("OLD_PARNTS_SUPORT_HSHLDCO", 0) or 0, - "institution_units": raw.get("INSTT_RECOMEND_HSHLDCO", 0) or 0, - "youth_units": raw.get("YGMN_HSHLDCO", 0) or 0, - "newborn_units": raw.get("NWBB_HSHLDCO", 0) or 0, - "top_amount": top, - } - - -def collect_all() -> Dict[str, int]: - """전체 수집 실행. 반환: {"new_count": N, "total_count": N}""" - if not API_KEY: - logger.warning("DATA_GO_KR_API_KEY 미설정 — 수집 건너뜀") - save_collect_log(0, 0, "API 키 미설정") - return {"new_count": 0, "total_count": 0} - - total_count = 0 - new_count = 0 - errors = [] - - for detail_ep, model_ep in DETAIL_ENDPOINTS: - try: - # 상세 공고 수집 - details = _api_call(detail_ep) - for raw in details: - parsed = _parse_apt_detail(raw) - if not parsed["house_manage_no"]: - continue - upsert_announcement(parsed) - total_count += 1 - - # 주택형별 상세 수집 - models = _api_call(model_ep) - for raw in models: - parsed = _parse_model(raw) - if not parsed["house_manage_no"]: - continue - upsert_model(parsed) - - except Exception as e: - logger.error(f"수집 에러 ({detail_ep}): {e}") - errors.append(f"{detail_ep}: {str(e)}") - - error_msg = "; ".join(errors) if errors else None - save_collect_log(new_count, total_count, error_msg) - logger.info(f"수집 완료: total={total_count}, new={new_count}, errors={len(errors)}") - return {"new_count": new_count, "total_count": total_count} -``` - -- [ ] **Step 2: Commit** - -```bash -git add realestate-lab/app/collector.py -git commit -m "feat(realestate-lab): 공공데이터포털 API 수집기" -``` - ---- - -### Task 5: 매칭 엔진 (matcher.py) - -**Files:** -- Create: `realestate-lab/app/matcher.py` - -- [ ] **Step 1: matcher.py 작성** - -```python -# realestate-lab/app/matcher.py -import json -import logging -from typing import Dict, Any, List - -from .db import ( - get_profile, get_announcements, save_match_result, clear_match_results, _conn, -) - -logger = logging.getLogger("realestate-lab") - - -def _check_eligible_types(profile: Dict[str, Any], ann: Dict[str, Any]) -> List[str]: - """프로필 기반 지원 가능 공급유형 판별.""" - types = [] - is_homeless = profile.get("is_homeless", False) - is_householder = profile.get("is_householder", False) - sub_months = profile.get("subscription_months", 0) or 0 - is_speculative = ann.get("is_speculative_area", "") == "Y" - - # 일반공급 1순위 - required_months = 24 if is_speculative else 12 - if is_homeless and is_householder and sub_months >= required_months: - types.append("일반1순위") - elif is_homeless: - types.append("일반2순위") - - # 특별공급 — 신혼부부 - if profile.get("is_newlywed") and is_homeless: - types.append("특별-신혼부부") - - # 특별공급 — 생애최초 - if profile.get("is_first_home") and is_homeless: - types.append("특별-생애최초") - - # 특별공급 — 다자녀 - children = profile.get("children_count", 0) or 0 - if children >= 2 and is_homeless: - types.append("특별-다자녀") - - # 특별공급 — 노부모부양 - if profile.get("has_dependents") and is_homeless: - types.append("특별-노부모부양") - - # 특별공급 — 청년 - age = profile.get("age", 0) or 0 - if 19 <= age <= 39 and is_homeless: - types.append("특별-청년") - - # 특별공급 — 신생아 - if profile.get("has_newborn") and is_homeless: - types.append("특별-신생아") - - return types - - -def _compute_score(profile: Dict[str, Any], ann: Dict[str, Any], models: List[Dict]) -> Dict[str, Any]: - """매칭 점수 산출 (0~100).""" - score = 0 - reasons = [] - - # 1. 지역 매칭 (30점) - pref_regions = profile.get("preferred_regions", []) - if pref_regions and ann.get("region_name"): - if ann["region_name"] in pref_regions: - score += 30 - reasons.append(f"지역 일치: {ann['region_name']}") - - # 2. 주택유형 매칭 (10점) - pref_types = profile.get("preferred_types", []) - type_map = {"01": "APT", "02": "오피스텔", "04": "무순위", "09": "민간사전청약", "10": "신혼희망타운"} - ann_type = type_map.get(ann.get("house_secd", ""), ann.get("house_secd", "")) - if pref_types and ann_type in pref_types: - score += 10 - reasons.append(f"유형 일치: {ann_type}") - - # 3. 면적 매칭 (15점) - min_area = profile.get("min_area") - max_area = profile.get("max_area") - if models and (min_area is not None or max_area is not None): - for m in models: - area = m.get("supply_area", 0) or 0 - in_range = True - if min_area and area < min_area: - in_range = False - if max_area and area > max_area: - in_range = False - if in_range and area > 0: - score += 15 - reasons.append(f"면적 범위 내: {area}㎡") - break - - # 4. 가격 매칭 (15점) - max_price = profile.get("max_price") - if models and max_price: - for m in models: - top = m.get("top_amount") - if top and top <= max_price: - score += 15 - reasons.append(f"가격 범위 내: {top}만원") - break - - # 5. 자격 매칭 (30점) - eligible = _check_eligible_types(profile, ann) - if eligible: - eligibility_score = min(len(eligible) * 10, 30) - score += eligibility_score - reasons.append(f"지원 가능: {', '.join(eligible)}") - - return { - "match_score": score, - "match_reasons": reasons, - "eligible_types": eligible, - } - - -def run_matching(): - """전체 매칭 실행 — 프로필과 모든 활성 공고를 매칭.""" - profile = get_profile() - if not profile: - logger.info("프로필 미설정 — 매칭 건너뜀") - return - - # 기존 매칭 결과 초기화 - clear_match_results() - - with _conn() as conn: - anns = conn.execute( - "SELECT * FROM announcements WHERE status IN ('청약예정', '청약중')" - ).fetchall() - - for ann_row in anns: - ann = {c: ann_row[c] for c in ann_row.keys()} - models = conn.execute( - "SELECT * FROM announcement_models WHERE house_manage_no = ? AND pblanc_no = ?", - (ann["house_manage_no"], ann["pblanc_no"]), - ).fetchall() - model_list = [dict(m) for m in models] - - result = _compute_score(profile, ann, model_list) - if result["match_score"] > 0: - save_match_result({ - "announcement_id": ann["id"], - "model_id": None, - "match_score": result["match_score"], - "match_reasons": result["match_reasons"], - "eligible_types": result["eligible_types"], - }) - - logger.info("매칭 완료") -``` - -- [ ] **Step 2: Commit** - -```bash -git add realestate-lab/app/matcher.py -git commit -m "feat(realestate-lab): 프로필 기반 매칭 엔진" -``` - ---- - -### Task 6: FastAPI 앱 (main.py) - -**Files:** -- Create: `realestate-lab/app/main.py` - -- [ ] **Step 1: main.py 작성** - -```python -# realestate-lab/app/main.py -import os -import logging -from contextlib import asynccontextmanager -from fastapi import FastAPI, Query, HTTPException -from fastapi.middleware.cors import CORSMiddleware -from apscheduler.schedulers.background import BackgroundScheduler - -from .db import ( - init_db, get_announcements, get_announcement, create_announcement, - update_announcement, delete_announcement, update_all_statuses, - get_profile, upsert_profile, get_matches, mark_match_read, - get_last_collect_log, get_dashboard, -) -from .collector import collect_all -from .matcher import run_matching -from .models import AnnouncementCreate, AnnouncementUpdate, ProfileUpdate - -logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(name)s] %(levelname)s %(message)s") -logger = logging.getLogger("realestate-lab") - -scheduler = BackgroundScheduler(timezone=os.getenv("TZ", "Asia/Seoul")) - - -def scheduled_collect(): - """매일 09:00 — 수집 + 매칭""" - logger.info("스케줄 수집 시작") - collect_all() - run_matching() - logger.info("스케줄 수집 + 매칭 완료") - - -def scheduled_status_update(): - """매일 00:00 — 상태 갱신 + 재매칭""" - logger.info("상태 갱신 시작") - update_all_statuses() - run_matching() - logger.info("상태 갱신 + 재매칭 완료") - - -@asynccontextmanager -async def lifespan(app: FastAPI): - init_db() - scheduler.add_job(scheduled_collect, "cron", hour=9, minute=0, id="collect") - scheduler.add_job(scheduled_status_update, "cron", hour=0, minute=0, id="status_update") - scheduler.start() - logger.info("realestate-lab 시작") - yield - scheduler.shutdown() - - -app = FastAPI(lifespan=lifespan) - -_cors_origins = os.getenv("CORS_ALLOW_ORIGINS", "http://localhost:3007,http://localhost:8080").split(",") -app.add_middleware( - CORSMiddleware, - allow_origins=[o.strip() for o in _cors_origins], - allow_credentials=False, - allow_methods=["GET", "POST", "PUT", "PATCH", "DELETE", "OPTIONS"], - allow_headers=["Content-Type"], -) - - -@app.get("/health") -def health(): - return {"status": "ok"} - - -# ── 공고 API ───────────────────────────────────────────────────────────────── - -@app.get("/api/realestate/announcements") -def api_announcements( - region: str = None, - status: str = None, - house_type: str = None, - matched_only: bool = False, - sort: str = "date", - page: int = Query(1, ge=1), - size: int = Query(20, ge=1, le=100), -): - return get_announcements(region, status, house_type, matched_only, sort, page, size) - - -@app.get("/api/realestate/announcements/{ann_id}") -def api_announcement_detail(ann_id: int): - ann = get_announcement(ann_id) - if not ann: - raise HTTPException(status_code=404, detail="Announcement not found") - return ann - - -@app.post("/api/realestate/announcements", status_code=201) -def api_announcement_create(body: AnnouncementCreate): - return create_announcement(body.model_dump()) - - -@app.put("/api/realestate/announcements/{ann_id}") -def api_announcement_update(ann_id: int, body: AnnouncementUpdate): - updated = update_announcement(ann_id, body.model_dump(exclude_none=True)) - if not updated: - raise HTTPException(status_code=404, detail="Announcement not found") - return updated - - -@app.delete("/api/realestate/announcements/{ann_id}") -def api_announcement_delete(ann_id: int): - if not delete_announcement(ann_id): - raise HTTPException(status_code=404, detail="Announcement not found") - return {"ok": True} - - -# ── 수집 API ───────────────────────────────────────────────────────────────── - -@app.post("/api/realestate/collect") -def api_collect(): - result = collect_all() - run_matching() - return result - - -@app.get("/api/realestate/collect/status") -def api_collect_status(): - log = get_last_collect_log() - return log if log else {"collected_at": None, "new_count": 0, "total_count": 0, "error": None} - - -# ── 프로필 API ─────────────────────────────────────────────────────────────── - -@app.get("/api/realestate/profile") -def api_profile_get(): - profile = get_profile() - return profile if profile else {} - - -@app.put("/api/realestate/profile") -def api_profile_update(body: ProfileUpdate): - return upsert_profile(body.model_dump(exclude_none=True)) - - -# ── 매칭 API ───────────────────────────────────────────────────────────────── - -@app.get("/api/realestate/matches") -def api_matches(page: int = Query(1, ge=1), size: int = Query(20, ge=1, le=100)): - return get_matches(page, size) - - -@app.post("/api/realestate/matches/refresh") -def api_matches_refresh(): - run_matching() - return {"ok": True} - - -@app.patch("/api/realestate/matches/{match_id}/read") -def api_match_read(match_id: int): - if not mark_match_read(match_id): - raise HTTPException(status_code=404, detail="Match not found") - return {"ok": True} - - -# ── 대시보드 API ───────────────────────────────────────────────────────────── - -@app.get("/api/realestate/dashboard") -def api_dashboard(): - return get_dashboard() -``` - -- [ ] **Step 2: Commit** - -```bash -git add realestate-lab/app/main.py -git commit -m "feat(realestate-lab): FastAPI 앱 + 스케줄러 + 전체 API 라우트" -``` - ---- - -### Task 7: 인프라 통합 (docker-compose, nginx, deploy script) - -**Files:** -- Modify: `docker-compose.yml` -- Modify: `nginx/default.conf` -- Modify: `scripts/deploy-nas.sh` - -- [ ] **Step 1: docker-compose.yml에 realestate-lab 서비스 추가** - -`blog-lab` 서비스 블록 뒤, `travel-proxy` 블록 앞에 추가: - -```yaml - realestate-lab: - build: - context: ./realestate-lab - container_name: realestate-lab - restart: unless-stopped - ports: - - "18800:8000" - environment: - - TZ=${TZ:-Asia/Seoul} - - DATA_GO_KR_API_KEY=${DATA_GO_KR_API_KEY:-} - - CORS_ALLOW_ORIGINS=${CORS_ALLOW_ORIGINS:-http://localhost:3007,http://localhost:8080} - volumes: - - ${REALESTATE_DATA_PATH:-./data/realestate}:/app/data - healthcheck: - test: ["CMD", "python", "-c", "import urllib.request; urllib.request.urlopen('http://localhost:8000/health')"] - interval: 30s - timeout: 5s - retries: 3 -``` - -- [ ] **Step 2: nginx/default.conf에 realestate 프록시 추가** - -`/api/music/` 블록 뒤에 추가: - -```nginx - # realestate API - location /api/realestate/ { - proxy_http_version 1.1; - proxy_set_header Host $host; - proxy_set_header X-Real-IP $remote_addr; - proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; - proxy_set_header X-Forwarded-Proto $scheme; - proxy_pass http://realestate-lab:8000/api/realestate/; - } -``` - -- [ ] **Step 3: scripts/deploy-nas.sh rsync 대상에 realestate-lab 추가** - -변경 전: -```bash -for dir in backend travel-proxy deployer stock-lab music-lab blog-lab nginx scripts; do -``` - -변경 후: -```bash -for dir in backend travel-proxy deployer stock-lab music-lab blog-lab realestate-lab nginx scripts; do -``` - -- [ ] **Step 4: nginx frontend 컨테이너 depends_on에 realestate-lab 추가** - -```yaml - depends_on: - - music-lab - - blog-lab - - realestate-lab -``` - -- [ ] **Step 5: Commit** - -```bash -git add docker-compose.yml nginx/default.conf scripts/deploy-nas.sh -git commit -m "infra: realestate-lab Docker/Nginx/배포 스크립트 통합" -``` - ---- - -### Task 8: lotto-backend에서 청약 코드 제거 - -**Files:** -- Modify: `backend/app/main.py` — 라인 22~27 import, 881~957 (realestate), 960~1064 (subscription) -- Modify: `backend/app/db.py` — 라인 184~214 (realestate_complexes 테이블), 216~300 (subscription 테이블), 837~963 (realestate CRUD), 966~1327 (subscription CRUD) - -- [ ] **Step 1: backend/app/main.py에서 청약 import 제거** - -라인 22~27 변경 — `# realestate` 와 `# subscription` import 블록 삭제: - -변경 전: -```python - # realestate - get_all_complexes, get_complex, create_complex, update_complex, delete_complex, - # subscription - get_all_subscription_items, create_subscription_item, - update_subscription_item, delete_subscription_item, - get_subscription_profile, upsert_subscription_profile, -``` - -변경 후: (해당 6줄 삭제) - -- [ ] **Step 2: backend/app/main.py에서 RealEstate API 섹션 삭제 (라인 881~957)** - -`# ── RealEstate API ──` 부터 `return {"ok": True}` (957번 라인)까지 전체 삭제: -- `VALID_STATUSES`, `VALID_PRIORITIES` 상수 -- `ComplexCreate`, `ComplexUpdate` 모델 -- `api_realestate_list`, `api_realestate_create`, `api_realestate_update`, `api_realestate_delete` 라우트 - -- [ ] **Step 3: backend/app/main.py에서 Subscription API 섹션 삭제 (라인 960~1064)** - -`# ── Subscription API ──` 부터 파일 끝 `upsert_subscription_profile` 호출까지 전체 삭제: -- `SubscriptionItemCreate`, `SubscriptionItemUpdate`, `SubscriptionProfile` 모델 -- `api_subscription_list`, `api_subscription_create`, `api_subscription_update`, `api_subscription_delete` 라우트 -- `api_subscription_profile_get`, `api_subscription_profile_put` 라우트 - -- [ ] **Step 4: backend/app/db.py에서 realestate_complexes 테이블 생성 삭제 (라인 184~214)** - -`# ── realestate_complexes 테이블 ──` 블록 전체 삭제 (CREATE TABLE + CREATE INDEX) - -- [ ] **Step 5: backend/app/db.py에서 subscription_items 테이블 생성 삭제 (라인 216~252)** - -`# ── subscription_items 테이블 ──` 블록 전체 삭제 (CREATE TABLE + CREATE INDEX) - -- [ ] **Step 6: backend/app/db.py에서 subscription_profile 테이블 생성 삭제 (라인 282~300)** - -`# ── subscription_profile 테이블 ──` 블록 전체 삭제 - -- [ ] **Step 7: backend/app/db.py에서 realestate_complexes CRUD 삭제 (라인 837~963)** - -`# ── realestate_complexes CRUD ──` 부터 `delete_complex` 함수 끝까지 전체 삭제: -- `_complex_row_to_dict`, `get_all_complexes`, `get_complex`, `create_complex`, `update_complex`, `delete_complex` - -- [ ] **Step 8: backend/app/db.py에서 subscription CRUD 삭제 (라인 966~1327)** - -`# ── subscription_items CRUD ──` 부터 `upsert_subscription_profile` 함수 끝까지 전체 삭제: -- `_SUB_ITEM_FIELD_MAP`, `_sub_item_row_to_dict`, `get_all_subscription_items`, `create_subscription_item`, `update_subscription_item`, `delete_subscription_item` -- `_profile_row_to_dict`, `get_subscription_profile`, `upsert_subscription_profile` - -- [ ] **Step 9: Commit** - -```bash -git add backend/app/main.py backend/app/db.py -git commit -m "refactor(lotto-backend): 청약 관련 코드 완전 제거 — realestate-lab으로 이관" -``` - ---- - -### Task 9: CLAUDE.md 업데이트 - -**Files:** -- Modify: `CLAUDE.md` - -- [ ] **Step 1: CLAUDE.md 서비스 목록에 realestate-lab 추가** - -Docker 서비스 & 포트 테이블에 추가: -``` -| `realestate-lab` | 18800 | 부동산 청약 자동 수집·매칭 API | -``` - -Nginx 라우팅 규칙 테이블에 추가: -``` -| `/api/realestate/` | `realestate-lab:8000` | 부동산 청약 API | -``` - -서비스별 핵심 정보 섹션에 realestate-lab 추가. - -lotto-lab 테이블 목록에서 `realestate_complexes`, `subscription_items`, `subscription_profile` 참조 삭제 (해당 테이블이 lotto.db 테이블 목록에 있는 경우). - -- [ ] **Step 2: Commit** - -```bash -git add CLAUDE.md -git commit -m "docs: CLAUDE.md에 realestate-lab 서비스 정보 추가" -``` - ---- - -### Task 10: 로컬 빌드 검증 - -- [ ] **Step 1: Docker 빌드 테스트** - -```bash -cd C:/Users/jaeoh/Desktop/workspace/web-backend -docker compose build realestate-lab -``` - -Expected: 빌드 성공 - -- [ ] **Step 2: 서비스 단독 기동 테스트** - -```bash -docker compose up -d realestate-lab -``` - -- [ ] **Step 3: 헬스체크** - -```bash -curl http://localhost:18800/health -``` - -Expected: `{"status": "ok"}` - -- [ ] **Step 4: 프로필 API 테스트** - -```bash -curl -X PUT http://localhost:18800/api/realestate/profile \ - -H "Content-Type: application/json" \ - -d '{"name":"test","age":30,"is_homeless":true,"preferred_regions":["서울"]}' -``` - -Expected: 프로필 JSON 응답 - -- [ ] **Step 5: 수동 수집 테스트** - -```bash -curl -X POST http://localhost:18800/api/realestate/collect -``` - -Expected: `{"new_count": N, "total_count": N}` (API 키 설정 시 실제 데이터 수집) - -- [ ] **Step 6: 대시보드 테스트** - -```bash -curl http://localhost:18800/api/realestate/dashboard -``` - -Expected: `{"active_count": N, "new_match_count": N, "upcoming": [...]}` - -- [ ] **Step 7: lotto-backend 정상 동작 확인** - -```bash -curl http://localhost:18000/health -``` - -Expected: `{"status":"ok"}` (청약 코드 제거 후에도 정상) diff --git a/docs/superpowers/plans/2026-04-08-music-lab-suno-enhancement.md b/docs/superpowers/plans/2026-04-08-music-lab-suno-enhancement.md deleted file mode 100644 index ac952bf..0000000 --- a/docs/superpowers/plans/2026-04-08-music-lab-suno-enhancement.md +++ /dev/null @@ -1,2594 +0,0 @@ -# Music Lab Suno API 전체 기능 확장 구현 계획 - -> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. - -**Goal:** Suno API의 미사용 기능을 전부 활용하여 music-lab을 완전한 AI 음악 프로덕션 스튜디오로 업그레이드한다. - -**Architecture:** 백엔드(music-lab)에 Suno API 신규 엔드포인트를 추가하고, 공통 폴링 헬퍼를 추출하여 중복을 제거한다. 프론트엔드(web-ui)는 1,725줄 단일 파일을 컴포넌트별로 분할하고 Phase별 UI를 추가한다. - -**Tech Stack:** Python 3.12, FastAPI, SQLite, React 18, Vanilla CSS, Vite - -**Spec:** `docs/superpowers/specs/2026-04-08-music-lab-suno-enhancement-design.md` - ---- - -## 파일 구조 - -### 백엔드 (web-backend/music-lab/) - -| 파일 | 작업 | -|------|------| -| `app/suno_provider.py` | 수정: V5_5 모델 추가, 공통 폴링 헬퍼 추출, 신규 파라미터 매핑, Phase 1~3 함수 추가 | -| `app/main.py` | 수정: GenerateRequest 확장, Phase 1~3 엔드포인트 추가 | -| `app/db.py` | 수정: 마이그레이션 컬럼 추가, 트랙 업데이트 함수 추가 | - -### 프론트엔드 (web-ui/src/) - -| 파일 | 작업 | -|------|------| -| `api.js` | 수정: Phase 1~3 API 함수 추가 | -| `pages/music/MusicStudio.jsx` | 수정: 컴포넌트 분할 후 메인 셸 역할 | -| `pages/music/MusicStudio.css` | 수정: 신규 컴포넌트 스타일 추가 | -| `pages/music/components/CreditsBadge.jsx` | 생성: 크레딧 잔액 배지 | -| `pages/music/components/CreateTab.jsx` | 생성: 생성 폼 (기존 코드 이동 + Phase 1 확장) | -| `pages/music/components/LyricsTab.jsx` | 생성: 가사 관리 (기존 코드 이동) | -| `pages/music/components/LibraryTab.jsx` | 생성: 라이브러리 (기존 코드 이동 + 더보기 메뉴) | -| `pages/music/components/AudioPlayer.jsx` | 생성: 오디오 플레이어 (기존 코드 이동) | -| `pages/music/components/CoverArtModal.jsx` | 생성: 커버 이미지 선택 모달 | -| `pages/music/components/StemModal.jsx` | 생성: 12스템 결과 모달 | -| `pages/music/components/SyncedLyricsPlayer.jsx` | 생성: 타임스탬프 가사 오버레이 | -| `pages/music/components/RemixTab.jsx` | 생성: Phase 3 업로드/리믹스 탭 | - ---- - -## Phase 1: 핵심 생성 강화 - -### Task 1: 백엔드 — suno_provider 리팩토링 + V5_5 + 신규 파라미터 - -**Files:** -- Modify: `music-lab/app/suno_provider.py` - -- [ ] **Step 1: V5_5 모델 추가 + 공통 폴링 헬퍼 추출** - -`music-lab/app/suno_provider.py` 상단의 SUNO_MODELS에 V5_5를 추가하고, 기존 `_poll_until_complete`를 범용 헬퍼로 리팩토링: - -```python -# SUNO_MODELS 리스트 끝에 추가 (line 31 뒤) - {"id": "V5_5", "name": "V5.5", "max_duration": "8분", "description": "커스텀 모델, 최신 음악성"}, -``` - -`_poll_until_complete` 함수를 범용화하여 다른 Suno 작업(WAV, 스템, 커버이미지 등)에도 재사용: - -```python -def _poll_suno_record( - record_info_path: str, - suno_task_id: str, - task_id: str, - max_attempts: int = POLL_MAX_ATTEMPTS, - interval: int = POLL_INTERVAL, - progress_msg_map: dict = None, -) -> Optional[dict]: - """범용 Suno 작업 폴링. SUCCESS 시 response 객체 반환. - - record_info_path: 예) "/generate/record-info", "/wav/record-info" - progress_msg_map: 상태별 메시지 오버라이드 (예: {"PENDING": "WAV 변환 대기 중..."}) - """ - error_statuses = { - "CREATE_TASK_FAILED", "GENERATE_AUDIO_FAILED", - "CALLBACK_EXCEPTION", "SENSITIVE_WORD_ERROR", - } - default_msgs = { - "PENDING": "대기열에서 대기 중...", - "TEXT_SUCCESS": "가사 생성 완료, 음악 생성 중...", - "FIRST_SUCCESS": "첫 번째 트랙 완료, 두 번째 생성 중...", - "GENERATING": "생성 중...", - } - msgs = {**default_msgs, **(progress_msg_map or {})} - - for attempt in range(max_attempts): - time.sleep(interval) - try: - resp = requests.get( - f"{SUNO_BASE_URL}{record_info_path}", - headers=_headers(), - params={"taskId": suno_task_id}, - timeout=15, - ) - if resp.status_code != 200: - continue - - body = resp.json() - if body.get("code") != 200: - continue - - data = body.get("data", {}) - status = data.get("status", "") - progress = min(15 + int((attempt / max_attempts) * 65), 79) - - if status == "SUCCESS": - return data.get("response", data) - elif status in error_statuses: - error_msg = data.get("errorMessage") or data.get("msg") or f"Suno 작업 실패 ({status})" - update_task(task_id, "failed", 0, "", error=error_msg) - return None - else: - msg = msgs.get(status, f"처리 중... ({status})") - if status == "FIRST_SUCCESS": - progress = max(progress, 60) - update_task(task_id, "processing", progress, msg) - - except Exception as e: - logger.warning("Suno poll error (attempt %d): %s", attempt, e) - continue - - update_task(task_id, "failed", 0, "", error="Suno 작업 타임아웃") - return None -``` - -기존 `_poll_until_complete`를 호출하는 곳(`run_suno_generation`, `run_suno_extend`, `run_vocal_removal`)을 `_poll_suno_record` 호출로 교체: - -```python -# run_suno_generation 내부 (기존: completed_tracks = _poll_until_complete(task_id, suno_task_id)) -response = _poll_suno_record("/generate/record-info", suno_task_id, task_id) -if not response: - return -completed_tracks = response.get("sunoData") or [] -if not completed_tracks: - update_task(task_id, "failed", 0, "", error="Suno 생성 완료했으나 트랙 데이터 없음") - return -``` - -`run_suno_extend`와 `run_vocal_removal`도 동일 패턴으로 교체. - -- [ ] **Step 2: _build_suno_payload에 신규 파라미터 매핑 추가** - -```python -def _build_suno_payload(params: dict) -> dict: - """프론트엔드 params → sunoapi.org 요청 형식으로 변환.""" - # ... 기존 로직 유지 ... - - # 신규 파라미터 매핑 (None이 아닌 경우에만 포함) - if params.get("vocal_gender"): - payload["vocalGender"] = params["vocal_gender"] - if params.get("negative_tags"): - payload["negativeTags"] = params["negative_tags"] - if params.get("style_weight") is not None: - payload["styleWeight"] = params["style_weight"] - if params.get("audio_weight") is not None: - payload["audioWeight"] = params["audio_weight"] - - return payload -``` - -이 4줄을 `_build_suno_payload` 함수의 `return payload` 직전에 추가. - -- [ ] **Step 3: 커버 이미지 생성 함수 추가** - -`suno_provider.py` 하단에 추가: - -```python -# ── 커버 이미지 생성 ───────────────────────────────────────────────────────── - -def run_cover_image(task_id: str, params: dict) -> None: - """Suno 곡의 커버 이미지 2장을 생성.""" - try: - if not SUNO_API_KEY: - update_task(task_id, "failed", 0, "", error="SUNO_API_KEY가 설정되지 않았습니다.") - return - - update_task(task_id, "processing", 5, "커버 이미지 생성 요청 중...") - - suno_task_id = params.get("suno_task_id", "") - if not suno_task_id: - update_task(task_id, "failed", 0, "", error="suno_task_id가 필요합니다") - return - - payload = { - "taskId": suno_task_id, - "callBackUrl": "https://example.com/noop", - } - - resp = requests.post( - f"{SUNO_BASE_URL}/suno/cover/generate", - headers=_headers(), - json=payload, - timeout=30, - ) - - if resp.status_code != 200: - update_task(task_id, "failed", 0, "", error=f"커버 이미지 API 오류: {resp.text[:300]}") - return - - body = resp.json() - if body.get("code") != 200: - update_task(task_id, "failed", 0, "", error=f"커버 이미지 거부: {body.get('msg', 'unknown')}") - return - - cover_task_id = body.get("data", {}).get("taskId", suno_task_id) - update_task(task_id, "processing", 15, "커버 이미지 생성 중...") - - response = _poll_suno_record( - "/suno/cover/record-info", cover_task_id, task_id, - max_attempts=30, interval=5, - progress_msg_map={"PENDING": "이미지 생성 대기 중...", "GENERATING": "이미지 생성 중..."}, - ) - if not response: - return - - images = response.get("images") or response.get("sunoData") or [] - image_urls = [] - if isinstance(images, list): - for img in images: - if isinstance(img, str): - image_urls.append(img) - elif isinstance(img, dict): - image_urls.append(img.get("imageUrl") or img.get("image_url", "")) - - update_task(task_id, "succeeded", 100, "커버 이미지 생성 완료", - audio_url=json.dumps(image_urls)) - - except Exception as e: - logger.exception("Cover image generation error for task %s", task_id) - update_task(task_id, "failed", 0, "", error=str(e)) -``` - -suno_provider.py 상단에 `import json` 추가 필요. - -- [ ] **Step 4: 크레딧 조회 폴백 로직** - -```python -def get_credits() -> Optional[dict]: - """Suno API 잔여 크레딧 조회. 두 엔드포인트 폴백.""" - if not SUNO_API_KEY: - return None - # 신규 엔드포인트 먼저 시도 - for path in ["/generate/credit", "/get-credits"]: - try: - resp = requests.get( - f"{SUNO_BASE_URL}{path}", - headers=_headers(), - timeout=15, - ) - if resp.status_code == 200: - body = resp.json() - data = body.get("data", body) - # /generate/credit은 정수 반환 - if isinstance(data, (int, float)): - return {"credits_left": int(data)} - return data - except Exception as e: - logger.warning("Suno credits API error (%s): %s", path, e) - return None -``` - -- [ ] **Step 5: 커밋** - -```bash -git add music-lab/app/suno_provider.py -git commit -m "refactor(music-lab): 공통 폴링 헬퍼 추출 + V5_5 모델 + 신규 파라미터 + 커버이미지" -``` - ---- - -### Task 2: 백엔드 — DB 마이그레이션 + main.py 엔드포인트 - -**Files:** -- Modify: `music-lab/app/db.py` -- Modify: `music-lab/app/main.py` - -- [ ] **Step 1: db.py 마이그레이션 + 업데이트 함수 추가** - -`db.py`의 `init_db()` 함수, 기존 마이그레이션 루프(line 72~) 뒤에 추가: - -```python - # Phase 1~3 신규 컬럼 마이그레이션 - for col, default in [ - ("cover_images", "'[]'"), - ("wav_url", "''"), - ("video_url", "''"), - ("stem_urls", "'{}'"), - ]: - try: - conn.execute(f"ALTER TABLE music_library ADD COLUMN {col} TEXT NOT NULL DEFAULT {default}") - except sqlite3.OperationalError: - pass -``` - -`_track_row_to_dict` 함수에 신규 컬럼 매핑 추가 (line 164 뒤): - -```python - "cover_images": json.loads(r["cover_images"]) if "cover_images" in keys and r["cover_images"] else [], - "wav_url": r["wav_url"] if "wav_url" in keys else "", - "video_url": r["video_url"] if "video_url" in keys else "", - "stem_urls": json.loads(r["stem_urls"]) if "stem_urls" in keys and r["stem_urls"] else {}, -``` - -파일 하단에 업데이트 함수 추가: - -```python -def update_track_cover_images(track_id: int, images: list) -> None: - with _conn() as conn: - conn.execute( - "UPDATE music_library SET cover_images=? WHERE id=?", - (json.dumps(images), track_id), - ) - - -def update_track_wav_url(track_id: int, wav_url: str) -> None: - with _conn() as conn: - conn.execute( - "UPDATE music_library SET wav_url=? WHERE id=?", - (wav_url, track_id), - ) - - -def update_track_video_url(track_id: int, video_url: str) -> None: - with _conn() as conn: - conn.execute( - "UPDATE music_library SET video_url=? WHERE id=?", - (video_url, track_id), - ) - - -def update_track_stem_urls(track_id: int, stems: dict) -> None: - with _conn() as conn: - conn.execute( - "UPDATE music_library SET stem_urls=? WHERE id=?", - (json.dumps(stems), track_id), - ) -``` - -- [ ] **Step 2: main.py — GenerateRequest 스키마 확장** - -`main.py`의 `GenerateRequest` 클래스(line 90~)에 필드 추가: - -```python -class GenerateRequest(BaseModel): - provider: str = "suno" - model: str = "V4" - title: str = "" - genre: str = "" - moods: List[str] = [] - instruments: List[str] = [] - duration_sec: Optional[int] = None - bpm: Optional[int] = None - key: str = "" - scale: str = "" - prompt: str = "" - # Suno 전용 - lyrics: str = "" - instrumental: bool = False - # Phase 1 신규 - vocal_gender: Optional[str] = None # "m" | "f" - negative_tags: Optional[str] = None # 제외 스타일 - style_weight: Optional[float] = None # 0.0~1.0 - audio_weight: Optional[float] = None # 0.0~1.0 -``` - -- [ ] **Step 3: main.py — 커버 이미지 엔드포인트 추가** - -`main.py` import에 `run_cover_image` 추가, 보컬분리 API 뒤에 엔드포인트 추가: - -```python -from .suno_provider import ( - run_suno_generation, run_suno_extend, run_vocal_removal, - run_cover_image, - generate_lyrics, get_credits, - SUNO_API_KEY, SUNO_MODELS, -) - - -# ── 커버 이미지 생성 API ──────────────────────────────────────────────────── - -class CoverImageRequest(BaseModel): - suno_task_id: str # Suno 생성 task ID - track_id: Optional[int] = None # 라이브러리 트랙 ID (결과 저장용) - - -@app.post("/api/music/cover-image") -def cover_image(req: CoverImageRequest, background_tasks: BackgroundTasks): - """Suno 곡의 커버 이미지 2장 생성.""" - if not SUNO_API_KEY: - raise HTTPException(status_code=400, detail="Suno API 키가 설정되지 않았습니다") - - task_id = str(uuid.uuid4()) - params = req.model_dump() - create_task(task_id, params, provider="suno") - background_tasks.add_task(run_cover_image, task_id, params) - return {"task_id": task_id, "provider": "suno"} -``` - -- [ ] **Step 4: 커밋** - -```bash -git add music-lab/app/db.py music-lab/app/main.py -git commit -m "feat(music-lab): Phase 1 DB 마이그레이션 + GenerateRequest 확장 + 커버이미지 엔드포인트" -``` - ---- - -### Task 3: 프론트엔드 — 컴포넌트 분할 - -**Files:** -- Modify: `web-ui/src/pages/music/MusicStudio.jsx` -- Create: `web-ui/src/pages/music/components/AudioPlayer.jsx` -- Create: `web-ui/src/pages/music/components/LyricsTab.jsx` -- Create: `web-ui/src/pages/music/components/CreditsBadge.jsx` - -이 태스크는 기존 MusicStudio.jsx에서 독립적인 컴포넌트를 별도 파일로 추출한다. 기능 변경 없이 구조만 변경. - -- [ ] **Step 1: AudioPlayer 컴포넌트 추출** - -`web-ui/src/pages/music/components/AudioPlayer.jsx` 생성: - -```jsx -import React, { useEffect, useRef, useState } from 'react'; - -const pad = (n) => String(Math.floor(n)).padStart(2, '0'); -export const fmtTime = (s) => `${pad(s / 60)}:${pad(s % 60)}`; - -const AudioPlayer = ({ audioUrl, totalSec, accentColor }) => { - const audioRef = useRef(null); - const [playing, setPlaying] = useState(false); - const [elapsed, setElapsed] = useState(0); - const [duration, setDuration] = useState(totalSec ?? 0); - const [volume, setVolume] = useState(1); - - const isFake = !audioUrl; - const timerRef = useRef(null); - const total = duration || totalSec || 60; - - const togglePlay = () => { - if (isFake) { - if (playing) { - clearInterval(timerRef.current); - setPlaying(false); - } else { - setPlaying(true); - timerRef.current = setInterval(() => { - setElapsed((e) => { - if (e >= total - 1) { - clearInterval(timerRef.current); - setPlaying(false); - return 0; - } - return e + 1; - }); - }, 1000); - } - return; - } - const el = audioRef.current; - if (!el) return; - playing ? el.pause() : el.play(); - }; - - const handleSeek = (e) => { - const rect = e.currentTarget.getBoundingClientRect(); - const ratio = (e.clientX - rect.left) / rect.width; - const newTime = ratio * total; - if (!isFake && audioRef.current) { - audioRef.current.currentTime = newTime; - } - setElapsed(newTime); - }; - - const handleVolumeChange = (e) => { - const v = Number(e.target.value); - setVolume(v); - if (!isFake && audioRef.current) audioRef.current.volume = v; - }; - - useEffect(() => () => clearInterval(timerRef.current), []); - - const progress = (elapsed / total) * 100; - - return ( -
- {!isFake && ( -