이창현
0522e48013
업로드 형식을 MIDI와 MusicXML과 MXL 및 PDF로 확장하고 파일별 검증을 추가했다. AlphaTab 악보 자산과 Nginx MIME 설정을 보완해 운영에서 오선과 음표가 표시되도록 수정했다. 공개 도메인 클래식과 난이도별 연습 편곡 7개를 같은 제목별로 묶고 출처와 운영 검증 결과를 문서화했다. 검사: TypeScript 검사와 Vitest 67개 및 Vite 운영 빌드 통과. 운영 브라우저 악보 표시와 네 가지 업로드 검증 형식 확인. 백업: /home/syskiller/piano/backups/bach-to-basics/20261003-1545-sheet-upload 운영 반영: 완료
332 lines
14 KiB
Python
332 lines
14 KiB
Python
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": "minuet-bwv-anh-114-beginner.mid",
|
|
"title": "미뉴에트 사장조 BWV Anh. 114",
|
|
"artist": "크리스티안 페촐트",
|
|
"description": "오른손 선율을 느린 속도로 먼저 익히는 입문 편곡입니다. 공개 도메인 원곡에서 연습용으로 구성했습니다.",
|
|
"difficulty": "beginner",
|
|
"genre": "바로크",
|
|
"arrangement": "melody",
|
|
"tags": ["미뉴에트", "사장조", "오른손", "입문", "공개 도메인"],
|
|
},
|
|
{
|
|
"storage_name": "minuet-bwv-anh-114.mid",
|
|
"title": "미뉴에트 사장조 BWV Anh. 114",
|
|
"artist": "크리스티안 페촐트",
|
|
"description": "안나 막달레나 바흐 음악수첩에 수록된 양손 원곡입니다. Mutopia Project 공개 도메인 자료를 사용했습니다.",
|
|
"difficulty": "easy",
|
|
"genre": "바로크",
|
|
"arrangement": "piano_solo",
|
|
"tags": ["미뉴에트", "사장조", "양손", "입문", "공개 도메인"],
|
|
},
|
|
{
|
|
"storage_name": "menuett-woo-82-easy.mid",
|
|
"title": "미뉴에트 내림마장조 WoO 82",
|
|
"artist": "루트비히 판 베토벤",
|
|
"description": "오른손 선율을 느린 속도로 익히는 쉬운 편곡입니다. 공개 도메인 원곡에서 연습용으로 구성했습니다.",
|
|
"difficulty": "easy",
|
|
"genre": "고전",
|
|
"arrangement": "melody",
|
|
"tags": ["미뉴에트", "내림마장조", "오른손", "쉬움", "공개 도메인"],
|
|
},
|
|
{
|
|
"storage_name": "menuett-woo-82.mid",
|
|
"title": "미뉴에트 내림마장조 WoO 82",
|
|
"artist": "루트비히 판 베토벤",
|
|
"description": "고전주의의 프레이징과 양손 균형을 익히는 미뉴에트입니다. Mutopia Project 공개 도메인 자료를 사용했습니다.",
|
|
"difficulty": "intermediate",
|
|
"genre": "고전",
|
|
"arrangement": "piano_solo",
|
|
"tags": ["미뉴에트", "내림마장조", "양손", "고전", "공개 도메인"],
|
|
},
|
|
{
|
|
"storage_name": "fur-elise-beginner.mid",
|
|
"title": "엘리제를 위하여 WoO 59",
|
|
"artist": "루트비히 판 베토벤",
|
|
"description": "유명한 첫 주제를 오른손으로 천천히 익히는 입문 편곡입니다. 공개 도메인 원곡에서 연습용으로 구성했습니다.",
|
|
"difficulty": "beginner",
|
|
"genre": "고전",
|
|
"arrangement": "melody",
|
|
"tags": ["오른손", "주제 연습", "입문", "베토벤", "공개 도메인"],
|
|
},
|
|
{
|
|
"storage_name": "fur-elise-easy.mid",
|
|
"title": "엘리제를 위하여 WoO 59",
|
|
"artist": "루트비히 판 베토벤",
|
|
"description": "첫 주제의 오른손 선율과 단순화한 왼손 근음을 함께 연습하는 쉬운 양손 편곡입니다.",
|
|
"difficulty": "easy",
|
|
"genre": "고전",
|
|
"arrangement": "accompaniment",
|
|
"tags": ["양손", "쉬운 반주", "주제 연습", "베토벤", "공개 도메인"],
|
|
},
|
|
{
|
|
"storage_name": "fur-elise.mid",
|
|
"title": "엘리제를 위하여 WoO 59",
|
|
"artist": "루트비히 판 베토벤",
|
|
"description": "오른손 아르페지오와 양손 전환을 연습하는 대표적인 피아노 소품입니다. Mutopia Project 공개 도메인 자료를 사용했습니다.",
|
|
"difficulty": "advanced",
|
|
"genre": "고전",
|
|
"arrangement": "piano_solo",
|
|
"tags": ["아르페지오", "양손", "피아노 소품", "베토벤", "공개 도메인"],
|
|
},
|
|
)
|
|
|
|
LEGACY_SAMPLE_KEYS = (
|
|
"first-snow-melody.mid",
|
|
"first-snow-accompaniment.mid",
|
|
"blue-breeze-walk.mid",
|
|
"little-morning-waltz.mid",
|
|
)
|
|
|
|
|
|
def _seed_samples() -> None:
|
|
source_root = Path(__file__).with_name("sample_midi")
|
|
with connect() as connection:
|
|
# 초기 시험용으로 만든 짧은 음형은 실제 공개 도메인 연습곡으로 교체한다.
|
|
# seed_key가 정확히 일치하는 기본 자료만 지우므로 회원 업로드에는 영향을 주지 않는다.
|
|
placeholders = ",".join("?" for _ in LEGACY_SAMPLE_KEYS)
|
|
connection.execute(
|
|
f"DELETE FROM midi_items WHERE owner_id IS NULL AND seed_key IN ({placeholders})",
|
|
LEGACY_SAMPLE_KEYS,
|
|
)
|
|
for legacy_name in LEGACY_SAMPLE_KEYS:
|
|
(MIDI_ROOT / legacy_name).unlink(missing_ok=True)
|
|
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() or hashlib.sha256(target.read_bytes()).digest() != hashlib.sha256(source.read_bytes()).digest():
|
|
shutil.copyfile(source, target)
|
|
existing = connection.execute(
|
|
"SELECT id FROM midi_items WHERE seed_key = ?", (sample["storage_name"],)
|
|
).fetchone()
|
|
if existing:
|
|
content = target.read_bytes()
|
|
connection.execute(
|
|
"""
|
|
UPDATE midi_items
|
|
SET title = ?, artist = ?, description = ?, original_name = ?,
|
|
storage_name = ?, size_bytes = ?, sha256 = ?, visibility = 'public',
|
|
min_tier = 0, difficulty = ?, genre = ?, arrangement = ?,
|
|
tags_json = ?, updated_at = ?
|
|
WHERE id = ?
|
|
""",
|
|
(
|
|
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(), existing["id"],
|
|
),
|
|
)
|
|
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()
|