Compare commits
2 Commits
a2bd26682e
...
535ffea45a
| Author | SHA1 | Date | |
|---|---|---|---|
| 535ffea45a | |||
| 9d5583935d |
@@ -272,6 +272,11 @@ def init_db() -> None:
|
||||
"""
|
||||
)
|
||||
|
||||
# ── 추가 인덱스 ───────────────────────────────────────────────────────
|
||||
conn.execute("CREATE INDEX IF NOT EXISTS idx_reco_based_checked ON recommendations(based_on_draw, checked)")
|
||||
conn.execute("CREATE INDEX IF NOT EXISTS idx_purchase_strategy ON purchase_history(source_strategy)")
|
||||
conn.execute("CREATE INDEX IF NOT EXISTS idx_purchase_checked ON purchase_history(draw_no, checked)")
|
||||
|
||||
|
||||
# ── todos CRUD ───────────────────────────────────────────────────────────────
|
||||
|
||||
@@ -512,8 +517,6 @@ def list_recommendations_ex(
|
||||
q: Optional[str] = None,
|
||||
sort: str = "id_desc", # id_desc|created_desc|favorite_desc
|
||||
) -> List[Dict[str, Any]]:
|
||||
import json
|
||||
|
||||
where = []
|
||||
args: list[Any] = []
|
||||
|
||||
@@ -810,7 +813,7 @@ def get_simulation_candidates(run_id: int, limit: int = 100) -> List[Dict[str, A
|
||||
# ── purchase_history CRUD ─────────────────────────────────────────────────────
|
||||
|
||||
def _purchase_row_to_dict(r) -> Dict[str, Any]:
|
||||
import json as _json
|
||||
|
||||
keys = r.keys()
|
||||
numbers_raw = r["numbers"] if "numbers" in keys else "[]"
|
||||
detail_raw = r["source_detail"] if "source_detail" in keys else "{}"
|
||||
@@ -823,12 +826,12 @@ def _purchase_row_to_dict(r) -> Dict[str, Any]:
|
||||
"prize": r["prize"],
|
||||
"note": r["note"],
|
||||
"created_at": r["created_at"],
|
||||
"numbers": _json.loads(numbers_raw) if numbers_raw else [],
|
||||
"numbers": json.loads(numbers_raw) if numbers_raw else [],
|
||||
"is_real": r["is_real"] if "is_real" in keys else 1,
|
||||
"source_strategy": r["source_strategy"] if "source_strategy" in keys else "manual",
|
||||
"source_detail": _json.loads(detail_raw) if detail_raw else {},
|
||||
"source_detail": json.loads(detail_raw) if detail_raw else {},
|
||||
"checked": r["checked"] if "checked" in keys else 0,
|
||||
"results": _json.loads(results_raw) if results_raw else [],
|
||||
"results": json.loads(results_raw) if results_raw else [],
|
||||
"total_prize": r["total_prize"] if "total_prize" in keys else 0,
|
||||
}
|
||||
|
||||
@@ -836,9 +839,9 @@ def _purchase_row_to_dict(r) -> Dict[str, Any]:
|
||||
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)
|
||||
|
||||
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(
|
||||
@@ -880,7 +883,7 @@ def get_purchases(draw_no: int = None, days: int = None,
|
||||
|
||||
|
||||
def update_purchase(purchase_id: int, data: Dict[str, Any]) -> Optional[Dict[str, Any]]:
|
||||
import json as _json
|
||||
|
||||
allowed = {"draw_no", "amount", "sets", "prize", "note", "numbers", "is_real", "source_strategy"}
|
||||
updates = {k: v for k, v in data.items() if k in allowed}
|
||||
if not updates:
|
||||
@@ -889,7 +892,7 @@ def update_purchase(purchase_id: int, data: Dict[str, Any]) -> Optional[Dict[str
|
||||
return _purchase_row_to_dict(row) if row else None
|
||||
# SQLite에 전달 전 타입 변환
|
||||
if "numbers" in updates:
|
||||
updates["numbers"] = _json.dumps(updates["numbers"], ensure_ascii=False)
|
||||
updates["numbers"] = json.dumps(updates["numbers"], ensure_ascii=False)
|
||||
if "is_real" in updates:
|
||||
updates["is_real"] = 1 if updates["is_real"] else 0
|
||||
set_clause = ", ".join(f"{k} = ?" for k in updates)
|
||||
@@ -911,7 +914,7 @@ def delete_purchase(purchase_id: int) -> bool:
|
||||
|
||||
|
||||
def get_purchase_stats() -> Dict[str, Any]:
|
||||
import json as _json
|
||||
|
||||
|
||||
def _calc_group(rows):
|
||||
if not rows:
|
||||
@@ -951,7 +954,7 @@ def get_purchase_stats() -> Dict[str, Any]:
|
||||
for r in srows:
|
||||
results_raw = r.get("results", "[]")
|
||||
try:
|
||||
results = _json.loads(results_raw) if isinstance(results_raw, str) else (results_raw or [])
|
||||
results = json.loads(results_raw) if isinstance(results_raw, str) else (results_raw or [])
|
||||
except Exception:
|
||||
results = []
|
||||
for res in results:
|
||||
@@ -1015,8 +1018,8 @@ def get_weekly_report(drw_no: int) -> Optional[Dict[str, Any]]:
|
||||
).fetchone()
|
||||
if not row:
|
||||
return None
|
||||
import json as _json
|
||||
return {"drw_no": row["drw_no"], "generated_at": row["generated_at"], **_json.loads(row["report"])}
|
||||
|
||||
return {"drw_no": row["drw_no"], "generated_at": row["generated_at"], **json.loads(row["report"])}
|
||||
|
||||
|
||||
def get_all_recommendation_numbers() -> List[List[int]]:
|
||||
@@ -1086,10 +1089,10 @@ def update_strategy_weight(strategy: str, weight: float, ema_score: float,
|
||||
|
||||
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),
|
||||
(json.dumps(results, ensure_ascii=False), total_prize, purchase_id),
|
||||
)
|
||||
|
||||
|
||||
@@ -24,26 +24,7 @@ from .db import (
|
||||
replace_best_picks,
|
||||
)
|
||||
from .analyzer import build_analysis_cache, build_number_weights, score_combination
|
||||
|
||||
|
||||
def _weighted_sample_6(weights: Dict[int, float]) -> List[int]:
|
||||
"""
|
||||
가중 확률 샘플링으로 중복 없이 6개 번호 추출.
|
||||
weights: {1: w1, 2: w2, ..., 45: w45}
|
||||
"""
|
||||
pool = list(range(1, 46))
|
||||
chosen: List[int] = []
|
||||
for _ in range(6):
|
||||
total = sum(weights[n] for n in pool)
|
||||
r = random.random() * total
|
||||
acc = 0.0
|
||||
for n in pool:
|
||||
acc += weights[n]
|
||||
if acc >= r:
|
||||
chosen.append(n)
|
||||
pool.remove(n)
|
||||
break
|
||||
return chosen
|
||||
from .utils import weighted_sample_6
|
||||
|
||||
|
||||
def run_simulation(
|
||||
@@ -82,7 +63,7 @@ def run_simulation(
|
||||
attempts = 0
|
||||
while len(candidates) < n_candidates and attempts < max_attempts:
|
||||
attempts += 1
|
||||
nums = _weighted_sample_6(weights)
|
||||
nums = weighted_sample_6(weights)
|
||||
key = tuple(sorted(nums))
|
||||
if key in seen_keys:
|
||||
continue
|
||||
|
||||
@@ -2,6 +2,8 @@ import random
|
||||
from collections import Counter
|
||||
from typing import Dict, Any, List, Tuple
|
||||
|
||||
from .utils import weighted_sample_6
|
||||
|
||||
def recommend_numbers(
|
||||
draws: List[Tuple[int, List[int]]],
|
||||
*,
|
||||
@@ -40,20 +42,7 @@ def recommend_numbers(
|
||||
weights[n] = max(w, 0.1)
|
||||
|
||||
# 중복 없이 6개 뽑기(가중 샘플링)
|
||||
chosen = []
|
||||
pool = list(range(1, 46))
|
||||
for _ in range(6):
|
||||
total = sum(weights[n] for n in pool)
|
||||
r = random.random() * total
|
||||
acc = 0.0
|
||||
for n in pool:
|
||||
acc += weights[n]
|
||||
if acc >= r:
|
||||
chosen.append(n)
|
||||
pool.remove(n)
|
||||
break
|
||||
|
||||
chosen_sorted = sorted(chosen)
|
||||
chosen_sorted = sorted(weighted_sample_6(weights))
|
||||
|
||||
explain = {
|
||||
"recent_window": recent_window,
|
||||
@@ -130,20 +119,7 @@ def recommend_with_heatmap(
|
||||
weights[n] = max(w, 0.1)
|
||||
|
||||
# 4. 가중 샘플링으로 6개 선택
|
||||
chosen = []
|
||||
pool = list(range(1, 46))
|
||||
for _ in range(6):
|
||||
total = sum(weights[n] for n in pool)
|
||||
r = random.random() * total
|
||||
acc = 0.0
|
||||
for n in pool:
|
||||
acc += weights[n]
|
||||
if acc >= r:
|
||||
chosen.append(n)
|
||||
pool.remove(n)
|
||||
break
|
||||
|
||||
chosen_sorted = sorted(chosen)
|
||||
chosen_sorted = sorted(weighted_sample_6(weights))
|
||||
|
||||
# 5. 설명 데이터
|
||||
explain = {
|
||||
|
||||
@@ -1,5 +1,26 @@
|
||||
import random
|
||||
from typing import List, Dict, Any, Tuple
|
||||
|
||||
|
||||
def weighted_sample_6(weights: Dict[int, float]) -> List[int]:
|
||||
"""
|
||||
가중 확률 샘플링으로 중복 없이 6개 번호 추출.
|
||||
weights: {1: w1, 2: w2, ..., 45: w45}
|
||||
"""
|
||||
pool = list(range(1, 46))
|
||||
chosen: List[int] = []
|
||||
for _ in range(6):
|
||||
total = sum(weights[n] for n in pool)
|
||||
r = random.random() * total
|
||||
acc = 0.0
|
||||
for n in pool:
|
||||
acc += weights[n]
|
||||
if acc >= r:
|
||||
chosen.append(n)
|
||||
pool.remove(n)
|
||||
break
|
||||
return chosen
|
||||
|
||||
def calc_metrics(numbers: List[int]) -> Dict[str, Any]:
|
||||
nums = sorted(numbers)
|
||||
s = sum(nums)
|
||||
|
||||
@@ -1,4 +1,5 @@
|
||||
FROM python:3.12-alpine
|
||||
ENV PYTHONUNBUFFERED=1
|
||||
|
||||
WORKDIR /app
|
||||
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
import os
|
||||
import uuid
|
||||
import logging
|
||||
from fastapi import FastAPI, HTTPException, BackgroundTasks
|
||||
from fastapi import FastAPI, HTTPException, BackgroundTasks, Query
|
||||
from fastapi.middleware.cors import CORSMiddleware
|
||||
from pydantic import BaseModel
|
||||
from typing import List, Optional
|
||||
@@ -94,7 +94,7 @@ def start_research(req: ResearchRequest, background_tasks: BackgroundTasks):
|
||||
|
||||
|
||||
@app.get("/api/blog-marketing/research/history")
|
||||
def list_research(limit: int = 30):
|
||||
def list_research(limit: int = Query(30, ge=1, le=100)):
|
||||
return {"analyses": get_keyword_analyses(limit)}
|
||||
|
||||
|
||||
@@ -285,7 +285,7 @@ def start_regenerate(post_id: int, background_tasks: BackgroundTasks):
|
||||
# ── 포스트 CRUD API ──────────────────────────────────────────────────────────
|
||||
|
||||
@app.get("/api/blog-marketing/posts")
|
||||
def list_posts(status: str = None, limit: int = 50):
|
||||
def list_posts(status: str = None, limit: int = Query(50, ge=1, le=100)):
|
||||
return {"posts": get_posts(status=status, limit=limit)}
|
||||
|
||||
|
||||
@@ -409,7 +409,7 @@ def start_market(post_id: int, background_tasks: BackgroundTasks):
|
||||
# ── 수익 추적 API ────────────────────────────────────────────────────────────
|
||||
|
||||
@app.get("/api/blog-marketing/commissions")
|
||||
def list_commissions(post_id: int = None, limit: int = 100):
|
||||
def list_commissions(post_id: int = None, limit: int = Query(100, ge=1, le=100)):
|
||||
return {"commissions": get_commissions(post_id=post_id, limit=limit)}
|
||||
|
||||
|
||||
|
||||
@@ -4,8 +4,6 @@ import asyncio
|
||||
import logging
|
||||
import re
|
||||
from typing import Any, Dict, List, Optional, Tuple
|
||||
from urllib.parse import urlparse
|
||||
|
||||
import httpx
|
||||
from bs4 import BeautifulSoup
|
||||
|
||||
|
||||
@@ -1,5 +1,6 @@
|
||||
import os, hmac, hashlib, subprocess, threading
|
||||
from fastapi import FastAPI, Request, HTTPException, BackgroundTasks
|
||||
from fastapi.responses import JSONResponse
|
||||
import logging
|
||||
|
||||
logging.basicConfig(
|
||||
@@ -64,8 +65,15 @@ async def webhook(req: Request, background_tasks: BackgroundTasks):
|
||||
if not verify(sig, body):
|
||||
raise HTTPException(401, "bad signature")
|
||||
|
||||
# 동시 배포 방지: 이미 진행 중이면 503 반환
|
||||
if _deploy_lock.locked():
|
||||
return JSONResponse(
|
||||
status_code=503,
|
||||
content={"ok": False, "message": "Deploy already in progress"},
|
||||
)
|
||||
|
||||
# ✅ 비동기 실행: Gitea에게는 즉시 OK 응답을 주고, 배포는 뒤에서 실행
|
||||
background_tasks.add_task(run_deploy_script)
|
||||
|
||||
|
||||
return {"ok": True, "message": "Deployment started in background"}
|
||||
|
||||
|
||||
@@ -121,7 +121,7 @@ services:
|
||||
- TRAVEL_THUMB_ROOT=${TRAVEL_THUMB_ROOT:-/data/thumbs}
|
||||
- TRAVEL_MEDIA_BASE=${TRAVEL_MEDIA_BASE:-/media/travel}
|
||||
- TRAVEL_CACHE_TTL=${TRAVEL_CACHE_TTL:-300}
|
||||
- CORS_ALLOW_ORIGINS=${CORS_ALLOW_ORIGINS:-*}
|
||||
- CORS_ALLOW_ORIGINS=${CORS_ALLOW_ORIGINS:-http://localhost:3007,http://localhost:8080}
|
||||
volumes:
|
||||
- ${PHOTO_PATH}:/data/travel:ro
|
||||
- ${RUNTIME_PATH}/travel-thumbs:/data/thumbs:rw
|
||||
|
||||
672
docs/superpowers/plans/2026-04-07-pet-lab.md
Normal file
672
docs/superpowers/plans/2026-04-07-pet-lab.md
Normal file
@@ -0,0 +1,672 @@
|
||||
# Pet 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:** Windows 데스크톱 펫 애플리케이션 — 화면 하단에 고정된 캐릭터가 마우스 시선을 추적하고 클릭/우클릭 상호작용을 지원한다.
|
||||
|
||||
**Architecture:** PyQt5 투명 프레임리스 윈도우에 캐릭터 이미지를 표시. QTimer 루프로 마우스 좌표를 폴링하여 이미지 기울기/반전으로 시선을 표현. 좌클릭(점프)/더블클릭(흔들기) 애니메이션과 우클릭 컨텍스트 메뉴 제공.
|
||||
|
||||
**Tech Stack:** Python 3.12, PyQt5
|
||||
|
||||
**Project Path:** `C:\Users\jaeoh\Desktop\workspace\pet-lab`
|
||||
|
||||
---
|
||||
|
||||
## File Structure
|
||||
|
||||
| 파일 | 역할 | 생성/수정 |
|
||||
|------|------|-----------|
|
||||
| `app/config.py` | 상수 정의 (크기, 위치, 애니메이션, 경로) | Create |
|
||||
| `app/eye_tracker.py` | 마우스→기울기 각도/반전 계산 (순수 함수) | Create |
|
||||
| `app/pet_widget.py` | 투명 윈도우 + 캐릭터 렌더링 + QTimer 루프 | Create |
|
||||
| `app/interaction.py` | 클릭 애니메이션 + 우클릭 메뉴 | Create |
|
||||
| `app/main.py` | 엔트리포인트 (QApplication 초기화) | Create |
|
||||
| `assets/characters/박뚱냥.png` | 캐릭터 이미지 | Copy |
|
||||
| `requirements.txt` | PyQt5 의존성 | Create |
|
||||
| `tests/test_eye_tracker.py` | eye_tracker 단위 테스트 | Create |
|
||||
| `tests/test_config.py` | config 상수 검증 테스트 | Create |
|
||||
|
||||
---
|
||||
|
||||
### Task 1: 프로젝트 초기화 + config.py
|
||||
|
||||
**Files:**
|
||||
- Create: `C:\Users\jaeoh\Desktop\workspace\pet-lab\requirements.txt`
|
||||
- Create: `C:\Users\jaeoh\Desktop\workspace\pet-lab\app\config.py`
|
||||
- Create: `C:\Users\jaeoh\Desktop\workspace\pet-lab\tests\test_config.py`
|
||||
- Copy: `Z:\homes\jaeoh\캐릭터\박뚱냥.jpg` → `C:\Users\jaeoh\Desktop\workspace\pet-lab\assets\characters\박뚱냥.png`
|
||||
|
||||
- [ ] **Step 1: 프로젝트 디렉토리 생성 및 git 초기화**
|
||||
|
||||
```bash
|
||||
mkdir -p "C:\Users\jaeoh\Desktop\workspace\pet-lab"/{app,assets/characters,tests}
|
||||
cd "C:\Users\jaeoh\Desktop\workspace\pet-lab"
|
||||
git init
|
||||
```
|
||||
|
||||
- [ ] **Step 2: 캐릭터 이미지 복사**
|
||||
|
||||
```bash
|
||||
cp "Z:\homes\jaeoh\캐릭터\박뚱냥.jpg" "C:\Users\jaeoh\Desktop\workspace\pet-lab\assets\characters\박뚱냥.png"
|
||||
```
|
||||
|
||||
참고: 원본이 .jpg이지만 투명 배경이 있는 이미지이므로 그대로 사용. 파일명은 .png으로 저장하되, 실제 포맷이 JPG라면 PyQt5의 QPixmap이 자동 감지하므로 문제없음.
|
||||
|
||||
- [ ] **Step 3: requirements.txt 생성**
|
||||
|
||||
```
|
||||
PyQt5>=5.15,<6.0
|
||||
pytest>=7.0
|
||||
```
|
||||
|
||||
- [ ] **Step 4: 가상환경 생성 및 의존성 설치**
|
||||
|
||||
```bash
|
||||
cd "C:\Users\jaeoh\Desktop\workspace\pet-lab"
|
||||
python -m venv venv
|
||||
venv\Scripts\activate
|
||||
pip install -r requirements.txt
|
||||
```
|
||||
|
||||
- [ ] **Step 5: config.py 작성**
|
||||
|
||||
```python
|
||||
"""pet-lab 설정 상수."""
|
||||
import os
|
||||
|
||||
# 캐릭터 크기 (높이 기준 px, 너비는 비율 유지)
|
||||
SIZES = {"small": 100, "medium": 150, "large": 200}
|
||||
DEFAULT_SIZE = "medium"
|
||||
|
||||
# 수평 위치 프리셋 (화면 너비 비율)
|
||||
POSITIONS = {"left": 0.1, "center": 0.5, "right": 0.9}
|
||||
DEFAULT_POSITION = "right"
|
||||
|
||||
# 시선 추적
|
||||
TIMER_INTERVAL_MS = 30
|
||||
MAX_TILT_ANGLE = 15.0
|
||||
|
||||
# 태스크바
|
||||
TASKBAR_HEIGHT = 48
|
||||
|
||||
# 애니메이션
|
||||
JUMP_HEIGHT = 30
|
||||
JUMP_DURATION_MS = 300
|
||||
SHAKE_OFFSET = 10
|
||||
SHAKE_DURATION_MS = 400
|
||||
|
||||
# 에셋 경로
|
||||
BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||||
CHARACTER_DIR = os.path.join(BASE_DIR, "assets", "characters")
|
||||
DEFAULT_CHARACTER = "박뚱냥.png"
|
||||
```
|
||||
|
||||
- [ ] **Step 6: test_config.py 작성**
|
||||
|
||||
```python
|
||||
"""config 상수 검증."""
|
||||
from app.config import SIZES, POSITIONS, DEFAULT_SIZE, DEFAULT_POSITION
|
||||
from app.config import TIMER_INTERVAL_MS, MAX_TILT_ANGLE, CHARACTER_DIR
|
||||
import os
|
||||
|
||||
|
||||
def test_sizes_has_three_presets():
|
||||
assert set(SIZES.keys()) == {"small", "medium", "large"}
|
||||
assert all(isinstance(v, int) and v > 0 for v in SIZES.values())
|
||||
|
||||
|
||||
def test_default_size_is_valid():
|
||||
assert DEFAULT_SIZE in SIZES
|
||||
|
||||
|
||||
def test_positions_has_three_presets():
|
||||
assert set(POSITIONS.keys()) == {"left", "center", "right"}
|
||||
assert all(0.0 < v < 1.0 for v in POSITIONS.values())
|
||||
|
||||
|
||||
def test_default_position_is_valid():
|
||||
assert DEFAULT_POSITION in POSITIONS
|
||||
|
||||
|
||||
def test_timer_interval_is_reasonable():
|
||||
assert 10 <= TIMER_INTERVAL_MS <= 100
|
||||
|
||||
|
||||
def test_max_tilt_angle_is_reasonable():
|
||||
assert 5.0 <= MAX_TILT_ANGLE <= 45.0
|
||||
|
||||
|
||||
def test_character_dir_exists():
|
||||
assert os.path.isdir(CHARACTER_DIR)
|
||||
```
|
||||
|
||||
- [ ] **Step 7: 테스트 실행**
|
||||
|
||||
```bash
|
||||
cd "C:\Users\jaeoh\Desktop\workspace\pet-lab"
|
||||
python -m pytest tests/test_config.py -v
|
||||
```
|
||||
|
||||
Expected: 7 passed
|
||||
|
||||
- [ ] **Step 8: .gitignore 생성 및 커밋**
|
||||
|
||||
`.gitignore`:
|
||||
```
|
||||
venv/
|
||||
__pycache__/
|
||||
*.pyc
|
||||
.pytest_cache/
|
||||
dist/
|
||||
build/
|
||||
*.spec
|
||||
```
|
||||
|
||||
```bash
|
||||
git add .
|
||||
git commit -m "feat: 프로젝트 초기화 — config, 캐릭터 에셋, 테스트"
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
### Task 2: eye_tracker.py — 시선 계산 모듈
|
||||
|
||||
**Files:**
|
||||
- Create: `C:\Users\jaeoh\Desktop\workspace\pet-lab\app\eye_tracker.py`
|
||||
- Create: `C:\Users\jaeoh\Desktop\workspace\pet-lab\tests\test_eye_tracker.py`
|
||||
|
||||
- [ ] **Step 1: test_eye_tracker.py 작성**
|
||||
|
||||
```python
|
||||
"""eye_tracker 시선 계산 테스트."""
|
||||
import math
|
||||
from app.eye_tracker import compute_gaze
|
||||
|
||||
|
||||
def test_mouse_right_of_character():
|
||||
"""마우스가 캐릭터 오른쪽 → 양수 기울기, flip=False."""
|
||||
angle, flip = compute_gaze(
|
||||
char_center_x=500, char_center_y=900,
|
||||
mouse_x=800, mouse_y=500,
|
||||
max_angle=15.0,
|
||||
)
|
||||
assert 0 < angle <= 15.0
|
||||
assert flip is False
|
||||
|
||||
|
||||
def test_mouse_left_of_character():
|
||||
"""마우스가 캐릭터 왼쪽 → 음수 기울기, flip=True."""
|
||||
angle, flip = compute_gaze(
|
||||
char_center_x=500, char_center_y=900,
|
||||
mouse_x=200, mouse_y=500,
|
||||
max_angle=15.0,
|
||||
)
|
||||
assert -15.0 <= angle < 0
|
||||
assert flip is True
|
||||
|
||||
|
||||
def test_mouse_directly_above():
|
||||
"""마우스가 캐릭터 바로 위 → 기울기 0, flip=False."""
|
||||
angle, flip = compute_gaze(
|
||||
char_center_x=500, char_center_y=900,
|
||||
mouse_x=500, mouse_y=100,
|
||||
max_angle=15.0,
|
||||
)
|
||||
assert angle == 0.0
|
||||
assert flip is False
|
||||
|
||||
|
||||
def test_mouse_at_character_position():
|
||||
"""마우스가 캐릭터 위치와 동일 → 기울기 0, flip=False."""
|
||||
angle, flip = compute_gaze(
|
||||
char_center_x=500, char_center_y=500,
|
||||
mouse_x=500, mouse_y=500,
|
||||
max_angle=15.0,
|
||||
)
|
||||
assert angle == 0.0
|
||||
assert flip is False
|
||||
|
||||
|
||||
def test_angle_clamped_to_max():
|
||||
"""기울기가 max_angle을 초과하지 않아야 한다."""
|
||||
angle, flip = compute_gaze(
|
||||
char_center_x=500, char_center_y=500,
|
||||
mouse_x=10000, mouse_y=500,
|
||||
max_angle=15.0,
|
||||
)
|
||||
assert abs(angle) <= 15.0
|
||||
|
||||
|
||||
def test_mouse_far_left():
|
||||
"""마우스가 매우 왼쪽 → 기울기 -max_angle에 근접."""
|
||||
angle, flip = compute_gaze(
|
||||
char_center_x=500, char_center_y=500,
|
||||
mouse_x=0, mouse_y=500,
|
||||
max_angle=15.0,
|
||||
)
|
||||
assert angle < 0
|
||||
assert flip is True
|
||||
```
|
||||
|
||||
- [ ] **Step 2: 테스트 실행 — 실패 확인**
|
||||
|
||||
```bash
|
||||
python -m pytest tests/test_eye_tracker.py -v
|
||||
```
|
||||
|
||||
Expected: FAIL with `ModuleNotFoundError: No module named 'app.eye_tracker'`
|
||||
|
||||
- [ ] **Step 3: eye_tracker.py 구현**
|
||||
|
||||
```python
|
||||
"""마우스 위치 기반 시선/기울기 계산 — 순수 함수 모듈."""
|
||||
import math
|
||||
|
||||
|
||||
def compute_gaze(
|
||||
char_center_x: float,
|
||||
char_center_y: float,
|
||||
mouse_x: float,
|
||||
mouse_y: float,
|
||||
max_angle: float = 15.0,
|
||||
) -> tuple[float, bool]:
|
||||
"""캐릭터 중심과 마우스 위치로 기울기 각도와 좌우 반전 여부를 계산한다.
|
||||
|
||||
Returns:
|
||||
(tilt_angle, flip_horizontal)
|
||||
- tilt_angle: -max_angle ~ +max_angle (도). 양수=우측 기울기, 음수=좌측 기울기.
|
||||
- flip_horizontal: True면 이미지를 좌우 반전 (마우스가 캐릭터 왼쪽).
|
||||
"""
|
||||
dx = mouse_x - char_center_x
|
||||
dy = mouse_y - char_center_y
|
||||
|
||||
if dx == 0 and dy == 0:
|
||||
return 0.0, False
|
||||
|
||||
# dx 방향의 비율로 기울기 결정 (atan2로 각도 → 비율 변환)
|
||||
angle_rad = math.atan2(abs(dx), max(abs(dy), 1))
|
||||
ratio = angle_rad / (math.pi / 2) # 0~1 범위
|
||||
tilt = ratio * max_angle
|
||||
|
||||
if dx < 0:
|
||||
tilt = -tilt
|
||||
|
||||
# max_angle 클램핑
|
||||
tilt = max(-max_angle, min(max_angle, tilt))
|
||||
|
||||
flip = dx < 0
|
||||
|
||||
return tilt, flip
|
||||
```
|
||||
|
||||
- [ ] **Step 4: 테스트 실행 — 통과 확인**
|
||||
|
||||
```bash
|
||||
python -m pytest tests/test_eye_tracker.py -v
|
||||
```
|
||||
|
||||
Expected: 6 passed
|
||||
|
||||
- [ ] **Step 5: 커밋**
|
||||
|
||||
```bash
|
||||
git add app/eye_tracker.py tests/test_eye_tracker.py
|
||||
git commit -m "feat: eye_tracker — 마우스 시선 기울기 계산 모듈"
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
### Task 3: pet_widget.py — 투명 윈도우 + 캐릭터 렌더링
|
||||
|
||||
**Files:**
|
||||
- Create: `C:\Users\jaeoh\Desktop\workspace\pet-lab\app\pet_widget.py`
|
||||
|
||||
- [ ] **Step 1: pet_widget.py 작성**
|
||||
|
||||
```python
|
||||
"""투명 윈도우 위에 캐릭터를 렌더링하고 시선을 추적하는 메인 위젯."""
|
||||
from PyQt5.QtWidgets import QWidget, QLabel, QApplication
|
||||
from PyQt5.QtCore import Qt, QTimer, QPoint
|
||||
from PyQt5.QtGui import QPixmap, QCursor, QTransform
|
||||
import os
|
||||
|
||||
from app.config import (
|
||||
SIZES, DEFAULT_SIZE, POSITIONS, DEFAULT_POSITION,
|
||||
TIMER_INTERVAL_MS, MAX_TILT_ANGLE, TASKBAR_HEIGHT,
|
||||
CHARACTER_DIR, DEFAULT_CHARACTER,
|
||||
)
|
||||
from app.eye_tracker import compute_gaze
|
||||
|
||||
|
||||
class PetWidget(QWidget):
|
||||
def __init__(self):
|
||||
super().__init__()
|
||||
self._size_key = DEFAULT_SIZE
|
||||
self._position_key = DEFAULT_POSITION
|
||||
self._always_on_top = True
|
||||
self._last_mouse_pos = None
|
||||
self._base_y = 0
|
||||
|
||||
self._init_window()
|
||||
self._load_character()
|
||||
self._position_on_screen()
|
||||
self._start_tracking()
|
||||
|
||||
def _init_window(self):
|
||||
flags = Qt.FramelessWindowHint | Qt.Tool
|
||||
if self._always_on_top:
|
||||
flags |= Qt.WindowStaysOnTopHint
|
||||
self.setWindowFlags(flags)
|
||||
self.setAttribute(Qt.WA_TranslucentBackground)
|
||||
|
||||
def _load_character(self):
|
||||
path = os.path.join(CHARACTER_DIR, DEFAULT_CHARACTER)
|
||||
self._original_pixmap = QPixmap(path)
|
||||
self._label = QLabel(self)
|
||||
self._apply_size()
|
||||
|
||||
def _apply_size(self):
|
||||
height = SIZES[self._size_key]
|
||||
scaled = self._original_pixmap.scaledToHeight(height, Qt.SmoothTransformation)
|
||||
self._label.setPixmap(scaled)
|
||||
self._label.setFixedSize(scaled.size())
|
||||
self.setFixedSize(scaled.size())
|
||||
|
||||
def _position_on_screen(self):
|
||||
screen = QApplication.primaryScreen().geometry()
|
||||
char_height = SIZES[self._size_key]
|
||||
self._base_y = screen.height() - TASKBAR_HEIGHT - char_height
|
||||
x_ratio = POSITIONS[self._position_key]
|
||||
x = int(screen.width() * x_ratio) - self.width() // 2
|
||||
self.move(x, self._base_y)
|
||||
|
||||
def _start_tracking(self):
|
||||
self._timer = QTimer(self)
|
||||
self._timer.timeout.connect(self._update_gaze)
|
||||
self._timer.start(TIMER_INTERVAL_MS)
|
||||
|
||||
def _update_gaze(self):
|
||||
mouse_pos = QCursor.pos()
|
||||
if self._last_mouse_pos == mouse_pos:
|
||||
return
|
||||
self._last_mouse_pos = mouse_pos
|
||||
|
||||
center = self.geometry().center()
|
||||
tilt, flip = compute_gaze(
|
||||
center.x(), center.y(),
|
||||
mouse_pos.x(), mouse_pos.y(),
|
||||
MAX_TILT_ANGLE,
|
||||
)
|
||||
|
||||
height = SIZES[self._size_key]
|
||||
scaled = self._original_pixmap.scaledToHeight(height, Qt.SmoothTransformation)
|
||||
|
||||
transform = QTransform()
|
||||
if flip:
|
||||
transform.scale(-1, 1)
|
||||
transform.rotate(tilt)
|
||||
|
||||
rotated = scaled.transformed(transform, Qt.SmoothTransformation)
|
||||
self._label.setPixmap(rotated)
|
||||
self._label.setFixedSize(rotated.size())
|
||||
self.setFixedSize(rotated.size())
|
||||
|
||||
# ── 크기/위치 변경 (interaction.py에서 호출) ──
|
||||
|
||||
def set_size(self, size_key: str):
|
||||
self._size_key = size_key
|
||||
self._apply_size()
|
||||
self._position_on_screen()
|
||||
|
||||
def set_position(self, position_key: str):
|
||||
self._position_key = position_key
|
||||
self._position_on_screen()
|
||||
|
||||
def toggle_always_on_top(self):
|
||||
self._always_on_top = not self._always_on_top
|
||||
flags = Qt.FramelessWindowHint | Qt.Tool
|
||||
if self._always_on_top:
|
||||
flags |= Qt.WindowStaysOnTopHint
|
||||
self.setWindowFlags(flags)
|
||||
self.show()
|
||||
|
||||
@property
|
||||
def always_on_top(self) -> bool:
|
||||
return self._always_on_top
|
||||
|
||||
@property
|
||||
def base_y(self) -> int:
|
||||
return self._base_y
|
||||
```
|
||||
|
||||
- [ ] **Step 2: 수동 테스트 — 투명 윈도우에 캐릭터 표시 확인**
|
||||
|
||||
임시 실행 스크립트:
|
||||
```bash
|
||||
cd "C:\Users\jaeoh\Desktop\workspace\pet-lab"
|
||||
python -c "
|
||||
import sys
|
||||
from PyQt5.QtWidgets import QApplication
|
||||
from app.pet_widget import PetWidget
|
||||
|
||||
app = QApplication(sys.argv)
|
||||
pet = PetWidget()
|
||||
pet.show()
|
||||
sys.exit(app.exec_())
|
||||
"
|
||||
```
|
||||
|
||||
Expected: 화면 우하단에 박뚱냥이 표시되고, 마우스 이동 시 기울기/반전이 바뀜.
|
||||
|
||||
- [ ] **Step 3: 커밋**
|
||||
|
||||
```bash
|
||||
git add app/pet_widget.py
|
||||
git commit -m "feat: pet_widget — 투명 윈도우 + 시선 추적 렌더링"
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
### Task 4: interaction.py — 클릭 반응 + 우클릭 메뉴
|
||||
|
||||
**Files:**
|
||||
- Create: `C:\Users\jaeoh\Desktop\workspace\pet-lab\app\interaction.py`
|
||||
- Modify: `C:\Users\jaeoh\Desktop\workspace\pet-lab\app\pet_widget.py` (마우스 이벤트 연결)
|
||||
|
||||
- [ ] **Step 1: interaction.py 작성**
|
||||
|
||||
```python
|
||||
"""클릭 애니메이션 + 우클릭 컨텍스트 메뉴."""
|
||||
from PyQt5.QtWidgets import QMenu, QAction, QApplication
|
||||
from PyQt5.QtCore import QPropertyAnimation, QEasingCurve, QPoint, QSequentialAnimationGroup
|
||||
|
||||
from app.config import (
|
||||
JUMP_HEIGHT, JUMP_DURATION_MS,
|
||||
SHAKE_OFFSET, SHAKE_DURATION_MS,
|
||||
SIZES, POSITIONS,
|
||||
)
|
||||
|
||||
|
||||
def play_jump(widget):
|
||||
"""좌클릭 — 위로 점프 후 복귀."""
|
||||
start = widget.pos()
|
||||
top = QPoint(start.x(), start.y() - JUMP_HEIGHT)
|
||||
|
||||
anim = QPropertyAnimation(widget, b"pos")
|
||||
anim.setDuration(JUMP_DURATION_MS)
|
||||
anim.setStartValue(start)
|
||||
anim.setKeyValueAt(0.4, top)
|
||||
anim.setEndValue(start)
|
||||
anim.setEasingCurve(QEasingCurve.OutBounce)
|
||||
|
||||
# prevent garbage collection
|
||||
widget._current_anim = anim
|
||||
anim.start()
|
||||
|
||||
|
||||
def play_shake(widget):
|
||||
"""더블클릭 — 좌우 흔들기."""
|
||||
start = widget.pos()
|
||||
left = QPoint(start.x() - SHAKE_OFFSET, start.y())
|
||||
right = QPoint(start.x() + SHAKE_OFFSET, start.y())
|
||||
|
||||
group = QSequentialAnimationGroup(widget)
|
||||
|
||||
for end_pos in [left, right, left, right, start]:
|
||||
anim = QPropertyAnimation(widget, b"pos")
|
||||
anim.setDuration(SHAKE_DURATION_MS // 5)
|
||||
anim.setEndValue(end_pos)
|
||||
group.addAnimation(anim)
|
||||
|
||||
widget._current_anim = group
|
||||
group.start()
|
||||
|
||||
|
||||
def show_context_menu(widget, global_pos):
|
||||
"""우클릭 — 컨텍스트 메뉴 표시."""
|
||||
menu = QMenu()
|
||||
|
||||
# 위치 서브메뉴
|
||||
pos_menu = menu.addMenu("위치")
|
||||
for key, label in [("left", "좌"), ("center", "중앙"), ("right", "우")]:
|
||||
action = pos_menu.addAction(label)
|
||||
action.triggered.connect(lambda checked, k=key: widget.set_position(k))
|
||||
|
||||
# 크기 서브메뉴
|
||||
size_menu = menu.addMenu("크기")
|
||||
for key, label in [("small", "소 (100px)"), ("medium", "중 (150px)"), ("large", "대 (200px)")]:
|
||||
action = size_menu.addAction(label)
|
||||
action.triggered.connect(lambda checked, k=key: widget.set_size(k))
|
||||
|
||||
# 항상 위 토글
|
||||
top_action = menu.addAction("항상 위" + (" ✓" if widget.always_on_top else ""))
|
||||
top_action.triggered.connect(widget.toggle_always_on_top)
|
||||
|
||||
menu.addSeparator()
|
||||
|
||||
# 종료
|
||||
quit_action = menu.addAction("종료")
|
||||
quit_action.triggered.connect(QApplication.quit)
|
||||
|
||||
menu.exec_(global_pos)
|
||||
```
|
||||
|
||||
- [ ] **Step 2: pet_widget.py에 마우스 이벤트 연결**
|
||||
|
||||
`pet_widget.py`의 `PetWidget` 클래스에 다음 메서드를 추가:
|
||||
|
||||
```python
|
||||
# ── 마우스 이벤트 (파일 하단, toggle_always_on_top 뒤에 추가) ──
|
||||
|
||||
def mousePressEvent(self, event):
|
||||
if event.button() == Qt.RightButton:
|
||||
from app.interaction import show_context_menu
|
||||
show_context_menu(self, event.globalPos())
|
||||
|
||||
def mouseDoubleClickEvent(self, event):
|
||||
if event.button() == Qt.LeftButton:
|
||||
from app.interaction import play_shake
|
||||
play_shake(self)
|
||||
|
||||
def mouseReleaseEvent(self, event):
|
||||
if event.button() == Qt.LeftButton:
|
||||
from app.interaction import play_jump
|
||||
play_jump(self)
|
||||
```
|
||||
|
||||
파일 상단 import에 추가 필요 없음 (lazy import 사용).
|
||||
|
||||
- [ ] **Step 3: 수동 테스트**
|
||||
|
||||
```bash
|
||||
cd "C:\Users\jaeoh\Desktop\workspace\pet-lab"
|
||||
python -c "
|
||||
import sys
|
||||
from PyQt5.QtWidgets import QApplication
|
||||
from app.pet_widget import PetWidget
|
||||
|
||||
app = QApplication(sys.argv)
|
||||
pet = PetWidget()
|
||||
pet.show()
|
||||
sys.exit(app.exec_())
|
||||
"
|
||||
```
|
||||
|
||||
테스트 항목:
|
||||
- 좌클릭 → 점프 애니메이션
|
||||
- 더블클릭 → 흔들기 애니메이션
|
||||
- 우클릭 → 메뉴 표시 (위치/크기/항상위/종료)
|
||||
- 메뉴에서 위치 변경 → 캐릭터 이동
|
||||
- 메뉴에서 크기 변경 → 캐릭터 크기 변경
|
||||
- 종료 → 앱 종료
|
||||
|
||||
- [ ] **Step 4: 커밋**
|
||||
|
||||
```bash
|
||||
git add app/interaction.py app/pet_widget.py
|
||||
git commit -m "feat: interaction — 클릭 점프/흔들기 + 우클릭 메뉴"
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
### Task 5: main.py — 엔트리포인트
|
||||
|
||||
**Files:**
|
||||
- Create: `C:\Users\jaeoh\Desktop\workspace\pet-lab\app\main.py`
|
||||
|
||||
- [ ] **Step 1: main.py 작성**
|
||||
|
||||
```python
|
||||
"""pet-lab 엔트리포인트."""
|
||||
import sys
|
||||
from PyQt5.QtWidgets import QApplication
|
||||
from app.pet_widget import PetWidget
|
||||
|
||||
|
||||
def main():
|
||||
app = QApplication(sys.argv)
|
||||
app.setQuitOnLastWindowClosed(False)
|
||||
|
||||
pet = PetWidget()
|
||||
pet.show()
|
||||
|
||||
sys.exit(app.exec_())
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
```
|
||||
|
||||
- [ ] **Step 2: 실행 확인**
|
||||
|
||||
```bash
|
||||
cd "C:\Users\jaeoh\Desktop\workspace\pet-lab"
|
||||
python -m app.main
|
||||
```
|
||||
|
||||
Expected: 박뚱냥이 화면 우하단에 표시되고, 시선 추적 + 클릭 반응 + 우클릭 메뉴 모두 동작.
|
||||
|
||||
- [ ] **Step 3: 커밋**
|
||||
|
||||
```bash
|
||||
git add app/main.py
|
||||
git commit -m "feat: main.py 엔트리포인트 — python -m app.main으로 실행"
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## Self-Review Checklist
|
||||
|
||||
**Spec coverage:**
|
||||
- [x] 투명 윈도우 (Task 3: `FramelessWindowHint`, `WA_TranslucentBackground`, `Tool`)
|
||||
- [x] 바닥 고정 (Task 3: `_position_on_screen`)
|
||||
- [x] 시선 추적 (Task 2: `compute_gaze`, Task 3: `_update_gaze`)
|
||||
- [x] 좌클릭 점프 (Task 4: `play_jump`)
|
||||
- [x] 더블클릭 흔들기 (Task 4: `play_shake`)
|
||||
- [x] 우클릭 메뉴 — 위치/크기/항상위/종료 (Task 4: `show_context_menu`)
|
||||
- [x] config 상수 (Task 1: `config.py`)
|
||||
- [x] 성능 최적화 — 마우스 변화 없으면 스킵 (Task 3: `_last_mouse_pos`)
|
||||
|
||||
**Placeholder scan:** 없음. 모든 step에 실제 코드 포함.
|
||||
|
||||
**Type consistency:** `compute_gaze` 시그니처 — Task 2 구현과 Task 3 호출 일치. `set_size`/`set_position` — Task 3 정의와 Task 4 호출 일치.
|
||||
@@ -1,4 +1,5 @@
|
||||
FROM python:3.12-alpine
|
||||
ENV PYTHONUNBUFFERED=1
|
||||
|
||||
WORKDIR /app
|
||||
COPY requirements.txt .
|
||||
|
||||
@@ -2,6 +2,11 @@ server {
|
||||
listen 80;
|
||||
server_name _;
|
||||
|
||||
# Security headers
|
||||
add_header X-Frame-Options "SAMEORIGIN" always;
|
||||
add_header X-Content-Type-Options "nosniff" always;
|
||||
add_header Referrer-Policy "strict-origin-when-cross-origin" always;
|
||||
|
||||
root /usr/share/nginx/html;
|
||||
index index.html;
|
||||
|
||||
|
||||
@@ -1,4 +1,5 @@
|
||||
FROM python:3.12-alpine
|
||||
ENV PYTHONUNBUFFERED=1
|
||||
|
||||
WORKDIR /app
|
||||
COPY requirements.txt .
|
||||
|
||||
@@ -183,12 +183,6 @@ def get_all_broker_cash() -> List[Dict[str, Any]]:
|
||||
return [dict(r) for r in rows]
|
||||
|
||||
|
||||
def get_broker_cash(broker: str) -> Dict[str, Any] | None:
|
||||
with _conn() as conn:
|
||||
row = conn.execute("SELECT * FROM broker_cash WHERE broker = ?", (broker,)).fetchone()
|
||||
return dict(row) if row else None
|
||||
|
||||
|
||||
def delete_broker_cash(broker: str) -> bool:
|
||||
with _conn() as conn:
|
||||
cur = conn.execute("DELETE FROM broker_cash WHERE broker = ?", (broker,))
|
||||
|
||||
@@ -17,11 +17,11 @@ from .db import (
|
||||
init_db, save_articles, get_latest_articles,
|
||||
add_portfolio_item, get_all_portfolio, get_portfolio_item,
|
||||
update_portfolio_item, delete_portfolio_item,
|
||||
upsert_broker_cash, get_all_broker_cash, get_broker_cash, delete_broker_cash,
|
||||
upsert_broker_cash, get_all_broker_cash, delete_broker_cash,
|
||||
upsert_asset_snapshot, get_asset_snapshots,
|
||||
add_sell_history, get_sell_history, update_sell_history, delete_sell_history,
|
||||
)
|
||||
from .scraper import fetch_market_news, fetch_major_indices, fetch_overseas_news
|
||||
from .scraper import fetch_market_news, fetch_major_indices
|
||||
from .price_fetcher import get_current_prices
|
||||
|
||||
app = FastAPI()
|
||||
@@ -118,17 +118,11 @@ def on_startup():
|
||||
|
||||
def run_scraping_job():
|
||||
logger.info("뉴스 스크래핑 시작")
|
||||
|
||||
# 1. 국내
|
||||
|
||||
articles_kr = fetch_market_news()
|
||||
count_kr = save_articles(articles_kr)
|
||||
|
||||
# 2. 해외 (임시 차단)
|
||||
# articles_world = fetch_overseas_news()
|
||||
# count_world = save_articles(articles_world)
|
||||
count_world = 0
|
||||
|
||||
logger.info(f"스크래핑 완료: 국내 {count_kr}건, 해외 {count_world}건")
|
||||
|
||||
logger.info(f"스크래핑 완료: 국내 {count_kr}건")
|
||||
|
||||
@app.get("/health")
|
||||
def health():
|
||||
@@ -156,14 +150,16 @@ def trigger_scrap():
|
||||
def get_balance():
|
||||
"""계좌 잔고 조회 (Windows AI Server Proxy)"""
|
||||
logger.info(f"Requesting Balance from {WINDOWS_AI_SERVER_URL}")
|
||||
resp = None
|
||||
try:
|
||||
resp = requests.get(f"{WINDOWS_AI_SERVER_URL}/trade/balance", timeout=5)
|
||||
if resp.status_code != 200:
|
||||
logger.error(f"Balance Error: {resp.status_code}")
|
||||
return JSONResponse(status_code=resp.status_code, content=resp.json())
|
||||
return resp.json()
|
||||
except requests.JSONDecodeError:
|
||||
return JSONResponse(status_code=resp.status_code, content={"error": f"Upstream error {resp.status_code}"})
|
||||
except ValueError:
|
||||
status = resp.status_code if resp is not None else 502
|
||||
return JSONResponse(status_code=status, content={"error": f"Upstream error {status}"})
|
||||
except Exception as e:
|
||||
logger.error(f"Balance Connection Failed: {e}")
|
||||
return JSONResponse(status_code=500, content={"error": "Connection Failed"})
|
||||
@@ -179,14 +175,16 @@ class OrderRequest(BaseModel):
|
||||
def order_stock(req: OrderRequest):
|
||||
"""주식 매수/매도 주문 (Windows AI Server Proxy)"""
|
||||
logger.info(f"Order Request: {req.action} {req.ticker} x{req.quantity}")
|
||||
resp = None
|
||||
try:
|
||||
resp = requests.post(f"{WINDOWS_AI_SERVER_URL}/trade/order", json=req.dict(), timeout=10)
|
||||
resp = requests.post(f"{WINDOWS_AI_SERVER_URL}/trade/order", json=req.model_dump(), timeout=10)
|
||||
if resp.status_code != 200:
|
||||
logger.error(f"Order Error: {resp.status_code}")
|
||||
return JSONResponse(status_code=resp.status_code, content=resp.json())
|
||||
return resp.json()
|
||||
except requests.JSONDecodeError:
|
||||
return JSONResponse(status_code=resp.status_code, content={"error": f"Upstream error {resp.status_code}"})
|
||||
except ValueError:
|
||||
status = resp.status_code if resp is not None else 502
|
||||
return JSONResponse(status_code=status, content={"error": f"Upstream error {status}"})
|
||||
except Exception as e:
|
||||
logger.error(f"Order Connection Failed: {e}")
|
||||
return JSONResponse(status_code=500, content={"error": "Connection Failed"})
|
||||
@@ -368,7 +366,7 @@ def update_portfolio(item_id: int, req: PortfolioUpdateRequest):
|
||||
"""포트폴리오 종목 수정"""
|
||||
if get_portfolio_item(item_id) is None:
|
||||
return JSONResponse(status_code=404, content={"error": "Item not found"})
|
||||
update_portfolio_item(item_id, **req.dict())
|
||||
update_portfolio_item(item_id, **req.model_dump())
|
||||
return {"ok": True}
|
||||
|
||||
|
||||
|
||||
@@ -79,59 +79,6 @@ def fetch_market_news() -> List[Dict[str, str]]:
|
||||
logger.error(f"국내 뉴스 스크래핑 실패: {e}")
|
||||
return []
|
||||
|
||||
def fetch_overseas_news() -> List[Dict[str, str]]:
|
||||
"""
|
||||
네이버 금융 해외증시 뉴스 크롤링 (모바일 API 사용)
|
||||
"""
|
||||
api_url = "https://api.stock.naver.com/news/overseas/mainnews"
|
||||
try:
|
||||
headers = {
|
||||
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko)"
|
||||
}
|
||||
resp = requests.get(api_url, headers=headers, timeout=10)
|
||||
resp.raise_for_status()
|
||||
|
||||
data = resp.json()
|
||||
if isinstance(data, list):
|
||||
items = data
|
||||
else:
|
||||
items = data.get("result", [])
|
||||
|
||||
articles = []
|
||||
for item in items:
|
||||
# API 키 매핑 (subject/title/tit, summary/subContent/sub_tit 등)
|
||||
title = item.get("subject") or item.get("title") or item.get("tit") or ""
|
||||
summary = item.get("summary") or item.get("subContent") or item.get("sub_tit") or ""
|
||||
press = item.get("officeName") or item.get("office_name") or item.get("cp_name") or ""
|
||||
|
||||
# 날짜 포맷팅 (20260126123000 -> 2026-01-26 12:30:00)
|
||||
raw_dt = str(item.get("dt", ""))
|
||||
if len(raw_dt) == 14:
|
||||
date = f"{raw_dt[:4]}-{raw_dt[4:6]}-{raw_dt[6:8]} {raw_dt[8:10]}:{raw_dt[10:12]}:{raw_dt[12:]}"
|
||||
else:
|
||||
date = raw_dt
|
||||
|
||||
# 링크 생성
|
||||
aid = item.get("articleId")
|
||||
oid = item.get("officeId")
|
||||
link = f"https://m.stock.naver.com/worldstock/news/read/{oid}/{aid}"
|
||||
|
||||
articles.append({
|
||||
"title": title,
|
||||
"link": link,
|
||||
"summary": summary,
|
||||
"press": press,
|
||||
"date": date,
|
||||
"crawled_at": time.strftime("%Y-%m-%d %H:%M:%S"),
|
||||
"category": "overseas"
|
||||
})
|
||||
|
||||
return articles
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"해외 뉴스 스크래핑 실패: {e}")
|
||||
return []
|
||||
|
||||
def fetch_major_indices() -> Dict[str, Any]:
|
||||
"""
|
||||
KOSPI, KOSDAQ, KOSPI200 등 주요 지표 (네이버 금융 홈)
|
||||
|
||||
Reference in New Issue
Block a user