Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
22 changes: 22 additions & 0 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -150,6 +150,28 @@ droidctx list-connectors
droidctx list-connectors --type GRAFANA
```

### `droidctx auto-sync`

Keep your context files fresh automatically. Uses launchd on macOS and cron on Linux.

```bash
# Enable auto-sync (runs every 30 minutes by default)
droidctx auto-sync enable --keyfile ./droidctx-context/credentials.yaml

# Custom interval (in minutes)
droidctx auto-sync enable --keyfile ./droidctx-context/credentials.yaml --interval 60

# Check status
droidctx auto-sync status

# Disable
droidctx auto-sync disable
```

Logs are written to `~/.config/droidctx/auto-sync.log`.

> **macOS note:** You may need to allow droidctx in System Settings → Privacy & Security the first time.

## Credentials Format

Create a YAML file with your connector credentials. Run `droidctx init` to generate a template with all supported types, or `droidctx detect` to auto-populate from your CLI tools.
Expand Down
81 changes: 81 additions & 0 deletions droidctx/auto_sync.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,81 @@
"""Auto-sync configuration and orchestration."""

import shutil
import sys
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Optional

import yaml

CONFIG_DIR = Path.home() / ".config" / "droidctx"
CONFIG_FILE = CONFIG_DIR / "auto-sync.yaml"
LOG_FILE = CONFIG_DIR / "auto-sync.log"


def load_config() -> dict[str, Any]:
"""Load auto-sync config from disk. Returns empty dict if missing."""
if not CONFIG_FILE.exists():
return {}
return yaml.safe_load(CONFIG_FILE.read_text()) or {}


def save_config(config: dict[str, Any]) -> None:
"""Write auto-sync config to disk, creating directory if needed."""
CONFIG_DIR.mkdir(parents=True, exist_ok=True)
CONFIG_FILE.write_text(yaml.dump(config, default_flow_style=False, sort_keys=False))


def resolve_paths(keyfile: Path, output_dir: Optional[Path]) -> tuple[Path, Path]:
"""Return (keyfile, output_dir) as absolute paths.

If output_dir is None, defaults to the keyfile's parent directory.
"""
keyfile = keyfile.resolve()
if output_dir is None:
output_dir = keyfile.parent.resolve()
else:
output_dir = output_dir.resolve()
return keyfile, output_dir


def find_droidctx_binary() -> Optional[str]:
"""Locate the droidctx executable on $PATH."""
return shutil.which("droidctx")


def get_last_run_time() -> Optional[str]:
"""Parse the log file and return the timestamp of the last run, or None."""
if not LOG_FILE.exists():
return None
try:
text = LOG_FILE.read_text().strip()
if not text:
return None
# Return the last non-empty line (most recent log entry)
for line in reversed(text.splitlines()):
line = line.strip()
if line:
return line
return None
except OSError:
return None


def build_config(
*,
keyfile: Path,
output_dir: Path,
interval_minutes: int,
droidctx_bin: str,
) -> dict[str, Any]:
"""Build a config dict ready to be saved."""
return {
"enabled": True,
"interval_minutes": interval_minutes,
"keyfile": str(keyfile),
"output_dir": str(output_dir),
"droidctx_bin": droidctx_bin,
"platform": sys.platform,
"created_at": datetime.now(timezone.utc).isoformat(),
}
101 changes: 101 additions & 0 deletions droidctx/main.py
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,12 @@
add_completion=False,
no_args_is_help=True,
)
auto_sync_app = typer.Typer(
name="auto-sync",
help="Manage automatic periodic syncing.",
no_args_is_help=True,
)
app.add_typer(auto_sync_app, name="auto-sync")
console = Console()


Expand Down Expand Up @@ -286,6 +292,101 @@ def list_connectors(
console.print(f"\n[dim]{len(CONNECTOR_CREDENTIALS)} connectors supported. Use --type <TYPE> for details.[/]\n")


@auto_sync_app.command()
def enable(
keyfile: Path = typer.Option(..., "--keyfile", "-k", help="Path to credentials YAML file"),
path: Optional[Path] = typer.Option(None, "--path", "-p", help="Output directory (default: same as keyfile dir)"),
interval: int = typer.Option(30, "--interval", "-i", help="Sync interval in minutes"),
):
"""Enable automatic periodic syncing."""
from droidctx.auto_sync import resolve_paths, find_droidctx_binary, save_config, build_config
from droidctx.scheduler import get_scheduler

keyfile_abs, output_dir = resolve_paths(keyfile, path)

if not keyfile_abs.exists():
console.print(f"[red]Keyfile not found: {keyfile_abs}[/]")
raise typer.Exit(1)

droidctx_bin = find_droidctx_binary()
if not droidctx_bin:
console.print("[red]Could not find 'droidctx' on PATH. Is it installed?[/]")
raise typer.Exit(1)

config = build_config(
keyfile=keyfile_abs,
output_dir=output_dir,
interval_minutes=interval,
droidctx_bin=droidctx_bin,
)

scheduler = get_scheduler()
# Re-enable: uninstall old job first
if scheduler.is_active():
scheduler.uninstall()
scheduler.install(config)
save_config(config)

console.print(f"[bold green]Auto-sync enabled[/] (every {interval} min)")
console.print(f" keyfile: {keyfile_abs}")
console.print(f" output_dir: {output_dir}")
console.print(f" binary: {droidctx_bin}")


@auto_sync_app.command()
def disable():
"""Disable automatic periodic syncing."""
from droidctx.auto_sync import load_config, save_config
from droidctx.scheduler import get_scheduler

config = load_config()
scheduler = get_scheduler()

if not config.get("enabled") and not scheduler.is_active():
console.print("[yellow]Auto-sync is not currently enabled.[/]")
raise typer.Exit(0)

if scheduler.is_active():
scheduler.uninstall()

config["enabled"] = False
save_config(config)
console.print("[bold green]Auto-sync disabled.[/]")


@auto_sync_app.command()
def status():
"""Show current auto-sync status."""
from droidctx.auto_sync import load_config, get_last_run_time
from droidctx.scheduler import get_scheduler

config = load_config()
if not config:
console.print("[yellow]Auto-sync has not been configured.[/]")
raise typer.Exit(0)

scheduler = get_scheduler()
active = scheduler.is_active()

table = Table(title="Auto-Sync Status")
table.add_column("Setting", style="bold")
table.add_column("Value")

table.add_row("Enabled", "[green]yes[/]" if config.get("enabled") and active else "[red]no[/]")
table.add_row("Interval", f"{config.get('interval_minutes', '?')} minutes")
table.add_row("Keyfile", str(config.get("keyfile", "?")))
table.add_row("Output dir", str(config.get("output_dir", "?")))
table.add_row("Binary", str(config.get("droidctx_bin", "?")))
table.add_row("Platform", str(config.get("platform", "?")))

last_run = get_last_run_time()
table.add_row("Last log entry", last_run or "[dim]none[/]")

console.print()
console.print(table)
console.print()


def _get_commented_reference(exclude: set[str] | None = None) -> str:
"""Return all connector templates as commented YAML for reference.

Expand Down
161 changes: 161 additions & 0 deletions droidctx/scheduler.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,161 @@
"""Platform-specific scheduler backends for auto-sync."""

import abc
import subprocess
import sys
import textwrap
from pathlib import Path
from typing import Any


class Scheduler(abc.ABC):
"""Abstract base for OS-level periodic job schedulers."""

@abc.abstractmethod
def install(self, config: dict[str, Any]) -> None:
"""Register the periodic sync job."""

@abc.abstractmethod
def uninstall(self) -> None:
"""Remove the periodic sync job."""

@abc.abstractmethod
def is_active(self) -> bool:
"""Return True if the job is currently registered."""


# ---------------------------------------------------------------------------
# macOS launchd
# ---------------------------------------------------------------------------

PLIST_PATH = Path.home() / "Library" / "LaunchAgents" / "io.drdroid.droidctx.auto-sync.plist"

_PLIST_TEMPLATE = textwrap.dedent("""\
<?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE plist PUBLIC "-//Apple//DTD PLIST 1.0//EN"
"http://www.apple.com/DTDs/PropertyList-1.0.dtd">
<plist version="1.0">
<dict>
<key>Label</key>
<string>io.drdroid.droidctx.auto-sync</string>
<key>ProgramArguments</key>
<array>
<string>{droidctx_bin}</string>
<string>sync</string>
<string>--keyfile</string>
<string>{keyfile}</string>
<string>--path</string>
<string>{output_dir}</string>
</array>
<key>StartInterval</key>
<integer>{interval_seconds}</integer>
<key>StandardOutPath</key>
<string>{log_file}</string>
<key>StandardErrorPath</key>
<string>{log_file}</string>
</dict>
</plist>
""")


class LaunchdScheduler(Scheduler):
"""macOS launchd backend."""

def install(self, config: dict[str, Any]) -> None:
from droidctx.auto_sync import LOG_FILE

plist_content = _PLIST_TEMPLATE.format(
droidctx_bin=config["droidctx_bin"],
keyfile=config["keyfile"],
output_dir=config["output_dir"],
interval_seconds=config["interval_minutes"] * 60,
log_file=str(LOG_FILE),
)
PLIST_PATH.parent.mkdir(parents=True, exist_ok=True)
PLIST_PATH.write_text(plist_content)
subprocess.run(
["launchctl", "load", str(PLIST_PATH)],
check=True,
capture_output=True,
)

def uninstall(self) -> None:
if PLIST_PATH.exists():
subprocess.run(
["launchctl", "unload", str(PLIST_PATH)],
check=True,
capture_output=True,
)
PLIST_PATH.unlink()

def is_active(self) -> bool:
return PLIST_PATH.exists()


# ---------------------------------------------------------------------------
# Linux cron
# ---------------------------------------------------------------------------

CRON_MARKER = "# droidctx-auto-sync"


class CronScheduler(Scheduler):
"""Linux crontab backend."""

def _read_crontab(self) -> str:
result = subprocess.run(
["crontab", "-l"],
capture_output=True,
text=True,
)
if result.returncode != 0:
return ""
return result.stdout

def _write_crontab(self, content: str) -> None:
subprocess.run(
["crontab", "-"],
input=content,
check=True,
text=True,
capture_output=True,
)

def install(self, config: dict[str, Any]) -> None:
from droidctx.auto_sync import LOG_FILE

# Remove old entry first
existing = self._read_crontab()
lines = [l for l in existing.splitlines() if CRON_MARKER not in l]

cmd = (
f"{config['droidctx_bin']} sync "
f"--keyfile {config['keyfile']} "
f"--path {config['output_dir']}"
)
cron_line = f"*/{config['interval_minutes']} * * * * {cmd} >> {LOG_FILE} 2>&1 {CRON_MARKER}"
lines.append(cron_line)

self._write_crontab("\n".join(lines) + "\n")

def uninstall(self) -> None:
existing = self._read_crontab()
lines = [l for l in existing.splitlines() if CRON_MARKER not in l]
self._write_crontab("\n".join(lines) + "\n" if lines else "")

def is_active(self) -> bool:
return CRON_MARKER in self._read_crontab()


# ---------------------------------------------------------------------------
# Factory
# ---------------------------------------------------------------------------

def get_scheduler() -> Scheduler:
"""Return the appropriate scheduler for the current platform."""
if sys.platform == "darwin":
return LaunchdScheduler()
elif sys.platform.startswith("linux"):
return CronScheduler()
else:
raise NotImplementedError(f"Auto-sync is not supported on {sys.platform}")
Loading