+14
-15
@@ -7,13 +7,12 @@ import tempfile
|
||||
from pathlib import Path
|
||||
|
||||
APPLICATION_ID = 0x56414D50
|
||||
SCHEMA_VERSION = 2
|
||||
SCHEMA_VERSION = 3
|
||||
COMMON_COLUMNS = """
|
||||
id TEXT PRIMARY KEY NOT NULL CHECK(length(trim(id)) BETWEEN 1 AND 160),
|
||||
title TEXT NOT NULL CHECK(length(trim(title)) BETWEEN 1 AND 200),
|
||||
composer TEXT NOT NULL,
|
||||
key_signature TEXT NOT NULL,
|
||||
authored_key_signature TEXT NOT NULL,
|
||||
source_path TEXT NOT NULL UNIQUE CHECK(length(source_path) > 0),
|
||||
source_json TEXT NOT NULL CHECK(json_valid(source_json)),
|
||||
measure_count INTEGER NOT NULL CHECK(measure_count > 0),
|
||||
@@ -34,19 +33,12 @@ def metadata(path: Path, root: Path) -> dict:
|
||||
for name in ('title', 'composer', 'key_signature', 'time_signature'):
|
||||
if not isinstance(meta[name], str):
|
||||
raise ValueError(f'{name} must be a string')
|
||||
authored = meta.get('authored_key_signature', '')
|
||||
if not isinstance(authored, str) or ('authored_key_signature' in meta and not authored):
|
||||
raise ValueError('authored_key_signature must be a nonempty string')
|
||||
key = meta['key_signature']
|
||||
if staff.get('arrangement_part_type') == 'practice' and key == 'No key' and 'variant_tonic' in meta:
|
||||
if not isinstance(meta['variant_tonic'], str):
|
||||
raise ValueError('variant_tonic must be a string')
|
||||
key = meta['variant_tonic'] + ' (no key)'
|
||||
if not isinstance(staff['name'], str):
|
||||
raise ValueError('staff name must be a string')
|
||||
relative = path.relative_to(root)
|
||||
return dict(id=path.stem, title=meta['title'], composer=meta['composer'], key_signature=key,
|
||||
authored_key_signature=authored, source_path=relative.as_posix(), source_json=text,
|
||||
source_path=relative.as_posix(), source_json=text,
|
||||
measure_count=len(staff['measures']), total_pages=len(presentation['pages']), part=staff['name'])
|
||||
|
||||
def populate(connection: sqlite3.Connection, root: Path) -> None:
|
||||
@@ -56,11 +48,11 @@ def populate(connection: sqlite3.Connection, root: Path) -> None:
|
||||
PRAGMA application_id = {APPLICATION_ID};
|
||||
PRAGMA user_version = {SCHEMA_VERSION};
|
||||
CREATE TABLE songs ({COMMON_COLUMNS}, part TEXT NOT NULL) STRICT;
|
||||
CREATE TABLE practice_exercises ({COMMON_COLUMNS}, exercise_id TEXT NOT NULL) STRICT;
|
||||
CREATE TABLE practice_exercises ({COMMON_COLUMNS}, exercise_id TEXT NOT NULL, part TEXT NOT NULL) STRICT;
|
||||
CREATE INDEX songs_menu ON songs(title COLLATE NOCASE, composer COLLATE NOCASE, part COLLATE NOCASE);
|
||||
CREATE INDEX practice_menu ON practice_exercises(exercise_id, id);
|
||||
''')
|
||||
authored_keys = {}
|
||||
practice_parts = {}
|
||||
for category, table, depth in (('songs', 'songs', 1), ('practice', 'practice_exercises', 2)):
|
||||
folder = root / category
|
||||
if (folder.exists() or folder.is_symlink()) and not folder.is_dir():
|
||||
@@ -73,13 +65,20 @@ def populate(connection: sqlite3.Connection, root: Path) -> None:
|
||||
row = metadata(path, root)
|
||||
if category == 'practice':
|
||||
identity = parts[0]
|
||||
if authored_keys.setdefault(identity, row['authored_key_signature']) != row['authored_key_signature']:
|
||||
raise ValueError(f'practice variants disagree on authored key: {identity}')
|
||||
row.pop('part')
|
||||
if path.stem not in ('acoustic', 'electric', 'vocal') or row['part'] != path.stem.capitalize():
|
||||
raise ValueError('practice requires an acoustic, electric, or vocal part')
|
||||
identity_metadata = (row['title'], row['composer'], row['key_signature'])
|
||||
group = practice_parts.setdefault(identity, (identity_metadata, set()))
|
||||
if group[0] != identity_metadata:
|
||||
raise ValueError(f'practice parts disagree on metadata: {identity}')
|
||||
group[1].add(path.stem)
|
||||
row.update(id=identity + '-' + path.stem, exercise_id=identity)
|
||||
connection.execute(f'INSERT INTO {table} ({", ".join(row)}) VALUES ({", ".join("?" for _ in row)})', tuple(row.values()))
|
||||
except (ValueError, KeyError, TypeError, sqlite3.Error) as error:
|
||||
raise ValueError(f'{path}: {error}') from error
|
||||
for identity, (_, parts) in practice_parts.items():
|
||||
if parts != {'acoustic', 'electric', 'vocal'}:
|
||||
raise ValueError(f'practice exercise requires all three parts: {identity}')
|
||||
connection.commit()
|
||||
if connection.execute('PRAGMA integrity_check').fetchall() != [('ok',)]:
|
||||
raise ValueError('catalogue integrity check failed')
|
||||
|
||||
Reference in New Issue
Block a user