forked from agessaman/meshcore-bot
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathbackup_database.py
More file actions
executable file
·224 lines (170 loc) · 6.33 KB
/
Copy pathbackup_database.py
File metadata and controls
executable file
·224 lines (170 loc) · 6.33 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
#!/usr/bin/env python3
"""
Database Backup Script for MeshCore Bot
Creates timestamped backups of all SQLite database files
"""
import sqlite3
import shutil
import os
import sys
from datetime import datetime
from pathlib import Path
def backup_database(db_path, backup_dir="backups"):
"""
Create a timestamped backup of a SQLite database
Args:
db_path: Path to the database file
backup_dir: Directory to store backups (default: backups)
Returns:
Path to the backup file if successful, None otherwise
"""
db_path = Path(db_path)
if not db_path.exists():
print(f"Warning: Database file {db_path} does not exist, skipping...")
return None
# Create backup directory if it doesn't exist
backup_path = Path(backup_dir)
backup_path.mkdir(exist_ok=True)
# Generate timestamp string
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
# Create backup filename with timestamp
db_name = db_path.stem
backup_filename = f"{db_name}_{timestamp}.db"
backup_file = backup_path / backup_filename
try:
# Use SQLite backup API for proper backup (handles WAL files correctly)
source_conn = sqlite3.connect(str(db_path))
backup_conn = sqlite3.connect(str(backup_file))
# Perform the backup
source_conn.backup(backup_conn)
# Close connections
backup_conn.close()
source_conn.close()
# Get file size for reporting
file_size = backup_file.stat().st_size
file_size_mb = file_size / (1024 * 1024)
print(f"✓ Backed up {db_path.name} -> {backup_file.name} ({file_size_mb:.2f} MB)")
return str(backup_file)
except sqlite3.Error as e:
print(f"✗ Error backing up {db_path.name}: {e}")
# Clean up failed backup file
if backup_file.exists():
backup_file.unlink()
return None
except Exception as e:
print(f"✗ Unexpected error backing up {db_path.name}: {e}")
if backup_file.exists():
backup_file.unlink()
return None
def check_database_status(db_path):
"""
Check if a database file is actually used (has tables)
Args:
db_path: Path to the database file
Returns:
Tuple of (is_used, table_count, file_size_mb)
"""
db_path = Path(db_path)
if not db_path.exists():
return (False, 0, 0.0)
file_size_mb = db_path.stat().st_size / (1024 * 1024)
try:
conn = sqlite3.connect(str(db_path))
cursor = conn.cursor()
# Get list of tables
cursor.execute("SELECT name FROM sqlite_master WHERE type='table'")
tables = cursor.fetchall()
table_count = len(tables)
conn.close()
# Consider database used if it has tables or is larger than 1KB
is_used = table_count > 0 or file_size_mb > 0.001
return (is_used, table_count, file_size_mb)
except sqlite3.Error:
return (False, 0, file_size_mb)
def find_database_files(root_dir=".", check_usage=True):
"""
Find all SQLite database files in the root directory
Args:
root_dir: Root directory to search (default: current directory)
check_usage: If True, check which databases are actually used
Returns:
List of database file paths, optionally sorted by priority (main database first)
"""
root = Path(root_dir)
db_files = []
# Find all .db files in root directory (not in subdirectories)
for db_file in root.glob("*.db"):
# Skip WAL and SHM files (they're not standalone databases)
if db_file.suffix == ".db":
db_files.append(db_file)
# Sort by priority: meshcore_bot.db first (main database), then others
def sort_key(db_file):
name = db_file.name.lower()
if name == "meshcore_bot.db":
return (0, name)
elif name == "bot_data.db":
return (1, name)
else:
return (2, name)
return sorted(db_files, key=sort_key)
def main():
"""Main backup function"""
# Get script directory (project root)
script_dir = Path(__file__).parent.absolute()
os.chdir(script_dir)
print("MeshCore Bot Database Backup")
print("=" * 50)
print(f"Working directory: {script_dir}")
print()
# Find all database files
db_files = find_database_files(script_dir)
if not db_files:
print("No database files found in the project root.")
return 1
print(f"Found {len(db_files)} database file(s):")
# Check which databases are actually used
active_databases = []
inactive_databases = []
for db_file in db_files:
is_used, table_count, file_size = check_database_status(db_file)
status = "✓ ACTIVE" if is_used else "○ EMPTY/UNUSED"
print(f" {status}: {db_file.name} ({table_count} tables, {file_size:.2f} MB)")
if is_used:
active_databases.append(db_file)
else:
inactive_databases.append(db_file)
print()
if not active_databases:
print("No active databases found. Nothing to backup.")
return 0
if inactive_databases:
print(f"Note: {len(inactive_databases)} database(s) appear to be empty/unused and will be skipped.")
print()
# Create backups (only for active databases)
backup_dir = script_dir / "backups"
successful_backups = []
failed_backups = []
for db_file in active_databases:
result = backup_database(db_file, backup_dir)
if result:
successful_backups.append(result)
else:
failed_backups.append(db_file.name)
print()
print("=" * 50)
print(f"Backup Summary:")
print(f" Successful: {len(successful_backups)}")
print(f" Failed: {len(failed_backups)}")
if successful_backups:
print(f"\nBackups saved to: {backup_dir}")
print("\nRecent backups:")
for backup in successful_backups:
print(f" - {Path(backup).name}")
if failed_backups:
print(f"\nFailed to backup:")
for db in failed_backups:
print(f" - {db}")
return 1
return 0
if __name__ == "__main__":
sys.exit(main())