dotfiles/bin/bin/stowage

#!/usr/bin/env python3
"""
originally stowage, by Keith Gaughan <https://github.com/kgaughan/>
modified by Andrew Williams <https://github.com/nikdoof/>

A dotfile package manager

Copyright (c) Keith Gaughan, 2017.
Copyright (c) Andrew Williams, 2021.

Permission is hereby granted, free of charge, to any person obtaining a copy of
this software and associated documentation files (the "Software"), to deal in
the Software without restriction, including without limitation the rights to
use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies
of the Software, and to permit persons to whom the Software is furnished to do
so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.
"""

import argparse
import fnmatch
import logging
import os
import re
import shutil
import sys
from pathlib import Path
from typing import Callable, List

logger = logging.getLogger(__name__)


def add(args: argparse.Namespace) -> None:
    """Add a file to a package by moving it and creating a symlink."""
    target = Path(args.target).resolve()
    file_path = Path(args.file).resolve()
    package = Path(args.repository, args.packages[0]).resolve()

    if not file_path.is_relative_to(target):
        logger.error("'%s' not under '%s'", args.file, args.target)
        sys.exit(1)

    rest = file_path.relative_to(target)
    dest_path = package / rest
    dest_dir = dest_path.parent

    if dest_path.exists():
        logger.error("file already exists in package: %s", dest_path)
        sys.exit(1)

    if not dest_dir.exists():
        logger.info("DIR %s", dest_dir)
        if not args.dry_run:
            dest_dir.mkdir(parents=True, mode=0o755, exist_ok=True)

    logger.info("SWAP %s %s", dest_path, file_path)
    if not args.dry_run:
        shutil.move(str(file_path), str(dest_path))
        try:
            file_path.symlink_to(dest_path)
        except OSError as e:
            logger.error("failed to create symlink: %s", e)
            # Attempt to restore the file
            try:
                shutil.move(str(dest_path), str(file_path))
            except Exception as restore_error:
                logger.error("failed to restore file: %s", restore_error)
            sys.exit(1)


def install(args: argparse.Namespace, is_excluded: Callable[[str], bool]) -> None:
    """Install packages by creating symlinks from repository to target."""
    for package in args.packages:
        package_dir = Path(args.repository, package)
        if not package_dir.is_dir():
            logger.warning("no such package: %s; skipping", package)
            continue

        # Walk the package
        for root, _, files in os.walk(package_dir, followlinks=True):
            root_path = Path(root)
            files = [filename for filename in files if not is_excluded(filename)]
            if not files:
                continue

            rest = root_path.relative_to(package_dir)
            dest = Path(args.target) / rest

            # Create the directory path
            if rest != Path("."):
                # If a non-directory exists with the same name and clobber is enabled, get rid of it.
                if dest.exists() and not dest.is_dir() and args.clobber:
                    logger.info("UNLINK %s", dest)
                    if not args.dry_run:
                        dest.unlink()

                # Make directory
                if not dest.exists():
                    logger.info("DIR %s", dest)
                    if not args.dry_run:
                        dest.mkdir(parents=True, mode=0o755, exist_ok=True)

            # Process files
            for filename in files:
                src_path = (root_path / filename).resolve()
                dest_path = dest / filename

                # Skip if the file exists and we're not clobbering
                if dest_path.exists() and not args.clobber:
                    logger.info("SKIP %s", dest_path)
                    continue

                # Does the file already exist?
                if dest_path.is_file() or dest_path.is_symlink():
                    logger.info("UNLINK %s", dest_path)
                    if not args.dry_run:
                        dest_path.unlink()

                # Link the file
                logger.info("LINK %s %s", src_path, dest_path)
                if not args.dry_run:
                    try:
                        dest_path.symlink_to(src_path)
                    except OSError as e:
                        logger.error("failed to create symlink %s: %s", dest_path, e)


