diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index 3caf2fab4..e08f0f67d 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -1,5 +1,3 @@ - - # Contribute to PasarGuard Thanks for considering contributing to **PasarGuard**! @@ -8,17 +6,17 @@ Thanks for considering contributing to **PasarGuard**! Please **don’t use GitHub Issues** to ask questions. Instead, use one of the following platforms: -* 💬 Telegram: [@Pasar_Guard](https://t.me/pasar_guard) -* 🗣️ GitHub Discussions: [PasarGuard Discussions](https://github.com/pasarguard/panel/discussions) +- 💬 Telegram: [@Pasar_Guard](https://t.me/pasar_guard) +- 🗣️ GitHub Discussions: [PasarGuard Discussions](https://github.com/pasarguard/panel/discussions) ## 🐞 Reporting Issues When reporting a bug or issue, please include: -* ✅ What you expected to happen -* ❌ What actually happened (include server logs or browser errors) -* ⚙️ Your `xray` JSON config and `.env` settings (censor sensitive info) -* 🔢 Your PasarGuard version and Docker version (if applicable) +- ✅ What you expected to happen +- ❌ What actually happened (include server logs or browser errors) +- ⚙️ Your `xray` JSON config and `.env` settings (censor sensitive info) +- 🔢 Your PasarGuard version and Docker version (if applicable) --- @@ -28,15 +26,15 @@ If there's no open issue for your idea, consider opening one for discussion **be You can contribute to any issue that: -* Has no PR linked -* Has no maintainer assigned +- Has no PR linked +- Has no maintainer assigned No need to ask for permission! ## 🔀 Branching Strategy -* Always branch off of the `next` branch -* Keep `master` stable and production-ready +- Always branch off of the `next` branch +- Keep `main` stable and production-ready --- @@ -46,6 +44,7 @@ No need to ask for permission! . ├── app # Backend code (FastAPI - Python) ├── cli # CLI code (Typer - Python) +├── tui # TUI code (Textual - Python) ├── dashboard # Frontend code (React - TypeScript) └── tests # API tests ``` @@ -56,10 +55,10 @@ No need to ask for permission! The backend is built with **FastAPI** and **SQLAlchemy**: -* **Pydantic models**: `app/models` -* **Database models & operations**: `app/db` -* **backend logic should go in**: `app/operations` -* **Migrations (Alembic)**: `app/db/migrations` +- **Pydantic models**: `app/models` +- **Database models & operations**: `app/db` +- **backend logic should go in**: `app/operations` +- **Migrations (Alembic)**: `app/db/migrations` 🧩 **Note**: Ensure **all backend logic is organized and implemented in the `operations` module**. This keeps route handling, database access, and service logic clearly separated and easier to maintain. @@ -67,8 +66,8 @@ The backend is built with **FastAPI** and **SQLAlchemy**: Enable the `DOCS` flag in your `.env` file to access: -* Swagger UI: [http://localhost:8000/docs](http://localhost:8000/docs) -* ReDoc: [http://localhost:8000/redoc](http://localhost:8000/redoc) +- Swagger UI: [http://localhost:8000/docs](http://localhost:8000/docs) +- ReDoc: [http://localhost:8000/redoc](http://localhost:8000/redoc) ### 🎯 Code Formatting @@ -95,8 +94,8 @@ make run-migration The frontend is located in the `dashboard` directory and is built using: -* **React + TypeScript** -* **Tailwind CSS (Shadcn UI)** +- **React + TypeScript** +- **Tailwind CSS (Shadcn UI)** To build: @@ -108,17 +107,25 @@ Remove the `dashboard/build` directory and restart the Python backend — the fr ### 🧩 Component Guidelines -* Follow **Tailwind + Shadcn** best practices -* Keep components **single-purpose** -* Prioritize **readability** and **maintainability** +- Follow **Tailwind + Shadcn** best practices +- Keep components **single-purpose** +- Prioritize **readability** and **maintainability** --- ## 🛠️ PasarGuard CLI -PasarGuard’s CLI is built using [Textual](https://textual.textualize.io/). +PasarGuard’s CLI is built using [Typer](https://typer.tiangolo.com/). + +- CLI codebase: `cli/` + +--- + +## 🛠️ PasarGuard TUI + +PasarGuard’s TUI is built using [Textual](https://textual.textualize.io/). -* CLI codebase: `cli/` +- TUI codebase: `tui/` --- diff --git a/Dockerfile b/Dockerfile index 7091558d3..62defc9a6 100644 --- a/Dockerfile +++ b/Dockerfile @@ -28,8 +28,12 @@ WORKDIR /code ENV PATH="/code/.venv/bin:$PATH" -COPY cli_wrapper.sh /usr/bin/marzban-cli -RUN chmod +x /usr/bin/marzban-cli +COPY cli_wrapper.sh /usr/bin/pasarguard-cli +RUN chmod +x /usr/bin/pasarguard-cli + +COPY tui_wrapper.sh /usr/bin/pasarguard-tui +RUN chmod +x /usr/bin/pasarguard-tui + RUN chmod +x /code/start.sh ENTRYPOINT ["/code/start.sh"] \ No newline at end of file diff --git a/Makefile b/Makefile index b78b1572d..d5c913da9 100644 --- a/Makefile +++ b/Makefile @@ -94,6 +94,12 @@ run: run-cli: @uv run pasarguard-cli.py +# run pasarguard-tui +.PHONY: run-tui +run-tui: + @uv run pasarguard-tui.py + + # Run tests .PHONY: test test: diff --git a/README-fa.md b/README-fa.md index dc4c52e21..fb3b6f31e 100644 --- a/README-fa.md +++ b/README-fa.md @@ -376,6 +376,18 @@ pasarguard cli [OPTIONS] COMMAND [ARGS]... برای اطلاعات بیشتر، می توانید [مستندات CLI پاسارگارد](./cli/README.md) را مطالعه کنید. +# رابط کاربری متنی (TUI) پاسارگارد + +پاسارگارد همچنین یک رابط کاربری متنی (TUI) برای مدیریت تعاملی مستقیم در ترمینال شما فراهم می کند. + +اگر پاسارگارد را با استفاده از اسکریپت نصب آسان نصب کرده اید، می توانید با اجرای دستور زیر به TUI دسترسی پیدا کنید: + +```bash +pasarguard tui +``` + +برای اطلاعات بیشتر، می توانید [مستندات TUI پاسارگارد](./tui/README.md) را مطالعه کنید. + # نود پروژه پاسارگارد [نود](https://github.com/PasarGuard/node) را معرفی می کند که توزیع زیرساخت را متحول می کند. با نود، می توانید زیرساخت خود را در چندین مکان توزیع کنید و از مزایایی مانند افزونگی، در دسترس بودن بالا، مقیاس پذیری و انعطاف پذیری بهره مند شوید. نود به کاربران این امکان را می دهد که به سرورهای مختلف متصل شوند و به آنها انعطاف پذیری انتخاب و اتصال به چندین سرور را به جای محدود شدن به تنها یک سرور ارائه می دهد. diff --git a/README-ru.md b/README-ru.md index 277ea40e6..401042784 100644 --- a/README-ru.md +++ b/README-ru.md @@ -397,6 +397,18 @@ pasarguard cli [OPTIONS] COMMAND [ARGS]... Для получения дополнительной информации можно ознакомиться с [документацией по PasarGuard CLI](./cli/README.md). +# Пользовательский интерфейс терминала (TUI) PasarGuard + +PasarGuard также предоставляет пользовательский интерфейс терминала (TUI) для интерактивного управления непосредственно в вашем терминале. + +Если вы установили PasarGuard с помощью простого установочного скрипта, вы можете получить доступ к TUI, запустив: + +```bash +pasarguard tui +``` + +Для получения дополнительной информации вы можете ознакомиться с [документацией по TUI PasarGuard](./tui/README.md). + # Node Проект PasarGuard представляет [node](https://github.com/PasarGuard/node), который помогает Вам в распределении инфраструктуры. С помощью node можно распределить инфраструктуру по нескольким узлам, получив такие преимущества, как высокая доступность, масштабируемость и гибкость. node позволяет пользователям подключаться к различным серверам, предоставляя им гибкость в выборе, а не ограничиваться только одним сервером. diff --git a/README-zh-cn.md b/README-zh-cn.md index 8ffe55c5b..9933bd13b 100644 --- a/README-zh-cn.md +++ b/README-zh-cn.md @@ -391,6 +391,18 @@ pasarguard cli [OPTIONS] COMMAND [ARGS]... 更多信息，您可以阅读[PasarGuard CLI的文档](./cli/README.md)。 +# PasarGuard 终端用户界面 (TUI) + +PasarGuard 还提供了一个终端用户界面 (TUI)，用于直接在您的终端中进行交互式管理。 + +如果您使用简易安装脚本安装了 PasarGuard，您可以通过运行以下命令访问 TUI： + +```bash +pasarguard tui +``` + +更多信息，您可以阅读 [PasarGuard TUI 的文档](./tui/README.md)。 + # Node PasarGuard项目引入了[node](https://github.com/PasarGuard/node)，它彻底改变了基础设施的分布。通过node，您可以将您的基础设施分布到多个位置，从而获得冗余、高可用性、可伸缩性、灵活性等好处。node使用户能够连接到不同的服务器，为他们提供了选择和连接多个服务器的灵活性，而不是仅限于一个服务器。 diff --git a/README.md b/README.md index 23007b85d..9ba5d55d4 100644 --- a/README.md +++ b/README.md @@ -385,6 +385,18 @@ pasarguard cli [OPTIONS] COMMAND [ARGS]... For more information, You can read [PasarGuard CLI's documentation](./cli/README.md). +# PasarGuard TUI + +PasarGuard also provides a Terminal User Interface (TUI) for interactive management directly within your terminal. + +If you've installed PasarGuard using the easy install script, you can access the TUI by running: + +```bash +pasarguard tui +``` + +For more information, you can read [PasarGuard TUI's documentation](./tui/README.md). + # Node The PasarGuard project introduces the [node](https://github.com/PasarGuard/node), which revolutionizes infrastructure distribution. With node, you can distribute your infrastructure across multiple locations, unlocking benefits such as redundancy, high availability, scalability, flexibility. node empowers users to connect to different servers, offering them the flexibility to choose and connect to multiple servers instead of being limited to only one server. diff --git a/app/utils/helpers.py b/app/utils/helpers.py index f91837d9f..4fd780111 100644 --- a/app/utils/helpers.py +++ b/app/utils/helpers.py @@ -57,15 +57,6 @@ def format_validation_error(error: ValidationError) -> str: return "\n".join([e["loc"][0].replace("_", " ").capitalize() + ": " + e["msg"] for e in error.errors()]) -def format_cli_validation_error(errors: ValidationError, notify: callable): - for error in errors.errors(): - for err in error["msg"].split(";"): - notify( - title=f"Error: {error['loc'][0].replace('_', ' ').capitalize()}", - message=err.strip(), - severity="error", - ) - def escape_tg_html(list: tuple[str]) -> tuple[str]: """Escapes HTML special characters for the telegram HTML parser.""" diff --git a/cli/README.md b/cli/README.md index b96344c0b..529542f84 100644 --- a/cli/README.md +++ b/cli/README.md @@ -1,42 +1,60 @@ # PasarGuard CLI -A modern, interactive command-line interface for managing PasarGuard, built with Textual. +A modern, type-safe command-line interface for managing PasarGuard, built with Typer. ## Features -- 🎯 Interactive TUI (Text User Interface) -- 📱 Responsive design with dark mode support -- ⌨️ Keyboard shortcuts for quick navigation -- 🔄 Real-time updates -- 📊 Rich data visualization -- 🔒 Secure admin management +- 🎯 Type-safe CLI with rich output +- 📊 Beautiful tables and panels +- 🔒 Secure admin management +- 📈 System status monitoring +- ⌨️ Interactive prompts and confirmations + +## Installation + +The CLI is included with PasarGuard and can be used directly: + +```bash +PasarGuard cli --help + +# Or from the project root +uv run PasarGuard-cli.py --help +``` ## Usage -### Starting the CLI +### General Commands ```bash -pasarguard cli +# Show version +pasarguard cli version + +# Show help +pasarguard cli --help ``` -### Keyboard Shortcuts +### Admin Management -#### Global Commands +```bash +# List all admins +pasarguard cli admins --list -- `q` - Quit the application -- `?` - Show help +# Create new admin +pasarguard cli admins --create username -#### Admin Section +# Delete admin +pasarguard cli admins --delete username -- `c` - Create new admin -- `m` - Modify admin -- `r` - Reset admin usage -- `d` - Delete admin -- `i` - Import admins from environment +# Modify admin (password and sudo status) +PasarGuard cli admins --modify username -### Admin Management +# Reset admin usage +pasarguard cli admins --reset-usage username +``` -- Create, modify, and delete admin accounts -- Reset admin usage statistics -- Import admins from environment variables -- View admin details and status +### System Information + +```bash +# Show system status +PasarGuard cli system +``` diff --git a/cli/__init__.py b/cli/__init__.py index 6ca458cdd..abcb32581 100644 --- a/cli/__init__.py +++ b/cli/__init__.py @@ -1,29 +1,49 @@ -from textual.screen import ModalScreen -from textual.widgets import Input +""" +PasarGuard CLI Package +A modern, type-safe CLI built with Typer for managing PasarGuard instances. +""" -class BaseModal(ModalScreen): - def __init__(self, *args, **kwargs): - super().__init__(*args, **kwargs) +from pydantic import ValidationError +from rich.console import Console +from rich.table import Table - async def key_left(self) -> None: - """Move focus left on arrow key press.""" - if not self.query_one(Input).has_focus: - self.app.action_focus_previous() +from app.models.admin import AdminDetails +from app.operation import OperatorType +from app.operation.admin import AdminOperation +from app.operation.system import SystemOperation - async def key_right(self) -> None: - """Move focus right on arrow key press.""" - if not self.query_one(Input).has_focus: - self.app.action_focus_next() +# Initialize console for rich output +console = Console() - async def key_down(self) -> None: - """Move focus down on arrow key press.""" - self.app.action_focus_next() +# system admin for CLI operations +SYSTEM_ADMIN = AdminDetails(username="cli", is_sudo=True, telegram_id=None, discord_webhook=None) - async def key_up(self) -> None: - """Move focus up on arrow key press.""" - self.app.action_focus_previous() - async def key_escape(self) -> None: - """Close modal when ESC is pressed.""" - self.dismiss() +def get_admin_operation() -> AdminOperation: + """Get admin operation instance.""" + return AdminOperation(OperatorType.CLI) + + +def get_system_operation() -> SystemOperation: + """Get node operation instance.""" + return SystemOperation(OperatorType.CLI) + + +class BaseCLI: + """Base class for CLI operations.""" + + def __init__(self): + self.console = console + + def create_table(self, title: str, columns: list) -> Table: + """Create a rich table with given columns.""" + table = Table(title=title) + for column in columns: + table.add_column(column["name"], style=column.get("style", "white")) + return table + + def format_cli_validation_error(self, errors: ValidationError): + for error in errors.errors(): + for err in error["msg"].split(";"): + self.console.print(f"[red]Error: {err}[/red]") diff --git a/cli/admin.py b/cli/admin.py index 497f001f5..4e2f8d939 100644 --- a/cli/admin.py +++ b/cli/admin.py @@ -1,447 +1,226 @@ -import asyncio +""" +Admin CLI Module -from decouple import UndefinedValueError, config +Handles admin account management through the command line interface. +""" + +import typer from pydantic import ValidationError -from rich.text import Text -from sqlalchemy import func, select -from textual.app import ComposeResult -from textual.containers import Container, Horizontal, Vertical -from textual.coordinate import Coordinate -from textual.widgets import Button, DataTable, Input, Static, Switch - -from app.db import AsyncSession + from app.db.base import get_db -from app.db.models import Admin, User -from app.models.admin import AdminCreate, AdminDetails, AdminModify -from app.operation import OperatorType -from app.operation.admin import AdminOperation -from app.utils.helpers import format_cli_validation_error, readable_datetime +from app.models.admin import AdminCreate, AdminModify from app.utils.system import readable_size -from cli import BaseModal - -SYSTEM_ADMIN = AdminDetails(username="cli", is_sudo=True, telegram_id=None, discord_webhook=None) - - -class AdminDelete(BaseModal): - def __init__( - self, db: AsyncSession, operation: AdminOperation, username: str, on_close: callable, *args, **kwargs - ) -> None: - super().__init__(*args, **kwargs) - self.db = db - self.operation = operation - self.username = username - self.on_close = on_close - - async def on_mount(self) -> None: - """Ensure the first button is focused.""" - yes_button = self.query_one("#no") - self.set_focus(yes_button) - - def compose(self) -> ComposeResult: - with Container(classes="modal-box-delete"): - yield Static("Are you sure about deleting this admin?", classes="title") - yield Horizontal( - Button("Yes", id="yes", variant="success"), - Button("No", id="no", variant="error"), - classes="button-container", - ) +from cli import SYSTEM_ADMIN, BaseCLI, console, get_admin_operation - async def on_button_pressed(self, event: Button.Pressed) -> None: - if event.button.id == "yes": - try: - await self.operation.remove_admin(self.db, self.username, SYSTEM_ADMIN) - self.on_close() - except ValueError as e: - self.notify(str(e), severity="error", title="Error") - await self.key_escape() - - -class AdminResetUsage(BaseModal): - def __init__( - self, db: AsyncSession, operation: AdminOperation, username: str, on_close: callable, *args, **kwargs - ) -> None: - super().__init__(*args, **kwargs) - self.db = db - self.operation = operation - self.username = username - self.on_close = on_close - - async def on_mount(self) -> None: - """Ensure the first button is focused.""" - reset_button = self.query_one("#cancel") - self.set_focus(reset_button) - - def compose(self) -> ComposeResult: - with Container(classes="modal-box-delete"): - yield Static("Are you sure about resetting this admin usage?", classes="title") - yield Horizontal( - Button("Reset", id="reset", variant="success"), - Button("Cancel", id="cancel", variant="error"), - classes="button-container", - ) - async def on_button_pressed(self, event: Button.Pressed) -> None: - if event.button.id == "reset": - try: - await self.operation.reset_admin_usage(self.db, self.username, SYSTEM_ADMIN) - self.notify("Admin usage reseted successfully", severity="success", title="Success") - self.on_close() - except ValueError as e: - self.notify(str(e), severity="error", title="Error") - await self.key_escape() - - -class AdminCreateModale(BaseModal): - def __init__(self, db: AsyncSession, operation: AdminOperation, on_close: callable, *args, **kwargs) -> None: - super().__init__(*args, **kwargs) - self.db = db - self.operation = operation - self.on_close = on_close - - def compose(self) -> ComposeResult: - with Container(classes="modal-box-form"): - yield Static("Create a new admin", classes="title") - yield Vertical( - Input(placeholder="Username", id="username"), - Input(placeholder="Password", password=True, id="password"), - Input(placeholder="Confirm Password", password=True, id="confirm_password"), - Input(placeholder="Telegram ID", id="telegram_id", type="integer"), - Input(placeholder="Discord ID", id="discord_id", type="integer"), - Input(placeholder="Discord Webhook", id="discord_webhook"), - Horizontal( - Static("Is sudo: ", classes="label"), - Switch(animate=False, id="is_sudo"), - classes="switch-container", - ), - classes="input-container", - ) - yield Horizontal( - Button("Create", id="create", variant="success"), - Button("Cancel", id="cancel", variant="error"), - classes="button-container", +class AdminCLI(BaseCLI): + """Admin CLI operations.""" + + async def list_admins(self, db): + """List all admin accounts.""" + admin_op = get_admin_operation() + admins = await admin_op.get_admins(db) + + if not admins: + self.console.print("[yellow]No admins found[/yellow]") + return + + table = self.create_table( + "Admin Accounts", + [ + {"name": "Username", "style": "cyan"}, + {"name": "Is Sudo", "style": "green"}, + {"name": "Used Traffic", "style": "blue"}, + {"name": "Is Disabled", "style": "red"}, + ], + ) + + for admin in admins: + table.add_row( + admin.username, + "✓" if admin.is_sudo else "✗", + readable_size(admin.used_traffic), + "✓" if admin.is_disabled else "✗", ) - async def on_mount(self) -> None: - """Ensure the first button is focused.""" - username_input = self.query_one("#username") - self.set_focus(username_input) - - async def key_enter(self) -> None: - """Create admin when Enter is pressed.""" - if not self.query_one("#is_sudo").has_focus and not self.query_one("#cancel").has_focus: - await self.on_button_pressed(Button.Pressed(self.query_one("#create"))) - - async def on_button_pressed(self, event: Button.Pressed) -> None: - if event.button.id == "create": - username = self.query_one("#username").value.strip() - password = self.query_one("#password").value.strip() - confirm_password = self.query_one("#confirm_password").value.strip() - telegram_id = self.query_one("#telegram_id").value or None - discord_webhook = self.query_one("#discord_webhook").value.strip() or None - discord_id = self.query_one("#discord_id").value or None - is_sudo = self.query_one("#is_sudo").value + self.console.print(table) + + async def create_admin(self, db, username: str): + """Create a new admin account.""" + admin_op = get_admin_operation() + + # Check if admin already exists + admins = await admin_op.get_admins(db) + if any(admin.username == username for admin in admins): + self.console.print(f"[red]Admin '{username}' already exists[/red]") + return + + while True: + # Get password + password = typer.prompt("Password", hide_input=True) + if not password: + self.console.print("[red]Password is required[/red]") + continue + + confirm_password = typer.prompt("Confirm Password", hide_input=True) if password != confirm_password: - self.notify("Password and confirm password do not match", severity="error", title="Error") - return + self.console.print("[red]Passwords do not match[/red]") + continue + try: - await self.operation.create_admin( - self.db, - AdminCreate( - username=username, - password=password, - telegram_id=telegram_id, - discord_webhook=discord_webhook, - discord_id=discord_id, - is_sudo=is_sudo, - ), - SYSTEM_ADMIN, - ) - self.notify("Admin created successfully", severity="success", title="Success") - await self.key_escape() - self.on_close() + # Create admin + new_admin = AdminCreate(username=username, password=password, is_sudo=False) + await admin_op.create_admin(db, new_admin, SYSTEM_ADMIN) + self.console.print(f"[green]Admin '{username}' created successfully[/green]") + break except ValidationError as e: - format_cli_validation_error(e, self.notify) - except ValueError as e: - self.notify(str(e), severity="error", title="Error") - elif event.button.id == "cancel": - await self.key_escape() - - -class AdminModifyModale(BaseModal): - def __init__( - self, db: AsyncSession, operation: AdminOperation, admin: Admin, on_close: callable, *args, **kwargs - ) -> None: - super().__init__(*args, **kwargs) - self.db = db - self.operation = operation - self.admin = admin - self.on_close = on_close - - def compose(self) -> ComposeResult: - with Container(classes="modal-box-form"): - yield Static("Modify admin", classes="title") - yield Vertical( - Input(placeholder="Username", id="username", disabled=True), - Input(placeholder="Password", password=True, id="password"), - Input(placeholder="Confirm Password", password=True, id="confirm_password"), - Input(placeholder="Telegram ID", id="telegram_id", type="integer"), - Input(placeholder="Discord ID", id="discord_id", type="integer"), - Input(placeholder="Discord Webhook", id="discord_webhook"), - Horizontal( - Static("Is sudo: ", classes="label"), - Switch(animate=False, id="is_sudo"), - Static("Is disabled: ", classes="label"), - Switch(animate=False, id="is_disabled"), - classes="switch-container", - ), - classes="input-container", - ) - yield Horizontal( - Button("Save", id="save", variant="success"), - Button("Cancel", id="cancel", variant="error"), - classes="button-container", - ) + self.format_cli_validation_error(e) + continue + except Exception as e: + self.console.print(f"[red]Error creating admin: {e}[/red]") + break + + async def delete_admin(self, db, username: str): + """Delete an admin account.""" + admin_op = get_admin_operation() + + # Check if admin exists + admins = await admin_op.get_admins(db) + if not any(admin.username == username for admin in admins): + self.console.print(f"[red]Admin '{username}' not found[/red]") + return - async def on_mount(self) -> None: - self.query_one("#username").value = self.admin.username - if self.admin.telegram_id: - self.query_one("#telegram_id").value = str(self.admin.telegram_id) - if self.admin.discord_webhook: - self.query_one("#discord_webhook").value = self.admin.discord_webhook - self.query_one("#is_sudo").value = self.admin.is_sudo - self.query_one("#is_disabled").value = self.admin.is_disabled - password_input = self.query_one("#password") - self.set_focus(password_input) - - async def key_enter(self) -> None: - """Save admin when Enter is pressed.""" - if ( - not self.query_one("#is_disabled").has_focus - and not self.query_one("#is_sudo").has_focus - and not self.query_one("#cancel").has_focus - ): - await self.on_button_pressed(Button.Pressed(self.query_one("#save"))) - - async def on_button_pressed(self, event: Button.Pressed) -> None: - if event.button.id == "save": - password = self.query_one("#password").value.strip() or None - confirm_password = self.query_one("#confirm_password").value.strip() or None - telegram_id = self.query_one("#telegram_id").value or None - discord_webhook = self.query_one("#discord_webhook").value.strip() or None - discord_id = self.query_one("#discord_id").value or None - is_sudo = self.query_one("#is_sudo").value - is_disabled = self.query_one("#is_disabled").value + if typer.confirm(f"Are you sure you want to delete admin '{username}'?"): + try: + await admin_op.remove_admin(db, username, SYSTEM_ADMIN) + self.console.print(f"[green]Admin '{username}' deleted successfully[/green]") + except Exception as e: + self.console.print(f"[red]Error deleting admin: {e}[/red]") + + async def modify_admin(self, db, username: str): + """Modify an admin account.""" + admin_op = get_admin_operation() + + # Check if admin exists + admins = await admin_op.get_admins(db) + if not any(admin.username == username for admin in admins): + self.console.print(f"[red]Admin '{username}' not found[/red]") + return - if password != confirm_password: - self.notify("Password and confirm password do not match", severity="error", title="Error") + # Get the current admin details + current_admin = next(admin for admin in admins if admin.username == username) + + self.console.print(f"[yellow]Modifying admin '{username}'[/yellow]") + self.console.print("[cyan]Current settings:[/cyan]") + self.console.print(f" Username: {current_admin.username}") + self.console.print(f" Is Sudo: {'✓' if current_admin.is_sudo else '✗'}") + + new_password = None + is_sudo = current_admin.is_sudo + # Password modification + if typer.confirm("Do you want to change the password?"): + new_password = typer.prompt("New password", hide_input=True) + confirm_password = typer.prompt("Confirm Password", hide_input=True) + if new_password != confirm_password: + self.console.print("[red]Passwords do not match[/red]") return + + # Sudo status modification + if typer.confirm(f"Do you want to change sudo status? (Current: {'✓' if current_admin.is_sudo else '✗'})"): + is_sudo = typer.confirm("Make this admin a sudo admin?") + + # Confirm changes + self.console.print("\n[cyan]Summary of changes:[/cyan]") + if new_password: + self.console.print(" Password: [yellow]Will be updated[/yellow]") + if is_sudo != current_admin.is_sudo: + self.console.print(f" Is Sudo: {'✓' if is_sudo else '✗'} [yellow](changed)[/yellow]") + + if typer.confirm("Do you want to apply these changes?"): try: - await self.operation.modify_admin( - self.db, - self.admin.username, - AdminModify( - password=password, - telegram_id=telegram_id, - discord_webhook=discord_webhook, - discord_id=discord_id, - is_sudo=is_sudo, - is_disabled=is_disabled, - ), - SYSTEM_ADMIN, - ) - self.notify("Admin modified successfully", severity="success", title="Success") - await self.key_escape() - self.on_close() - except ValidationError as e: - format_cli_validation_error(e, self.notify) - except ValueError as e: - self.notify(str(e), severity="error", title="Error") - elif event.button.id == "cancel": - await self.key_escape() - - -class AdminContent(Static): - def __init__(self, *args, **kwargs) -> None: - super().__init__(*args, **kwargs) - self.db: AsyncSession = None - self.admin_operator = AdminOperation(OperatorType.CLI) - self.table: DataTable = None - self.no_admins: Static = None - self.current_page = 1 - self.page_size = 10 - self.total_admins = 0 - - BINDINGS = [ - ("c", "create_admin", "Create admin"), - ("m", "modify_admin", "Modify admin"), - ("r", "reset_admin_usage", "Reset admin usage"), - ("d", "delete_admin", "Delete admin"), - ("i", "import_from_env", "Import from env"), - ("p", "previous_page", "Previous page"), - ("n", "next_page", "Next page"), - ] - - def compose(self) -> ComposeResult: - yield DataTable(id="admin-list") - yield Static( - "No admin found\n\nCreate an admin by pressing 'c'\n\nhelp by pressing '?'", - classes="title box", - id="no-admins", - ) - yield Static("", id="pagination-info", classes="pagination-info") - - async def on_mount(self) -> None: - self.db = await anext(get_db()) - self.table = self.query_one("#admin-list") - self.no_admins = self.query_one("#no-admins") - self.pagination_info = self.query_one("#pagination-info") - self.no_admins.styles.display = "none" - self.table.styles.display = "none" - self.table.cursor_type = "row" - self.table.styles.text_align = "center" - await self.admins_list() - - def _center_text(self, text, width): - padding = width - len(text) - left_padding = padding // 2 - right_padding = padding - left_padding - return " " * left_padding + text + " " * right_padding - - async def admins_list(self): - self.table.clear() - self.table.columns.clear() - columns = ( - "Username", - "Used Traffic", - "Lifetime Used Traffic", - "Users Usage", - "Is sudo", - "Is disabled", - "Created at", - "Telegram ID", - "Discord ID", - "Discord Webhook", - ) - self.total_admins = await self.admin_operator.get_admins_count(self.db) - offset = (self.current_page - 1) * self.page_size - limit = self.page_size - admins = await self.admin_operator.get_admins(self.db, offset=offset, limit=limit) - if not admins: - self.no_admins.styles.display = "block" - self.pagination_info.update("") - return + # Interactive modification + modified_admin = AdminModify(is_sudo=is_sudo, password=new_password) + await admin_op.modify_admin(db, username, modified_admin, SYSTEM_ADMIN) + self.console.print(f"[green]Admin '{username}' modified successfully[/green]") + except Exception as e: + self.console.print(f"[red]Error modifying admin: {e}[/red]") else: - self.no_admins.styles.display = "none" - self.table.styles.display = "block" - users_usages = await asyncio.gather(*[self.calculate_admin_usage(admin.id) for admin in admins]) + self.console.print("[yellow]Modification cancelled[/yellow]") - admins_data = [ - ( - admin.username, - readable_size(admin.used_traffic), - readable_size(admin.lifetime_used_traffic), - users_usages[i], - "✔️" if admin.is_sudo else "✖️", - "✔️" if admin.is_disabled else "✖️", - readable_datetime(admin.created_at), - str(admin.telegram_id or "✖️"), - str(admin.discord_id or "✖️"), - str(admin.discord_webhook or "✖️"), - ) - for i, admin in enumerate(admins) - ] - column_widths = [ - max(len(str(columns[i])), max(len(str(row[i])) for row in admins_data)) for i in range(len(columns)) - ] - - centered_columns = [self._center_text(column, column_widths[i]) for i, column in enumerate(columns)] - self.table.add_columns(*centered_columns) - i = 1 - for row, adnin in zip(admins_data, admins): - centered_row = [self._center_text(str(cell), column_widths[i]) for i, cell in enumerate(row)] - label = Text(f"{i + offset}") - i += 1 - self.table.add_row(*centered_row, key=adnin.username, label=label) - - total_pages = (self.total_admins + self.page_size - 1) // self.page_size - self.pagination_info.update(f"Page {self.current_page}/{total_pages} (Total admins: {self.total_admins})\nPress `n` for go to the next page and `p` to back to previose page") - - @property - def selected_admin(self): - return self.table.coordinate_to_cell_key(Coordinate(self.table.cursor_row, 0)).row_key.value - - async def action_delete_admin(self): - if not self.table.columns: + async def reset_admin_usage(self, db, username: str): + """Reset admin usage statistics.""" + admin_op = get_admin_operation() + + # Check if admin exists + admins = await admin_op.get_admins(db) + if not any(admin.username == username for admin in admins): + self.console.print(f"[red]Admin '{username}' not found[/red]") return - self.app.push_screen(AdminDelete(self.db, self.admin_operator, self.selected_admin, self._refresh_table)) - def _refresh_table(self): - self.run_worker(self.admins_list) + if typer.confirm(f"Are you sure you want to reset usage for admin '{username}'?"): + try: + await admin_op.reset_admin_usage(db, username, SYSTEM_ADMIN) + self.console.print(f"[green]Usage reset for admin '{username}'[/green]") + except Exception as e: + self.console.print(f"[red]Error resetting usage: {e}[/red]") - async def action_create_admin(self): - self.app.push_screen(AdminCreateModale(self.db, self.admin_operator, self._refresh_table)) - async def action_modify_admin(self): - if not self.table.columns: - return - admin = await self.admin_operator.get_validated_admin(self.db, username=self.selected_admin) - self.app.push_screen(AdminModifyModale(self.db, self.admin_operator, admin, self._refresh_table)) +# CLI commands +async def list_admins(): + """List all admin accounts.""" + admin_cli = AdminCLI() + async for db in get_db(): + try: + await admin_cli.list_admins(db) + except Exception as e: + console.print(f"[red]Error: {e}[/red]") + finally: + break + - async def action_import_from_env(self): +async def create_admin(username: str): + """Create a new admin account.""" + admin_cli = AdminCLI() + async for db in get_db(): try: - username, password = config("SUDO_USERNAME"), config("SUDO_PASSWORD") - except UndefinedValueError: - self.notify( - "Unable to get SUDO_USERNAME and/or SUDO_PASSWORD.\n" - "Make sure you have set them in the env file or as environment variables.", - severity="error", - title="Error", - ) - return - if not (username and password): - self.notify( - "Unable to retrieve username and password.\nMake sure both SUDO_USERNAME and SUDO_PASSWORD are set.", - severity="error", - title="Error", - ) - return + await admin_cli.create_admin(db, username) + except Exception as e: + console.print(f"[red]Error: {e}[/red]") + finally: + break + + +async def delete_admin(username: str): + """Delete an admin account.""" + admin_cli = AdminCLI() + async for db in get_db(): try: - await self.admin_operator.create_admin( - self.db, - AdminCreate(username=username, password=password, is_sudo=True), - SYSTEM_ADMIN, - ) - self.notify("Admin created successfully", severity="success", title="Success") - self._refresh_table() - except ValidationError as e: - format_cli_validation_error(e, self.notify) - except ValueError as e: - self.notify(str(e), severity="error", title="Error") - - async def action_reset_admin_usage(self): - if not self.table.columns: - return - self.app.push_screen(AdminResetUsage(self.db, self.admin_operator, self.selected_admin, self._refresh_table)) - - async def action_previous_page(self): - if self.current_page > 1: - self.current_page -= 1 - await self.admins_list() - - async def action_next_page(self): - total_pages = (self.total_admins + self.page_size - 1) // self.page_size - if self.current_page < total_pages: - self.current_page += 1 - await self.admins_list() - - async def calculate_admin_usage(self, admin_id: int) -> str: - usage = await self.db.execute(select(func.sum(User.used_traffic)).filter_by(admin_id=admin_id)) - return readable_size(int(usage.scalar() or 0)) - - async def key_enter(self) -> None: - if self.table.columns: - await self.action_modify_admin() - - async def on_prune(self, event): - await self.db.close() - return await super().on_prune(event) + await admin_cli.delete_admin(db, username) + except Exception as e: + console.print(f"[red]Error: {e}[/red]") + finally: + break + + +async def modify_admin(username: str): + """Modify an admin account.""" + admin_cli = AdminCLI() + async for db in get_db(): + try: + await admin_cli.modify_admin(db, username) + except Exception as e: + console.print(f"[red]Error: {e}[/red]") + finally: + break + + +async def reset_admin_usage(username: str): + """Reset admin usage statistics.""" + admin_cli = AdminCLI() + async for db in get_db(): + try: + await admin_cli.reset_admin_usage(db, username) + except Exception as e: + console.print(f"[red]Error: {e}[/red]") + finally: + break diff --git a/cli/main.py b/cli/main.py new file mode 100644 index 000000000..cbd3fdb79 --- /dev/null +++ b/cli/main.py @@ -0,0 +1,63 @@ +#!/usr/bin/env python3 +""" +PasarGuard CLI - Command Line Interface for PasarGuard Management + +A modern, type-safe CLI built with Typer for managing PasarGuard instances. +""" + +import asyncio +from typing import Optional + +import typer + +from cli import console +from cli.admin import create_admin, delete_admin, list_admins, modify_admin, reset_admin_usage +from cli.system import show_status + +# Initialize Typer app +app = typer.Typer( + name="PasarGuard", + help="PasarGuard CLI - Command Line Interface for PasarGuard Management", + add_completion=False, + rich_markup_mode="rich", +) + + +@app.command() +def version(): + """Show PasarGuard version.""" + from app import __version__ + + console.print(f"[bold blue]PasarGuard[/bold blue] version [bold green]{__version__}[/bold green]") + + +@app.command() +def admins( + list: bool = typer.Option(False, "--list", "-l", help="List all admins"), + create: Optional[str] = typer.Option(None, "--create", "-c", help="Create new admin"), + delete: Optional[str] = typer.Option(None, "--delete", "-d", help="Delete admin"), + modify: Optional[str] = typer.Option(None, "--modify", "-m", help="Modify admin"), + reset_usage: Optional[str] = typer.Option(None, "--reset-usage", "-r", help="Reset admin usage"), +): + """List & manage admin accounts.""" + + if list or not any([create, delete, modify, reset_usage]): + asyncio.run(list_admins()) + elif create: + asyncio.run(create_admin(create)) + elif delete: + asyncio.run(delete_admin(delete)) + elif modify: + asyncio.run(modify_admin(modify)) + elif reset_usage: + asyncio.run(reset_admin_usage(reset_usage)) + + +@app.command() +def system(): + """Show system status.""" + asyncio.run(show_status()) + + +if __name__ == "__main__": + app() diff --git a/cli/system.py b/cli/system.py new file mode 100644 index 000000000..4399704f7 --- /dev/null +++ b/cli/system.py @@ -0,0 +1,58 @@ +""" +System CLI Module + +Handles system status and information through the command line interface. +""" + +from rich.panel import Panel + +from app.db.base import get_db +from app.utils.system import readable_size +from cli import SYSTEM_ADMIN, BaseCLI, console, get_system_operation + + +class SystemCLI(BaseCLI): + """System CLI operations.""" + + async def show_status(self, db): + """Show system status.""" + system_op = get_system_operation() + stats = await system_op.get_system_stats(db, SYSTEM_ADMIN) + + status_text = ( + f"[bold]System Statistics[/bold]\n\n" + f"CPU Usage: [green]{stats.cpu_usage:.1f}%[/green]\n" + f"Memory Usage: [green]{stats.mem_used / stats.mem_total * 100:.1f}%[/green] " + f"([cyan]{readable_size(stats.mem_used)}[/cyan] / [cyan]{readable_size(stats.mem_total)}[/cyan])\n" + f"CPU Cores: [magenta]{stats.cpu_cores}[/magenta]\n" + f"Total Users: [blue]{stats.total_user}[/blue]\n" + f"Active Users: [green]{stats.active_users}[/green]\n" + f"Online Users: [yellow]{stats.online_users}[/yellow]\n" + f"On Hold Users: [yellow]{stats.on_hold_users}[/yellow]\n" + f"Disabled Users: [red]{stats.disabled_users}[/red]\n" + f"Expired Users: [red]{stats.expired_users}[/red]\n" + f"Limited Users: [yellow]{stats.limited_users}[/yellow]\n" + f"Data Usage (In): [blue]{readable_size(stats.incoming_bandwidth)}[/blue]\n" + f"Data Usage (Out): [blue]{readable_size(stats.outgoing_bandwidth)}[/blue]" + ) + + panel = Panel( + status_text, + title="System Information", + border_style="blue", + ) + + self.console.print(panel) + + +# CLI commands +async def show_status(): + """Show system status.""" + system_cli = SystemCLI() + async for db in get_db(): + try: + await system_cli.show_status(db) + except Exception as e: + console.print(f"[red]Error: {e}[/red]") + finally: + break diff --git a/pasarguard-cli.py b/pasarguard-cli.py index a0859ae61..d54fdb7d6 100644 --- a/pasarguard-cli.py +++ b/pasarguard-cli.py @@ -1,53 +1,24 @@ -#! /usr/bin/env python3 -from textual.app import App, ComposeResult -from textual.widgets import Footer, Header - -from cli.help import HelpModal -from config import DEBUG - - -class PasarGuardCLI(App): - """A Textual app to manage pasarguard""" - - CSS_PATH = "cli/style.tcss" - ENABLE_COMMAND_PALETTE = DEBUG - - def __init__(self, *args, **kwargs) -> None: - super().__init__(*args, **kwargs) - self.theme = "textual-dark" - - BINDINGS = [ - ("ctrl+c", "quit", "Quit"), - ("q", "quit", "Quit"), - ("?", "help", "Help"), - ] - - def compose(self) -> ComposeResult: - """Create child widgets for the app.""" - from cli.admin import AdminContent - - yield Header() - yield AdminContent(id="admin-content") - yield Footer() - - def on_mount(self) -> None: - """Called when the app is mounted.""" - self.action_show_admins() - - def action_show_admins(self) -> None: - """Show the admins section.""" - self.query_one("#admin-content") - - async def action_quit(self) -> None: - """An action to quit the app.""" - self.exit() - - def action_help(self) -> None: - """Show help information in a modal.""" - admin_content = self.query_one("#admin-content") - self.push_screen(HelpModal(self.BINDINGS, admin_content.BINDINGS)) - - -if __name__ == "__main__": - app = PasarGuardCLI() - app.run() +#!/usr/bin/env python3 +""" +PasarGuard CLI Wrapper +This script provides a simple entry point for the PasarGuard CLI. +""" + +import sys +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).parent)) + +try: + from cli.main import app + + if len(sys.argv) == 1: + sys.argv.append("--help") + app() +except ImportError as e: + print(f"Error importing CLI: {e}") + print("Make sure you're running this from the PasarGuard project directory.") + sys.exit(1) +except Exception as e: + print(f"Error running CLI: {e}") + sys.exit(1) diff --git a/pasarguard-tui.py b/pasarguard-tui.py new file mode 100644 index 000000000..a5fab073b --- /dev/null +++ b/pasarguard-tui.py @@ -0,0 +1,53 @@ +#! /usr/bin/env python3 +from textual.app import App, ComposeResult +from textual.widgets import Footer, Header + +from config import DEBUG +from tui.help import HelpModal + + +class PasarGuardTUI(App): + """A Textual app to manage pasarguard""" + + CSS_PATH = "tui/style.tcss" + ENABLE_COMMAND_PALETTE = DEBUG + + def __init__(self, *args, **kwargs) -> None: + super().__init__(*args, **kwargs) + self.theme = "textual-dark" + + BINDINGS = [ + ("ctrl+c", "quit", "Quit"), + ("q", "quit", "Quit"), + ("?", "help", "Help"), + ] + + def compose(self) -> ComposeResult: + """Create child widgets for the app.""" + from tui.admin import AdminContent + + yield Header() + yield AdminContent(id="admin-content") + yield Footer() + + def on_mount(self) -> None: + """Called when the app is mounted.""" + self.action_show_admins() + + def action_show_admins(self) -> None: + """Show the admins section.""" + self.query_one("#admin-content") + + async def action_quit(self) -> None: + """An action to quit the app.""" + self.exit() + + def action_help(self) -> None: + """Show help information in a modal.""" + admin_content = self.query_one("#admin-content") + self.push_screen(HelpModal(self.BINDINGS, admin_content.BINDINGS)) + + +if __name__ == "__main__": + app = PasarGuardTUI() + app.run() diff --git a/pyproject.toml b/pyproject.toml index c08b1d20e..d14220056 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -46,6 +46,7 @@ dependencies = [ "aiorwlock>=1.5.0", "gozargah-node-bridge>=0.0.44", "aiomonitor>=0.7.1", + "typer>=0.17.3", ] [tool.ruff] diff --git a/tui/README.md b/tui/README.md new file mode 100644 index 000000000..55dd11307 --- /dev/null +++ b/tui/README.md @@ -0,0 +1,42 @@ +# PasarGuard TUI + +A modern, interactive command-line interface for managing PasarGuard, built with Textual. + +## Features + +- 🎯 Interactive TUI (Text User Interface) +- 📱 Responsive design with dark mode support +- ⌨️ Keyboard shortcuts for quick navigation +- 🔄 Real-time updates +- 📊 Rich data visualization +- 🔒 Secure admin management + +## Usage + +### Starting the TUI + +```bash +pasarguard tui +``` + +### Keyboard Shortcuts + +#### Global Commands + +- `q` - Quit the application +- `?` - Show help + +#### Admin Section + +- `c` - Create new admin +- `m` - Modify admin +- `r` - Reset admin usage +- `d` - Delete admin +- `i` - Import admins from environment + +### Admin Management + +- Create, modify, and delete admin accounts +- Reset admin usage statistics +- Import admins from environment variables +- View admin details and status diff --git a/tui/__init__.py b/tui/__init__.py new file mode 100644 index 000000000..6ca458cdd --- /dev/null +++ b/tui/__init__.py @@ -0,0 +1,29 @@ +from textual.screen import ModalScreen +from textual.widgets import Input + + +class BaseModal(ModalScreen): + def __init__(self, *args, **kwargs): + super().__init__(*args, **kwargs) + + async def key_left(self) -> None: + """Move focus left on arrow key press.""" + if not self.query_one(Input).has_focus: + self.app.action_focus_previous() + + async def key_right(self) -> None: + """Move focus right on arrow key press.""" + if not self.query_one(Input).has_focus: + self.app.action_focus_next() + + async def key_down(self) -> None: + """Move focus down on arrow key press.""" + self.app.action_focus_next() + + async def key_up(self) -> None: + """Move focus up on arrow key press.""" + self.app.action_focus_previous() + + async def key_escape(self) -> None: + """Close modal when ESC is pressed.""" + self.dismiss() diff --git a/tui/admin.py b/tui/admin.py new file mode 100644 index 000000000..04b3d1732 --- /dev/null +++ b/tui/admin.py @@ -0,0 +1,481 @@ +import asyncio + +from decouple import UndefinedValueError, config +from pydantic import ValidationError +from rich.text import Text +from sqlalchemy import func, select +from textual.app import ComposeResult +from textual.containers import Container, Horizontal, Vertical +from textual.coordinate import Coordinate +from textual.widgets import Button, DataTable, Input, Static, Switch + +from app.db import AsyncSession +from app.db.base import get_db +from app.db.models import Admin, User +from app.models.admin import AdminCreate, AdminDetails, AdminModify +from app.operation import OperatorType +from app.operation.admin import AdminOperation +from app.utils.helpers import readable_datetime +from app.utils.system import readable_size +from tui import BaseModal + +SYSTEM_ADMIN = AdminDetails(username="tui", is_sudo=True, telegram_id=None, discord_webhook=None) + + +class AdminDelete(BaseModal): + def __init__( + self, db: AsyncSession, operation: AdminOperation, username: str, on_close: callable, *args, **kwargs + ) -> None: + super().__init__(*args, **kwargs) + self.db = db + self.operation = operation + self.username = username + self.on_close = on_close + + async def on_mount(self) -> None: + """Ensure the first button is focused.""" + yes_button = self.query_one("#no") + self.set_focus(yes_button) + + def compose(self) -> ComposeResult: + with Container(classes="modal-box-delete"): + yield Static("Are you sure about deleting this admin?", classes="title") + yield Horizontal( + Button("Yes", id="yes", variant="success"), + Button("No", id="no", variant="error"), + classes="button-container", + ) + + async def on_button_pressed(self, event: Button.Pressed) -> None: + if event.button.id == "yes": + try: + await self.operation.remove_admin(self.db, self.username, SYSTEM_ADMIN) + self.on_close() + except ValueError as e: + self.notify(str(e), severity="error", title="Error") + await self.key_escape() + + +class AdminResetUsage(BaseModal): + def __init__( + self, db: AsyncSession, operation: AdminOperation, username: str, on_close: callable, *args, **kwargs + ) -> None: + super().__init__(*args, **kwargs) + self.db = db + self.operation = operation + self.username = username + self.on_close = on_close + + async def on_mount(self) -> None: + """Ensure the first button is focused.""" + reset_button = self.query_one("#cancel") + self.set_focus(reset_button) + + def compose(self) -> ComposeResult: + with Container(classes="modal-box-delete"): + yield Static("Are you sure about resetting this admin usage?", classes="title") + yield Horizontal( + Button("Reset", id="reset", variant="success"), + Button("Cancel", id="cancel", variant="error"), + classes="button-container", + ) + + async def on_button_pressed(self, event: Button.Pressed) -> None: + if event.button.id == "reset": + try: + await self.operation.reset_admin_usage(self.db, self.username, SYSTEM_ADMIN) + self.notify("Admin usage reseted successfully", severity="success", title="Success") + self.on_close() + except ValueError as e: + self.notify(str(e), severity="error", title="Error") + await self.key_escape() + + +class AdminCreateModale(BaseModal): + def __init__( + self, + db: AsyncSession, + operation: AdminOperation, + on_close: callable, + format_tui_validation_error: callable, + *args, + **kwargs, + ) -> None: + super().__init__(*args, **kwargs) + self.db = db + self.operation = operation + self.on_close = on_close + self.format_tui_validation_error = format_tui_validation_error + + def compose(self) -> ComposeResult: + with Container(classes="modal-box-form"): + yield Static("Create a new admin", classes="title") + yield Vertical( + Input(placeholder="Username", id="username"), + Input(placeholder="Password", password=True, id="password"), + Input(placeholder="Confirm Password", password=True, id="confirm_password"), + Input(placeholder="Telegram ID", id="telegram_id", type="integer"), + Input(placeholder="Discord ID", id="discord_id", type="integer"), + Input(placeholder="Discord Webhook", id="discord_webhook"), + Horizontal( + Static("Is sudo: ", classes="label"), + Switch(animate=False, id="is_sudo"), + classes="switch-container", + ), + classes="input-container", + ) + yield Horizontal( + Button("Create", id="create", variant="success"), + Button("Cancel", id="cancel", variant="error"), + classes="button-container", + ) + + async def on_mount(self) -> None: + """Ensure the first button is focused.""" + username_input = self.query_one("#username") + self.set_focus(username_input) + + async def key_enter(self) -> None: + """Create admin when Enter is pressed.""" + if not self.query_one("#is_sudo").has_focus and not self.query_one("#cancel").has_focus: + await self.on_button_pressed(Button.Pressed(self.query_one("#create"))) + + async def on_button_pressed(self, event: Button.Pressed) -> None: + if event.button.id == "create": + username = self.query_one("#username").value.strip() + password = self.query_one("#password").value.strip() + confirm_password = self.query_one("#confirm_password").value.strip() + telegram_id = self.query_one("#telegram_id").value or None + discord_webhook = self.query_one("#discord_webhook").value.strip() or None + discord_id = self.query_one("#discord_id").value or None + is_sudo = self.query_one("#is_sudo").value + if password != confirm_password: + self.notify("Password and confirm password do not match", severity="error", title="Error") + return + try: + await self.operation.create_admin( + self.db, + AdminCreate( + username=username, + password=password, + telegram_id=telegram_id, + discord_webhook=discord_webhook, + discord_id=discord_id, + is_sudo=is_sudo, + ), + SYSTEM_ADMIN, + ) + self.notify("Admin created successfully", severity="success", title="Success") + await self.key_escape() + self.on_close() + except ValidationError as e: + self.format_tui_validation_error(e) + except ValueError as e: + self.notify(str(e), severity="error", title="Error") + elif event.button.id == "cancel": + await self.key_escape() + + +class AdminModifyModale(BaseModal): + def __init__( + self, + db: AsyncSession, + operation: AdminOperation, + admin: Admin, + on_close: callable, + format_tui_validation_error: callable, + *args, + **kwargs, + ) -> None: + super().__init__(*args, **kwargs) + self.db = db + self.operation = operation + self.admin = admin + self.on_close = on_close + self.format_tui_validation_error = format_tui_validation_error + + def compose(self) -> ComposeResult: + with Container(classes="modal-box-form"): + yield Static("Modify admin", classes="title") + yield Vertical( + Input(placeholder="Username", id="username", disabled=True), + Input(placeholder="Password", password=True, id="password"), + Input(placeholder="Confirm Password", password=True, id="confirm_password"), + Input(placeholder="Telegram ID", id="telegram_id", type="integer"), + Input(placeholder="Discord ID", id="discord_id", type="integer"), + Input(placeholder="Discord Webhook", id="discord_webhook"), + Horizontal( + Static("Is sudo: ", classes="label"), + Switch(animate=False, id="is_sudo"), + Static("Is disabled: ", classes="label"), + Switch(animate=False, id="is_disabled"), + classes="switch-container", + ), + classes="input-container", + ) + yield Horizontal( + Button("Save", id="save", variant="success"), + Button("Cancel", id="cancel", variant="error"), + classes="button-container", + ) + + async def on_mount(self) -> None: + self.query_one("#username").value = self.admin.username + if self.admin.telegram_id: + self.query_one("#telegram_id").value = str(self.admin.telegram_id) + if self.admin.discord_webhook: + self.query_one("#discord_webhook").value = self.admin.discord_webhook + self.query_one("#is_sudo").value = self.admin.is_sudo + self.query_one("#is_disabled").value = self.admin.is_disabled + password_input = self.query_one("#password") + self.set_focus(password_input) + + async def key_enter(self) -> None: + """Save admin when Enter is pressed.""" + if ( + not self.query_one("#is_disabled").has_focus + and not self.query_one("#is_sudo").has_focus + and not self.query_one("#cancel").has_focus + ): + await self.on_button_pressed(Button.Pressed(self.query_one("#save"))) + + async def on_button_pressed(self, event: Button.Pressed) -> None: + if event.button.id == "save": + password = self.query_one("#password").value.strip() or None + confirm_password = self.query_one("#confirm_password").value.strip() or None + telegram_id = self.query_one("#telegram_id").value or None + discord_webhook = self.query_one("#discord_webhook").value.strip() or None + discord_id = self.query_one("#discord_id").value or None + is_sudo = self.query_one("#is_sudo").value + is_disabled = self.query_one("#is_disabled").value + + if password != confirm_password: + self.notify("Password and confirm password do not match", severity="error", title="Error") + return + try: + await self.operation.modify_admin( + self.db, + self.admin.username, + AdminModify( + password=password, + telegram_id=telegram_id, + discord_webhook=discord_webhook, + discord_id=discord_id, + is_sudo=is_sudo, + is_disabled=is_disabled, + ), + SYSTEM_ADMIN, + ) + self.notify("Admin modified successfully", severity="success", title="Success") + await self.key_escape() + self.on_close() + except ValidationError as e: + self.format_tui_validation_error(e) + except ValueError as e: + self.notify(str(e), severity="error", title="Error") + elif event.button.id == "cancel": + await self.key_escape() + + +class AdminContent(Static): + def __init__(self, *args, **kwargs) -> None: + super().__init__(*args, **kwargs) + self.db: AsyncSession = None + self.admin_operator = AdminOperation(OperatorType.CLI) + self.table: DataTable = None + self.no_admins: Static = None + self.current_page = 1 + self.page_size = 10 + self.total_admins = 0 + + BINDINGS = [ + ("c", "create_admin", "Create admin"), + ("m", "modify_admin", "Modify admin"), + ("r", "reset_admin_usage", "Reset admin usage"), + ("d", "delete_admin", "Delete admin"), + ("i", "import_from_env", "Import from env"), + ("p", "previous_page", "Previous page"), + ("n", "next_page", "Next page"), + ] + + def compose(self) -> ComposeResult: + yield DataTable(id="admin-list") + yield Static( + "No admin found\n\nCreate an admin by pressing 'c'\n\nhelp by pressing '?'", + classes="title box", + id="no-admins", + ) + yield Static("", id="pagination-info", classes="pagination-info") + + async def on_mount(self) -> None: + self.db = await anext(get_db()) + self.table = self.query_one("#admin-list") + self.no_admins = self.query_one("#no-admins") + self.pagination_info = self.query_one("#pagination-info") + self.no_admins.styles.display = "none" + self.table.styles.display = "none" + self.table.cursor_type = "row" + self.table.styles.text_align = "center" + await self.admins_list() + + def _center_text(self, text, width): + padding = width - len(text) + left_padding = padding // 2 + right_padding = padding - left_padding + return " " * left_padding + text + " " * right_padding + + async def admins_list(self): + self.table.clear() + self.table.columns.clear() + columns = ( + "Username", + "Used Traffic", + "Lifetime Used Traffic", + "Users Usage", + "Is sudo", + "Is disabled", + "Created at", + "Telegram ID", + "Discord ID", + "Discord Webhook", + ) + self.total_admins = await self.admin_operator.get_admins_count(self.db) + offset = (self.current_page - 1) * self.page_size + limit = self.page_size + admins = await self.admin_operator.get_admins(self.db, offset=offset, limit=limit) + if not admins: + self.no_admins.styles.display = "block" + self.pagination_info.update("") + return + else: + self.no_admins.styles.display = "none" + self.table.styles.display = "block" + users_usages = await asyncio.gather(*[self.calculate_admin_usage(admin.id) for admin in admins]) + + admins_data = [ + ( + admin.username, + readable_size(admin.used_traffic), + readable_size(admin.lifetime_used_traffic), + users_usages[i], + "✔️" if admin.is_sudo else "✖️", + "✔️" if admin.is_disabled else "✖️", + readable_datetime(admin.created_at), + str(admin.telegram_id or "✖️"), + str(admin.discord_id or "✖️"), + str(admin.discord_webhook or "✖️"), + ) + for i, admin in enumerate(admins) + ] + column_widths = [ + max(len(str(columns[i])), max(len(str(row[i])) for row in admins_data)) for i in range(len(columns)) + ] + + centered_columns = [self._center_text(column, column_widths[i]) for i, column in enumerate(columns)] + self.table.add_columns(*centered_columns) + i = 1 + for row, adnin in zip(admins_data, admins): + centered_row = [self._center_text(str(cell), column_widths[i]) for i, cell in enumerate(row)] + label = Text(f"{i + offset}") + i += 1 + self.table.add_row(*centered_row, key=adnin.username, label=label) + + total_pages = (self.total_admins + self.page_size - 1) // self.page_size + self.pagination_info.update( + f"Page {self.current_page}/{total_pages} (Total admins: {self.total_admins})\nPress `n` for go to the next page and `p` to back to previose page" + ) + + @property + def selected_admin(self): + return self.table.coordinate_to_cell_key(Coordinate(self.table.cursor_row, 0)).row_key.value + + async def action_delete_admin(self): + if not self.table.columns: + return + self.app.push_screen(AdminDelete(self.db, self.admin_operator, self.selected_admin, self._refresh_table)) + + def _refresh_table(self): + self.run_worker(self.admins_list) + + async def action_create_admin(self): + self.app.push_screen( + AdminCreateModale(self.db, self.admin_operator, self._refresh_table, self.format_tui_validation_error) + ) + + async def action_modify_admin(self): + if not self.table.columns: + return + admin = await self.admin_operator.get_validated_admin(self.db, username=self.selected_admin) + self.app.push_screen( + AdminModifyModale( + self.db, self.admin_operator, admin, self._refresh_table, self.format_tui_validation_error + ) + ) + + async def action_import_from_env(self): + try: + username, password = config("SUDO_USERNAME"), config("SUDO_PASSWORD") + except UndefinedValueError: + self.notify( + "Unable to get SUDO_USERNAME and/or SUDO_PASSWORD.\n" + "Make sure you have set them in the env file or as environment variables.", + severity="error", + title="Error", + ) + return + if not (username and password): + self.notify( + "Unable to retrieve username and password.\nMake sure both SUDO_USERNAME and SUDO_PASSWORD are set.", + severity="error", + title="Error", + ) + return + try: + await self.admin_operator.create_admin( + self.db, + AdminCreate(username=username, password=password, is_sudo=True), + SYSTEM_ADMIN, + ) + self.notify("Admin created successfully", severity="success", title="Success") + self._refresh_table() + except ValidationError as e: + self.format_tui_validation_error(e) + except ValueError as e: + self.notify(str(e), severity="error", title="Error") + + async def action_reset_admin_usage(self): + if not self.table.columns: + return + self.app.push_screen(AdminResetUsage(self.db, self.admin_operator, self.selected_admin, self._refresh_table)) + + async def action_previous_page(self): + if self.current_page > 1: + self.current_page -= 1 + await self.admins_list() + + async def action_next_page(self): + total_pages = (self.total_admins + self.page_size - 1) // self.page_size + if self.current_page < total_pages: + self.current_page += 1 + await self.admins_list() + + async def calculate_admin_usage(self, admin_id: int) -> str: + usage = await self.db.execute(select(func.sum(User.used_traffic)).filter_by(admin_id=admin_id)) + return readable_size(int(usage.scalar() or 0)) + + async def key_enter(self) -> None: + if self.table.columns: + await self.action_modify_admin() + + async def on_prune(self, event): + await self.db.close() + return await super().on_prune(event) + + def format_tui_validation_error(self, errors: ValidationError): + for error in errors.errors(): + for err in error["msg"].split(";"): + self.notify( + title=f"Error: {error['loc'][0].replace('_', ' ').capitalize()}", + message=err.strip(), + severity="error", + ) diff --git a/cli/help.py b/tui/help.py similarity index 98% rename from cli/help.py rename to tui/help.py index bf059fe67..98b89b85e 100644 --- a/cli/help.py +++ b/tui/help.py @@ -1,7 +1,7 @@ from textual.app import ComposeResult from textual.widgets import Static, Button from textual.containers import Container -from cli import BaseModal +from tui import BaseModal class HelpModal(BaseModal): diff --git a/cli/style.tcss b/tui/style.tcss similarity index 100% rename from cli/style.tcss rename to tui/style.tcss diff --git a/tui_wrapper.sh b/tui_wrapper.sh new file mode 100644 index 000000000..24f9c583e --- /dev/null +++ b/tui_wrapper.sh @@ -0,0 +1,2 @@ +#!/bin/bash +python /code/pasarguard-tui.py diff --git a/uv.lock b/uv.lock index 58cc4e7dc..6d6c41bb8 100644 --- a/uv.lock +++ b/uv.lock @@ -944,6 +944,7 @@ dependencies = [ { name = "sqlalchemy", extra = ["asyncio"] }, { name = "sse-starlette" }, { name = "textual", extra = ["syntax"] }, + { name = "typer" }, { name = "uvicorn" }, { name = "uvloop", marker = "sys_platform != 'win32'" }, { name = "websocket-client" }, @@ -1000,6 +1001,7 @@ requires-dist = [ { name = "sqlalchemy", extras = ["asyncio"], specifier = ">=2.0.40" }, { name = "sse-starlette", specifier = ">=2.1.3" }, { name = "textual", extras = ["syntax"], specifier = ">=2.1.2" }, + { name = "typer", specifier = ">=0.17.3" }, { name = "uvicorn", specifier = "==0.27.0.post1" }, { name = "uvloop", marker = "sys_platform != 'win32'", specifier = ">=0.21.0" }, { name = "websocket-client", specifier = "==1.7.0" }, @@ -1401,6 +1403,15 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/00/db/c376b0661c24cf770cb8815268190668ec1330eba8374a126ceef8c72d55/ruff-0.12.5-py3-none-win_arm64.whl", hash = "sha256:48cdbfc633de2c5c37d9f090ba3b352d1576b0015bfc3bc98eaf230275b7e805", size = 11951564, upload-time = "2025-07-24T13:26:34.994Z" }, ] +[[package]] +name = "shellingham" +version = "1.5.4" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/58/15/8b3609fd3830ef7b27b655beb4b4e9c62313a4e8da8c676e142cc210d58e/shellingham-1.5.4.tar.gz", hash = "sha256:8dbca0739d487e5bd35ab3ca4b36e11c4078f3a234bfce294b0a0291363404de", size = 10310, upload-time = "2023-10-24T04:13:40.426Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/e0/f9/0595336914c5619e5f28a1fb793285925a8cd4b432c9da0a987836c7f822/shellingham-1.5.4-py2.py3-none-any.whl", hash = "sha256:7ecfff8f2fd72616f7481040475a65b2bf8af90a56c89140852d1120324e8686", size = 9755, upload-time = "2023-10-24T04:13:38.866Z" }, +] + [[package]] name = "six" version = "1.17.0" @@ -1836,6 +1847,21 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/62/fa/b25e688df5b4e024bc3627bc3f951524ef9c8b0756f0646411efa5063a10/tree_sitter_yaml-0.7.1-cp310-abi3-win_arm64.whl", hash = "sha256:298ade69ad61f76bb3e50ced809650ec30521a51aa2708166b176419ccb0a6ba", size = 43801, upload-time = "2025-05-22T13:34:55.471Z" }, ] +[[package]] +name = "typer" +version = "0.17.3" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "click" }, + { name = "rich" }, + { name = "shellingham" }, + { name = "typing-extensions" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/dd/82/f4bfed3bc18c6ebd6f828320811bbe4098f92a31adf4040bee59c4ae02ea/typer-0.17.3.tar.gz", hash = "sha256:0c600503d472bcf98d29914d4dcd67f80c24cc245395e2e00ba3603c9332e8ba", size = 103517, upload-time = "2025-08-30T12:35:24.05Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/ca/e8/b3d537470e8404659a6335e7af868e90657efb73916ef31ddf3d8b9cb237/typer-0.17.3-py3-none-any.whl", hash = "sha256:643919a79182ab7ac7581056d93c6a2b865b026adf2872c4d02c72758e6f095b", size = 46494, upload-time = "2025-08-30T12:35:22.391Z" }, +] + [[package]] name = "typing-extensions" version = "4.14.1"