232 lines
7.3 KiB
Python
232 lines
7.3 KiB
Python
'''
|
|
____ ___ _ _ _
|
|
/ ___| / _ \| | (_) |_ ___
|
|
\___ \| | | | | | | __/ _ \
|
|
___) | |_| | |___| | || __/
|
|
|____/ \__\_\_____|_|\__\___|
|
|
|
|
Script to delete a table from sqlite database
|
|
/usr/bin/python3 /var/www/nebuleair_pro_4g/sqlite/delete_table.py table_name [--confirm]
|
|
|
|
Available tables are:
|
|
data_NPM
|
|
data_NPM_5channels
|
|
data_BME280
|
|
data_envea
|
|
timestamp_table
|
|
data_MPPT
|
|
data_WIND
|
|
modem_status
|
|
config_table
|
|
envea_sondes_table
|
|
|
|
Examples:
|
|
# Will ask for confirmation
|
|
python3 delete_table.py data_NPM
|
|
|
|
# Skip confirmation prompt
|
|
python3 delete_table.py data_NPM --confirm
|
|
|
|
# List all tables
|
|
python3 delete_table.py --list
|
|
'''
|
|
|
|
import sqlite3
|
|
import sys
|
|
import os
|
|
from datetime import datetime
|
|
|
|
def list_tables(cursor):
|
|
"""List all tables in the database"""
|
|
cursor.execute("SELECT name FROM sqlite_master WHERE type='table' ORDER BY name")
|
|
tables = cursor.fetchall()
|
|
|
|
print("\n📋 Available tables:")
|
|
print("-" * 40)
|
|
for table in tables:
|
|
# Get row count for each table
|
|
cursor.execute(f"SELECT COUNT(*) FROM {table[0]}")
|
|
count = cursor.fetchone()[0]
|
|
print(f" {table[0]} ({count} rows)")
|
|
print("-" * 40)
|
|
|
|
def get_table_info(cursor, table_name):
|
|
"""Get information about a table"""
|
|
try:
|
|
# Check if table exists
|
|
cursor.execute("SELECT name FROM sqlite_master WHERE type='table' AND name=?", (table_name,))
|
|
if not cursor.fetchone():
|
|
return None
|
|
|
|
# Get row count
|
|
cursor.execute(f"SELECT COUNT(*) FROM {table_name}")
|
|
row_count = cursor.fetchone()[0]
|
|
|
|
# Get table schema
|
|
cursor.execute(f"PRAGMA table_info({table_name})")
|
|
columns = cursor.fetchall()
|
|
|
|
return {
|
|
'row_count': row_count,
|
|
'columns': columns
|
|
}
|
|
except sqlite3.Error as e:
|
|
print(f"Error getting table info: {e}")
|
|
return None
|
|
|
|
def backup_table(cursor, table_name, db_path):
|
|
"""Create a backup of the table before deletion"""
|
|
try:
|
|
backup_dir = os.path.dirname(db_path)
|
|
timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
|
|
backup_file = os.path.join(backup_dir, f"{table_name}_backup_{timestamp}.sql")
|
|
|
|
# Get table schema
|
|
cursor.execute(f"SELECT sql FROM sqlite_master WHERE type='table' AND name=?", (table_name,))
|
|
create_sql = cursor.fetchone()
|
|
|
|
if create_sql:
|
|
with open(backup_file, 'w') as f:
|
|
# Write table creation SQL
|
|
f.write(f"-- Backup of table {table_name} created on {datetime.now()}\n")
|
|
f.write(f"{create_sql[0]};\n\n")
|
|
|
|
# Write data
|
|
cursor.execute(f"SELECT * FROM {table_name}")
|
|
rows = cursor.fetchall()
|
|
|
|
if rows:
|
|
# Get column names
|
|
cursor.execute(f"PRAGMA table_info({table_name})")
|
|
columns = [col[1] for col in cursor.fetchall()]
|
|
|
|
f.write(f"-- Data for table {table_name}\n")
|
|
for row in rows:
|
|
values = []
|
|
for value in row:
|
|
if value is None:
|
|
values.append('NULL')
|
|
elif isinstance(value, str):
|
|
escaped_value = value.replace("'", "''")
|
|
values.append(f"'{escaped_value}'")
|
|
else:
|
|
values.append(str(value))
|
|
|
|
f.write(f"INSERT INTO {table_name} ({', '.join(columns)}) VALUES ({', '.join(values)});\n")
|
|
|
|
print(f"✓ Table backed up to: {backup_file}")
|
|
return backup_file
|
|
except Exception as e:
|
|
print(f"⚠️ Backup failed: {e}")
|
|
return None
|
|
|
|
def delete_table(cursor, table_name, create_backup=True, db_path=None):
|
|
"""Delete a table from the database"""
|
|
|
|
# Get table info first
|
|
table_info = get_table_info(cursor, table_name)
|
|
if not table_info:
|
|
print(f"❌ Table '{table_name}' does not exist!")
|
|
return False
|
|
|
|
print(f"\n📊 Table Information:")
|
|
print(f" Name: {table_name}")
|
|
print(f" Rows: {table_info['row_count']}")
|
|
print(f" Columns: {len(table_info['columns'])}")
|
|
|
|
# Create backup if requested
|
|
backup_file = None
|
|
if create_backup and db_path:
|
|
print(f"\n💾 Creating backup...")
|
|
backup_file = backup_table(cursor, table_name, db_path)
|
|
|
|
try:
|
|
# Delete the table
|
|
cursor.execute(f"DROP TABLE {table_name}")
|
|
print(f"\n✅ Table '{table_name}' deleted successfully!")
|
|
|
|
if backup_file:
|
|
print(f" Backup saved: {backup_file}")
|
|
|
|
return True
|
|
|
|
except sqlite3.Error as e:
|
|
print(f"❌ Error deleting table: {e}")
|
|
return False
|
|
|
|
def main():
|
|
if len(sys.argv) < 2:
|
|
print("Usage: python3 delete_table.py <table_name> [--confirm] [--no-backup]")
|
|
print(" python3 delete_table.py --list")
|
|
sys.exit(1)
|
|
|
|
db_path = "/var/www/nebuleair_pro_4g/sqlite/sensors.db"
|
|
|
|
# Check if database exists
|
|
if not os.path.exists(db_path):
|
|
print(f"❌ Database not found: {db_path}")
|
|
sys.exit(1)
|
|
|
|
# Parse arguments
|
|
args = sys.argv[1:]
|
|
|
|
if '--list' in args:
|
|
# List all tables
|
|
conn = sqlite3.connect(db_path)
|
|
cursor = conn.cursor()
|
|
list_tables(cursor)
|
|
conn.close()
|
|
return
|
|
|
|
table_name = args[0]
|
|
skip_confirmation = '--confirm' in args
|
|
create_backup = '--no-backup' not in args
|
|
|
|
try:
|
|
# Connect to database
|
|
conn = sqlite3.connect(db_path)
|
|
cursor = conn.cursor()
|
|
|
|
# List available tables first
|
|
list_tables(cursor)
|
|
|
|
# Check if table exists
|
|
table_info = get_table_info(cursor, table_name)
|
|
if not table_info:
|
|
print(f"\n❌ Table '{table_name}' does not exist!")
|
|
conn.close()
|
|
sys.exit(1)
|
|
|
|
# Confirmation prompt
|
|
if not skip_confirmation:
|
|
print(f"\n⚠️ WARNING: You are about to delete table '{table_name}'")
|
|
print(f" This table contains {table_info['row_count']} rows")
|
|
if create_backup:
|
|
print(f" A backup will be created before deletion")
|
|
else:
|
|
print(f" NO BACKUP will be created (--no-backup flag used)")
|
|
|
|
response = input(f"\nAre you sure you want to delete '{table_name}'? (yes/no): ").lower().strip()
|
|
|
|
if response not in ['yes', 'y']:
|
|
print("❌ Operation cancelled")
|
|
conn.close()
|
|
sys.exit(0)
|
|
|
|
# Perform deletion
|
|
success = delete_table(cursor, table_name, create_backup, db_path)
|
|
|
|
if success:
|
|
conn.commit()
|
|
print(f"\n🎉 Operation completed successfully!")
|
|
else:
|
|
print(f"\n❌ Operation failed!")
|
|
|
|
conn.close()
|
|
|
|
except Exception as e:
|
|
print(f"❌ Error: {e}")
|
|
sys.exit(1)
|
|
|
|
if __name__ == "__main__":
|
|
main() |