actually fixed auto backups now

This commit is contained in:
Lilac-Rose
2025-11-16 16:09:54 +01:00
parent 08700104b0
commit 35b2ee7760

View File

@@ -21,6 +21,7 @@ class BackupXP(commands.Cog):
self.db_dir = os.path.join(self.base_dir, "databases")
self.backup_dir = os.path.join(self.base_dir, "backups")
self.last_backup_file = os.path.join(self.backup_dir, "last_backup.txt")
self.last_auto_backup_file = os.path.join(self.backup_dir, "last_auto_backup.txt")
os.makedirs(self.backup_dir, exist_ok=True)
# Start the daily check task
self.auto_backup_task.start()
@@ -44,73 +45,109 @@ class BackupXP(commands.Cog):
success, message = await self.create_backup()
await interaction.followup.send(message)
@tasks.loop(hours=1)
@tasks.loop(minutes=15)
async def auto_backup_task(self):
"""Runs hourly and checks if it's 10 AM EST and a weekly backup is due"""
"""Runs every 15 minutes and checks if it's 10 AM EST and a daily backup is due"""
# Get current time in EST
now_est = datetime.now(EST)
# Check if it's 10 AM EST (hour 10, minute 0-59)
if now_est.hour == BACKUP_HOUR:
print(f"[Backup] Time check: {now_est.strftime('%Y-%m-%d %I:%M %p %Z')} (Hour: {now_est.hour}, Minute: {now_est.minute})")
# Check if it's between 10:00 AM and 10:15 AM EST
if now_est.hour == BACKUP_HOUR and now_est.minute < 15:
print(f"[Backup] Inside backup window - Checking for backup")
await self.check_last_backup()
else:
print(f"[Backup] Outside backup window (need hour={BACKUP_HOUR} and minute<15)")
@auto_backup_task.before_loop
async def before_auto_backup(self):
"""Wait until the bot is ready before starting the loop."""
await self.bot.wait_until_ready()
print(f"[Backup] Auto backup task started. Will check every 15 minutes for 10 AM EST backup window.")
now_est = datetime.now(EST)
print(f"[Backup] Current time: {now_est.strftime('%Y-%m-%d %I:%M %p %Z')}")
async def check_last_backup(self):
"""Check if a backup has been done today at 10 AM EST"""
"""Check if an auto backup has been done today at 10 AM EST"""
now = datetime.now()
now_est = datetime.now(EST)
if not os.path.exists(self.last_backup_file):
# No previous backup — make initial backup
await self.create_backup(log_channel=True, reason="Startup initial backup")
print(f"[Backup] Checking auto backup status at {now_est.strftime('%Y-%m-%d %I:%M %p %Z')}")
if not os.path.exists(self.last_auto_backup_file):
# No previous auto backup — make initial backup
print("[Backup] No last_auto_backup.txt found, creating initial auto backup")
await self.create_backup(log_channel=True, reason="Auto daily backup (10 AM EST)", is_auto=True)
await self.cleanup_old_backups()
return
with open(self.last_backup_file, "r") as f:
with open(self.last_auto_backup_file, "r") as f:
try:
last_time = datetime.fromisoformat(f.read().strip())
except Exception:
last_time = datetime.min
# Check if we've already backed up today
# Check if we've already done an auto backup today
last_time_est = last_time.astimezone(EST)
print(f"[Backup] Last auto backup: {last_time_est.strftime('%Y-%m-%d %I:%M %p %Z')}")
print(f"[Backup] Last auto backup date: {last_time_est.date()}, Today: {now_est.date()}")
if last_time_est.date() != now_est.date():
# Haven't backed up today yet, so do it now
await self.create_backup(log_channel=True, reason="Auto daily backup (10 AM EST)")
# Haven't done auto backup today yet, so do it now
print("[Backup] Starting daily auto backup...")
await self.create_backup(log_channel=True, reason="Auto daily backup (10 AM EST)", is_auto=True)
await self.cleanup_old_backups()
else:
print("[Backup] Already auto-backed up today, skipping")
async def create_backup(self, log_channel=False, reason=None):
async def create_backup(self, log_channel=False, reason=None, is_auto=False):
"""Handles the actual backup logic"""
lifetime_db = os.path.join(self.db_dir, "lifetime.db")
annual_db = os.path.join(self.db_dir, "annual.db")
print(f"[Backup] Looking for databases:")
print(f"[Backup] Lifetime: {lifetime_db}")
print(f"[Backup] Annual: {annual_db}")
print(f"[Backup] Lifetime exists: {os.path.exists(lifetime_db)}")
print(f"[Backup] Annual exists: {os.path.exists(annual_db)}")
missing = [db for db in [lifetime_db, annual_db] if not os.path.exists(db)]
if missing:
return False, f"❌ Missing database files: {', '.join(os.path.basename(m) for m in missing)}"
error_msg = f"❌ Missing database files: {', '.join(os.path.basename(m) for m in missing)}"
print(f"[Backup] {error_msg}")
return False, error_msg
timestamp = datetime.now().strftime("%Y-%m-%d_%H-%M-%S")
lifetime_backup = os.path.join(self.backup_dir, f"lifetime_{timestamp}.db")
annual_backup = os.path.join(self.backup_dir, f"annual_{timestamp}.db")
prefix = "auto_" if is_auto else ""
lifetime_backup = os.path.join(self.backup_dir, f"{prefix}lifetime_{timestamp}.db")
annual_backup = os.path.join(self.backup_dir, f"{prefix}annual_{timestamp}.db")
try:
shutil.copy2(lifetime_db, lifetime_backup)
shutil.copy2(annual_db, annual_backup)
print(f"[Backup] Files copied successfully")
print(f"[Backup] {os.path.basename(lifetime_backup)}")
print(f"[Backup] {os.path.basename(annual_backup)}")
# Record last backup time
with open(self.last_backup_file, "w") as f:
f.write(datetime.now().isoformat())
# Record last auto backup time if this was automatic
if is_auto:
with open(self.last_auto_backup_file, "w") as f:
f.write(datetime.now().isoformat())
# File sizes
lifetime_size = os.path.getsize(lifetime_db) / (1024 * 1024)
annual_size = os.path.getsize(annual_db) / (1024 * 1024)
total_size = lifetime_size + annual_size
backup_type = "Auto Backup" if is_auto else "Manual Backup"
msg = (
f"✅ Databases backed up successfully! ({reason or 'Manual backup'})\n"
f" **{backup_type}** - Databases backed up successfully!\n"
f"**Lifetime:** `{os.path.basename(lifetime_backup)}` ({lifetime_size:.2f} MB)\n"
f"**Annual:** `{os.path.basename(annual_backup)}` ({annual_size:.2f} MB)\n"
f"**Total size:** {total_size:.2f} MB"
@@ -120,12 +157,18 @@ class BackupXP(commands.Cog):
# Send to notification channel only
notification_channel = self.bot.get_channel(NOTIFICATION_CHANNEL_ID)
if notification_channel:
print(f"[Backup] Sending notification to channel {NOTIFICATION_CHANNEL_ID}")
await notification_channel.send(msg)
else:
print(f"[Backup] Could not find notification channel {NOTIFICATION_CHANNEL_ID}")
print(f"[Backup] Backup completed successfully")
return True, msg
except Exception as e:
return False, f"❌ Backup failed: `{e}`"
error_msg = f"❌ Backup failed: `{e}`"
print(f"[Backup] {error_msg}")
return False, error_msg
async def cleanup_old_backups(self):
"""Delete backup files older than MAX_BACKUP_AGE (30 days)"""