이창현
8298f5af26
CI / Frontend - type-check & test (push) Canceled after 0s
비슷한 임시 연습곡 다섯 개를 교체했다. 부르크뮐러와 슈만 원본 30곡의 출처와 권리 정보를 보존했다. 각 MIDI에 양손 연습용 도입부와 마무리를 넣고 난이도를 음표 밀도로 분류했다. MIDI 구조 검사 30개와 서버 격리 테스트 5개를 통과했다. 운영 백업: /home/syskiller/piano/backups/bach-to-basics/20261009-practice-30-intro-ending. 백엔드만 운영에 반영했고 HTTPS 공개 목록과 30개 다운로드를 확인했다. 회원 업로드 2건은 보존됐다. 실제 건반 연주감 검사는 남아 있다.
348 lines
15 KiB
Python
348 lines
15 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": "piano_solo",
|
|
"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": "piano_solo",
|
|
"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": "piano_solo",
|
|
"tags": ["양손", "주제 연습", "입문", "베토벤", "공개 도메인"],
|
|
},
|
|
{
|
|
"storage_name": "fur-elise-easy.mid",
|
|
"title": "엘리제를 위하여 WoO 59",
|
|
"artist": "루트비히 판 베토벤",
|
|
"description": "첫 주제의 오른손 선율과 단순화한 왼손 근음을 함께 연습하는 쉬운 양손 편곡입니다.",
|
|
"difficulty": "easy",
|
|
"genre": "고전",
|
|
"arrangement": "piano_solo",
|
|
"tags": ["양손", "쉬운 양손", "주제 연습", "베토벤", "공개 도메인"],
|
|
},
|
|
{
|
|
"storage_name": "fur-elise.mid",
|
|
"title": "엘리제를 위하여 WoO 59",
|
|
"artist": "루트비히 판 베토벤",
|
|
"description": "오른손 아르페지오와 양손 전환을 연습하는 대표적인 피아노 소품입니다. Mutopia Project 공개 도메인 자료를 사용했습니다.",
|
|
"difficulty": "advanced",
|
|
"genre": "고전",
|
|
"arrangement": "piano_solo",
|
|
"tags": ["아르페지오", "양손", "피아노 소품", "베토벤", "공개 도메인"],
|
|
},
|
|
)
|
|
|
|
# 별도 목록은 원본 MIDI와 권리 표시를 함께 보관한다. 화면에 쓰는 필드만 기본 샘플에 합친다.
|
|
PRACTICE_COLLECTION = json.loads(
|
|
(Path(__file__).with_name("sample_midi") / "practice_collection.json").read_text(encoding="utf-8")
|
|
)
|
|
SAMPLES += tuple(
|
|
{key: item[key] for key in (
|
|
"storage_name", "title", "artist", "description", "difficulty", "genre", "arrangement", "tags"
|
|
)}
|
|
for item in PRACTICE_COLLECTION
|
|
)
|
|
|
|
LEGACY_SAMPLE_KEYS = (
|
|
"first-snow-melody.mid",
|
|
"first-snow-accompaniment.mid",
|
|
"blue-breeze-walk.mid",
|
|
"little-morning-waltz.mid",
|
|
"starlight-steps-beginner.mid",
|
|
"starlight-steps-easy.mid",
|
|
"evening-waltz-beginner.mid",
|
|
"sunrise-arpeggio-easy.mid",
|
|
"black-key-pentatonic-beginner.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()
|