89 lines
3.1 KiB
Python
89 lines
3.1 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Backup Manager for Git Agent
|
|
Handles backup branch creation and management
|
|
"""
|
|
|
|
import os
|
|
import logging
|
|
from datetime import datetime, timezone
|
|
from typing import Dict, Any
|
|
|
|
class BackupManager:
|
|
"""Manages backup branch operations"""
|
|
|
|
def __init__(self, config: Dict[str, Any], logger: logging.Logger):
|
|
self.config = config
|
|
self.logger = logger
|
|
self.backup_config = config.get('backup', {})
|
|
self.prefix = self.backup_config.get('branch_prefix', 'backup-')
|
|
|
|
def create_backup_branch(self) -> str:
|
|
"""Create a new backup branch with timestamp"""
|
|
timestamp = datetime.now(timezone.utc)
|
|
branch_name = f"{self.prefix}{timestamp.strftime('%Y-%m-%d-%H')}"
|
|
|
|
# Get current directory from git utils
|
|
current_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
|
|
|
# Create backup branch
|
|
import subprocess
|
|
try:
|
|
# Create and checkout new branch
|
|
result = subprocess.run(
|
|
['git', 'checkout', '-b', branch_name],
|
|
cwd=current_dir,
|
|
capture_output=True,
|
|
text=True,
|
|
check=False
|
|
)
|
|
|
|
if result.returncode == 0:
|
|
self.logger.info(f"✅ Created backup branch: {branch_name}")
|
|
return branch_name
|
|
else:
|
|
# Branch might already exist, just checkout
|
|
result = subprocess.run(
|
|
['git', 'checkout', branch_name],
|
|
cwd=current_dir,
|
|
capture_output=True,
|
|
text=True,
|
|
check=False
|
|
)
|
|
|
|
if result.returncode == 0:
|
|
self.logger.info(f"✅ Using existing backup branch: {branch_name}")
|
|
return branch_name
|
|
else:
|
|
self.logger.error(f"❌ Failed to create/checkout backup branch: {result.stderr}")
|
|
return None
|
|
|
|
except Exception as e:
|
|
self.logger.error(f"❌ Exception creating backup branch: {e}")
|
|
return None
|
|
|
|
def get_backup_count(self) -> int:
|
|
"""Get current number of backup branches"""
|
|
current_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
|
try:
|
|
result = subprocess.run(
|
|
['git', 'branch', '-a'],
|
|
cwd=current_dir,
|
|
capture_output=True,
|
|
text=True,
|
|
check=False
|
|
)
|
|
|
|
if result.returncode == 0:
|
|
branches = result.stdout.strip().split('\n')
|
|
backup_branches = [
|
|
b.strip().replace('* ', '').replace('remotes/origin/', '')
|
|
for b in branches
|
|
if b.strip() and self.prefix in b
|
|
]
|
|
return len(backup_branches)
|
|
|
|
except Exception as e:
|
|
self.logger.error(f"❌ Error counting backup branches: {e}")
|
|
|
|
return 0 |