src/databases.py - master - yeexzdev/frontiersystem

src/databases.py
95 строк3 KB
yeexzdev
initial commit
15 май 2026, 20:25
15 май 2026, 20:25bc97861
import sqlite3
import logging
import os
from datetime import datetime
logger = logging.getLogger(__name__)

class CardsDB:
    def __init__(self, db_path):
        self.path = db_path
        self._init_db()

    def _connect(self):
        return sqlite3.connect(self.path, check_same_thread=False)

    def _init_db(self):
        with self._connect() as conn:
            conn.execute('''CREATE TABLE IF NOT EXISTS users 
                         (user_id INTEGER PRIMARY KEY, card_uid TEXT UNIQUE, taken TEXT NULL)''')
            conn.commit()

    def check_user_card(self, card_uid):
        with self._connect() as conn:
            res = conn.execute('SELECT user_id FROM users WHERE (card_uid = ? AND taken = ?)', 
                             (card_uid, "yes")).fetchone()
            return res[0] if res else None

    def get_free_card_id(self):
        with self._connect() as conn:
            cur = conn.cursor()
            cur.execute("SELECT user_id FROM users WHERE (taken = 'no' OR taken IS NULL) AND user_id != 1 LIMIT 1")
            res = cur.fetchone()
            if res:
                cur.execute("UPDATE users SET taken = 'yes' WHERE user_id = ?", (res[0],))
                conn.commit()
                return res[0]
            return None

class UsersDB:
    def __init__(self, db_path):
        self.path = db_path
        self._init_db()

    def _connect(self):
        return sqlite3.connect(self.path, check_same_thread=False)

    def _init_db(self):
        with self._connect() as conn:
            conn.execute('''CREATE TABLE IF NOT EXISTS userslist 
                         (user_id INTEGER PRIMARY KEY, name TEXT, pnumber TEXT, 
                          role TEXT, role2 TEXT, id TEXT, pin_hash TEXT)''')
            conn.commit()

    def is_registered(self, vk_id):
        with self._connect() as conn:
            res = conn.execute("SELECT 1 FROM userslist WHERE user_id = ?", (vk_id,)).fetchone()
            return res is not None

    def add_user(self, vk_id, name, phone, r1, r2, pool_id, pin_hash):
        with self._connect() as conn:
            conn.execute("INSERT INTO userslist VALUES (?, ?, ?, ?, ?, ?, ?)", 
                         (vk_id, name, phone, r1, r2, pool_id, pin_hash))
            conn.commit()

    def get_auth_data(self, pool_id):
        """Возвращает хэш пина, ВК ID и роль для проверки доступа"""
        with self._connect() as conn:
            query = "SELECT pin_hash, user_id, role FROM userslist WHERE id = ?"
            res = conn.execute(query, (pool_id,)).fetchone()
            return res
        

class LogsDB:
    def __init__(self, db_path):
        self.path = db_path
        self._init_db()

    def _connect(self):
        return sqlite3.connect(self.path, check_same_thread=False)

    def _init_db(self):
        with self._connect() as conn:
            conn.execute('''CREATE TABLE IF NOT EXISTS logs 
                         (id INTEGER PRIMARY KEY AUTOINCREMENT, 
                          dt TEXT, 
                          contour TEXT, 
                          card_id TEXT, 
                          status TEXT)''')
            conn.commit()

    def add_log(self, contour, card_id, status):
        dt = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        with self._connect() as conn:
            conn.execute("INSERT INTO logs (dt, contour, card_id, status) VALUES (?, ?, ?, ?)",
                         (dt, contour, card_id, status))
            conn.commit()