from config.database import get_connection, release_connection def _row_to_dict(row): """ Chuyển đổi một dòng kết quả từ tuple (từ câu lệnh JOIN) thành dictionary. """ if not row: return None device_dict = { "id": str(row[0]), "device_type_id": str(row[1]), "name": row[2], "description": row[3], "ip_address": row[4], "port": row[5], "latitude": row[6], "longitude": row[7], "color": row[8], "avatar_url": row[9], "is_active": row[10], "created": row[11].isoformat() if row[11] else None, "modified": row[12].isoformat() if row[12] else None } # Nếu câu truy vấn có JOIN với device_type if len(row) > 13: device_dict["device_type_name"] = row[13] device_dict["device_type_icon"] = row[14] return device_dict def find_all_devices(): """ Lấy danh sách tất cả các thiết bị trong database, JOIN với bảng device_type để hiển thị tên loại thiết bị và icon đại diện. """ conn = get_connection() cur = None try: cur = conn.cursor() cur.execute(""" SELECT d.id, d.device_type_id, d.name, d.description, d.ip_address, d.port, d.latitude, d.longitude, d.color, d.avatar_url, d.is_active, d.created, d.modified, dt.name as device_type_name, dt.icon_url as device_type_icon FROM device d JOIN device_type dt ON d.device_type_id = dt.id ORDER BY d.created DESC """) rows = cur.fetchall() devices = [] for row in rows: devices.append(_row_to_dict(row)) return devices finally: if cur: cur.close() release_connection(conn) def find_device_by_id(device_id): """ Tìm thiết bị theo ID, JOIN với device_type để lấy thông tin chi tiết. """ conn = get_connection() cur = None try: cur = conn.cursor() cur.execute(""" SELECT d.id, d.device_type_id, d.name, d.description, d.ip_address, d.port, d.latitude, d.longitude, d.color, d.avatar_url, d.is_active, d.created, d.modified, dt.name as device_type_name, dt.icon_url as device_type_icon FROM device d JOIN device_type dt ON d.device_type_id = dt.id WHERE d.id = %s """, (device_id,)) row = cur.fetchone() return _row_to_dict(row) if row else None finally: if cur: cur.close() release_connection(conn) def find_device_by_name(name): """ Tìm thiết bị theo tên (không phân biệt hoa thường) để phục vụ validate trùng tên. """ conn = get_connection() cur = None try: cur = conn.cursor() cur.execute(""" SELECT id, device_type_id, name, description, ip_address, port, latitude, longitude, color, avatar_url, is_active, created, modified FROM device WHERE LOWER(name) = LOWER(%s) """, (name,)) row = cur.fetchone() return _row_to_dict(row) if row else None finally: if cur: cur.close() release_connection(conn) def find_device_by_ip(ip_address): """ Tìm thiết bị theo địa chỉ IP để phục vụ validate tránh trùng lặp IP thiết bị. """ conn = get_connection() cur = None try: cur = conn.cursor() cur.execute(""" SELECT id, device_type_id, name, description, ip_address, port, latitude, longitude, color, avatar_url, is_active, created, modified FROM device WHERE ip_address = %s """, (ip_address,)) row = cur.fetchone() return _row_to_dict(row) if row else None finally: if cur: cur.close() release_connection(conn) def insert_device(data): """ Tạo mới một thiết bị mạng (chỉ ghi nhận vào bảng device). """ conn = get_connection() cur = None try: cur = conn.cursor() # 1. Thêm thiết bị vào bảng device cur.execute(""" INSERT INTO device (device_type_id, name, description, ip_address, port, latitude, longitude, color, avatar_url, is_active) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s) RETURNING id, device_type_id, name, description, ip_address, port, latitude, longitude, color, avatar_url, is_active, created, modified """, ( data["device_type_id"], data["name"], data.get("description"), data["ip_address"], data.get("port"), data["latitude"], data["longitude"], data["color"], data.get("avatar_url"), data.get("is_active", True) )) device_row = cur.fetchone() device_id = device_row[0] # 2. Lấy lại thiết bị kèm thông tin DeviceType đã JOIN để trả về đầy đủ dữ liệu cur.execute(""" SELECT d.id, d.device_type_id, d.name, d.description, d.ip_address, d.port, d.latitude, d.longitude, d.color, d.avatar_url, d.is_active, d.created, d.modified, dt.name as device_type_name, dt.icon_url as device_type_icon FROM device d JOIN device_type dt ON d.device_type_id = dt.id WHERE d.id = %s """, (device_id,)) full_row = cur.fetchone() # Commit transaction conn.commit() return _row_to_dict(full_row) except Exception: conn.rollback() raise finally: if cur: cur.close() release_connection(conn) def update_device_db(device_id, data): """ Cập nhật thông tin chi tiết của một thiết bị. """ conn = get_connection() cur = None try: cur = conn.cursor() update_fields = [] params = [] fields_list = [ "device_type_id", "name", "description", "ip_address", "port", "latitude", "longitude", "color", "avatar_url", "is_active" ] for key in fields_list: if key in data: update_fields.append(f"{key} = %s") params.append(data[key]) if not update_fields: return find_device_by_id(device_id) update_fields.append("modified = CURRENT_TIMESTAMP") sql = f""" UPDATE device SET {', '.join(update_fields)} WHERE id = %s RETURNING id, device_type_id, name, description, ip_address, port, latitude, longitude, color, avatar_url, is_active, created, modified """ params.append(device_id) cur.execute(sql, tuple(params)) row = cur.fetchone() conn.commit() if row: # Lấy chi tiết có kèm JOIN device_type để trả về đầy đủ return find_device_by_id(device_id) return None except Exception: conn.rollback() raise finally: if cur: cur.close() release_connection(conn) def delete_device_db(device_id): """ Xóa thiết bị khỏi database. Do có khóa ngoại với Cascade Delete, tất cả các dòng liên quan trong monitor_config, alert_config, device_status, alert_log sẽ tự động bị xóa. """ conn = get_connection() cur = None try: cur = conn.cursor() cur.execute(""" DELETE FROM device WHERE id = %s """, (device_id,)) conn.commit() except Exception: conn.rollback() raise finally: if cur: cur.close() release_connection(conn)