chpark
d16456cb92
# 사용자별 격리 - JWT 토큰에 uid 추가 (auth.get_uid 헬퍼) - PostgreSQL — exchange_credentials/automation_config/trades/signal_events 에 user_id BIGINT - SQLite user_settings 테이블 신설 (글로벌 settings 는 옛 호환) - 모든 DB 함수 시그니처에 user_id 인자 추가 — 다른 사용자 데이터 절대 접근 불가 - alert_state — 모든 dict key 가 (user_id, ...) tuple 로 계층화 - core_logic alert_loop — 활성 사용자 순회 + 각자 settings/symbol/텔레그램 적용 - ensure_user_defaults() / ensure_user_automation() — 첫 사용 시 자동 시드 # 사용자 관리 (admin only) - users_db: delete_user / admin_reset_password / set_role - /api/users POST DELETE PUT password PUT role (본인 강등 / 마지막 admin 보호) - /admin/users 페이지 — 등록/삭제/role 토글/비번 reset 모달 - 사이드바 adminOnly 필터 — admin role 만 메뉴 노출 # 대시보드 개선 - 모바일 / 범례 토글 (모바일 60 캔들, 데스크톱 200) - 트레이드 이력: open 트레이드 실시간 PnL% (Binance ticker 호출 + 방향별 계산) - 메트릭 카드 분리 (실거래 vs 실시간 open) # 안정성 - api.ts: error.detail array/object 안전 처리 ([object Object] 방지) - Chart.tsx: Plotly yaxis title 객체 형태 + 모바일 height 동적 조정 Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
94 lines
3.6 KiB
Python
94 lines
3.6 KiB
Python
from fastapi import APIRouter, Depends, HTTPException
|
|
from pydantic import BaseModel
|
|
from typing import Optional
|
|
import users_db
|
|
from ..auth import require_user, require_admin
|
|
|
|
router = APIRouter()
|
|
|
|
|
|
def _serialize(r):
|
|
return {
|
|
"id": r["id"], "username": r["username"], "role": r["role"],
|
|
"created_at": str(r["created_at"]) if r.get("created_at") else None,
|
|
"last_login_at": str(r["last_login_at"]) if r.get("last_login_at") else None,
|
|
}
|
|
|
|
|
|
@router.get("")
|
|
def list_users(_: dict = Depends(require_admin)):
|
|
return [_serialize(r) for r in users_db.list_users()]
|
|
|
|
|
|
class CreateUserIn(BaseModel):
|
|
username: str
|
|
password: str
|
|
role: str = "user"
|
|
|
|
|
|
@router.post("")
|
|
def create_user(body: CreateUserIn, _: dict = Depends(require_admin)):
|
|
if body.role not in ("admin", "user"):
|
|
raise HTTPException(status_code=400, detail="role 은 admin / user")
|
|
if len(body.password) < 6:
|
|
raise HTTPException(status_code=400, detail="비밀번호 6자 이상")
|
|
if not body.username.strip():
|
|
raise HTTPException(status_code=400, detail="username 필수")
|
|
uid = users_db.create_user(body.username.strip(), body.password, body.role)
|
|
if uid is None:
|
|
raise HTTPException(status_code=400, detail="생성 실패 (이미 존재할 수 있음)")
|
|
return {"id": uid, "username": body.username.strip(), "role": body.role}
|
|
|
|
|
|
@router.delete("/{user_id}")
|
|
def delete_user(user_id: int, payload: dict = Depends(require_admin)):
|
|
me = payload.get("sub")
|
|
# 본인 삭제 방지
|
|
target = next((u for u in users_db.list_users() if u["id"] == user_id), None)
|
|
if target and target["username"] == me:
|
|
raise HTTPException(status_code=400, detail="본인 계정은 삭제 불가")
|
|
# admin 이 단 한명이면 삭제 방지
|
|
if target and target["role"] == "admin":
|
|
admin_count = sum(1 for u in users_db.list_users() if u["role"] == "admin")
|
|
if admin_count <= 1:
|
|
raise HTTPException(status_code=400, detail="마지막 admin 은 삭제 불가")
|
|
if not users_db.delete_user(user_id):
|
|
raise HTTPException(status_code=404, detail="not found")
|
|
return {"ok": True}
|
|
|
|
|
|
class ResetPasswordIn(BaseModel):
|
|
new_password: str
|
|
|
|
|
|
@router.put("/{user_id}/password")
|
|
def reset_password(user_id: int, body: ResetPasswordIn, _: dict = Depends(require_admin)):
|
|
if len(body.new_password) < 6:
|
|
raise HTTPException(status_code=400, detail="비밀번호 6자 이상")
|
|
if not users_db.admin_reset_password(user_id, body.new_password):
|
|
raise HTTPException(status_code=404, detail="not found")
|
|
return {"ok": True}
|
|
|
|
|
|
class SetRoleIn(BaseModel):
|
|
role: str
|
|
|
|
|
|
@router.put("/{user_id}/role")
|
|
def set_role(user_id: int, body: SetRoleIn, payload: dict = Depends(require_admin)):
|
|
if body.role not in ("admin", "user"):
|
|
raise HTTPException(status_code=400, detail="role 은 admin / user")
|
|
me = payload.get("sub")
|
|
target = next((u for u in users_db.list_users() if u["id"] == user_id), None)
|
|
# 본인 role 강등 방지
|
|
if target and target["username"] == me and body.role != "admin":
|
|
raise HTTPException(status_code=400, detail="본인 role 변경 불가")
|
|
# 마지막 admin 강등 방지
|
|
if target and target["role"] == "admin" and body.role != "admin":
|
|
admin_count = sum(1 for u in users_db.list_users() if u["role"] == "admin")
|
|
if admin_count <= 1:
|
|
raise HTTPException(status_code=400, detail="마지막 admin role 변경 불가")
|
|
if not users_db.set_role(user_id, body.role):
|
|
raise HTTPException(status_code=404, detail="not found")
|
|
return {"ok": True}
|