회원과 곡 보관함 기능을 추가한다
회원 가입과 로그인 및 등급별 접근과 공개 범위를 구현했다.`nMIDI 업로드와 검색 및 같은 곡의 버전 묶음을 추가했다.`n비밀번호 변경과 회원 탈퇴 및 관리자 회원 관리를 포함했다.`n직접 만든 공개 샘플 MIDI 네 곡과 운영 문서를 추가했다.
This commit is contained in:
이창현
@@ -0,0 +1,267 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import base64
|
||||
import hashlib
|
||||
import hmac
|
||||
import json
|
||||
import os
|
||||
import secrets
|
||||
import shutil
|
||||
import sqlite3
|
||||
from contextlib import contextmanager
|
||||
from datetime import datetime, timedelta, timezone
|
||||
from pathlib import Path
|
||||
from typing import Iterator
|
||||
|
||||
|
||||
DATA_ROOT = Path(os.environ.get("LIBRARY_DATA_DIR", "/data"))
|
||||
DATABASE_PATH = DATA_ROOT / "library.sqlite3"
|
||||
MIDI_ROOT = DATA_ROOT / "midi"
|
||||
SESSION_DAYS = int(os.environ.get("SESSION_DAYS", "30"))
|
||||
|
||||
|
||||
def utc_now() -> str:
|
||||
return datetime.now(timezone.utc).isoformat()
|
||||
|
||||
|
||||
@contextmanager
|
||||
def connect() -> Iterator[sqlite3.Connection]:
|
||||
connection = sqlite3.connect(DATABASE_PATH, timeout=20)
|
||||
connection.row_factory = sqlite3.Row
|
||||
connection.execute("PRAGMA foreign_keys = ON")
|
||||
connection.execute("PRAGMA journal_mode = WAL")
|
||||
try:
|
||||
yield connection
|
||||
connection.commit()
|
||||
except Exception:
|
||||
connection.rollback()
|
||||
raise
|
||||
finally:
|
||||
connection.close()
|
||||
|
||||
|
||||
def hash_password(password: str) -> str:
|
||||
salt = secrets.token_bytes(16)
|
||||
digest = hashlib.scrypt(password.encode("utf-8"), salt=salt, n=2**14, r=8, p=1, dklen=32)
|
||||
return "scrypt$16384$8$1$%s$%s" % (
|
||||
base64.urlsafe_b64encode(salt).decode("ascii"),
|
||||
base64.urlsafe_b64encode(digest).decode("ascii"),
|
||||
)
|
||||
|
||||
|
||||
def verify_password(password: str, stored: str) -> bool:
|
||||
try:
|
||||
algorithm, n, r, p, salt_text, digest_text = stored.split("$", 5)
|
||||
if algorithm != "scrypt":
|
||||
return False
|
||||
salt = base64.urlsafe_b64decode(salt_text.encode("ascii"))
|
||||
expected = base64.urlsafe_b64decode(digest_text.encode("ascii"))
|
||||
actual = hashlib.scrypt(
|
||||
password.encode("utf-8"), salt=salt, n=int(n), r=int(r), p=int(p), dklen=len(expected)
|
||||
)
|
||||
return hmac.compare_digest(actual, expected)
|
||||
except (ValueError, TypeError):
|
||||
return False
|
||||
|
||||
|
||||
def token_digest(token: str) -> str:
|
||||
return hashlib.sha256(token.encode("utf-8")).hexdigest()
|
||||
|
||||
|
||||
def create_session(user_id: int) -> tuple[str, str]:
|
||||
token = secrets.token_urlsafe(36)
|
||||
csrf_token = secrets.token_urlsafe(24)
|
||||
expires_at = (datetime.now(timezone.utc) + timedelta(days=SESSION_DAYS)).isoformat()
|
||||
with connect() as connection:
|
||||
connection.execute("DELETE FROM sessions WHERE expires_at <= ?", (utc_now(),))
|
||||
connection.execute(
|
||||
"INSERT INTO sessions (token_hash, user_id, csrf_token, expires_at, created_at) VALUES (?, ?, ?, ?, ?)",
|
||||
(token_digest(token), user_id, csrf_token, expires_at, utc_now()),
|
||||
)
|
||||
return token, csrf_token
|
||||
|
||||
|
||||
def get_user_for_session(token: str | None) -> dict | None:
|
||||
if not token:
|
||||
return None
|
||||
with connect() as connection:
|
||||
row = connection.execute(
|
||||
"""
|
||||
SELECT u.id, u.username, u.display_name, u.email, u.tier, u.role, u.status,
|
||||
s.csrf_token, s.expires_at
|
||||
FROM sessions s
|
||||
JOIN users u ON u.id = s.user_id
|
||||
WHERE s.token_hash = ? AND s.expires_at > ?
|
||||
""",
|
||||
(token_digest(token), utc_now()),
|
||||
).fetchone()
|
||||
if row is None or row["status"] != "active":
|
||||
return None
|
||||
return dict(row)
|
||||
|
||||
|
||||
def delete_session(token: str | None) -> None:
|
||||
if not token:
|
||||
return
|
||||
with connect() as connection:
|
||||
connection.execute("DELETE FROM sessions WHERE token_hash = ?", (token_digest(token),))
|
||||
|
||||
|
||||
SAMPLES = (
|
||||
{
|
||||
"storage_name": "first-snow-melody.mid",
|
||||
"title": "첫눈이 내리는 밤",
|
||||
"artist": "ITOSM 연습곡",
|
||||
"description": "오른손으로 천천히 익히는 다장조 멜로디 연습곡",
|
||||
"difficulty": "beginner",
|
||||
"genre": "연습곡",
|
||||
"arrangement": "melody",
|
||||
"tags": ["다장조", "오른손", "겨울", "입문"],
|
||||
},
|
||||
{
|
||||
"storage_name": "first-snow-accompaniment.mid",
|
||||
"title": "첫눈이 내리는 밤",
|
||||
"artist": "ITOSM 연습곡",
|
||||
"description": "같은 곡을 양손 반주형으로 편곡한 중급 연습 버전",
|
||||
"difficulty": "intermediate",
|
||||
"genre": "연습곡",
|
||||
"arrangement": "accompaniment",
|
||||
"tags": ["다장조", "양손", "반주", "겨울"],
|
||||
},
|
||||
{
|
||||
"storage_name": "blue-breeze-walk.mid",
|
||||
"title": "파란 바람 산책",
|
||||
"artist": "ITOSM 연습곡",
|
||||
"description": "가벼운 리듬으로 연주하는 쉬운 멜로디",
|
||||
"difficulty": "easy",
|
||||
"genre": "가요풍",
|
||||
"arrangement": "melody",
|
||||
"tags": ["멜로디", "리듬", "쉬움", "가요풍"],
|
||||
},
|
||||
{
|
||||
"storage_name": "little-morning-waltz.mid",
|
||||
"title": "아침의 작은 왈츠",
|
||||
"artist": "ITOSM 연습곡",
|
||||
"description": "3박자와 왼손 화음을 함께 익히는 피아노 독주곡",
|
||||
"difficulty": "intermediate",
|
||||
"genre": "왈츠",
|
||||
"arrangement": "piano_solo",
|
||||
"tags": ["3박자", "왈츠", "양손", "화음"],
|
||||
},
|
||||
)
|
||||
|
||||
|
||||
def _seed_samples() -> None:
|
||||
source_root = Path(__file__).with_name("sample_midi")
|
||||
with connect() as connection:
|
||||
for sample in SAMPLES:
|
||||
source = source_root / sample["storage_name"]
|
||||
if not source.exists():
|
||||
continue
|
||||
target = MIDI_ROOT / sample["storage_name"]
|
||||
if not target.exists():
|
||||
shutil.copyfile(source, target)
|
||||
existing = connection.execute(
|
||||
"SELECT id FROM midi_items WHERE seed_key = ?", (sample["storage_name"],)
|
||||
).fetchone()
|
||||
if existing:
|
||||
continue
|
||||
content = target.read_bytes()
|
||||
connection.execute(
|
||||
"""
|
||||
INSERT INTO midi_items (
|
||||
id, owner_id, seed_key, title, artist, description, original_name,
|
||||
storage_name, size_bytes, sha256, visibility, min_tier, difficulty,
|
||||
genre, arrangement, tags_json, created_at, updated_at
|
||||
) VALUES (?, NULL, ?, ?, ?, ?, ?, ?, ?, ?, 'public', 0, ?, ?, ?, ?, ?, ?)
|
||||
""",
|
||||
(
|
||||
secrets.token_hex(16), sample["storage_name"], sample["title"], sample["artist"],
|
||||
sample["description"], sample["storage_name"], sample["storage_name"], len(content),
|
||||
hashlib.sha256(content).hexdigest(), sample["difficulty"], sample["genre"],
|
||||
sample["arrangement"], json.dumps(sample["tags"], ensure_ascii=False), utc_now(), utc_now(),
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
def initialize_database() -> None:
|
||||
DATA_ROOT.mkdir(parents=True, exist_ok=True)
|
||||
MIDI_ROOT.mkdir(parents=True, exist_ok=True)
|
||||
DATA_ROOT.chmod(0o750)
|
||||
MIDI_ROOT.chmod(0o750)
|
||||
with connect() as connection:
|
||||
connection.executescript(
|
||||
"""
|
||||
CREATE TABLE IF NOT EXISTS users (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
username TEXT NOT NULL UNIQUE COLLATE NOCASE,
|
||||
display_name TEXT NOT NULL,
|
||||
email TEXT UNIQUE COLLATE NOCASE,
|
||||
password_hash TEXT NOT NULL,
|
||||
tier INTEGER NOT NULL DEFAULT 1 CHECK (tier BETWEEN 1 AND 4),
|
||||
role TEXT NOT NULL DEFAULT 'member' CHECK (role IN ('member', 'moderator', 'admin')),
|
||||
status TEXT NOT NULL DEFAULT 'active' CHECK (status IN ('active', 'suspended')),
|
||||
created_at TEXT NOT NULL
|
||||
);
|
||||
CREATE TABLE IF NOT EXISTS sessions (
|
||||
token_hash TEXT PRIMARY KEY,
|
||||
user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE,
|
||||
csrf_token TEXT NOT NULL,
|
||||
expires_at TEXT NOT NULL,
|
||||
created_at TEXT NOT NULL
|
||||
);
|
||||
CREATE INDEX IF NOT EXISTS idx_sessions_user ON sessions(user_id);
|
||||
CREATE TABLE IF NOT EXISTS midi_items (
|
||||
id TEXT PRIMARY KEY,
|
||||
owner_id INTEGER REFERENCES users(id) ON DELETE SET NULL,
|
||||
seed_key TEXT UNIQUE,
|
||||
title TEXT NOT NULL,
|
||||
artist TEXT NOT NULL DEFAULT '',
|
||||
description TEXT NOT NULL DEFAULT '',
|
||||
original_name TEXT NOT NULL,
|
||||
storage_name TEXT NOT NULL UNIQUE,
|
||||
size_bytes INTEGER NOT NULL,
|
||||
sha256 TEXT NOT NULL,
|
||||
visibility TEXT NOT NULL CHECK (visibility IN ('public', 'members', 'shared', 'private')),
|
||||
min_tier INTEGER NOT NULL DEFAULT 0 CHECK (min_tier BETWEEN 0 AND 4),
|
||||
difficulty TEXT NOT NULL CHECK (difficulty IN ('beginner', 'easy', 'intermediate', 'advanced', 'expert')),
|
||||
genre TEXT NOT NULL DEFAULT '',
|
||||
arrangement TEXT NOT NULL CHECK (arrangement IN ('melody', 'accompaniment', 'piano_solo', 'duet', 'exercise')),
|
||||
tags_json TEXT NOT NULL DEFAULT '[]',
|
||||
created_at TEXT NOT NULL,
|
||||
updated_at TEXT NOT NULL
|
||||
);
|
||||
CREATE INDEX IF NOT EXISTS idx_midi_title ON midi_items(title COLLATE NOCASE);
|
||||
CREATE INDEX IF NOT EXISTS idx_midi_owner ON midi_items(owner_id);
|
||||
CREATE TABLE IF NOT EXISTS midi_shares (
|
||||
midi_id TEXT NOT NULL REFERENCES midi_items(id) ON DELETE CASCADE,
|
||||
user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE,
|
||||
PRIMARY KEY (midi_id, user_id)
|
||||
);
|
||||
"""
|
||||
)
|
||||
DATABASE_PATH.chmod(0o600)
|
||||
admin_username = os.environ.get("BOOTSTRAP_ADMIN_USERNAME", "").strip()
|
||||
password_file = os.environ.get("BOOTSTRAP_ADMIN_PASSWORD_FILE", "").strip()
|
||||
if admin_username and password_file:
|
||||
existing = connection.execute(
|
||||
"SELECT id FROM users WHERE username = ? COLLATE NOCASE", (admin_username,)
|
||||
).fetchone()
|
||||
if existing is None:
|
||||
password = Path(password_file).read_text(encoding="utf-8").strip()
|
||||
if len(password) < 10:
|
||||
raise RuntimeError("초기 관리자 비밀번호는 10자 이상이어야 합니다")
|
||||
connection.execute(
|
||||
"""
|
||||
INSERT INTO users (
|
||||
username, display_name, password_hash, tier, role, status, created_at
|
||||
) VALUES (?, ?, ?, 4, 'admin', 'active', ?)
|
||||
""",
|
||||
(
|
||||
admin_username,
|
||||
os.environ.get("BOOTSTRAP_ADMIN_DISPLAY_NAME", "관리자").strip() or "관리자",
|
||||
hash_password(password),
|
||||
utc_now(),
|
||||
),
|
||||
)
|
||||
_seed_samples()
|
||||
+7
-4
@@ -33,7 +33,8 @@ except ImportError:
|
||||
from fastapi import Depends, FastAPI, HTTPException, Request, Response, Security
|
||||
from fastapi.middleware.cors import CORSMiddleware
|
||||
from fastapi.security import APIKeyHeader
|
||||
from routers import export_routes, fingering, omr, transcribe, youtube
|
||||
from library_db import initialize_database
|
||||
from routers import export_routes, fingering, library, omr, transcribe, youtube
|
||||
|
||||
logging.basicConfig(level=logging.WARNING)
|
||||
|
||||
@@ -106,6 +107,7 @@ def _startup_checks() -> None:
|
||||
@asynccontextmanager
|
||||
async def _lifespan(_app: FastAPI):
|
||||
_startup_checks()
|
||||
initialize_database()
|
||||
yield
|
||||
|
||||
|
||||
@@ -131,9 +133,9 @@ _allowed_origins = [o.strip() for o in _raw_origins.split(",") if o.strip()]
|
||||
app.add_middleware(
|
||||
CORSMiddleware,
|
||||
allow_origins=_allowed_origins,
|
||||
allow_credentials=False,
|
||||
allow_methods=["GET", "POST"],
|
||||
allow_headers=["Content-Type", "X-API-Key"],
|
||||
allow_credentials=True,
|
||||
allow_methods=["GET", "POST", "PATCH", "DELETE"],
|
||||
allow_headers=["Content-Type", "X-API-Key", "X-CSRF-Token"],
|
||||
)
|
||||
|
||||
# ── Security-headers middleware ───────────────────────────────────────────────
|
||||
@@ -266,6 +268,7 @@ app.include_router(omr.router, prefix="/omr", tags=["omr"])
|
||||
app.include_router(youtube.router, prefix="/youtube", tags=["youtube"])
|
||||
app.include_router(fingering.router, prefix="/fingering", tags=["fingering"])
|
||||
app.include_router(export_routes.router, prefix="/export", tags=["export"])
|
||||
app.include_router(library.router, prefix="/library", tags=["library"])
|
||||
|
||||
|
||||
@app.get("/health")
|
||||
|
||||
@@ -0,0 +1,593 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import hmac
|
||||
import json
|
||||
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:
|
||||
response.set_cookie(
|
||||
COOKIE_NAME,
|
||||
token,
|
||||
max_age=30 * 24 * 60 * 60,
|
||||
httponly=True,
|
||||
secure=True,
|
||||
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)
|
||||
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
@@ -0,0 +1,55 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import importlib
|
||||
import os
|
||||
import tempfile
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
class LibraryDatabaseTest(unittest.TestCase):
|
||||
def setUp(self) -> None:
|
||||
self.temp = tempfile.TemporaryDirectory()
|
||||
os.environ["LIBRARY_DATA_DIR"] = self.temp.name
|
||||
os.environ.pop("BOOTSTRAP_ADMIN_USERNAME", None)
|
||||
os.environ.pop("BOOTSTRAP_ADMIN_PASSWORD_FILE", None)
|
||||
import library_db
|
||||
|
||||
self.db = importlib.reload(library_db)
|
||||
self.db.initialize_database()
|
||||
|
||||
def tearDown(self) -> None:
|
||||
self.temp.cleanup()
|
||||
|
||||
def test_password_hash_is_salted_and_verifiable(self) -> None:
|
||||
first = self.db.hash_password("safe-password-123")
|
||||
second = self.db.hash_password("safe-password-123")
|
||||
self.assertNotEqual(first, second)
|
||||
self.assertTrue(self.db.verify_password("safe-password-123", first))
|
||||
self.assertFalse(self.db.verify_password("wrong-password", first))
|
||||
|
||||
def test_session_round_trip(self) -> None:
|
||||
with self.db.connect() as connection:
|
||||
cursor = connection.execute(
|
||||
"""
|
||||
INSERT INTO users (username, display_name, password_hash, created_at)
|
||||
VALUES ('tester', '테스터', ?, ?)
|
||||
""",
|
||||
(self.db.hash_password("safe-password-123"), self.db.utc_now()),
|
||||
)
|
||||
user_id = cursor.lastrowid
|
||||
token, csrf = self.db.create_session(user_id)
|
||||
user = self.db.get_user_for_session(token)
|
||||
self.assertIsNotNone(user)
|
||||
self.assertEqual(user["username"], "tester")
|
||||
self.assertEqual(user["csrf_token"], csrf)
|
||||
self.db.delete_session(token)
|
||||
self.assertIsNone(self.db.get_user_for_session(token))
|
||||
|
||||
def test_database_directory_is_private_project_data(self) -> None:
|
||||
self.assertTrue(Path(self.temp.name, "library.sqlite3").is_file())
|
||||
self.assertTrue(Path(self.temp.name, "midi").is_dir())
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,72 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
OUTPUT = Path(__file__).resolve().parents[1] / "sample_midi"
|
||||
|
||||
|
||||
def variable_length(value: int) -> bytes:
|
||||
data = [value & 0x7F]
|
||||
value >>= 7
|
||||
while value:
|
||||
data.insert(0, (value & 0x7F) | 0x80)
|
||||
value >>= 7
|
||||
return bytes(data)
|
||||
|
||||
|
||||
def track(events: list[tuple[int, bytes]], tempo: int = 500_000, program: int = 0) -> bytes:
|
||||
body = bytearray()
|
||||
body.extend(b"\x00\xff\x51\x03" + tempo.to_bytes(3, "big"))
|
||||
body.extend(b"\x00\xff\x58\x04\x04\x02\x18\x08")
|
||||
body.extend(bytes([0, 0xC0, program]))
|
||||
for delta, message in events:
|
||||
body.extend(variable_length(delta))
|
||||
body.extend(message)
|
||||
body.extend(b"\x00\xff\x2f\x00")
|
||||
return b"MTrk" + len(body).to_bytes(4, "big") + body
|
||||
|
||||
|
||||
def midi_file(events: list[tuple[int, bytes]], tempo: int = 500_000) -> bytes:
|
||||
header = b"MThd" + (6).to_bytes(4, "big") + (0).to_bytes(2, "big") + (1).to_bytes(2, "big") + (480).to_bytes(2, "big")
|
||||
return header + track(events, tempo)
|
||||
|
||||
|
||||
def melody(notes: list[int], beats: int = 1, velocity: int = 82) -> list[tuple[int, bytes]]:
|
||||
result: list[tuple[int, bytes]] = []
|
||||
for note in notes:
|
||||
result.append((0, bytes([0x90, note, velocity])))
|
||||
result.append((480 * beats, bytes([0x80, note, 0])))
|
||||
return result
|
||||
|
||||
|
||||
def chord_progression(chords: list[list[int]], melody_notes: list[int]) -> list[tuple[int, bytes]]:
|
||||
result: list[tuple[int, bytes]] = []
|
||||
for index, note in enumerate(melody_notes):
|
||||
chord = chords[(index // 4) % len(chords)]
|
||||
for chord_note in chord:
|
||||
result.append((0, bytes([0x90, chord_note, 52])))
|
||||
result.append((0, bytes([0x90, note, 86])))
|
||||
result.append((480, bytes([0x80, note, 0])))
|
||||
for chord_note in chord:
|
||||
result.append((0, bytes([0x80, chord_note, 0])))
|
||||
return result
|
||||
|
||||
|
||||
def main() -> None:
|
||||
OUTPUT.mkdir(parents=True, exist_ok=True)
|
||||
first_snow = [64, 67, 69, 67, 64, 62, 60, 62, 64, 67, 72, 71, 69, 67, 64, 60]
|
||||
blue_breeze = [60, 62, 64, 67, 69, 67, 64, 62, 64, 65, 67, 72, 69, 67, 64, 60]
|
||||
morning_waltz = [64, 67, 72, 71, 67, 64, 62, 65, 69, 67, 64, 62, 60, 64, 67, 72]
|
||||
(OUTPUT / "first-snow-melody.mid").write_bytes(midi_file(melody(first_snow), 560_000))
|
||||
(OUTPUT / "first-snow-accompaniment.mid").write_bytes(
|
||||
midi_file(chord_progression([[48, 52, 55], [45, 48, 52], [41, 45, 48], [43, 47, 50]], first_snow), 560_000)
|
||||
)
|
||||
(OUTPUT / "blue-breeze-walk.mid").write_bytes(midi_file(melody(blue_breeze), 470_000))
|
||||
(OUTPUT / "little-morning-waltz.mid").write_bytes(
|
||||
midi_file(chord_progression([[48, 52, 55], [43, 47, 50], [45, 48, 52]], morning_waltz), 620_000)
|
||||
)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user