def uninstall(args: argparse.Namespace, is_excluded: Callable[[str], bool]) -> None:
    """Uninstall packages by removing symlinks."""
    dirs: List[Path] = []
    for package in args.packages:
        package_dir = Path(args.repository, package)
        if not package_dir.is_dir():
            logger.warning("no such package: %s; skipping", package)
            continue

        for root, _, files in os.walk(package_dir, followlinks=True):
            root_path = Path(root)
            files = [filename for filename in files if not is_excluded(filename)]
            if not files:
                continue

            rest = root_path.relative_to(package_dir)
            dest = Path(args.target) / rest

            if rest != Path("."):
                dirs.append(dest)

            for filename in files:
                dest_path = dest / filename
                if not dest_path.exists():
                    logger.debug("does not exist: %s", dest_path)
                    continue

                if dest_path.is_symlink():
                    src_path = (root_path / filename).resolve()
                    try:
                        if dest_path.resolve() == src_path:
                            logger.info("UNLINK %s", dest_path)
                            if not args.dry_run:
                                dest_path.unlink()
                        else:
                            logger.info("SKIP %s (points elsewhere)", dest_path)
                    except (OSError, RuntimeError) as e:
                        logger.warning("error checking symlink %s: %s", dest_path, e)
                else:
                    logger.info("SKIP %s (not a symlink)", dest_path)

    # Delete the directories if empty.
    for dir_path in sorted(dirs, key=lambda p: len(str(p)), reverse=True):
        if not dir_path.exists():
            continue
        try:
            logger.info("RMDIR %s", dir_path)
            if not args.dry_run:
                dir_path.rmdir()
        except OSError:
            logger.debug("directory not empty: %s", dir_path)


def make_argparser() -> argparse.ArgumentParser:
    """Create and configure the argument parser."""
    parser = argparse.ArgumentParser(description="A dotfile package manager.")
    parser.add_argument("--verbose", "-v", action="store_true", help="Verbose output")
    parser.add_argument("--dry-run", "-n", action="store_true", help="Dry run.")
    parser.add_argument(
        "--target",
        "-t",
        default=str(Path.home()),
        help="Target directory in which to place symlinks",
    )
    parser.add_argument(
        "--repository",
        "-r",
        default=str(Path.home() / ".dotfiles"),
        help="The location of the dotfile repository",
    )
    parser.add_argument(
        "--exclude",
        "-x",
        action="append",
        default=[],
        metavar="GLOB",
        help="Glob pattern of files to exclude",
    )
    parser.add_argument(
        "--clobber",
        action="store_true",
        help="Replace files even if they exist.",
    )

    subparsers = parser.add_subparsers(dest="command", help="sub-command help")

    # List
    subparsers.add_parser("list", help="List packages in the repository")

    # Add
    parser_add = subparsers.add_parser("add", help="Add a file to a package")
    parser_add.add_argument("file", metavar="FILE", help="File to stow")
    parser_add.add_argument(
        "packages", metavar="PACKAGE", nargs="+", help="Packages to install"
    )

    # Uninstall
    parser_uninstall = subparsers.add_parser("uninstall", help="Remove a package")
    parser_uninstall.add_argument(
        "packages", metavar="PACKAGE", nargs="+", help="Packages to uninstall"
    )

    # Install
    parser_install = subparsers.add_parser("install", help="Install packages")
    parser_install.add_argument(
        "packages", metavar="PACKAGE", nargs="+", help="Packages to install"
    )

    return parser


def main() -> None:
    """Main entry point for the stowage script."""
    parser = make_argparser()
    args = parser.parse_args()

    # Configure logging
    log_level = logging.INFO if args.verbose or args.dry_run else logging.WARNING
    logging.basicConfig(
        level=log_level,
        format='%(message)s',
        stream=sys.stdout
    )

    # Validate repository exists
    repo_path = Path(args.repository)
    if not repo_path.exists():
        logger.error("repository does not exist: %s", args.repository)
        sys.exit(1)

    # Validate target exists
    target_path = Path(args.target)
    if not target_path.exists():
        logger.error("target directory does not exist: %s", args.target)
        sys.exit(1)

    exclude = [re.compile(fnmatch.translate(pattern)) for pattern in args.exclude]

    def is_excluded(filename: str) -> bool:
        """Check if a filename matches any exclusion pattern."""
        return any(pattern.match(filename) for pattern in exclude)

    if args.command == "list":
        if not repo_path.is_dir():
            logger.error("repository is not a directory: %s", args.repository)
            sys.exit(1)
        packages = sorted([
            item.name for item in repo_path.iterdir()
            if item.is_dir() and not item.name.startswith(".")
        ])
        if packages:
            for package in packages:
                print(package)
        else:
            logger.info("no packages found in repository")
    elif args.command == "add":
        if len(args.packages) > 1:
            parser.error("add only works with a single package")
        file_path = Path(args.file)
        # Handle both absolute and relative paths
        if not file_path.is_absolute():
            file_path = Path(args.target) / file_path
        if not file_path.is_file():
            parser.error(f"no such file: {args.file}")
        args.file = str(file_path)
        add(args)
    elif args.command == "install":
        install(args, is_excluded)
    elif args.command == "uninstall":
        uninstall(args, is_excluded)
    else:
        parser.print_help()
        sys.exit(1)


if __name__ == "__main__":
    main()