Add CLI modules: validate, sync, merge, history
This commit is contained in:
283
confsync/cli/sync.py
Normal file
283
confsync/cli/sync.py
Normal file
@@ -0,0 +1,283 @@
|
||||
"""Sync command for cross-machine synchronization."""
|
||||
|
||||
from pathlib import Path
|
||||
from typing import Optional
|
||||
import typer
|
||||
from rich.console import Console
|
||||
from rich.panel import Panel
|
||||
from rich.table import Table
|
||||
|
||||
from confsync.core.manifest import ManifestBuilder
|
||||
from confsync.utils.encryption import EncryptionManager
|
||||
from confsync.utils.git_utils import GitManager
|
||||
|
||||
sync_cmd = typer.Typer(
|
||||
name="sync",
|
||||
help="Synchronize configurations between machines",
|
||||
no_args_is_help=True,
|
||||
)
|
||||
|
||||
console = Console()
|
||||
|
||||
|
||||
@sync_cmd.command("push")
|
||||
def sync_push(
|
||||
repo: str = typer.Option(
|
||||
..., "--repo", "-r",
|
||||
help="Remote repository URL"
|
||||
),
|
||||
branch: str = typer.Option(
|
||||
"main", "--branch", "-b",
|
||||
help="Branch to push to"
|
||||
),
|
||||
message: str = typer.Option(
|
||||
"Update configurations", "--message", "-m",
|
||||
help="Commit message"
|
||||
),
|
||||
encrypt: bool = typer.Option(
|
||||
False, "--encrypt", "-e",
|
||||
help="Encrypt configurations before pushing"
|
||||
),
|
||||
passphrase: Optional[str] = typer.Option(
|
||||
None, "--passphrase", "-p",
|
||||
help="Encryption passphrase"
|
||||
),
|
||||
manifest: str = typer.Option(
|
||||
"confsync_manifest.yaml",
|
||||
"--manifest", "-M",
|
||||
help="Path to manifest file"
|
||||
),
|
||||
):
|
||||
"""Push configurations to remote repository."""
|
||||
if not Path(manifest).exists():
|
||||
console.print(f"[red]Error:[/red] Manifest file not found: {manifest}")
|
||||
return
|
||||
|
||||
git = GitManager()
|
||||
builder = ManifestBuilder()
|
||||
manifest_obj = builder.load_manifest(manifest)
|
||||
|
||||
try:
|
||||
if encrypt:
|
||||
console.print("[bold]Encrypting configurations...[/bold]")
|
||||
enc_manager = EncryptionManager(passphrase=passphrase)
|
||||
temp_manifest_path = "/tmp/manifest_temp.yaml"
|
||||
builder.save_manifest(manifest_obj, temp_manifest_path)
|
||||
enc_manager.encrypt_manifest(temp_manifest_path)
|
||||
|
||||
console.print("[bold]Initializing local repository...[/bold]")
|
||||
if not git.init_repo("/tmp/confsync_local"):
|
||||
console.print("[red]Error: Failed to initialize repository[/red]")
|
||||
return
|
||||
|
||||
git.add_files([manifest])
|
||||
|
||||
console.print("[bold]Committing changes...[/bold]")
|
||||
if git.commit(message):
|
||||
console.print("[green]Committed to local repository[/green]")
|
||||
else:
|
||||
console.print("[yellow]No changes to commit[/yellow]")
|
||||
|
||||
if not git.add_remote("origin", repo):
|
||||
console.print("[yellow]Remote already exists or could not add[/yellow]")
|
||||
|
||||
console.print(f"[bold]Pushing to {repo}...[/bold]")
|
||||
if git.push("origin", branch):
|
||||
console.print(Panel.fit(
|
||||
"[bold green]Successfully pushed configurations![/bold green]",
|
||||
style="green",
|
||||
subtitle=f"Branch: {branch}",
|
||||
))
|
||||
else:
|
||||
console.print("[red]Error: Failed to push to remote[/red]")
|
||||
|
||||
except Exception as e:
|
||||
console.print(f"[red]Error during push:[/red] {e}")
|
||||
|
||||
|
||||
@sync_cmd.command("pull")
|
||||
def sync_pull(
|
||||
repo: str = typer.Option(
|
||||
..., "--repo", "-r",
|
||||
help="Remote repository URL"
|
||||
),
|
||||
branch: str = typer.Option(
|
||||
"main", "--branch", "-b",
|
||||
help="Branch to pull from"
|
||||
),
|
||||
output: str = typer.Option(
|
||||
"confsync_manifest.yaml",
|
||||
"--output", "-o",
|
||||
help="Output file for manifest"
|
||||
),
|
||||
decrypt: bool = typer.Option(
|
||||
False, "--decrypt", "-d",
|
||||
help="Decrypt configurations after pulling"
|
||||
),
|
||||
passphrase: Optional[str] = typer.Option(
|
||||
None, "--passphrase", "-p",
|
||||
help="Decryption passphrase"
|
||||
),
|
||||
):
|
||||
"""Pull configurations from remote repository."""
|
||||
git = GitManager()
|
||||
|
||||
try:
|
||||
console.print(f"[bold]Cloning from {repo}...[/bold]")
|
||||
if not git.clone_repo(repo, "/tmp/confsync_remote", branch):
|
||||
console.print("[red]Error: Failed to clone repository[/red]")
|
||||
return
|
||||
|
||||
manifest_path = "/tmp/confsync_remote/confsync_manifest.yaml"
|
||||
if not Path(manifest_path).exists():
|
||||
console.print("[red]Error: No manifest found in repository[/red]")
|
||||
return
|
||||
|
||||
if decrypt:
|
||||
console.print("[bold]Decrypting configurations...[/bold]")
|
||||
enc_manager = EncryptionManager(passphrase=passphrase)
|
||||
content = Path(manifest_path).read_text()
|
||||
decrypted = enc_manager.decrypt_manifest(content)
|
||||
Path(output).write_text(decrypted)
|
||||
console.print(f"[green]Decrypted manifest to {output}[/green]")
|
||||
else:
|
||||
import shutil
|
||||
shutil.copy(manifest_path, output)
|
||||
console.print(f"[green]Manifest saved to {output}[/green]")
|
||||
|
||||
console.print(Panel.fit(
|
||||
"[bold green]Successfully pulled configurations![/bold green]",
|
||||
style="green",
|
||||
subtitle=f"Branch: {branch}",
|
||||
))
|
||||
|
||||
except Exception as e:
|
||||
console.print(f"[red]Error during pull:[/red] {e}")
|
||||
|
||||
|
||||
@sync_cmd.command("status")
|
||||
def sync_status(
|
||||
repo: str = typer.Option(
|
||||
None, "--repo", "-r",
|
||||
help="Remote repository URL to check remote status"
|
||||
),
|
||||
manifest: str = typer.Option(
|
||||
"confsync_manifest.yaml",
|
||||
"--manifest", "-M",
|
||||
help="Path to manifest file"
|
||||
),
|
||||
):
|
||||
"""Check sync status."""
|
||||
git = GitManager()
|
||||
|
||||
local_manifest = Path(manifest)
|
||||
if local_manifest.exists():
|
||||
builder = ManifestBuilder()
|
||||
local_manifest_obj = builder.load_manifest(manifest)
|
||||
local_files = len(local_manifest_obj.entries)
|
||||
else:
|
||||
local_files = 0
|
||||
|
||||
console.print("[bold]Sync Status[/bold]")
|
||||
console.print("-" * 40)
|
||||
|
||||
table = Table()
|
||||
table.add_column("Status", style="cyan")
|
||||
table.add_column("Details", style="magenta")
|
||||
|
||||
table.add_row("Local Manifest", str(local_files) + " files" if local_files else "Not found")
|
||||
|
||||
if repo:
|
||||
git = GitManager()
|
||||
if git.clone_repo(repo, "/tmp/confsync_status_check"):
|
||||
remote_manifest_path = "/tmp/confsync_status_check/confsync_manifest.yaml"
|
||||
if Path(remote_manifest_path).exists():
|
||||
remote_manifest_obj = builder.load_manifest(remote_manifest_path)
|
||||
remote_files = len(remote_manifest_obj.entries)
|
||||
table.add_row("Remote Repository", str(remote_files) + " files")
|
||||
|
||||
if local_manifest_obj and local_files != remote_files:
|
||||
status = "Out of sync"
|
||||
style = "yellow"
|
||||
else:
|
||||
status = "In sync"
|
||||
style = "green"
|
||||
table.add_row("Sync Status", f"[{style}]{status}[/{style}]")
|
||||
else:
|
||||
table.add_row("Remote Repository", "No manifest found")
|
||||
else:
|
||||
table.add_row("Remote Repository", "Could not access")
|
||||
|
||||
console.print(table)
|
||||
|
||||
|
||||
@sync_cmd.command("encrypt")
|
||||
def sync_encrypt(
|
||||
path: str = typer.Argument(
|
||||
...,
|
||||
help="Path to file or manifest to encrypt"
|
||||
),
|
||||
output: Optional[str] = typer.Option(
|
||||
None, "--output", "-o",
|
||||
help="Output file path"
|
||||
),
|
||||
passphrase: Optional[str] = typer.Option(
|
||||
None, "--passphrase", "-p",
|
||||
help="Encryption passphrase"
|
||||
),
|
||||
):
|
||||
"""Encrypt a file or manifest."""
|
||||
if not Path(path).exists():
|
||||
console.print(f"[red]Error:[/red] File not found: {path}")
|
||||
return
|
||||
|
||||
enc_manager = EncryptionManager(passphrase=passphrase)
|
||||
|
||||
try:
|
||||
if path.endswith('.yaml') or path.endswith('.yml'):
|
||||
content = Path(path).read_text()
|
||||
encrypted = enc_manager.encrypt_manifest(content)
|
||||
output_path = output or path.replace('.yaml', '.enc.yaml')
|
||||
Path(output_path).write_text(encrypted)
|
||||
console.print(f"[green]Encrypted manifest to {output_path}[/green]")
|
||||
else:
|
||||
output_path = enc_manager.encrypt_file(path, output)
|
||||
console.print(f"[green]Encrypted file to {output_path}[/green]")
|
||||
except Exception as e:
|
||||
console.print(f"[red]Error encrypting:[/red] {e}")
|
||||
|
||||
|
||||
@sync_cmd.command("decrypt")
|
||||
def sync_decrypt(
|
||||
path: str = typer.Argument(
|
||||
...,
|
||||
help="Path to encrypted file"
|
||||
),
|
||||
output: Optional[str] = typer.Option(
|
||||
None, "--output", "-o",
|
||||
help="Output file path"
|
||||
),
|
||||
passphrase: Optional[str] = typer.Option(
|
||||
None, "--passphrase", "-p",
|
||||
help="Decryption passphrase"
|
||||
),
|
||||
):
|
||||
"""Decrypt a file or manifest."""
|
||||
if not Path(path).exists():
|
||||
console.print(f"[red]Error:[/red] File not found: {path}")
|
||||
return
|
||||
|
||||
enc_manager = EncryptionManager(passphrase=passphrase)
|
||||
|
||||
try:
|
||||
if path.endswith('.enc.yaml') or path.endswith('.enc.yml'):
|
||||
content = Path(path).read_text()
|
||||
decrypted = enc_manager.decrypt_manifest(content)
|
||||
output_path = output or path.replace('.enc.yaml', '.yaml').replace('.enc.yml', '.yml')
|
||||
Path(output_path).write_text(decrypted)
|
||||
console.print(f"[green]Decrypted manifest to {output_path}[/green]")
|
||||
else:
|
||||
output_path = enc_manager.decrypt_file(path, output)
|
||||
console.print(f"[green]Decrypted file to {output_path}[/green]")
|
||||
except Exception as e:
|
||||
console.print(f"[red]Error decrypting:[/red] {e}")
|
||||
Reference in New Issue
Block a user