diff options
Diffstat (limited to 'dots_manager')
| -rw-r--r-- | dots_manager/args.py | 43 | ||||
| -rw-r--r-- | dots_manager/cli.py | 30 | ||||
| -rw-r--r-- | dots_manager/config.py | 145 | ||||
| -rw-r--r-- | dots_manager/env.py | 56 | ||||
| -rw-r--r-- | dots_manager/kawaii_logger.py | 53 | ||||
| -rw-r--r-- | dots_manager/parallel.py | 21 | ||||
| -rw-r--r-- | dots_manager/shell.py | 3 | ||||
| -rw-r--r-- | dots_manager/stow.py | 12 | ||||
| -rw-r--r-- | dots_manager/template.py | 18 | ||||
| -rw-r--r-- | dots_manager/utils.py | 21 |
10 files changed, 226 insertions, 176 deletions
diff --git a/dots_manager/args.py b/dots_manager/args.py deleted file mode 100644 index 40fc42f..0000000 --- a/dots_manager/args.py +++ /dev/null @@ -1,43 +0,0 @@ -import argparse -from pathlib import Path -from dots_manager.config import Config - - -def parse_arguments() -> argparse.Namespace: - parser = argparse.ArgumentParser(description="cute dotfiles manager ✧˖°") - - parser.add_argument( - "--compile", action="store_true", help="action: compile dotfiles" - ) - parser.add_argument( - "--source", - type=Path, - default=Config.default_source_dir, - help=f"default: '{Config.default_source_dir}'", - ) - parser.add_argument( - "--comp", - type=Path, - default=Config.default_compiled_dir, - help=f"default: '{Config.default_compiled_dir}'", - ) - parser.add_argument( - "--stow", action="store_true", help="action: stow compiled dotfiles" - ) - parser.add_argument( - "--clean", action="store_true", help="action: clean stowed dotfiles" - ) - parser.add_argument( - "--target", - type=Path, - default=Config.default_target_dir, - help=f"default: '{Config.default_target_dir}'", - ) - parser.add_argument( - "--verbose", - "-v", - action="store_true", - default=False, - help="enable verbose logging", - ) - return parser.parse_args() diff --git a/dots_manager/cli.py b/dots_manager/cli.py index f152a96..6621954 100644 --- a/dots_manager/cli.py +++ b/dots_manager/cli.py @@ -1,16 +1,28 @@ -from dots_manager.args import parse_arguments -from dots_manager.env import initialize_environment +import sys +import shutil +from dots_manager.config import Environment, parse_arguments from dots_manager.template import compile_dotfiles from dots_manager.stow import apply_stow_operation_to_packages def main(): args = parse_arguments() - env = initialize_environment(args) + env = Environment.from_argv(args) - if args.clean: - apply_stow_operation_to_packages(args.comp, args.target, "-D", env) - if args.compile: - compile_dotfiles(args.source, args.comp, env) - if args.stow: - apply_stow_operation_to_packages(args.comp, args.target, "--no-folding", env) + if args.clean and not ( + apply_stow_operation_to_packages(args.output, args.target, "--delete", env) + and (not args.output.exists() or shutil.rmtree(args.output) is None) + ): + env.logger.error("could not clean up stowed dotfiles <_mood.sad>") + sys.exit(1) + if args.compile and not compile_dotfiles(args.source, args.output, env): + env.logger.error("could not compile dotfiles <_mood.sad>") + sys.exit(1) + if args.stow and not apply_stow_operation_to_packages( + args.output, args.target, "--no-folding", env + ): + env.logger.error("could not stow dotfile packages <_mood.sad>") + sys.exit(1) + + env.logger.info("done! <_mood.happy>") + sys.exit(0) diff --git a/dots_manager/config.py b/dots_manager/config.py index d6194f6..96d3293 100644 --- a/dots_manager/config.py +++ b/dots_manager/config.py @@ -1,14 +1,147 @@ +import argparse +import os +import json +import logging from dataclasses import dataclass +from typing import Dict, Any, Optional from pathlib import Path -import os +from dots_manager.shell import run_shell_command +from dots_manager.utils import merge_dicts +from dots_manager.kawaii_logger import setup_logger + + +@dataclass(frozen=True) +class Environment: + platform: str + system_name: str + context: Dict[str, Any] + logger: logging.Logger + + @staticmethod + def from_argv(args: argparse.Namespace): + logger = setup_logger(verbose=args.verbose, logger_name="dots_manager") + + platform = args.platform or run_shell_command( + [str(args.helper_scripts / "platform.sh")], logger + ) + if not platform: + raise ValueError("failed to determine platform... ") + os.environ["PLATFORM"] = platform + + system_name: Optional[str] = args.system_name or run_shell_command( + [str(args.helper_scripts / "system_name.sh")], logger + ) + if not system_name: + raise ValueError("failed to determine system name... ") + + context = load_context(platform, system_name, args.context, logger) + return Environment(platform, system_name, context, logger) @dataclass(frozen=True) -class Config: +class Constants: default_target_dir: Path = Path.home() - default_source_dir: Path = Path(__file__).parent.parent / "dots" - default_compiled_dir: Path = Path("./.compiled_dotfiles/") - script_dir: Path = Path("home/scripts") - contexts: Path = Path("contexts.json") + + dots_repo_dir: Path = Path.home() / Path("dotfiles") + default_source_dir: Path = dots_repo_dir / Path("dots") + default_compiled_dir: Path = dots_repo_dir / Path(".compiled_dotfiles") + + default_script_dir: Path = default_source_dir / Path("home/scripts") + default_context: Path = dots_repo_dir / Path("context.json") + template_extension: str = ".j2" max_workers: int = (os.cpu_count() or 1) * 2 + + global_context_key: str = "_global" + platform_default_context_key: str = "_default" + + +def load_context( + platform: str, system_name: str, context_file: Path, logger: logging.Logger +): + logger.info(f"reading context file: {context_file} ✧*:｡ﾟ✧") + + context = json.loads(context_file.read_text()) + + global_context = context.get(Constants.global_context_key, {}) + platform_defaults = context.get(platform, {}).get( + Constants.platform_default_context_key, {} + ) + defaults = merge_dicts(global_context, platform_defaults) + + system_config = context.get(platform, {}).get(system_name, {}) + if not system_config: + logger.warning( + f"could not find context for 'contexts.{platform}.{system_name}' in {context_file.absolute()}" + ) + return { + "platform": platform, + "system_name": system_name, + **merge_dicts(defaults, system_config), + } + + +def parse_arguments() -> argparse.Namespace: + parser = argparse.ArgumentParser( + description="૮ ․ ․ ྀིა emprespresso's dotfiles manager 🐧✧˖°" + ) + parser.add_argument( + "--verbose", + "-v", + action="store_true", + default=os.environ.get("DEBUG", "false").lower() in ("y", "yes", "true", "t"), + help="enables verbose logging.", + ) + + parser.add_argument("--compile", action="store_true", help="compile le dotfiles.") + parser.add_argument( + "--source", + type=Path, + default=Constants.default_source_dir, + help=f"where to look for templated dotfile packages. default: '{Constants.default_source_dir}'.", + ) + parser.add_argument( + "--output", + type=Path, + default=Constants.default_compiled_dir, + help=f"where to store compiled, stowable dotfile packages. default: '{Constants.default_compiled_dir}'.", + ) + parser.add_argument( + "--context", + type=Path, + default=Constants.default_context, + help=f"path to contexts, stored as json. default: '{Constants.default_context}'.", + ) + parser.add_argument( + "--helper-scripts", + type=Path, + default=Constants.default_script_dir, + help="where to find executable scripts to determine system info (device name, platform, etc.).", + ) + parser.add_argument( + "--system-name", + type=str, + default=None, + help="the system's name. when unspecified, inferred via the hostname, or on osx, computername.", + ) + parser.add_argument( + "--platform", + type=str, + default=None, + help="the system's os platform (i.e. osx, linux, bsd, windows, etc.). when unspecified, inferred via $OSTYPE.", + ) + + parser.add_argument( + "--stow", action="store_true", help="action: stow compiled dotfiles" + ) + parser.add_argument( + "--clean", action="store_true", help="action: clean stowed dotfiles" + ) + parser.add_argument( + "--target", + type=Path, + default=Constants.default_target_dir, + help=f"where the dotfile packages' symlinks will be stowed. default: '{Constants.default_target_dir}'", + ) + + return parser.parse_args() diff --git a/dots_manager/env.py b/dots_manager/env.py deleted file mode 100644 index 4d89c9e..0000000 --- a/dots_manager/env.py +++ /dev/null @@ -1,56 +0,0 @@ -import os -import json -import logging -from pathlib import Path -from dataclasses import dataclass -from typing import Dict, Any -from dots_manager.config import Config -from dots_manager.shell import run_shell_command -from dots_manager.utils import merge_dicts -from dots_manager.kawaii_logger import setup_logger - - -@dataclass(frozen=True) -class Environment: - platform: str - system_name: str - context: Dict[str, Any] - logger: logging.Logger - - -def initialize_environment(args) -> Environment: - logger = setup_logger(args.verbose) - scripts = args.source / Config.script_dir - platform = run_shell_command([str(scripts / "platform.sh")], logger) - if not platform: - raise ValueError("failed to determine platform... ") - os.environ["PLATFORM"] = platform - system_name = run_shell_command([str(scripts / "system_name.sh")], logger) - if not system_name: - raise ValueError("failed to determine system name... ") - context = load_context(platform, system_name, Config.contexts, logger) - return Environment(platform, system_name, context, logger) - - -def load_context(platform, system_name, context_file, logger): - try: - logger.info(f"reading context file: {context_file} ✧*:｡ﾟ✧") - with open(context_file) as f: - contexts = json.load(f) - - global_config = contexts.get("_global", {}) - platform_defaults = contexts.get(platform, {}).get("_default", {}) - defaults = merge_dicts(global_config, platform_defaults) - system_config = contexts.get(platform, {}).get(system_name, {}) - if not system_config: - logger.warning( - f"couldn’t find system-specific config for {platform}.{system_name} (ɐ•́ﾟ•̀ɐ)" - ) - return { - "platform": platform, - "system_name": system_name, - **merge_dicts(defaults, system_config), - } - except (FileNotFoundError, json.JSONDecodeError) as e: - logger.error(f"error loading context: {e} ⋆ฺ°☁｡⋆ฺ °★ °｡") - return {} diff --git a/dots_manager/kawaii_logger.py b/dots_manager/kawaii_logger.py index f82da23..f13ea14 100644 --- a/dots_manager/kawaii_logger.py +++ b/dots_manager/kawaii_logger.py @@ -2,20 +2,22 @@ import logging import sys import random -MOOD_EMOTICONS = { - "debug": ["(=^･^=)", "(=・ｪ・=)", "(=^-ω-^=)", "(=^‥^=)"], - "info": ["(＾• ω •＾)", "(=^･ω･^)y＝", "(≚ᄌ≚)", "(｡♥‿♥｡)"], - "warning": ["(；￣Д￣)", "(￢_￢;)", "(･ัω･ั)", "(・_・ヾ"], - "error": ["(╥﹏╥)", "(≧Д≦)", "(；′⌒`)", "(Ｔ▽Ｔ)"], +MOOD_KAOMOJI = { + "<_mood.happy>": ["(ﾉ◕ヮ◕)ﾉ*:･ﾟ✧", "(＾▽＾)", "( ˘⌣˘)♡(˘⌣˘ )"], + "<_mood.excited>": ["(๑˃ᴗ˂)ﻭ"], + "<_mood.sad>": ["(｡•́︿•̀｡)", "(╯︵╰,)", "(ಥ﹏ಥ)", "(︶︹︺)"], + "<_mood.anxious>": ["(ノдヽ)", "(◎_◎;)", "(・_・;)", "(ﾟДﾟ;)"], } -MOOD_SUFFIXES = { - "_happy": ["(ﾉ◕ヮ◕)ﾉ*:･ﾟ✧", "(๑˃ᴗ˂)ﻭ", "(＾▽＾)", "( ˘⌣˘)♡(˘⌣˘ )"], - "_sad": ["(｡•́︿•̀｡)", "(╯︵╰,)", "(ಥ﹏ಥ)", "(︶︹︺)"], - "_anxious": ["(ノдヽ)", "(◎_◎;)", "(・_・;)", "(ﾟДﾟ;)"], +LEVEL_KAOMOJI = { + logging.DEBUG: ["(=^･^=)", "(=・ｪ・=)", "(=^-ω-^=)", "(=^‥^=)"], + logging.INFO: ["(＾• ω •＾)", "(=^･ω･^)y＝", "(≚ᄌ≚)", "(｡♥‿♥｡)"], + logging.WARNING: ["(；￣Д￣)", "(￢_￢;)", "(･ัω･ั)", "(・_・ヾ"], + logging.ERROR: ["(╥﹏╥)", "(≧Д≦)", "(；′⌒`)", "(Ｔ▽Ｔ)"], + logging.CRITICAL: ["(╯°□°)╯︵ ┻━┻"], } -LEVEL_COLORS = { +LEVEL_ANSI_STYLES = { logging.DEBUG: "\033[95m", # light magenta logging.INFO: "\033[96m", # light cyan logging.WARNING: "\033[93m", # light yellow @@ -23,32 +25,33 @@ LEVEL_COLORS = { logging.CRITICAL: "\033[35m", # magenta } -RESET_COLOR = "\033[0m" +RESET_ANSI = "\033[0m" class KawaiiFormatter(logging.Formatter): - def format(self, record): - level = record.levelname.lower() - color = LEVEL_COLORS.get(record.levelno, "") - base_emotes = MOOD_EMOTICONS.get(level, []) - mood_emotes = [] - - emote_pool = mood_emotes if mood_emotes else base_emotes - emote = random.choice(emote_pool) if emote_pool else "(・ω・)" - + def format(self, record: logging.LogRecord): message = record.msg.strip() ts = self.formatTime(record, "%Y-%m-%d %H:%M:%S") lvl = record.levelname filename = record.filename lineno = str(record.lineno) - formatted = f"[{ts}] {color}[{lvl} {filename}:{lineno}]{RESET_COLOR} {color}{message}{RESET_COLOR} {emote}" - return f"{formatted}{RESET_COLOR}" + split_last_word = message.rsplit(" ", 1) + last_word = split_last_word[-1] if split_last_word else "" + emote_pool = LEVEL_KAOMOJI.get(record.levelno, []) + if last_word in MOOD_KAOMOJI: + message = split_last_word[0] if len(split_last_word) > 1 else "" + emote_pool = MOOD_KAOMOJI.get(last_word, []) + emote = random.choice(emote_pool) if emote_pool else "(・ω・)" + + color = LEVEL_ANSI_STYLES.get(record.levelno, "") + + formatted = f"[{ts}] {color}[{lvl} {filename}:{lineno}]{RESET_ANSI} {color}{message}{RESET_ANSI} {emote}" + return f"{formatted}{RESET_ANSI}" -def setup_logger(verbose: bool = False) -> logging.Logger: - """sets up a super cute logger with sparkles and cat faces ✨""" - logger = logging.getLogger("dotfiles") +def setup_logger(verbose: bool = False, logger_name: str = "") -> logging.Logger: + logger = logging.getLogger(logger_name) logger.setLevel(logging.DEBUG if verbose else logging.INFO) handler = logging.StreamHandler(sys.stdout) diff --git a/dots_manager/parallel.py b/dots_manager/parallel.py deleted file mode 100644 index 8c85660..0000000 --- a/dots_manager/parallel.py +++ /dev/null @@ -1,21 +0,0 @@ -from concurrent.futures import ThreadPoolExecutor, as_completed -from typing import Callable, List, Optional, TypeVar -from dots_manager.config import Config -from dots_manager.env import Environment - -T = TypeVar("T") -R = TypeVar("R") - - -def parallelize( - worker: Callable[[T], R], - items: List[T], - env: Environment, - executor: Optional[ThreadPoolExecutor] = None, -) -> List[R]: - if executor is None: - executor = ThreadPoolExecutor(max_workers=Config.max_workers) - with executor as exec: - futures = [exec.submit(worker, item) for item in items] - env.logger.info(f"submitted {len(futures)} tasks to executor ₰˜.༄") - return [f.result() for f in as_completed(futures)] diff --git a/dots_manager/shell.py b/dots_manager/shell.py index afd705a..09e84b2 100644 --- a/dots_manager/shell.py +++ b/dots_manager/shell.py @@ -9,6 +9,9 @@ def run_shell_command(cmd: List[str], logger: logging.Logger) -> Optional[str]: return subprocess.run( cmd, capture_output=True, text=True, check=True ).stdout.strip() + except FileNotFoundError as e: + logger.error(f"excecutable not found: {cmd[0]}, {e}") + return None except subprocess.CalledProcessError as e: logger.error(f"command failed: {cmd}, {e.stderr}") return None diff --git a/dots_manager/stow.py b/dots_manager/stow.py index b78e2fc..da4fbbb 100644 --- a/dots_manager/stow.py +++ b/dots_manager/stow.py @@ -1,9 +1,8 @@ from pathlib import Path from typing import Literal -from dots_manager.env import Environment +from dots_manager.config import Environment from dots_manager.shell import run_shell_command -from dots_manager.parallel import parallelize -from dots_manager.utils import is_some +from dots_manager.utils import is_some, parallelize def list_stowable_packages(packages: Path) -> list[Path]: @@ -18,9 +17,12 @@ def list_stowable_packages(packages: Path) -> list[Path]: def apply_stow_operation_to_packages( packages: Path, target: Path, - stow_op: Literal["-D", "--no-folding"], + stow_op: Literal["--delete", "--no-folding"], env: Environment, ) -> bool: + if not packages.exists(): + env.logger.warn("nothing to clean up <_mood.anxious>") + return True if not run_shell_command(["stow", "--version"], env.logger): env.logger.error("stow not installed D:") return False @@ -35,6 +37,6 @@ def apply_stow_operation_to_packages( results = parallelize( lambda command: is_some(run_shell_command, command, env.logger)[0], commands, - env, + env.logger, ) return len(commands) == sum(1 if x else 0 for x in results) diff --git a/dots_manager/template.py b/dots_manager/template.py index 4c62500..13aecf7 100644 --- a/dots_manager/template.py +++ b/dots_manager/template.py @@ -2,20 +2,18 @@ from pathlib import Path from typing import Optional import shutil import jinja2 -from dots_manager.config import Config -from dots_manager.parallel import parallelize -from dots_manager.utils import is_some -from dots_manager.env import Environment +from dots_manager.config import Constants, Environment +from dots_manager.utils import is_some, parallelize def is_template(path: Path) -> bool: - return path.suffix == Config.template_extension + return path.suffix == Constants.template_extension def find_templates_under(destination: Path) -> list[Path]: return [ t - for t in destination.glob("**/*" + Config.template_extension) + for t in destination.glob("**/*" + Constants.template_extension) if t.is_file() and is_template(t) ] @@ -47,15 +45,15 @@ def render( source: Path, jinja_env: jinja2.Environment, env: Environment ) -> Optional[str]: try: - env.logger.debug(f"rendering template {source} ✿.｡.:･") + env.logger.debug(f"rendering template {source} <_mood.happy>") return jinja_env.from_string(source.read_text()).render(**env.context) except Exception as e: - env.logger.error(f"couldn’t render {source}: {e}") + env.logger.error(f"couldn’t render {source}: {e} <_mood.sad>") return None def compile_dotfiles(source: Path, output: Path, env: Environment) -> bool: - env.logger.info(f"compiling {source} to {output}") + env.logger.info(f"compiling {source} to {output} <_mood.anxious>") output.mkdir(exist_ok=True, parents=True) shutil.copytree(source, output, dirs_exist_ok=True) @@ -68,6 +66,6 @@ def compile_dotfiles(source: Path, output: Path, env: Environment) -> bool: )[0] and (template.unlink() is None), find_templates_under(output), - env, + env.logger, ) return all(tasks) diff --git a/dots_manager/utils.py b/dots_manager/utils.py index e8210cc..1b900f0 100644 --- a/dots_manager/utils.py +++ b/dots_manager/utils.py @@ -1,8 +1,11 @@ -from typing import Callable, Optional, TypeVar, Tuple, ParamSpec, Dict, Any +import logging +from typing import Callable, Optional, TypeVar, Tuple, ParamSpec, Dict, Any, List +from concurrent.futures import ThreadPoolExecutor, as_completed from functools import reduce P = ParamSpec("P") T = TypeVar("T") +R = TypeVar("R") def is_some( @@ -23,3 +26,19 @@ def merge_dicts(*dicts: Dict[str, Any]) -> Dict[str, Any]: return out return reduce(merge, dicts, {}) + + +def parallelize( + worker: Callable[[T], R], + items: List[T], + logger: logging.Logger, + executor: Optional[ThreadPoolExecutor] = None, +) -> List[R]: + if executor is None: + from dots_manager.config import Constants + + executor = ThreadPoolExecutor(max_workers=Constants.max_workers) + with executor as exec: + futures = [exec.submit(worker, item) for item in items] + logger.info(f"submitted {len(futures)} tasks to executor <_mood.excited>") + return [f.result() for f in as_completed(futures)] |