67 lines
1.5 KiB
Python
67 lines
1.5 KiB
Python
import sqlite3
|
|
|
|
from database.paths import (
|
|
L1_DB,
|
|
L2_DB,
|
|
L2_SCHEMA,
|
|
L3_DB,
|
|
L3_SCHEMA,
|
|
RUNTIME_DATA_ROOT,
|
|
ensure_runtime_directories,
|
|
)
|
|
|
|
|
|
def _apply_schema(database_path, schema_path):
|
|
db = sqlite3.connect(str(database_path))
|
|
try:
|
|
db.execute('PRAGMA foreign_keys = ON')
|
|
db.executescript(schema_path.read_text(encoding='utf-8'))
|
|
result = db.execute('PRAGMA quick_check').fetchone()[0]
|
|
if result != 'ok':
|
|
raise RuntimeError(
|
|
f'{database_path.name} quick_check failed: {result}'
|
|
)
|
|
db.commit()
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
def bootstrap_runtime():
|
|
ensure_runtime_directories()
|
|
|
|
l1 = sqlite3.connect(str(L1_DB))
|
|
try:
|
|
l1.execute(
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS raw_iframe_network (
|
|
match_id TEXT PRIMARY KEY,
|
|
content TEXT NOT NULL,
|
|
processed_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
|
|
)
|
|
"""
|
|
)
|
|
l1.commit()
|
|
finally:
|
|
l1.close()
|
|
|
|
_apply_schema(L2_DB, L2_SCHEMA)
|
|
_apply_schema(L3_DB, L3_SCHEMA)
|
|
|
|
from web.database import initialize_web_db
|
|
initialize_web_db()
|
|
|
|
return {
|
|
'data_root': str(RUNTIME_DATA_ROOT),
|
|
'l1': str(L1_DB),
|
|
'l2': str(L2_DB),
|
|
'l3': str(L3_DB),
|
|
}
|
|
|
|
|
|
if __name__ == '__main__':
|
|
result = bootstrap_runtime()
|
|
print('YRTV runtime initialized')
|
|
for key, value in result.items():
|
|
print(f' {key}: {value}')
|
|
|