dekun
62 lines
1.7 KiB
Python
62 lines
1.7 KiB
Python
"""SQLite 注册表."""
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import sqlite3
|
|
from typing import Any
|
|
|
|
|
|
def db_path() -> str:
|
|
return (os.getenv("DB_PATH") or "hub.db").strip() or "hub.db"
|
|
|
|
|
|
def get_db() -> sqlite3.Connection:
|
|
conn = sqlite3.connect(db_path(), timeout=30)
|
|
conn.row_factory = sqlite3.Row
|
|
return conn
|
|
|
|
|
|
def init_db() -> None:
|
|
conn = get_db()
|
|
try:
|
|
conn.execute(
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS hub_instances (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
name TEXT NOT NULL,
|
|
base_url TEXT NOT NULL,
|
|
bridge_token TEXT NOT NULL,
|
|
embed_path TEXT NOT NULL DEFAULT '/embed',
|
|
tags TEXT DEFAULT '',
|
|
enabled INTEGER NOT NULL DEFAULT 1,
|
|
sort_order INTEGER NOT NULL DEFAULT 0,
|
|
created_at TEXT NOT NULL,
|
|
updated_at TEXT NOT NULL
|
|
)
|
|
"""
|
|
)
|
|
conn.execute(
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS hub_cash_samples (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
instance_id INTEGER NOT NULL,
|
|
ts_ms INTEGER NOT NULL,
|
|
cash_usdt REAL NOT NULL,
|
|
UNIQUE(instance_id, ts_ms)
|
|
)
|
|
"""
|
|
)
|
|
conn.execute(
|
|
"CREATE INDEX IF NOT EXISTS idx_hub_cash_samples_inst_ts "
|
|
"ON hub_cash_samples(instance_id, ts_ms)"
|
|
)
|
|
conn.commit()
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def row_to_dict(row: sqlite3.Row | None) -> dict[str, Any] | None:
|
|
if row is None:
|
|
return None
|
|
return {k: row[k] for k in row.keys()}
|