feat: add web dashboard with slide-in drawer navigation
- Flask backend with 23 API routes (entertainment, telegram, docker, system) - Alpine.js + Tailwind CSS dark-mode SPA with 4 tabs - pos-dashboard CLI tool with port/config management - Mobile slide-in drawer with swipe-to-close - Sticky header stays pinned on scroll - Tab completion fixes for category-less tools - POST /api/telegram/commands endpoint for adding commands
This commit is contained in:
@@ -0,0 +1,179 @@
|
||||
"""
|
||||
Telegram Listener API blueprint.
|
||||
|
||||
Routes under ``/api/telegram/`` that manage the Telegram bot listener service.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import re
|
||||
import subprocess
|
||||
from pathlib import Path
|
||||
|
||||
from flask import Blueprint, jsonify, request
|
||||
|
||||
from lib.runner import run_pos, run_cmd
|
||||
from lib.parsers import parse_telegram_commands
|
||||
from lib.config import load_telegram_env, mask_secrets
|
||||
|
||||
telegram_bp = Blueprint("telegram", __name__, url_prefix="/api/telegram")
|
||||
|
||||
_SERVICE = "pos-telegram-listener.service"
|
||||
|
||||
|
||||
# ── Helpers ──────────────────────────────────────────────────────────────────
|
||||
|
||||
def _systemd_user(*args: str, timeout: int = 15) -> dict:
|
||||
"""Run a ``systemctl --user`` command."""
|
||||
return run_cmd(["systemctl", "--user"] + list(args), timeout=timeout)
|
||||
|
||||
|
||||
# ── Routes ───────────────────────────────────────────────────────────────────
|
||||
|
||||
@telegram_bp.route("/status", methods=["GET"])
|
||||
def telegram_status():
|
||||
"""Check the systemd user service status for the Telegram listener.
|
||||
|
||||
Returns::
|
||||
|
||||
{"running": bool, "active_state": str, "sub_state": str}
|
||||
"""
|
||||
# Check active state
|
||||
res = _systemd_user("is-active", _SERVICE)
|
||||
running = res["returncode"] == 0 and res["stdout"].strip() == "active"
|
||||
|
||||
# Get detailed properties
|
||||
props: dict = {"active_state": res["stdout"].strip(), "sub_state": ""}
|
||||
detail = _systemd_user("show", _SERVICE, "--property=ActiveState,SubState,MainPID")
|
||||
if detail["returncode"] == 0:
|
||||
for line in detail["stdout"].splitlines():
|
||||
if "=" in line:
|
||||
k, v = line.split("=", 1)
|
||||
props[k.strip().lower()] = v.strip()
|
||||
|
||||
return jsonify({
|
||||
"running": running,
|
||||
"active_state": props.get("activestate", "unknown"),
|
||||
"sub_state": props.get("substate", "unknown"),
|
||||
"pid": props.get("mainpid", ""),
|
||||
})
|
||||
|
||||
|
||||
@telegram_bp.route("/enable", methods=["POST"])
|
||||
def telegram_enable():
|
||||
"""Enable the Telegram listener service."""
|
||||
result = run_pos(["communication", "telegram-listener", "--enable"], timeout=30)
|
||||
if result.get("error") and result["returncode"] != 0:
|
||||
return jsonify({
|
||||
"error": result["error"],
|
||||
"stderr": result.get("stderr", ""),
|
||||
}), 500
|
||||
|
||||
return jsonify({"message": result["stdout"].strip() or "telegram listener enabled"})
|
||||
|
||||
|
||||
@telegram_bp.route("/disable", methods=["POST"])
|
||||
def telegram_disable():
|
||||
"""Disable the Telegram listener service."""
|
||||
result = run_pos(["communication", "telegram-listener", "--disable"], timeout=30)
|
||||
if result.get("error") and result["returncode"] != 0:
|
||||
return jsonify({
|
||||
"error": result["error"],
|
||||
"stderr": result.get("stderr", ""),
|
||||
}), 500
|
||||
|
||||
return jsonify({"message": result["stdout"].strip() or "telegram listener disabled"})
|
||||
|
||||
|
||||
@telegram_bp.route("/commands", methods=["GET"])
|
||||
def telegram_commands():
|
||||
"""Read the telegram_commands.env and return parsed command map."""
|
||||
config_dir = Path(
|
||||
__import__("os").environ.get(
|
||||
"XDG_CONFIG_HOME",
|
||||
__import__("os").path.expanduser("~/.config"),
|
||||
)
|
||||
) / "linux_post_install"
|
||||
map_file = config_dir / "telegram_commands.env"
|
||||
|
||||
if not map_file.is_file():
|
||||
return jsonify({"commands": [], "map_file": str(map_file)})
|
||||
|
||||
text = map_file.read_text(encoding="utf-8", errors="replace")
|
||||
commands = parse_telegram_commands(text)
|
||||
return jsonify({"commands": commands, "map_file": str(map_file)})
|
||||
|
||||
|
||||
@telegram_bp.route("/commands", methods=["POST"])
|
||||
def telegram_commands_add():
|
||||
"""Add a command to the telegram_commands.env map file.
|
||||
|
||||
Expects JSON body: ``{"command": "/status", "description": "...", "script": "..."}``
|
||||
"""
|
||||
body = request.get_json(silent=True) or {}
|
||||
command = (body.get("command") or "").strip()
|
||||
description = (body.get("description") or "").strip()
|
||||
script = (body.get("script") or "").strip()
|
||||
|
||||
if not command:
|
||||
return jsonify({"error": "missing command name"}), 400
|
||||
if not script:
|
||||
return jsonify({"error": "missing bash script"}), 400
|
||||
|
||||
# Validate command starts with /
|
||||
if not command.startswith("/"):
|
||||
command = "/" + command
|
||||
|
||||
# Validate command name (alphanumeric, hyphens, underscores)
|
||||
if not re.match(r"^/[a-zA-Z0-9_-]+$", command):
|
||||
return jsonify({"error": f"invalid command name: {command} — use only letters, numbers, hyphens"}), 400
|
||||
|
||||
# Build the map line
|
||||
if description:
|
||||
line = f"{command}::{description}={script}"
|
||||
else:
|
||||
line = f"{command}={script}"
|
||||
|
||||
# Write to the map file
|
||||
config_dir = Path(
|
||||
os.environ.get(
|
||||
"XDG_CONFIG_HOME",
|
||||
os.path.expanduser("~/.config"),
|
||||
)
|
||||
) / "linux_post_install"
|
||||
map_file = config_dir / "telegram_commands.env"
|
||||
|
||||
# Check if command already exists
|
||||
if map_file.is_file():
|
||||
existing = map_file.read_text(encoding="utf-8", errors="replace")
|
||||
for existing_line in existing.splitlines():
|
||||
existing_line = existing_line.strip()
|
||||
if not existing_line or existing_line.startswith("#"):
|
||||
continue
|
||||
existing_cmd = existing_line.split("=")[0].split("::")[0].strip()
|
||||
if existing_cmd == command:
|
||||
return jsonify({"error": f"command {command} already exists — delete it first"}), 409
|
||||
|
||||
# Append the new command
|
||||
config_dir.mkdir(parents=True, exist_ok=True)
|
||||
with open(map_file, "a", encoding="utf-8") as f:
|
||||
f.write(line + "\n")
|
||||
|
||||
return jsonify({"ok": True, "command": command, "message": f"Command {command} added"})
|
||||
|
||||
|
||||
@telegram_bp.route("/commands/sync", methods=["POST"])
|
||||
def telegram_commands_sync():
|
||||
"""Push mapped commands to the bot's "/" menu (setMyCommands)."""
|
||||
result = run_pos(
|
||||
["communication", "telegram-listener", "--sync-commands"],
|
||||
timeout=30,
|
||||
)
|
||||
if result.get("error") and result["returncode"] != 0:
|
||||
return jsonify({
|
||||
"error": result["error"],
|
||||
"stderr": result.get("stderr", ""),
|
||||
}), 500
|
||||
|
||||
return jsonify({"message": result["stdout"].strip() or "commands synced"})
|
||||
Reference in New Issue
Block a user