From f6a7b89192a6bb02382e8d2b0d6090f7590b3bd6 Mon Sep 17 00:00:00 2001 From: Akshaj Singhal Date: Wed, 3 Jun 2026 11:10:23 +0000 Subject: [PATCH] refactor(kernel): fix None leak and migrate core utilities to /core --- .gitignore | 4 + .npmignore | 15 -- tests/test_database.py | 10 +- trushell/__main__.py | 4 +- trushell/cli.py | 128 ++----------- trushell/commands/__init__.py | 1 + trushell/commands/chronoterm.py | 45 +++++ trushell/commands/core.py | 91 ++++++++++ trushell/commands/editor.py | 26 +++ trushell/commands/joke.py | 70 ++++++++ trushell/commands/tasks.py | 76 ++++++++ trushell/config/builtin_commands.md | 17 ++ trushell/config/plugins.md | 5 + trushell/core/__init__.py | 1 + trushell/{ => core}/database.py | 3 +- trushell/{model.py => core/models.py} | 0 trushell/{ => core}/settings.py | 4 +- trushell/core/trukernel.py | 250 ++++++++++++++++++++++++++ trushell/plugins/git_enhancer/main.py | 8 + trushell/project.py | 24 +-- trushell/pyfunny.py | 59 ------ trushell/todocli.py | 4 +- trushell/utils/__init__.py | 1 + trushell/utils/logger.py | 41 +++++ trushell/utils/manifest_parser.py | 133 ++++++++++++++ 25 files changed, 803 insertions(+), 217 deletions(-) delete mode 100644 .npmignore create mode 100644 trushell/commands/__init__.py create mode 100644 trushell/commands/chronoterm.py create mode 100644 trushell/commands/core.py create mode 100644 trushell/commands/editor.py create mode 100644 trushell/commands/joke.py create mode 100644 trushell/commands/tasks.py create mode 100644 trushell/config/builtin_commands.md create mode 100644 trushell/config/plugins.md create mode 100644 trushell/core/__init__.py rename trushell/{ => core}/database.py (98%) rename trushell/{model.py => core/models.py} (100%) rename trushell/{ => core}/settings.py (97%) create mode 100644 trushell/core/trukernel.py create mode 100644 trushell/plugins/git_enhancer/main.py delete mode 100644 trushell/pyfunny.py create mode 100644 trushell/utils/__init__.py create mode 100644 trushell/utils/logger.py create mode 100644 trushell/utils/manifest_parser.py diff --git a/.gitignore b/.gitignore index 92680b2..a6b6f27 100644 --- a/.gitignore +++ b/.gitignore @@ -13,3 +13,7 @@ dist/ build/ .idea/ .vscode/ + +# Ignore runtime TruShell user configuration and local runtime state +trushell/config/my_command_config.md +.trushell/ diff --git a/.npmignore b/.npmignore deleted file mode 100644 index 68b5234..0000000 --- a/.npmignore +++ /dev/null @@ -1,15 +0,0 @@ -# Exclude Python bytecode and caches from npm bundles -__pycache__/ -*.pyc -*.pyo -*.pyd - -# Exclude development metadata and tests from runtime package -tests/ -.pytest_cache/ -poetry.lock -package-lock.json -pyproject.toml -.gitignore -.npmignore -.github/ diff --git a/tests/test_database.py b/tests/test_database.py index 44360f0..f4c1d64 100644 --- a/tests/test_database.py +++ b/tests/test_database.py @@ -1,10 +1,10 @@ -from trushell.database import _create_table, get_all_todos, get_db_connection, insert_todo -from trushell.model import Todo +from trushell.core.database import _create_table, get_all_todos, get_db_connection, insert_todo +from trushell.core.models import Todo def test_get_db_connection_returns_fresh_connection(monkeypatch, tmp_path) -> None: db_path = tmp_path / "todos.db" - monkeypatch.setattr("trushell.database.DB_PATH", db_path) + monkeypatch.setattr("trushell.core.database.DB_PATH", db_path) conn_one = get_db_connection() conn_two = get_db_connection() @@ -17,7 +17,7 @@ def test_get_db_connection_returns_fresh_connection(monkeypatch, tmp_path) -> No def test_insert_todo_assigns_sequential_positions(monkeypatch, tmp_path) -> None: db_path = tmp_path / "todos.db" - monkeypatch.setattr("trushell.database.DB_PATH", db_path) + monkeypatch.setattr("trushell.core.database.DB_PATH", db_path) _create_table() insert_todo(Todo(task="first", category="work")) @@ -31,7 +31,7 @@ def test_insert_todo_assigns_sequential_positions(monkeypatch, tmp_path) -> None def test_get_all_todos_works_with_local_connections(monkeypatch, tmp_path) -> None: db_path = tmp_path / "todos.db" - monkeypatch.setattr("trushell.database.DB_PATH", db_path) + monkeypatch.setattr("trushell.core.database.DB_PATH", db_path) _create_table() insert_todo(Todo(task="alpha", category="study")) diff --git a/trushell/__main__.py b/trushell/__main__.py index d5cf0fd..714dd0c 100644 --- a/trushell/__main__.py +++ b/trushell/__main__.py @@ -1,4 +1,4 @@ -from .cli import app +from .cli import app_with_lower if __name__ == "__main__": - app() + app_with_lower() diff --git a/trushell/cli.py b/trushell/cli.py index d7d08b9..b748996 100644 --- a/trushell/cli.py +++ b/trushell/cli.py @@ -1,38 +1,30 @@ from __future__ import annotations -import shlex import sys import typer from . import __version__ +from .core.trukernel import TruKernel from .project import run_interactive_shell -from .pyfunny import joke, joke_trex -from .settings import launch_settings -from .todocli import addtask, completetask, deletetask, showtask, updatetask -from .chronoterm.shell import app as chronoterm_app -app = typer.Typer(name="trushell", help="TruShell: jokes, todos, time, and more.") +app = typer.Typer(name="trushell", help="TruShell manifest-driven launcher.") +kernel = TruKernel() + def app_with_lower() -> None: - """Entry point that normalizes command name to lowercase for case-insensitive invocation.""" - # Normalize the command name to lowercase for case-insensitive behavior - if len(sys.argv) > 0: - sys.argv[0] = sys.argv[0].lower() - # Invoke the main Typer app + """Entry point that normalizes the first argument to lowercase for case-insensitive invocation.""" + if len(sys.argv) > 1: + sys.argv[1] = sys.argv[1].lower() + if sys.argv[1] not in {"--help", "-h", "version"}: + raw = " ".join(sys.argv[1:]) + kernel.execute_command(raw) + return app() -def _invoke_chronoterm_command(command: str) -> None: - try: - chronoterm_app(shlex.split(command)) - except SystemExit: - return - except Exception as error: - typer.secho(f"Error running ChronoTerm command: {error}", fg=typer.colors.RED) - @app.callback(invoke_without_command=True) def main(ctx: typer.Context) -> None: - """Launch the TruShell interactive REPL when no command is provided.""" + """Launch the REPL when no command is provided.""" if ctx.invoked_subcommand is None: run_interactive_shell() @@ -41,99 +33,3 @@ def main(ctx: typer.Context) -> None: def version() -> None: """Show the installed TruShell version.""" typer.echo(__version__) - - -@app.command("joke") -def cli_joke() -> None: - """Tell a random joke with ASCII art.""" - typer.echo(joke()) - - -@app.command("joke-trex") -def cli_joke_trex() -> None: - """Tell a T-Rex joke with sound.""" - typer.echo(joke_trex()) - - -@app.command("addtask") -def cli_addtask(task: str, category: str) -> None: - """Add a todo task.""" - addtask(task, category) - - -@app.command("deletetask") -def cli_deletetask(position: int) -> None: - """Delete a todo task by position.""" - deletetask(position) - - -@app.command("updatetask") -def cli_updatetask(position: int, task: str | None = None, category: str | None = None) -> None: - """Update the text or category of a todo task.""" - updatetask(position, task, category) - - -@app.command("completetask") -def cli_completetask(position: int) -> None: - """Mark a todo task as complete.""" - completetask(position) - - -@app.command("showtasks") -def cli_showtasks() -> None: - """Show all todo tasks.""" - showtask() - - -@app.command("settings") -def cli_settings() -> None: - """Open the interactive settings manager.""" - launch_settings() - - -@app.command("now") -def cli_now() -> None: - """Show the current local time.""" - _invoke_chronoterm_command("now") - - -@app.command("time") -def cli_time() -> None: - """Show the current time in an ASCII clock.""" - _invoke_chronoterm_command("time") - - -@app.command("world") -def cli_world() -> None: - """Show world time for favorite timezones.""" - _invoke_chronoterm_command("world") - - -@app.command("tz") -def cli_tz(action: str = typer.Argument("list", help="list | add | remove"), name: str | None = typer.Argument(None, help="IANA timezone name, e.g. Europe/London")) -> None: - """Manage favorite timezones.""" - command = "tz" - if name: - command += f" {action} {name}" - else: - command += f" {action}" - _invoke_chronoterm_command(command) - - -@app.command("alarm") -def cli_alarm(action: str = typer.Argument("list", help="list | add | remove"), time: str | None = typer.Argument(None, help="Time as HH:MM or YYYY-MM-DD HH:MM"), tz: str | None = typer.Option(None, "--tz", help="Timezone name"), label: str | None = typer.Option(None, "--label", help="Alarm label")) -> None: - """Manage alarms.""" - command = "alarm " + action - if time: - command += f" {time}" - if tz: - command += f" --tz {tz}" - if label: - command += f" --label {label}" - _invoke_chronoterm_command(command) - - -@app.command("sw") -def cli_sw(action: str = typer.Argument("show", help="start | pause | lap | reset | show")) -> None: - """Control the stopwatch.""" - _invoke_chronoterm_command(f"sw {action}") diff --git a/trushell/commands/__init__.py b/trushell/commands/__init__.py new file mode 100644 index 0000000..7ae6482 --- /dev/null +++ b/trushell/commands/__init__.py @@ -0,0 +1 @@ +"""Built-in TruShell commands implemented as manifest-driven wrappers.""" diff --git a/trushell/commands/chronoterm.py b/trushell/commands/chronoterm.py new file mode 100644 index 0000000..c05f9d7 --- /dev/null +++ b/trushell/commands/chronoterm.py @@ -0,0 +1,45 @@ +from __future__ import annotations + +import shlex + +from trushell.chronoterm.shell import app as chronoterm_app + + +def _run_chrono_command(raw_command: str) -> None: + try: + chronoterm_app(shlex.split(raw_command)) + except SystemExit: + return + except Exception as error: + print(f"ChronoTerm error: {error}") + + +def run_now(_: str) -> None: + _run_chrono_command("now") + + +def run_time(_: str) -> None: + _run_chrono_command("time") + + +def run_world(_: str) -> None: + _run_chrono_command("world") + + +def run_tz(args: str) -> None: + if not args.strip(): + _run_chrono_command("tz list") + return + _run_chrono_command(f"tz {args.strip()}") + + +def run_alarm(args: str) -> None: + if not args.strip(): + _run_chrono_command("alarm list") + return + _run_chrono_command(f"alarm {args.strip()}") + + +def run_sw(args: str) -> None: + command = "sw" if not args.strip() else f"sw {args.strip()}" + _run_chrono_command(command) diff --git a/trushell/commands/core.py b/trushell/commands/core.py new file mode 100644 index 0000000..42807f1 --- /dev/null +++ b/trushell/commands/core.py @@ -0,0 +1,91 @@ +from __future__ import annotations + +import os +import subprocess + + +def run_help(_: str) -> None: + """Display available commands by reading the manifest registry.""" + # Import here to avoid circular dependency if trukernel imports core + from trushell.core.trukernel import get_kernel + kernel = get_kernel() + cmds = sorted(kernel.registry.keys()) + print("Available commands:") + # Print in columns + col_width = max(len(c) for c in cmds) + 2 + cols = 4 + for i, cmd in enumerate(cmds): + print(f" {cmd:<{col_width}}", end="") + if (i + 1) % cols == 0: + print() + if len(cmds) % cols != 0: + print() + print("\nType 'help ' for more info (coming soon).") + + +def run_exit(_: str) -> str: + """Signal TruKernel to exit the REPL loop cleanly. + + Returns a special sentinel string instead of raising SystemExit. + The kernel checks for this return value and breaks its loop. + """ + return "__TRUSHELL_EXIT__" + + +def run_settings(args: str) -> None: + """Open the TruShell settings manager.""" + try: + from trushell.core.settings import launch_settings + launch_settings() + except ImportError: + print("Settings module not available.") + except Exception as e: + print(f"Settings error: {e}") + + +def run_os_passthrough(args: str) -> int: + """Generic wrapper that passes args directly to the OS shell. + + Args: + args: The full argument string after the command name. + e.g., for 'ls -la /tmp', args = '-la /tmp' + + Returns: + The subprocess return code (0 = success). + """ + if not args: + print("Usage: [args]") + return 1 + try: + result = subprocess.run(args, shell=True) + return result.returncode + except FileNotFoundError: + print(f"Command not found in PATH.") + return 127 + except Exception as e: + print(f"OS command error: {e}") + return 1 + + +def run_cd_command(args: str) -> None: + """Change directory. Updates Python process CWD so subsequent + os_passthrough commands inherit the new directory. + + Note: This does NOT update the terminal emulator's displayed path. + That is an inherent limitation of Python-based shells. + """ + if not args: + target = os.path.expanduser("~") + else: + target = os.path.expandvars(os.path.expanduser(args.strip())) + + try: + os.chdir(target) + # Print new path so user knows cd succeeded + print(os.getcwd()) + except FileNotFoundError: + print(f"cd: no such file or directory: {target}") + except PermissionError: + print(f"cd: permission denied: {target}") + except NotADirectoryError: + print(f"cd: not a directory: {target}") \ No newline at end of file diff --git a/trushell/commands/editor.py b/trushell/commands/editor.py new file mode 100644 index 0000000..d18c639 --- /dev/null +++ b/trushell/commands/editor.py @@ -0,0 +1,26 @@ +from __future__ import annotations + +from pathlib import Path + + +def run_edit_command(args: str) -> None: + """Open a file in the TruShell editor from a manifest-driven command.""" + if not args.strip(): + print("Usage: edit ") + return + + file_path = Path(args.strip()).expanduser() + initial_text = "" + if file_path.exists(): + try: + initial_text = file_path.read_text(encoding="utf-8") + except OSError as error: + print(f"Editor error: {error}") + return + + try: + from trushell.project import TruShellEditor + + TruShellEditor(str(file_path), initial_text=initial_text).run() + except Exception as error: + print(f"Editor error: {error}") diff --git a/trushell/commands/joke.py b/trushell/commands/joke.py new file mode 100644 index 0000000..3dda450 --- /dev/null +++ b/trushell/commands/joke.py @@ -0,0 +1,70 @@ +from __future__ import annotations + +from pathlib import Path +from typing import Tuple + +import cowsay +import pyjokes + + +def _sound_path(filename: str) -> Path: + return Path(__file__).resolve().parents[1] / "chronoterm" / "sounds" / filename + + +def _play_sound(filename: str) -> None: + try: + # lazy import to avoid optional dependency at kernel import time + from trushell.chronoterm.sound import play_alarm + + sound_path = _sound_path(filename) + if not sound_path.exists(): + return + play_alarm() + except Exception: + # best-effort: do not raise from sound playing + return + + +def _joke_preferences() -> Tuple[str, str]: + try: + from trushell.chronoterm.state import StateStore + + state = StateStore().load() + return state.joke_character or "cow", state.joke_sound or "cow-sound.mp3" + except Exception: + return "cow", "cow-sound.mp3" + + +def run_joke_command(_: str) -> str: + """Return a formatted joke string (cow art by default).""" + joke_text = pyjokes.get_joke() + character_name, sound_file = _joke_preferences() + _play_sound(sound_file) + speaker = getattr(cowsay, character_name, cowsay.cow) + return speaker(joke_text) + + +def run_joke_trex_command(_: str) -> str: + """Return a T-Rex joke string.""" + joke_text = pyjokes.get_joke() + _play_sound("trex-sound.mp3") + return cowsay.trex(joke_text) +from __future__ import annotations + +from trushell.pyfunny import joke, joke_trex + + +def run_joke_command(_: str) -> None: + """Tell a random joke with the default TruShell humor engine.""" + try: + print(joke()) + except Exception as error: + print(f"Joke error: {error}") + + +def run_joke_trex_command(_: str) -> None: + """Tell a T-Rex joke with sound.""" + try: + print(joke_trex()) + except Exception as error: + print(f"Joke error: {error}") diff --git a/trushell/commands/tasks.py b/trushell/commands/tasks.py new file mode 100644 index 0000000..0eb4e05 --- /dev/null +++ b/trushell/commands/tasks.py @@ -0,0 +1,76 @@ +from __future__ import annotations + +from typing import Callable + +from trushell.core.database import complete_todo, get_all_todos, insert_todo +from trushell.core.models import Todo + + +def add_task(args: str) -> None: + """Add a new task to the todo list. The full remainder is treated as the task.""" + if not args.strip(): + print("Usage: task add ") + return + + task_text = args.strip() + todo = Todo(task=task_text, category="General") + insert_todo(todo) + print("Task added.") + + +def show_tasks(_: str) -> None: + """Display the current todo list.""" + tasks = get_all_todos() + if not tasks: + print("No tasks found.") + return + + for index, task in enumerate(tasks, start=1): + status = "✅" if task.status == 2 else "❌" + print(f"{index}. {task.task} [{task.category}] {status}") + + +def complete_task(args: str) -> None: + """Mark a todo item as complete.""" + if not args.strip() or not args.strip().isdigit(): + print("Usage: task done ") + return + + index = int(args.strip()) - 1 + try: + complete_todo(index) + print("Task completed.") + except Exception as error: + print(f"Task error: {error}") + + +def list_tasks(_: str) -> None: + """Alias for show_tasks.""" + show_tasks("") + + +def run_task_command(args: str) -> None: + """Dispatch task subcommands from the manifest-driven task wrapper.""" + subcommands: dict[str, Callable[[str], None]] = { + "add": add_task, + "show": show_tasks, + "done": complete_task, + "list": list_tasks, + } + + if not args.strip(): + print("Usage: task [options]") + return + + parts = args.split(maxsplit=1) + subcmd = parts[0].lower() + subargs = parts[1] if len(parts) > 1 else "" + + handler = subcommands.get(subcmd) + if handler: + try: + handler(subargs) + except Exception as error: + print(f"Task error: {error}") + else: + print(f"Unknown task subcommand: {subcmd}") diff --git a/trushell/config/builtin_commands.md b/trushell/config/builtin_commands.md new file mode 100644 index 0000000..3c26636 --- /dev/null +++ b/trushell/config/builtin_commands.md @@ -0,0 +1,17 @@ +# Built-in command registry for TruShell. +# Each entry is one line and must include {cmd: name}, "function()", and [path]. + +{cmd: help}; "run_help()"; [trushell/commands/core.py]; +{cmd: exit}; "run_exit()"; [trushell/commands/core.py]; +{cmd: edit}; "run_edit_command()"; [trushell/commands/editor.py]; +{cmd: settings}; "run_settings()"; [trushell/commands/core.py]; +{cmd: task}; "run_task_command()"; [trushell/commands/tasks.py]; +{cmd: joke}; "run_joke_command()"; [trushell/commands/joke.py]; +{cmd: joke-trex}; "run_joke_trex_command()"; [trushell/commands/joke.py]; +{cmd: now}; "run_now()"; [trushell/commands/chronoterm.py]; +{cmd: time}; "run_time()"; [trushell/commands/chronoterm.py]; +{cmd: world}; "run_world()"; [trushell/commands/chronoterm.py]; +{cmd: tz}; "run_tz()"; [trushell/commands/chronoterm.py]; +{cmd: alarm}; "run_alarm()"; [trushell/commands/chronoterm.py]; +{cmd: sw}; "run_sw()"; [trushell/commands/chronoterm.py]; + diff --git a/trushell/config/plugins.md b/trushell/config/plugins.md new file mode 100644 index 0000000..393315a --- /dev/null +++ b/trushell/config/plugins.md @@ -0,0 +1,5 @@ +# User-installed plugins are registered here. +# Plugins can optionally run immediately with {lifecycle: on_load}. + +{cmd: gstatus}; "plugin_init()"; [trushell/plugins/git_enhancer/main.py]; {lifecycle: on_load}; +{cmd: theme}; "load_theme()"; [~/.trushell/plugins/theme_engine/loader.py]; {version:1.2}; diff --git a/trushell/core/__init__.py b/trushell/core/__init__.py new file mode 100644 index 0000000..f7e06b1 --- /dev/null +++ b/trushell/core/__init__.py @@ -0,0 +1 @@ +"""Core TruShell kernel package.""" diff --git a/trushell/database.py b/trushell/core/database.py similarity index 98% rename from trushell/database.py rename to trushell/core/database.py index 594b1fa..886092b 100644 --- a/trushell/database.py +++ b/trushell/core/database.py @@ -6,7 +6,7 @@ from platformdirs import user_data_dir -from .model import Todo +from trushell.core.models import Todo APP_NAME = "TruShell" APP_AUTHOR = "AkshajSinghal" @@ -14,6 +14,7 @@ DATA_DIR.mkdir(parents=True, exist_ok=True) DB_PATH = DATA_DIR / "todos.db" + def get_db_connection() -> sqlite3.Connection: return sqlite3.connect(DB_PATH, check_same_thread=False) diff --git a/trushell/model.py b/trushell/core/models.py similarity index 100% rename from trushell/model.py rename to trushell/core/models.py diff --git a/trushell/settings.py b/trushell/core/settings.py similarity index 97% rename from trushell/settings.py rename to trushell/core/settings.py index a61f213..fb369d7 100644 --- a/trushell/settings.py +++ b/trushell/core/settings.py @@ -5,7 +5,7 @@ import typer -from .chronoterm.state import StateStore +from trushell.chronoterm.state import StateStore COMMANDS = ["time", "world", "joke"] TIME_TEMPLATES = [ @@ -54,7 +54,7 @@ def _select_option(prompt: str, options: Sequence[str]) -> str | None: def _available_sound_files() -> list[str]: - sounds_dir = Path(__file__).resolve().parent / "chronoterm" / "sounds" + sounds_dir = Path(__file__).resolve().parents[1] / "chronoterm" / "sounds" if not sounds_dir.exists(): return [] return sorted([path.name for path in sounds_dir.iterdir() if path.is_file()]) diff --git a/trushell/core/trukernel.py b/trushell/core/trukernel.py new file mode 100644 index 0000000..012368b --- /dev/null +++ b/trushell/core/trukernel.py @@ -0,0 +1,250 @@ +from __future__ import annotations + +import importlib.util +import os +import subprocess +import sys +from pathlib import Path +from typing import Any + +from ..utils.logger import get_logger +from ..utils.manifest_parser import parse_manifest + +# Sentinel value returned when user types 'exit' +EXIT_SENTINEL = "__TRUSHELL_EXIT__" + + +_kernel_instance: TruKernel | None = None + +class TruKernel: + """Central manifest-driven dispatch engine for TruShell.""" + + def __init__(self) -> None: + global _kernel_instance + self.logger = get_logger(__name__) + self.base_dir = Path(__file__).resolve().parents[2] + self.registry: dict[str, dict[str, Any]] = {} + self._loaded_modules: dict[str, Any] = {} + self._load_manifests() + _kernel_instance = self # Store reference for get_kernel() + + # ------------------------------------------------------------------ # + # Manifest Loading + # ------------------------------------------------------------------ # + def _manifest_path(self, filename: str) -> Path: + return self.base_dir / "trushell" / "config" / filename + + def _load_manifests(self) -> None: + builtins = parse_manifest( + self._manifest_path("builtin_commands.md"), source="builtin" + ) + plugins = parse_manifest( + self._manifest_path("plugins.md"), source="plugin" + ) + aliases = parse_manifest( + self._manifest_path("my_command_config.md"), source="alias" + ) + + for entry in builtins: + self._register(entry) + for entry in plugins: + self._register(entry) + for entry in aliases: + self._register(entry, override=True) + + # Run on_load plugins during init + for entry in plugins: + if entry["meta"].get("lifecycle") == "on_load": + self._execute_entry(entry, args="", init=True) + + def _register(self, entry: dict[str, Any], override: bool = False) -> None: + command = entry["command"] + if override or command not in self.registry: + self.registry[command] = entry + self.logger.debug( + "Registered '%s' from %s", command, entry.get("source") + ) + else: + self.logger.debug( + "Command '%s' already registered from %s; skipping %s", + command, + self.registry[command].get("source"), + entry.get("source"), + ) + + # ------------------------------------------------------------------ # + # Module Importing (Lazy + Cached) + # ------------------------------------------------------------------ # + def _resolve_module_path(self, raw_path: str) -> Path: + path = Path(raw_path).expanduser() + if path.is_absolute(): + return path + return (self.base_dir / path).resolve() + + def _import_module(self, path: str): + resolved = self._resolve_module_path(path) + if not resolved.exists(): + raise FileNotFoundError(f"Module not found: {resolved}") + + # Use resolved path string as key — no hash collision risk + module_key = str(resolved) + if module_key in self._loaded_modules: + return self._loaded_modules[module_key] + + spec = importlib.util.spec_from_file_location(module_key, resolved) + if spec is None or spec.loader is None: + raise ImportError(f"Cannot load spec: {resolved}") + + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) # type: ignore[attr-defined] + self._loaded_modules[module_key] = module + self.logger.debug("Loaded module: %s", resolved) + return module + + # ------------------------------------------------------------------ # + # Command Execution + # ------------------------------------------------------------------ # + def _execute_entry( + self, entry: dict[str, Any], args: str = "", init: bool = False + ) -> str | bool: + """Execute a registered command entry. + + Returns: + EXIT_SENTINEL – user typed 'exit', REPL should stop + True – command executed successfully + False – command failed (logged, error printed) + """ + command = entry["command"] + try: + module = self._import_module(entry["path"]) + func = getattr(module, entry["function"]) + except (FileNotFoundError, ImportError, AttributeError) as err: + if not init: + print("Command not available. Check manifest and module path.") + self.logger.warning( + "Failed to load '%s' from %s: %s", command, entry["path"], err + ) + return False + + try: + result = func(args) + # Check for exit sentinel + if result == EXIT_SENTINEL: + self.logger.info("Exit command received") + return EXIT_SENTINEL + + # Only print if the function explicitly returned a value + # This prevents the kernel from printing 'None' when functions + # perform their own prints and return None. + if result is not None: + print(result) + + self.logger.debug("Dispatched '%s' → %s", command, entry["path"]) + return True + except SystemExit: + # Should never happen if commands use sentinel, but guard anyway + self.logger.info("SystemExit caught from '%s'", command) + return EXIT_SENTINEL + except Exception as err: + self.logger.exception("Unhandled exception in '%s'", command) + print(f"Command error: {err}") + return False + + def execute_command(self, raw_command: str) -> str | bool: + """Parse and dispatch a single user input line. + + Returns: + EXIT_SENTINEL – stop the REPL loop + True – command handled (success or logged failure) + False – command not found AND os passthrough also failed + """ + trimmed = raw_command.strip() + if not trimmed: + return True # blank line is fine + + parts = trimmed.split(maxsplit=1) + command = parts[0].lower() + args = parts[1] if len(parts) > 1 else "" + + # 1. Try manifest registry + entry = self.registry.get(command) + if entry is not None: + return self._execute_entry(entry, args=args) + + # 2. Fall back to OS passthrough for unknown commands + return self._os_passthrough(command, args) + + # ------------------------------------------------------------------ # + # OS Passthrough Fallback + # ------------------------------------------------------------------ # + def _os_passthrough(self, command: str, args: str) -> bool: + """Try to run an unregistered command via the host OS shell. + + Special-cases 'cd' because it must change the Python process CWD. + """ + full_cmd = f"{command} {args}".strip() + + if command == "cd": + return self._handle_cd(args) + + try: + result = subprocess.run( + full_cmd, + shell=True, + capture_output=True, + text=True, + ) + if result.stdout: + print(result.stdout, end="") + if result.returncode != 0: + stderr = result.stderr.strip() + if result.returncode == 127: + print("Command not found. Type 'help' for available commands.") + elif stderr: + print(stderr, end="", file=sys.stderr) + return False + if result.stderr: + print(result.stderr, end="", file=sys.stderr) + return True + except FileNotFoundError: + print(f"Command not found: {command}") + self.logger.warning("OS command not found: %s", command) + return False + except Exception as err: + print(f"OS command error: {err}") + self.logger.exception("OS passthrough failed: %s", full_cmd) + return False + + def _handle_cd(self, args: str) -> bool: + """Change directory in the Python process so subprocesses inherit it.""" + target = os.path.expandvars(os.path.expanduser(args.strip())) if args else os.path.expanduser("~") + try: + os.chdir(target) + print(os.getcwd()) # Visual feedback that cd worked + return True + except FileNotFoundError: + print(f"cd: no such file or directory: {target}") + return False + except PermissionError: + print(f"cd: permission denied: {target}") + return False + except NotADirectoryError: + print(f"cd: not a directory: {target}") + return False + + # ------------------------------------------------------------------ # + # Public Helpers + # ------------------------------------------------------------------ # + def list_commands(self) -> list[str]: + return sorted(self.registry) + + +def get_kernel() -> TruKernel: + """Return the global TruKernel instance. + + Used by commands like help that need a live registry. + """ + global _kernel_instance + if _kernel_instance is None: + _kernel_instance = TruKernel() + return _kernel_instance \ No newline at end of file diff --git a/trushell/plugins/git_enhancer/main.py b/trushell/plugins/git_enhancer/main.py new file mode 100644 index 0000000..093986c --- /dev/null +++ b/trushell/plugins/git_enhancer/main.py @@ -0,0 +1,8 @@ +from __future__ import annotations + + +def plugin_init(_: str) -> None: + """Sample plugin initializer for the TruShell manifest-driven architecture.""" + # This placeholder plugin is intentionally lightweight. Actual plugin logic + # should handle its own errors and user-facing messages. + return diff --git a/trushell/project.py b/trushell/project.py index c9fb100..2664d20 100644 --- a/trushell/project.py +++ b/trushell/project.py @@ -277,27 +277,21 @@ def _handle_os_fallback(raw_command: str) -> bool: def run_interactive_shell() -> None: """Persistent REPL loop for the TruShell core.""" + from .core.trukernel import EXIT_SENTINEL, TruKernel + + kernel = TruKernel() + typer.secho("Entering TruShell. Type 'exit' to quit.", fg=typer.colors.CYAN) while True: try: - raw_command, command, argument = _prompt_command() + raw_command, command, _ = _prompt_command() except (KeyboardInterrupt, EOFError): typer.echo("") break - local_result = _handle_local_command(command, argument) - if local_result == "exit": + result = kernel.execute_command(raw_command) + if result == EXIT_SENTINEL: break - if local_result == "handled": - continue - if _handle_chronoterm_command(raw_command, command): - continue - if _handle_cd_command(raw_command): - continue - if _handle_edit_command(raw_command): - continue - if _handle_os_fallback(raw_command): - continue - - typer.secho(f"Unknown command: {command}", fg=typer.colors.RED) + if result is False: + typer.secho(f"Unknown command: {command}", fg=typer.colors.RED) diff --git a/trushell/pyfunny.py b/trushell/pyfunny.py deleted file mode 100644 index 32c0c1d..0000000 --- a/trushell/pyfunny.py +++ /dev/null @@ -1,59 +0,0 @@ -from __future__ import annotations - -from pathlib import Path - -import cowsay -import pyjokes -import typer - -from .chronoterm.state import StateStore -from .chronoterm.sound import play_alarm - -DEFAULT_JOKE_CHARACTER = "cow" -DEFAULT_JOKE_SOUND = "cow-sound.mp3" - - -def _sound_path(filename: str) -> Path: - return Path(__file__).resolve().parent / "chronoterm" / "sounds" / filename - - -def _play_sound(filename: str) -> None: - sound_path = _sound_path(filename) - - if not sound_path.exists(): - typer.secho(f"Sound file missing: {sound_path}", fg=typer.colors.YELLOW) - return - - # Note: Your play_alarm() currently plays a system beep/tone. - # If you want it to play specific MP3s, you'd need to update play_alarm - # to accept a file path. For now, this just triggers the alarm sound - # as a notification that a joke is coming. - try: - play_alarm() - except Exception: - typer.secho("Unable to play sound. Continuing without audio.", fg=typer.colors.YELLOW) - - -def _joke_preferences() -> tuple[str, str]: - state = StateStore().load() - return state.joke_character or DEFAULT_JOKE_CHARACTER, state.joke_sound or DEFAULT_JOKE_SOUND - - -def _render_joke(character_name: str, text: str) -> str: - speaker = getattr(cowsay, character_name, None) - if not callable(speaker): - speaker = cowsay.cow - return speaker(text) - - -def joke() -> str: - joke_text = pyjokes.get_joke() - character_name, sound_file = _joke_preferences() - _play_sound(sound_file) - return _render_joke(character_name, joke_text) - - -def joke_trex() -> str: - joke_text = pyjokes.get_joke() - _play_sound("trex-sound.mp3") - return cowsay.trex(joke_text) \ No newline at end of file diff --git a/trushell/todocli.py b/trushell/todocli.py index bb13e4d..a70359f 100644 --- a/trushell/todocli.py +++ b/trushell/todocli.py @@ -4,8 +4,8 @@ from rich.console import Console from rich.table import Table -from .database import complete_todo, delete_todo, get_all_todos, insert_todo, update_todo -from .model import Todo +from trushell.core.database import complete_todo, delete_todo, get_all_todos, insert_todo, update_todo +from trushell.core.models import Todo console = Console() app = typer.Typer(name="todo", help="Manage todo tasks.") diff --git a/trushell/utils/__init__.py b/trushell/utils/__init__.py new file mode 100644 index 0000000..e65caa3 --- /dev/null +++ b/trushell/utils/__init__.py @@ -0,0 +1 @@ +"""Utility helpers for TruShell.""" diff --git a/trushell/utils/logger.py b/trushell/utils/logger.py new file mode 100644 index 0000000..774d58c --- /dev/null +++ b/trushell/utils/logger.py @@ -0,0 +1,41 @@ +from __future__ import annotations + +import logging +from logging.handlers import RotatingFileHandler +from pathlib import Path + + +def get_logger(name: str = "trushell") -> logging.Logger: + """Create or return the centralized TruShell logger.""" + logger = logging.getLogger(name) + if logger.handlers: + return logger + + log_dir = Path.home() / ".trushell" / "logs" + log_dir.mkdir(parents=True, exist_ok=True) + log_file = log_dir / "trushell.log" + + formatter = logging.Formatter( + "[%(asctime)s] %(levelname)s | %(module)s | %(message)s", + datefmt="%Y-%m-%d %H:%M:%S", + ) + + file_handler = RotatingFileHandler( + log_file, + maxBytes=5 * 1024 * 1024, + backupCount=3, + encoding="utf-8", + ) + file_handler.setLevel(logging.DEBUG) + file_handler.setFormatter(formatter) + + console_handler = logging.StreamHandler() + console_handler.setLevel(logging.WARNING) + console_handler.setFormatter(formatter) + + logger.setLevel(logging.DEBUG) + logger.addHandler(file_handler) + logger.addHandler(console_handler) + logger.propagate = False + + return logger diff --git a/trushell/utils/manifest_parser.py b/trushell/utils/manifest_parser.py new file mode 100644 index 0000000..5cbfe77 --- /dev/null +++ b/trushell/utils/manifest_parser.py @@ -0,0 +1,133 @@ +from __future__ import annotations + +import logging +import os +from pathlib import Path +from typing import Any + +from .logger import get_logger + + +def _strip_quotes(value: str) -> str: + value = value.strip() + if value.startswith('"') and value.endswith('"'): + value = value[1:-1].strip() + return value + + +def _strip_brackets(value: str) -> str: + value = value.strip() + if value.startswith("[") and value.endswith("]"): + value = value[1:-1].strip() + return value + + +def _parse_meta_token(token: str) -> tuple[str, str] | None: + content = token.strip()[1:-1].strip() + if ":" not in content: + return None + key, value = content.split(":", 1) + return key.strip().lower(), value.strip() + + +def parse_manifest(manifest_path: Path, source: str = "manifest") -> list[dict[str, Any]]: + """Parse a manifest file and return a list of validated command entries.""" + logger = get_logger() + entries: list[dict[str, Any]] = [] + + if not manifest_path.exists(): + logger.warning("Manifest file not found: %s", manifest_path) + return entries + + try: + with manifest_path.open("r", encoding="utf-8") as handle: + lines = handle.readlines() + except OSError as error: + logger.warning("Unable to read manifest %s: %s", manifest_path, error) + return entries + + for line_number, raw_line in enumerate(lines, start=1): + line = raw_line.strip() + if not line or line.startswith("#"): + continue + + parts = [part.strip() for part in line.split(";") if part.strip()] + command: str | None = None + function: str | None = None + path: str | None = None + meta: dict[str, str] = {} + malformed = False + + for part in parts: + if part.startswith("#"): + break + if part.startswith("{") and part.endswith("}"): + parsed = _parse_meta_token(part) + if parsed is None: + logger.warning( + "Malformed metadata in %s at line %s: %s", + manifest_path, + line_number, + part, + ) + malformed = True + break + key, value = parsed + if key in {"cmd", "alias"}: + command = value.lower() + else: + meta[key] = value + elif part.startswith('"') and part.endswith('"'): + function_name = _strip_quotes(part) + function_name = function_name.removesuffix("()") + if not function_name: + logger.warning( + "Empty function name in %s at line %s: %s", + manifest_path, + line_number, + part, + ) + malformed = True + break + function = function_name + elif part.startswith("[") and part.endswith("]"): + raw_path = _strip_brackets(part) + if not raw_path: + logger.warning( + "Empty path in %s at line %s: %s", + manifest_path, + line_number, + part, + ) + malformed = True + break + path = os.path.expanduser(raw_path) + else: + logger.warning( + "Unknown manifest token in %s at line %s: %s", + manifest_path, + line_number, + part, + ) + malformed = True + break + + if malformed or command is None or function is None or path is None: + logger.warning( + "Skipping invalid manifest entry in %s at line %s.", + manifest_path, + line_number, + ) + continue + + entries.append( + { + "command": command, + "function": function, + "path": path, + "meta": meta, + "source": source, + } + ) + + return entries