data/db.py - main - alex.piter/zigbridge

data/db.py
111 строк3 KB
Alexey Lapko
fix bug
27 май 2026, 16:53
27 май 2026, 16:538cb2844
import sqlite3
import os

DEFAULT_DB = os.path.join(os.path.dirname(__file__), '..', 'database.db')
# Allow overriding DB path via env var (useful in container where /data may be mounted)
DB_PATH = os.environ.get('ZIGBRIDGE_DB', os.environ.get('DB_PATH', DEFAULT_DB))


def get_connection():
    # Ensure directory exists for DB file
    db_dir = os.path.dirname(DB_PATH)
    try:
        os.makedirs(db_dir, exist_ok=True)
    except Exception:
        pass
    return sqlite3.connect(DB_PATH)


def create_database():
    init_request = """
    CREATE TABLE IF NOT EXISTS records(
    cluster_id INTEGER NOT NULL UNIQUE,
    device_id TEXT UNIQUE,
    status INTEGER DEFAULT 0,
    enabled INTEGER DEFAULT 0
    )
    """
    conn = get_connection()
    cursor = conn.cursor()
    cursor.execute(init_request)
    conn.commit()
    cursor.close()
    conn.close()


def init():
    try:
        conn = get_connection()
        conn.cursor().execute("SELECT * FROM records LIMIT 1").close()
        conn.close()
    except Exception:
        create_database()


def get_devices():
    conn = get_connection()
    cursor = conn.cursor()
    cursor.execute("SELECT cluster_id, device_id, status, enabled FROM records")
    data = cursor.fetchall()
    cursor.close()
    conn.close()
    result = []
    for i in data:
        result.append({
            'cluster_id': i[0],
            'ha_device': i[1],
            'status': False if i[2] == 0 else True,
            'enabled': False if i[3] == 0 else True
        })
    return result

def get_device(device_id):
    conn = get_connection()
    cursor = conn.cursor()
    cursor.execute("SELECT cluster_id, device_id, status, enabled FROM records WHERE device_id=?", (device_id,))
    data = cursor.fetchone()
    cursor.close()
    conn.close()
    result = None
    if data:
        result = {
        'cluster_id': data[0],
        'ha_device': data[1],
        'status': False if data[2] == 0 else True,
        'enabled': False if data[3] == 0 else True
    }
    return result

def delete_device(cluster_id):
    conn = get_connection()
    cursor = conn.cursor()
    cursor.execute("DELETE FROM records WHERE cluster_id=?", (cluster_id,))
    conn.commit()
    cursor.close()
    conn.close()


def change_device(data):
    # data expected to contain keys: 'ha_device', 'status', 'enabled', 'cluster_id'
    conn = get_connection()
    cursor = conn.cursor()
    cursor.execute(
        "UPDATE records SET device_id=?, status=?, enabled=? WHERE cluster_id=?",
        (data['ha_device'], 1 if data.get('status') else 0, 1 if data.get('enabled') else 0, data['cluster_id'])
    )
    conn.commit()
    cursor.close()
    conn.close()


def add_device(data):
    conn = get_connection()
    cursor = conn.cursor()
    cursor.execute(
        "INSERT OR IGNORE INTO records(cluster_id,device_id,status,enabled) VALUES (?, ?, 0, 1)",
        (data['cluster_id'], data['ha_device'])
    )
    conn.commit()
    cursor.close()
    conn.close()
    return True