Add CLI modules: validate, sync, merge, history
This commit is contained in:
208
confsync/cli/merge.py
Normal file
208
confsync/cli/merge.py
Normal file
@@ -0,0 +1,208 @@
|
||||
"""Merge command for intelligent configuration merging."""
|
||||
|
||||
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.merger import Merger
|
||||
from confsync.core.manifest import ManifestBuilder
|
||||
|
||||
merge_cmd = typer.Typer(
|
||||
name="merge",
|
||||
help="Merge configuration files",
|
||||
no_args_is_help=True,
|
||||
)
|
||||
|
||||
console = Console()
|
||||
|
||||
|
||||
@merge_cmd.command("file")
|
||||
def merge_file(
|
||||
local: str = typer.Argument(..., help="Path to local configuration file"),
|
||||
remote: str = typer.Argument(..., help="Path to remote configuration file"),
|
||||
base: Optional[str] = typer.Option(
|
||||
None, "--base", "-b",
|
||||
help="Path to base/common configuration file"
|
||||
),
|
||||
strategy: str = typer.Option(
|
||||
"three_way", "--strategy", "-s",
|
||||
help="Merge strategy (keep_local, keep_remote, keep_common, three_way, union)"
|
||||
),
|
||||
output: str = typer.Option(
|
||||
"merged_config.yaml", "--output", "-o",
|
||||
help="Output file for merged configuration"
|
||||
),
|
||||
):
|
||||
"""Merge two configuration files."""
|
||||
local_path = Path(local)
|
||||
remote_path = Path(remote)
|
||||
base_path = Path(base) if base else None
|
||||
|
||||
if not local_path.exists():
|
||||
console.print(f"[red]Error:[/red] Local file not found: {local}")
|
||||
return
|
||||
|
||||
if not remote_path.exists():
|
||||
console.print(f"[red]Error:[/red] Remote file not found: {remote}")
|
||||
return
|
||||
|
||||
local_content = local_path.read_text()
|
||||
remote_content = remote_path.read_text()
|
||||
base_content = base_path.read_text() if base_path and base_path.exists() else ""
|
||||
|
||||
merger = Merger()
|
||||
success, merged, conflict = merger.merge_file(
|
||||
local_content, remote_content, base_content, local, strategy
|
||||
)
|
||||
|
||||
if success:
|
||||
Path(output).write_text(merged)
|
||||
console.print(Panel.fit(
|
||||
"[bold green]Merge Successful![/bold green]",
|
||||
style="green",
|
||||
subtitle=f"Output: {output}",
|
||||
))
|
||||
else:
|
||||
console.print(Panel.fit(
|
||||
"[bold yellow]Merge Conflicts Detected![/bold yellow]",
|
||||
style="yellow",
|
||||
))
|
||||
console.print("\n[bold]Conflicts:[/bold]")
|
||||
if conflict:
|
||||
console.print(conflict.format_conflict())
|
||||
|
||||
conflict_output = output + ".conflict"
|
||||
Path(conflict_output).write_text(merged)
|
||||
console.print(f"\n[yellow]Merged with conflicts written to {conflict_output}[/yellow]")
|
||||
|
||||
|
||||
@merge_cmd.command("manifest")
|
||||
def merge_manifest(
|
||||
local_manifest: str = typer.Argument(
|
||||
...,
|
||||
help="Path to local manifest"
|
||||
),
|
||||
remote_manifest: str = typer.Argument(
|
||||
...,
|
||||
help="Path to remote manifest"
|
||||
),
|
||||
base_manifest: Optional[str] = typer.Option(
|
||||
None, "--base", "-b",
|
||||
help="Path to base/common manifest"
|
||||
),
|
||||
strategy: str = typer.Option(
|
||||
"local_wins", "--strategy", "-s",
|
||||
help="Merge strategy (local_wins, remote_wins)"
|
||||
),
|
||||
output: str = typer.Option(
|
||||
"merged_manifest.yaml", "--output", "-o",
|
||||
help="Output file for merged manifest"
|
||||
),
|
||||
):
|
||||
"""Merge two manifests."""
|
||||
local_path = Path(local_manifest)
|
||||
remote_path = Path(remote_manifest)
|
||||
|
||||
if not local_path.exists():
|
||||
console.print(f"[red]Error:[/red] Local manifest not found: {local_manifest}")
|
||||
return
|
||||
|
||||
if not remote_path.exists():
|
||||
console.print(f"[red]Error:[/red] Remote manifest not found: {remote_manifest}")
|
||||
return
|
||||
|
||||
builder = ManifestBuilder()
|
||||
local_manifest_obj = builder.load_manifest(local_manifest)
|
||||
remote_manifest_obj = builder.load_manifest(remote_manifest)
|
||||
|
||||
base_manifest_obj = None
|
||||
if base_manifest:
|
||||
base_path = Path(base_manifest)
|
||||
if base_path.exists():
|
||||
base_manifest_obj = builder.load_manifest(base_manifest)
|
||||
|
||||
merged = builder.merge_manifests(
|
||||
base_manifest_obj or ManifestBuilder().build_from_detected([]),
|
||||
remote_manifest_obj,
|
||||
local_manifest_obj,
|
||||
strategy
|
||||
)
|
||||
|
||||
builder.save_manifest(merged, output)
|
||||
|
||||
console.print(Panel.fit(
|
||||
"[bold green]Manifest Merge Complete![/bold green]",
|
||||
style="green",
|
||||
subtitle=f"Output: {output}",
|
||||
))
|
||||
console.print(f"Total entries: {len(merged.entries)}")
|
||||
|
||||
|
||||
@merge_cmd.command("preview")
|
||||
def merge_preview(
|
||||
local: str = typer.Argument(..., help="Path to local configuration file"),
|
||||
remote: str = typer.Argument(..., help="Path to remote configuration file"),
|
||||
base: Optional[str] = typer.Option(
|
||||
None, "--base", "-b",
|
||||
help="Path to base/common configuration file"
|
||||
),
|
||||
strategy: str = typer.Option(
|
||||
"three_way", "--strategy", "-s",
|
||||
help="Merge strategy to preview"
|
||||
),
|
||||
):
|
||||
"""Preview merge result without writing files."""
|
||||
local_path = Path(local)
|
||||
remote_path = Path(remote)
|
||||
base_path = Path(base) if base else None
|
||||
|
||||
local_content = local_path.read_text() if local_path.exists() else ""
|
||||
remote_content = remote_path.read_text() if remote_path.exists() else ""
|
||||
base_content = base_path.read_text() if base_path and base_path.exists() else ""
|
||||
|
||||
merger = Merger()
|
||||
success, merged, conflict = merger.merge_file(
|
||||
local_content, remote_content, base_content, local, strategy
|
||||
)
|
||||
|
||||
console.print("[bold]Merge Preview[/bold]")
|
||||
console.print("-" * 40)
|
||||
|
||||
if success:
|
||||
console.print("[green]No conflicts would occur[/green]")
|
||||
else:
|
||||
console.print("[yellow]Conflicts would occur:[/yellow]")
|
||||
if conflict:
|
||||
console.print(conflict.format_conflict())
|
||||
|
||||
console.print("\n[bold]Merged Content Preview:[/bold]")
|
||||
lines = merged.split('\n')
|
||||
for i, line in enumerate(lines[:50], 1):
|
||||
console.print(f"{i:4d} | {line}")
|
||||
if len(lines) > 50:
|
||||
console.print(f"... ({len(lines) - 50} more lines)")
|
||||
|
||||
|
||||
@merge_cmd.command("strategies")
|
||||
def list_strategies():
|
||||
"""List available merge strategies."""
|
||||
table = Table(title="Available Merge Strategies")
|
||||
table.add_column("Strategy", style="cyan")
|
||||
table.add_column("Description", style="white")
|
||||
table.add_column("Use Case", style="magenta")
|
||||
|
||||
strategies = [
|
||||
("keep_local", "Keep local changes, ignore remote", "Preserve local overrides"),
|
||||
("keep_remote", "Keep remote changes, ignore local", "Overwrite with remote"),
|
||||
("keep_common", "Keep only common lines", "Find shared configuration"),
|
||||
("three_way", "Standard three-way merge", "When both sides have changes"),
|
||||
("union", "Combine both, remove duplicates", "Union of all settings"),
|
||||
]
|
||||
|
||||
for strategy, description, use_case in strategies:
|
||||
table.add_row(strategy, description, use_case)
|
||||
|
||||
console.print(table)
|
||||
Reference in New Issue
Block a user