gahusb/web-page-backend
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor: 전체 코드베이스 감사 기반 리팩토링 — 버그 수정, 데드코드 제거, 보안 강화

Browse Source 
P0 버그 수정:
- stock-lab: trade 엔드포인트 NameError 수정 (resp 미정의)
- deployer: 동시 배포 시 HTTP 200 → 503 반환

P1 데드코드 제거:
- stock-lab: fetch_overseas_news(), get_broker_cash() 제거
- blog-lab: 미사용 urlparse import 제거
- lotto-lab: 중복 inline import json 7곳 제거

P2 성능/효율 개선:
- lotto-lab: 가중 샘플링 3중 복사 → utils.weighted_sample_6() 통합
- lotto-lab: DB 인덱스 3개 추가 (recommendations, purchase_history)
- stock-lab: Pydantic .dict() → .model_dump() 호환
- blog-lab: 페이지네이션 상한(le=100) 추가

P3 보안/인프라:
- nginx: X-Frame-Options, X-Content-Type-Options, Referrer-Policy 헤더 추가
- docker-compose: travel-proxy CORS 와일드카드 → localhost 전용
- Dockerfile: music-lab, blog-lab, realestate-lab에 PYTHONUNBUFFERED 추가

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
gahusb
2026-04-07 04:10:14 +09:00
parent 9d5583935d
commit 535ffea45a
15 changed files with 84 additions and 150 deletions
Download Patch File Download Diff File Expand all files Collapse all files

37
backend/app/db.py
View File

@@ -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),
)

23
backend/app/generator.py
View File

@@ -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

32
backend/app/recommender.py
View File

@@ -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 = {

21
backend/app/utils.py
View File

@@ -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)