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)