"""xcsift binary installer and manager.
This module handles automatic installation of the xcsift CLI tool
from GitHub releases if it's not already available on the system.
"""
import asyncio
import os
import platform
import shutil
import stat
import sys
import tarfile
import zipfile
from pathlib import Path
import httpx
# GitHub repository for xcsift releases
XCSIFT_REPO = "ldomaradzki/xcsift"
GITHUB_API_URL = f"https://api.github.com/repos/{XCSIFT_REPO}/releases/latest"
# Cache directory for downloaded binary
CACHE_DIR = Path.home() / ".local" / "share" / "xcsift-mcp" / "bin"
class XcsiftNotFoundError(Exception):
"""Raised when xcsift cannot be found or installed."""
pass
def get_xcsift_path() -> str | None:
"""Check if xcsift is available in PATH or cache.
Returns:
Path to xcsift binary if found, None otherwise.
"""
# Check PATH first
path = shutil.which("xcsift")
if path:
return path
# Check common Homebrew locations
common_paths = [
"/usr/local/bin/xcsift",
"/opt/homebrew/bin/xcsift",
str(CACHE_DIR / "xcsift"),
]
for p in common_paths:
if os.path.isfile(p) and os.access(p, os.X_OK):
return p
return None
def _get_platform_asset_name() -> str:
"""Get the expected asset name for the current platform.
Returns:
Asset filename pattern to look for in GitHub releases.
Raises:
XcsiftNotFoundError: If platform is not supported.
"""
system = platform.system().lower()
machine = platform.machine().lower()
if system != "darwin":
raise XcsiftNotFoundError(
f"xcsift is only available for macOS. Current platform: {system}"
)
# Map architecture names
if machine in ("arm64", "aarch64"):
arch = "arm64"
elif machine in ("x86_64", "amd64"):
arch = "x86_64"
else:
raise XcsiftNotFoundError(f"Unsupported architecture: {machine}")
return f"xcsift-{arch}-apple-darwin"
async def _download_file(url: str, dest: Path) -> None:
"""Download a file from URL to destination.
Args:
url: URL to download from.
dest: Destination path for the file.
"""
async with httpx.AsyncClient(follow_redirects=True) as client:
response = await client.get(url)
response.raise_for_status()
dest.write_bytes(response.content)
async def _get_latest_release_url() -> tuple[str, str]:
"""Get the download URL for the latest xcsift release.
Returns:
Tuple of (download_url, version).
Raises:
XcsiftNotFoundError: If no suitable release is found.
"""
asset_pattern = _get_platform_asset_name()
async with httpx.AsyncClient() as client:
response = await client.get(
GITHUB_API_URL,
headers={"Accept": "application/vnd.github.v3+json"},
)
if response.status_code == 404:
raise XcsiftNotFoundError(
f"No releases found for {XCSIFT_REPO}. "
"Please install xcsift manually: brew install xcsift"
)
response.raise_for_status()
release = response.json()
version = release.get("tag_name", "unknown")
assets = release.get("assets", [])
# Find matching asset
for asset in assets:
name = asset.get("name", "")
if asset_pattern in name:
return asset["browser_download_url"], version
raise XcsiftNotFoundError(
f"No compatible binary found for {asset_pattern} in release {version}. "
f"Available assets: {[a['name'] for a in assets]}. "
"Please install xcsift manually: brew install xcsift"
)
async def download_xcsift() -> str:
"""Download and install xcsift from GitHub releases.
Returns:
Path to the installed xcsift binary.
Raises:
XcsiftNotFoundError: If download or installation fails.
"""
try:
download_url, version = await _get_latest_release_url()
except httpx.HTTPError as e:
raise XcsiftNotFoundError(f"Failed to fetch release info: {e}") from e
# Create cache directory
CACHE_DIR.mkdir(parents=True, exist_ok=True)
# Download the archive
archive_name = download_url.split("/")[-1]
archive_path = CACHE_DIR / archive_name
try:
await _download_file(download_url, archive_path)
except httpx.HTTPError as e:
raise XcsiftNotFoundError(f"Failed to download xcsift: {e}") from e
# Extract binary
binary_path = CACHE_DIR / "xcsift"
try:
if archive_name.endswith(".tar.gz") or archive_name.endswith(".tgz"):
with tarfile.open(archive_path, "r:gz") as tar:
# Find the xcsift binary in the archive
for member in tar.getmembers():
if member.name.endswith("xcsift") and member.isfile():
# Extract to cache dir
member.name = "xcsift"
tar.extract(member, CACHE_DIR)
break
else:
raise XcsiftNotFoundError("xcsift binary not found in archive")
elif archive_name.endswith(".zip"):
with zipfile.ZipFile(archive_path, "r") as zf:
for name in zf.namelist():
if name.endswith("xcsift"):
with zf.open(name) as src, open(binary_path, "wb") as dst:
dst.write(src.read())
break
else:
raise XcsiftNotFoundError("xcsift binary not found in archive")
else:
# Assume it's the raw binary
shutil.move(str(archive_path), str(binary_path))
finally:
# Clean up archive
if archive_path.exists():
archive_path.unlink()
# Make executable
binary_path.chmod(binary_path.stat().st_mode | stat.S_IXUSR | stat.S_IXGRP | stat.S_IXOTH)
return str(binary_path)
async def ensure_xcsift() -> str:
"""Ensure xcsift is available, installing if needed.
Returns:
Path to the xcsift binary.
Raises:
XcsiftNotFoundError: If xcsift cannot be found or installed.
"""
# Check if already available
path = get_xcsift_path()
if path:
return path
# Try to download
return await download_xcsift()
def ensure_xcsift_sync() -> str:
"""Synchronous wrapper for ensure_xcsift().
Returns:
Path to the xcsift binary.
Raises:
XcsiftNotFoundError: If xcsift cannot be found or installed.
"""
return asyncio.run(ensure_xcsift())
async def get_xcsift_version(xcsift_path: str | None = None) -> str:
"""Get the version of xcsift.
Args:
xcsift_path: Path to xcsift binary. If None, will be auto-detected.
Returns:
Version string.
"""
if xcsift_path is None:
xcsift_path = await ensure_xcsift()
proc = await asyncio.create_subprocess_exec(
xcsift_path,
"--version",
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
)
stdout, _ = await proc.communicate()
return stdout.decode().strip()
