이창현
3541ca454f
CI / Frontend - type-check & test (push) Canceled after 0s
운영에서는 Secure 쿠키를 강제한다.`nHTTP로 실행하는 격리 QA에서만 환경 변수로 해제할 수 있게 했다.
596 lines
25 KiB
Python
596 lines
25 KiB
Python
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import hmac
|
|
import json
|
|
import os
|
|
import re
|
|
import secrets
|
|
from pathlib import Path
|
|
from urllib.parse import quote
|
|
|
|
from fastapi import APIRouter, Cookie, File, Form, HTTPException, Query, Request, Response, UploadFile
|
|
from fastapi.responses import FileResponse
|
|
from pydantic import BaseModel, Field
|
|
|
|
from library_db import (
|
|
MIDI_ROOT,
|
|
connect,
|
|
create_session,
|
|
delete_session,
|
|
get_user_for_session,
|
|
hash_password,
|
|
utc_now,
|
|
verify_password,
|
|
)
|
|
|
|
|
|
router = APIRouter()
|
|
COOKIE_NAME = "b2b_session"
|
|
MAX_MIDI_BYTES = 10 * 1024 * 1024
|
|
MAX_USER_FILES = 100
|
|
MAX_USER_BYTES = 500 * 1024 * 1024
|
|
USERNAME_RE = re.compile(r"^[A-Za-z0-9_-]{3,30}$")
|
|
DIFFICULTIES = {"beginner", "easy", "intermediate", "advanced", "expert"}
|
|
ARRANGEMENTS = {"melody", "accompaniment", "piano_solo", "duet", "exercise"}
|
|
VISIBILITIES = {"public", "members", "shared", "private"}
|
|
|
|
|
|
class AuthPayload(BaseModel):
|
|
username: str = Field(min_length=3, max_length=30)
|
|
password: str = Field(min_length=10, max_length=128)
|
|
|
|
|
|
class RegisterPayload(AuthPayload):
|
|
display_name: str = Field(min_length=2, max_length=30)
|
|
email: str | None = Field(default=None, max_length=120)
|
|
|
|
|
|
class PasswordChangePayload(BaseModel):
|
|
current_password: str = Field(min_length=1, max_length=128)
|
|
new_password: str = Field(min_length=10, max_length=128)
|
|
|
|
|
|
class WithdrawPayload(BaseModel):
|
|
password: str = Field(min_length=1, max_length=128)
|
|
confirmation: str
|
|
|
|
|
|
class MidiUpdatePayload(BaseModel):
|
|
title: str | None = Field(default=None, min_length=1, max_length=120)
|
|
artist: str | None = Field(default=None, max_length=80)
|
|
description: str | None = Field(default=None, max_length=500)
|
|
visibility: str | None = None
|
|
min_tier: int | None = Field(default=None, ge=0, le=4)
|
|
difficulty: str | None = None
|
|
genre: str | None = Field(default=None, max_length=40)
|
|
arrangement: str | None = None
|
|
tags: list[str] | None = None
|
|
shared_usernames: list[str] | None = None
|
|
|
|
|
|
class UserUpdatePayload(BaseModel):
|
|
tier: int | None = Field(default=None, ge=1, le=4)
|
|
role: str | None = None
|
|
status: str | None = None
|
|
|
|
|
|
def _public_user(user: dict | None) -> dict | None:
|
|
if user is None:
|
|
return None
|
|
return {
|
|
"id": user["id"],
|
|
"username": user["username"],
|
|
"display_name": user["display_name"],
|
|
"email": user.get("email"),
|
|
"tier": user["tier"],
|
|
"role": user["role"],
|
|
}
|
|
|
|
|
|
def _session_user(session_token: str | None) -> dict | None:
|
|
return get_user_for_session(session_token)
|
|
|
|
|
|
def _require_user(session_token: str | None) -> dict:
|
|
user = _session_user(session_token)
|
|
if user is None:
|
|
raise HTTPException(status_code=401, detail="로그인이 필요합니다")
|
|
return user
|
|
|
|
|
|
def _require_csrf(request: Request, user: dict) -> None:
|
|
supplied = request.headers.get("x-csrf-token", "")
|
|
if not supplied or not hmac.compare_digest(supplied, user["csrf_token"]):
|
|
raise HTTPException(status_code=403, detail="요청 확인 값이 올바르지 않습니다")
|
|
|
|
|
|
def _require_admin(session_token: str | None) -> dict:
|
|
user = _require_user(session_token)
|
|
if user["role"] != "admin":
|
|
raise HTTPException(status_code=403, detail="관리자 권한이 필요합니다")
|
|
return user
|
|
|
|
|
|
def _normalize_tags(tags: str | list[str]) -> list[str]:
|
|
values = tags.split(",") if isinstance(tags, str) else tags
|
|
cleaned: list[str] = []
|
|
for value in values:
|
|
item = value.strip()
|
|
if not item or item in cleaned:
|
|
continue
|
|
if len(item) > 30:
|
|
raise HTTPException(status_code=400, detail="태그는 30자 이하로 입력하세요")
|
|
cleaned.append(item)
|
|
if len(cleaned) > 12:
|
|
raise HTTPException(status_code=400, detail="태그는 최대 12개까지 사용할 수 있습니다")
|
|
return cleaned
|
|
|
|
|
|
def _validate_metadata(visibility: str, min_tier: int, difficulty: str, arrangement: str) -> None:
|
|
if visibility not in VISIBILITIES:
|
|
raise HTTPException(status_code=400, detail="공개 범위가 올바르지 않습니다")
|
|
if difficulty not in DIFFICULTIES:
|
|
raise HTTPException(status_code=400, detail="난이도가 올바르지 않습니다")
|
|
if arrangement not in ARRANGEMENTS:
|
|
raise HTTPException(status_code=400, detail="편곡 형태가 올바르지 않습니다")
|
|
if not 0 <= min_tier <= 4:
|
|
raise HTTPException(status_code=400, detail="회원 등급 범위가 올바르지 않습니다")
|
|
if visibility == "public" and min_tier != 0:
|
|
raise HTTPException(status_code=400, detail="전체 공개 곡은 등급 제한을 사용할 수 없습니다")
|
|
if visibility != "public" and min_tier < 1:
|
|
raise HTTPException(status_code=400, detail="회원 전용 곡은 최소 등급이 1 이상이어야 합니다")
|
|
|
|
|
|
def _resolve_share_ids(connection, usernames: list[str]) -> list[int]:
|
|
normalized = [name.strip() for name in usernames if name.strip()]
|
|
if not normalized:
|
|
return []
|
|
placeholders = ",".join("?" for _ in normalized)
|
|
rows = connection.execute(
|
|
f"SELECT id, username FROM users WHERE username COLLATE NOCASE IN ({placeholders}) AND status = 'active'",
|
|
normalized,
|
|
).fetchall()
|
|
found = {row["username"].lower(): row["id"] for row in rows}
|
|
missing = [name for name in normalized if name.lower() not in found]
|
|
if missing:
|
|
raise HTTPException(status_code=400, detail="존재하지 않는 회원: " + ", ".join(missing))
|
|
return list(dict.fromkeys(found.values()))
|
|
|
|
|
|
def _can_access(row: dict, user: dict | None, shared_ids: set[int]) -> bool:
|
|
if user and user["role"] == "admin":
|
|
return True
|
|
if user and row["owner_id"] == user["id"]:
|
|
return True
|
|
if row["visibility"] == "public":
|
|
return True
|
|
if user is None or user["tier"] < row["min_tier"]:
|
|
return False
|
|
if row["visibility"] == "members":
|
|
return True
|
|
if row["visibility"] == "shared":
|
|
return user["id"] in shared_ids
|
|
return False
|
|
|
|
|
|
def _serialize_item(row: dict, user: dict | None, shared_usernames: list[str]) -> dict:
|
|
tags = json.loads(row["tags_json"] or "[]")
|
|
can_edit = bool(user and (user["role"] == "admin" or row["owner_id"] == user["id"]))
|
|
return {
|
|
"id": row["id"],
|
|
"title": row["title"],
|
|
"artist": row["artist"],
|
|
"description": row["description"],
|
|
"original_name": row["original_name"],
|
|
"size_bytes": row["size_bytes"],
|
|
"visibility": row["visibility"],
|
|
"min_tier": row["min_tier"],
|
|
"difficulty": row["difficulty"],
|
|
"genre": row["genre"],
|
|
"arrangement": row["arrangement"],
|
|
"tags": tags,
|
|
"owner": row["owner_name"] or "ITOSM 피아노 학습실",
|
|
"owner_username": row["owner_username"],
|
|
"shared_usernames": shared_usernames if can_edit else [],
|
|
"can_edit": can_edit,
|
|
"created_at": row["created_at"],
|
|
"group_key": f"{row['artist'].strip().lower()}::{row['title'].strip().lower()}",
|
|
}
|
|
|
|
|
|
def _load_item(item_id: str) -> tuple[dict, set[int], list[str]]:
|
|
with connect() as connection:
|
|
row = connection.execute(
|
|
"""
|
|
SELECT m.*, u.display_name AS owner_name, u.username AS owner_username
|
|
FROM midi_items m LEFT JOIN users u ON u.id = m.owner_id
|
|
WHERE m.id = ?
|
|
""",
|
|
(item_id,),
|
|
).fetchone()
|
|
if row is None:
|
|
raise HTTPException(status_code=404, detail="곡을 찾을 수 없습니다")
|
|
shares = connection.execute(
|
|
"SELECT s.user_id, u.username FROM midi_shares s JOIN users u ON u.id = s.user_id WHERE s.midi_id = ?",
|
|
(item_id,),
|
|
).fetchall()
|
|
return dict(row), {share["user_id"] for share in shares}, [share["username"] for share in shares]
|
|
|
|
|
|
@router.post("/auth/register")
|
|
def register(payload: RegisterPayload, response: Response):
|
|
username = payload.username.strip()
|
|
display_name = payload.display_name.strip()
|
|
email = payload.email.strip().lower() if payload.email else None
|
|
if not USERNAME_RE.fullmatch(username):
|
|
raise HTTPException(status_code=400, detail="아이디는 영문과 숫자 및 밑줄과 하이픈으로 3~30자까지 입력하세요")
|
|
if email and ("@" not in email or email.startswith("@") or email.endswith("@")):
|
|
raise HTTPException(status_code=400, detail="이메일 형식이 올바르지 않습니다")
|
|
with connect() as connection:
|
|
exists = connection.execute(
|
|
"SELECT 1 FROM users WHERE username = ? COLLATE NOCASE OR (? IS NOT NULL AND email = ? COLLATE NOCASE)",
|
|
(username, email, email),
|
|
).fetchone()
|
|
if exists:
|
|
raise HTTPException(status_code=409, detail="이미 사용 중인 아이디 또는 이메일입니다")
|
|
cursor = connection.execute(
|
|
"INSERT INTO users (username, display_name, email, password_hash, created_at) VALUES (?, ?, ?, ?, ?)",
|
|
(username, display_name, email, hash_password(payload.password), utc_now()),
|
|
)
|
|
user_id = cursor.lastrowid
|
|
token, csrf_token = create_session(user_id)
|
|
_set_cookie(response, token)
|
|
user = get_user_for_session(token)
|
|
return {"user": _public_user(user), "csrf_token": csrf_token}
|
|
|
|
|
|
@router.post("/auth/login")
|
|
def login(payload: AuthPayload, response: Response):
|
|
with connect() as connection:
|
|
row = connection.execute(
|
|
"SELECT * FROM users WHERE username = ? COLLATE NOCASE", (payload.username.strip(),)
|
|
).fetchone()
|
|
if row is None or not verify_password(payload.password, row["password_hash"]):
|
|
raise HTTPException(status_code=401, detail="아이디 또는 비밀번호가 올바르지 않습니다")
|
|
if row["status"] != "active":
|
|
raise HTTPException(status_code=403, detail="사용이 중지된 계정입니다")
|
|
token, csrf_token = create_session(row["id"])
|
|
_set_cookie(response, token)
|
|
user = get_user_for_session(token)
|
|
return {"user": _public_user(user), "csrf_token": csrf_token}
|
|
|
|
|
|
def _set_cookie(response: Response, token: str) -> None:
|
|
secure_cookie = os.environ.get("SESSION_COOKIE_SECURE", "1").strip().lower() not in {"0", "false", "no"}
|
|
response.set_cookie(
|
|
COOKIE_NAME,
|
|
token,
|
|
max_age=30 * 24 * 60 * 60,
|
|
httponly=True,
|
|
secure=secure_cookie,
|
|
samesite="lax",
|
|
path="/",
|
|
)
|
|
|
|
|
|
@router.get("/auth/me")
|
|
def me(session_token: str | None = Cookie(default=None, alias=COOKIE_NAME)):
|
|
user = _session_user(session_token)
|
|
return {"user": _public_user(user), "csrf_token": user["csrf_token"] if user else None}
|
|
|
|
|
|
@router.post("/auth/logout")
|
|
def logout(request: Request, response: Response, session_token: str | None = Cookie(default=None, alias=COOKIE_NAME)):
|
|
user = _require_user(session_token)
|
|
_require_csrf(request, user)
|
|
delete_session(session_token)
|
|
response.delete_cookie(COOKIE_NAME, path="/")
|
|
return {"ok": True}
|
|
|
|
|
|
@router.post("/auth/change-password")
|
|
def change_password(
|
|
payload: PasswordChangePayload,
|
|
request: Request,
|
|
response: Response,
|
|
session_token: str | None = Cookie(default=None, alias=COOKIE_NAME),
|
|
):
|
|
user = _require_user(session_token)
|
|
_require_csrf(request, user)
|
|
with connect() as connection:
|
|
row = connection.execute("SELECT password_hash FROM users WHERE id = ?", (user["id"],)).fetchone()
|
|
if row is None or not verify_password(payload.current_password, row["password_hash"]):
|
|
raise HTTPException(status_code=400, detail="현재 비밀번호가 올바르지 않습니다")
|
|
if verify_password(payload.new_password, row["password_hash"]):
|
|
raise HTTPException(status_code=400, detail="새 비밀번호는 현재 비밀번호와 달라야 합니다")
|
|
connection.execute(
|
|
"UPDATE users SET password_hash = ? WHERE id = ?",
|
|
(hash_password(payload.new_password), user["id"]),
|
|
)
|
|
connection.execute("DELETE FROM sessions WHERE user_id = ?", (user["id"],))
|
|
response.delete_cookie(COOKIE_NAME, path="/")
|
|
return {"ok": True, "message": "비밀번호가 변경되었습니다. 다시 로그인해 주세요."}
|
|
|
|
|
|
@router.post("/auth/withdraw")
|
|
def withdraw(
|
|
payload: WithdrawPayload,
|
|
request: Request,
|
|
response: Response,
|
|
session_token: str | None = Cookie(default=None, alias=COOKIE_NAME),
|
|
):
|
|
user = _require_user(session_token)
|
|
_require_csrf(request, user)
|
|
if payload.confirmation != "회원 탈퇴":
|
|
raise HTTPException(status_code=400, detail="확인란에 회원 탈퇴를 정확히 입력하세요")
|
|
with connect() as connection:
|
|
row = connection.execute("SELECT password_hash FROM users WHERE id = ?", (user["id"],)).fetchone()
|
|
if row is None or not verify_password(payload.password, row["password_hash"]):
|
|
raise HTTPException(status_code=400, detail="비밀번호가 올바르지 않습니다")
|
|
files = connection.execute(
|
|
"SELECT storage_name FROM midi_items WHERE owner_id = ?", (user["id"],)
|
|
).fetchall()
|
|
connection.execute("DELETE FROM midi_items WHERE owner_id = ?", (user["id"],))
|
|
connection.execute("DELETE FROM users WHERE id = ?", (user["id"],))
|
|
for item in files:
|
|
(MIDI_ROOT / item["storage_name"]).unlink(missing_ok=True)
|
|
response.delete_cookie(COOKIE_NAME, path="/")
|
|
return {"ok": True, "message": "회원 탈퇴가 완료되었습니다"}
|
|
|
|
|
|
@router.get("/midi")
|
|
def list_midi(
|
|
search: str = Query(default="", max_length=100),
|
|
genre: str = Query(default="", max_length=40),
|
|
difficulty: str = Query(default="", max_length=20),
|
|
arrangement: str = Query(default="", max_length=30),
|
|
tag: str = Query(default="", max_length=30),
|
|
session_token: str | None = Cookie(default=None, alias=COOKIE_NAME),
|
|
):
|
|
user = _session_user(session_token)
|
|
clauses = ["1=1"]
|
|
params: list[object] = []
|
|
if search:
|
|
clauses.append("(m.title LIKE ? OR m.artist LIKE ? OR m.description LIKE ? OR m.tags_json LIKE ?)")
|
|
needle = f"%{search.strip()}%"
|
|
params.extend([needle, needle, needle, needle])
|
|
if genre:
|
|
clauses.append("m.genre = ?")
|
|
params.append(genre)
|
|
if difficulty:
|
|
clauses.append("m.difficulty = ?")
|
|
params.append(difficulty)
|
|
if arrangement:
|
|
clauses.append("m.arrangement = ?")
|
|
params.append(arrangement)
|
|
if tag:
|
|
clauses.append("m.tags_json LIKE ?")
|
|
params.append(f'%"{tag}"%')
|
|
with connect() as connection:
|
|
rows = connection.execute(
|
|
f"""
|
|
SELECT m.*, u.display_name AS owner_name, u.username AS owner_username
|
|
FROM midi_items m LEFT JOIN users u ON u.id = m.owner_id
|
|
WHERE {' AND '.join(clauses)}
|
|
ORDER BY m.artist COLLATE NOCASE, m.title COLLATE NOCASE, m.difficulty, m.created_at
|
|
""",
|
|
params,
|
|
).fetchall()
|
|
share_rows = connection.execute("SELECT midi_id, user_id FROM midi_shares").fetchall()
|
|
username_rows = connection.execute(
|
|
"SELECT s.midi_id, u.username FROM midi_shares s JOIN users u ON u.id = s.user_id"
|
|
).fetchall()
|
|
shared_ids: dict[str, set[int]] = {}
|
|
shared_names: dict[str, list[str]] = {}
|
|
for share in share_rows:
|
|
shared_ids.setdefault(share["midi_id"], set()).add(share["user_id"])
|
|
for share in username_rows:
|
|
shared_names.setdefault(share["midi_id"], []).append(share["username"])
|
|
items = []
|
|
for source in rows:
|
|
row = dict(source)
|
|
if _can_access(row, user, shared_ids.get(row["id"], set())):
|
|
items.append(_serialize_item(row, user, shared_names.get(row["id"], [])))
|
|
facets = {
|
|
"genres": sorted({item["genre"] for item in items if item["genre"]}),
|
|
"tags": sorted({tag_value for item in items for tag_value in item["tags"]}),
|
|
}
|
|
return {"items": items, "facets": facets}
|
|
|
|
|
|
@router.get("/midi/{item_id}/file")
|
|
def download_midi(item_id: str, session_token: str | None = Cookie(default=None, alias=COOKIE_NAME)):
|
|
user = _session_user(session_token)
|
|
row, shared_ids, _ = _load_item(item_id)
|
|
if not _can_access(row, user, shared_ids):
|
|
raise HTTPException(status_code=403, detail="이 곡을 사용할 권한이 없습니다")
|
|
path = MIDI_ROOT / row["storage_name"]
|
|
if not path.is_file() or path.parent.resolve() != MIDI_ROOT.resolve():
|
|
raise HTTPException(status_code=404, detail="MIDI 파일을 찾을 수 없습니다")
|
|
encoded = quote(row["original_name"].replace("\r", "").replace("\n", ""))
|
|
return FileResponse(
|
|
path,
|
|
media_type="audio/midi",
|
|
headers={"Content-Disposition": f"inline; filename*=UTF-8''{encoded}"},
|
|
)
|
|
|
|
|
|
@router.post("/midi")
|
|
async def upload_midi(
|
|
request: Request,
|
|
file: UploadFile = File(...),
|
|
title: str = Form(...),
|
|
artist: str = Form(""),
|
|
description: str = Form(""),
|
|
visibility: str = Form("private"),
|
|
min_tier: int = Form(1),
|
|
difficulty: str = Form("beginner"),
|
|
genre: str = Form(""),
|
|
arrangement: str = Form("melody"),
|
|
tags: str = Form(""),
|
|
shared_usernames: str = Form(""),
|
|
session_token: str | None = Cookie(default=None, alias=COOKIE_NAME),
|
|
):
|
|
user = _require_user(session_token)
|
|
_require_csrf(request, user)
|
|
clean_title = title.strip()
|
|
if not clean_title or len(clean_title) > 120:
|
|
raise HTTPException(status_code=400, detail="곡 제목은 1~120자로 입력하세요")
|
|
_validate_metadata(visibility, min_tier, difficulty, arrangement)
|
|
original_name = Path(file.filename or "upload.mid").name
|
|
if Path(original_name).suffix.lower() not in {".mid", ".midi"}:
|
|
raise HTTPException(status_code=400, detail="MIDI 파일만 업로드할 수 있습니다")
|
|
content = await file.read(MAX_MIDI_BYTES + 1)
|
|
if len(content) > MAX_MIDI_BYTES:
|
|
raise HTTPException(status_code=413, detail="MIDI 파일은 10MB 이하만 업로드할 수 있습니다")
|
|
if len(content) < 14 or not content.startswith(b"MThd"):
|
|
raise HTTPException(status_code=400, detail="올바른 MIDI 파일이 아닙니다")
|
|
item_id = secrets.token_hex(16)
|
|
storage_name = f"{item_id}.mid"
|
|
target = MIDI_ROOT / storage_name
|
|
tag_values = _normalize_tags(tags)
|
|
share_names = [name.strip() for name in shared_usernames.split(",") if name.strip()]
|
|
with connect() as connection:
|
|
usage = connection.execute(
|
|
"SELECT COUNT(*) AS file_count, COALESCE(SUM(size_bytes), 0) AS total_bytes FROM midi_items WHERE owner_id = ?",
|
|
(user["id"],),
|
|
).fetchone()
|
|
if usage["file_count"] >= MAX_USER_FILES or usage["total_bytes"] + len(content) > MAX_USER_BYTES:
|
|
raise HTTPException(status_code=413, detail="회원별 보관 한도를 초과했습니다")
|
|
share_ids = _resolve_share_ids(connection, share_names) if visibility == "shared" else []
|
|
target.write_bytes(content)
|
|
target.chmod(0o640)
|
|
try:
|
|
connection.execute(
|
|
"""
|
|
INSERT INTO midi_items (
|
|
id, owner_id, title, artist, description, original_name, storage_name,
|
|
size_bytes, sha256, visibility, min_tier, difficulty, genre, arrangement,
|
|
tags_json, created_at, updated_at
|
|
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
|
""",
|
|
(
|
|
item_id, user["id"], clean_title, artist.strip()[:80], description.strip()[:500],
|
|
original_name, storage_name, len(content), hashlib.sha256(content).hexdigest(),
|
|
visibility, min_tier, difficulty, genre.strip()[:40], arrangement,
|
|
json.dumps(tag_values, ensure_ascii=False), utc_now(), utc_now(),
|
|
),
|
|
)
|
|
connection.executemany(
|
|
"INSERT INTO midi_shares (midi_id, user_id) VALUES (?, ?)",
|
|
[(item_id, share_id) for share_id in share_ids],
|
|
)
|
|
except Exception:
|
|
target.unlink(missing_ok=True)
|
|
raise
|
|
row, _, names = _load_item(item_id)
|
|
return _serialize_item(row, user, names)
|
|
|
|
|
|
@router.patch("/midi/{item_id}")
|
|
def update_midi(
|
|
item_id: str,
|
|
payload: MidiUpdatePayload,
|
|
request: Request,
|
|
session_token: str | None = Cookie(default=None, alias=COOKIE_NAME),
|
|
):
|
|
user = _require_user(session_token)
|
|
_require_csrf(request, user)
|
|
row, _, _ = _load_item(item_id)
|
|
if user["role"] != "admin" and row["owner_id"] != user["id"]:
|
|
raise HTTPException(status_code=403, detail="본인이 올린 곡만 수정할 수 있습니다")
|
|
values = payload.model_dump(exclude_unset=True)
|
|
visibility = values.get("visibility", row["visibility"])
|
|
min_tier = values.get("min_tier", row["min_tier"])
|
|
difficulty = values.get("difficulty", row["difficulty"])
|
|
arrangement = values.get("arrangement", row["arrangement"])
|
|
_validate_metadata(visibility, min_tier, difficulty, arrangement)
|
|
if "tags" in values:
|
|
values["tags_json"] = json.dumps(_normalize_tags(values.pop("tags")), ensure_ascii=False)
|
|
share_names = values.pop("shared_usernames", None)
|
|
allowed_columns = {
|
|
"title", "artist", "description", "visibility", "min_tier", "difficulty",
|
|
"genre", "arrangement", "tags_json",
|
|
}
|
|
updates = {key: value.strip() if isinstance(value, str) else value for key, value in values.items() if key in allowed_columns}
|
|
if "title" in updates and not updates["title"]:
|
|
raise HTTPException(status_code=400, detail="곡 제목은 비워둘 수 없습니다")
|
|
with connect() as connection:
|
|
if updates:
|
|
assignments = ", ".join(f"{column} = ?" for column in updates)
|
|
connection.execute(
|
|
f"UPDATE midi_items SET {assignments}, updated_at = ? WHERE id = ?",
|
|
[*updates.values(), utc_now(), item_id],
|
|
)
|
|
if share_names is not None or visibility != "shared":
|
|
connection.execute("DELETE FROM midi_shares WHERE midi_id = ?", (item_id,))
|
|
if visibility == "shared":
|
|
ids = _resolve_share_ids(connection, share_names or [])
|
|
connection.executemany(
|
|
"INSERT INTO midi_shares (midi_id, user_id) VALUES (?, ?)",
|
|
[(item_id, user_id) for user_id in ids],
|
|
)
|
|
updated, _, names = _load_item(item_id)
|
|
return _serialize_item(updated, user, names)
|
|
|
|
|
|
@router.delete("/midi/{item_id}")
|
|
def delete_midi(
|
|
item_id: str,
|
|
request: Request,
|
|
session_token: str | None = Cookie(default=None, alias=COOKIE_NAME),
|
|
):
|
|
user = _require_user(session_token)
|
|
_require_csrf(request, user)
|
|
row, _, _ = _load_item(item_id)
|
|
if row["seed_key"]:
|
|
raise HTTPException(status_code=400, detail="기본 공개 샘플은 삭제할 수 없습니다")
|
|
if user["role"] != "admin" and row["owner_id"] != user["id"]:
|
|
raise HTTPException(status_code=403, detail="본인이 올린 곡만 삭제할 수 있습니다")
|
|
with connect() as connection:
|
|
connection.execute("DELETE FROM midi_items WHERE id = ?", (item_id,))
|
|
(MIDI_ROOT / row["storage_name"]).unlink(missing_ok=True)
|
|
return {"ok": True}
|
|
|
|
|
|
@router.get("/admin/users")
|
|
def list_users(session_token: str | None = Cookie(default=None, alias=COOKIE_NAME)):
|
|
_require_admin(session_token)
|
|
with connect() as connection:
|
|
rows = connection.execute(
|
|
"SELECT id, username, display_name, email, tier, role, status, created_at FROM users ORDER BY created_at"
|
|
).fetchall()
|
|
return {"users": [dict(row) for row in rows]}
|
|
|
|
|
|
@router.patch("/admin/users/{user_id}")
|
|
def update_user(
|
|
user_id: int,
|
|
payload: UserUpdatePayload,
|
|
request: Request,
|
|
session_token: str | None = Cookie(default=None, alias=COOKIE_NAME),
|
|
):
|
|
admin = _require_admin(session_token)
|
|
_require_csrf(request, admin)
|
|
values = payload.model_dump(exclude_unset=True)
|
|
if values.get("role") not in {None, "member", "moderator", "admin"}:
|
|
raise HTTPException(status_code=400, detail="회원 역할이 올바르지 않습니다")
|
|
if values.get("status") not in {None, "active", "suspended"}:
|
|
raise HTTPException(status_code=400, detail="회원 상태가 올바르지 않습니다")
|
|
if user_id == admin["id"] and ("role" in values or values.get("status") == "suspended"):
|
|
raise HTTPException(status_code=400, detail="현재 관리자 계정의 역할이나 상태는 직접 변경할 수 없습니다")
|
|
if not values:
|
|
raise HTTPException(status_code=400, detail="변경할 값이 없습니다")
|
|
with connect() as connection:
|
|
exists = connection.execute("SELECT id FROM users WHERE id = ?", (user_id,)).fetchone()
|
|
if not exists:
|
|
raise HTTPException(status_code=404, detail="회원을 찾을 수 없습니다")
|
|
assignments = ", ".join(f"{column} = ?" for column in values)
|
|
connection.execute(f"UPDATE users SET {assignments} WHERE id = ?", [*values.values(), user_id])
|
|
row = connection.execute(
|
|
"SELECT id, username, display_name, email, tier, role, status, created_at FROM users WHERE id = ?",
|
|
(user_id,),
|
|
).fetchone()
|
|
return dict(row)
|