сборка для Ильи

database.py

@@ -0,0 +1,202 @@
+import sqlite3
+import os
+
+DB_PATH = os.path.join(os.path.dirname(__file__), 'app_data.db')
+
+def init_db():
+    conn = sqlite3.connect(DB_PATH)
+    cursor = conn.cursor()
+    cursor.execute('''
+        CREATE TABLE IF NOT EXISTS address_list_items (
+            id INTEGER PRIMARY KEY AUTOINCREMENT,
+            ip TEXT UNIQUE NOT NULL,
+            list_name TEXT NOT NULL,
+            comment TEXT,
+            mikrotik_id TEXT,
+            enabled BOOLEAN DEFAULT TRUE,
+            last_sync DATETIME DEFAULT CURRENT_TIMESTAMP
+        )
+    ''')
+    cursor.execute('''
+        CREATE TABLE IF NOT EXISTS admins (
+            id INTEGER PRIMARY KEY AUTOINCREMENT,
+            username TEXT UNIQUE NOT NULL,
+            password_hash TEXT NOT NULL,
+            created_at DATETIME DEFAULT CURRENT_TIMESTAMP
+        )
+    ''')
+    cursor.execute('''
+        CREATE TABLE IF NOT EXISTS named_lists (
+            id INTEGER PRIMARY KEY AUTOINCREMENT,
+            name TEXT UNIQUE NOT NULL
+        )
+    ''')
+    
+    # Миграция: добавляем существующие уникальные имена списков в новую таблицу
+    cursor.execute('SELECT DISTINCT list_name FROM address_list_items')
+    existing_lists = cursor.fetchall()
+    for row in existing_lists:
+        cursor.execute('INSERT OR IGNORE INTO named_lists (name) VALUES (?)', (row[0],))
+    
+    # Если список пуст, создаем дефолтный
+    cursor.execute('SELECT COUNT(*) FROM named_lists')
+    if cursor.fetchone()[0] == 0:
+        cursor.execute('INSERT INTO named_lists (name) VALUES (?)', ('trusted',))
+
+    conn.commit()
+    conn.close()
+
+def get_db_connection():
+    conn = sqlite3.connect(DB_PATH)
+    conn.row_factory = sqlite3.Row
+    return conn
+
+def get_item_by_ip(ip):
+    conn = get_db_connection()
+    item = conn.execute('SELECT * FROM address_list_items WHERE ip = ?', (ip,)).fetchone()
+    conn.close()
+    return item
+
+def get_all_items(list_name=None):
+    conn = get_db_connection()
+    if list_name:
+        items = conn.execute('SELECT * FROM address_list_items WHERE list_name = ?', (list_name,)).fetchall()
+    else:
+        items = conn.execute('SELECT * FROM address_list_items').fetchall()
+    conn.close()
+    return [dict(i) for i in items]
+
+def add_item(ip, list_name, comment, mikrotik_id, enabled=True):
+    conn = get_db_connection()
+    try:
+        cursor = conn.cursor()
+        cursor.execute(
+            'INSERT INTO address_list_items (ip, list_name, comment, mikrotik_id, enabled) VALUES (?, ?, ?, ?, ?)',
+            (ip, list_name, comment, mikrotik_id, enabled)
+        )
+        conn.commit()
+        return True, cursor.lastrowid
+    except sqlite3.IntegrityError:
+        return False, "IP already exists"
+    finally:
+        conn.close()
+
+def update_item_status(ip, enabled):
+    conn = get_db_connection()
+    conn.execute(
+        'UPDATE address_list_items SET enabled = ?, last_sync = CURRENT_TIMESTAMP WHERE ip = ?',
+        (enabled, ip)
+    )
+    conn.commit()
+    conn.close()
+
+def update_mikrotik_id(ip, mikrotik_id):
+    conn = get_db_connection()
+    conn.execute(
+        'UPDATE address_list_items SET mikrotik_id = ? WHERE ip = ?',
+        (mikrotik_id, ip)
+    )
+    conn.commit()
+    conn.close()
+
+def delete_item_by_ip(ip):
+    conn = get_db_connection()
+    conn.execute('DELETE FROM address_list_items WHERE ip = ?', (ip,))
+    conn.commit()
+    conn.close()
+
+def delete_item_by_id(db_id):
+    conn = get_db_connection()
+    cursor = conn.cursor()
+    item = conn.execute('SELECT ip, mikrotik_id FROM address_list_items WHERE id = ?', (db_id,)).fetchone()
+    if item:
+        conn.execute('DELETE FROM address_list_items WHERE id = ?', (db_id,))
+        conn.commit()
+        conn.close()
+        return dict(item)
+    conn.close()
+    return None
+
+def get_item_by_id(db_id):
+    conn = get_db_connection()
+    item = conn.execute('SELECT * FROM address_list_items WHERE id = ?', (db_id,)).fetchone()
+    conn.close()
+    return item
+
+def get_unique_lists():
+    conn = get_db_connection()
+    lists = conn.execute('SELECT name FROM named_lists ORDER BY name').fetchall()
+    conn.close()
+    return [row['name'] for row in lists]
+
+def get_named_lists_detailed():
+    conn = get_db_connection()
+    lists = conn.execute('SELECT * FROM named_lists ORDER BY name').fetchall()
+    conn.close()
+    return [dict(row) for row in lists]
+
+def add_named_list(name):
+    conn = get_db_connection()
+    try:
+        conn.execute('INSERT INTO named_lists (name) VALUES (?)', (name,))
+        conn.commit()
+        return True, ""
+    except sqlite3.IntegrityError:
+        return False, "List already exists"
+    finally:
+        conn.close()
+
+def delete_named_list(list_id):
+    conn = get_db_connection()
+    try:
+        # Проверяем, используются ли записи из этого списка
+        list_name_row = conn.execute('SELECT name FROM named_lists WHERE id = ?', (list_id,)).fetchone()
+        if not list_name_row:
+            return False, "Not found"
+        
+        usage_count = conn.execute('SELECT COUNT(*) FROM address_list_items WHERE list_name = ?', (list_name_row['name'],)).fetchone()[0]
+        if usage_count > 0:
+            return False, f"Cannot delete: {usage_count} addresses are still in this list"
+
+        conn.execute('DELETE FROM named_lists WHERE id = ?', (list_id,))
+        conn.commit()
+        return True, ""
+    finally:
+        conn.close()
+
+# --- Admin Operations ---
+
+def get_admin(username):
+    conn = get_db_connection()
+    admin = conn.execute('SELECT * FROM admins WHERE username = ?', (username,)).fetchone()
+    conn.close()
+    return admin
+
+def add_admin(username, password_hash):
+    conn = get_db_connection()
+    try:
+        conn.execute('INSERT INTO admins (username, password_hash) VALUES (?, ?)', (username, password_hash))
+        conn.commit()
+        return True
+    except sqlite3.IntegrityError:
+        return False
+    finally:
+        conn.close()
+
+def update_admin(old_username, new_username, new_password_hash):
+    conn = get_db_connection()
+    try:
+        conn.execute(
+            'UPDATE admins SET username = ?, password_hash = ? WHERE username = ?',
+            (new_username, new_password_hash, old_username)
+        )
+        conn.commit()
+        return True
+    except Exception:
+        return False
+    finally:
+        conn.close()
+
+if __name__ == '__main__':
+    init_db()
+    print("Database initialized.")