mirror of
https://github.com/andrey271192/keenetic_ssh-web.git
synced 2026-11-04 00:43:06 +00:00
keenetic_ssh-web: веб-панель CLI на Entware (порт 2001), расписание, вывод команд
- Flask + Waitress, пароль и ALLOWED_IPS - install.sh / uninstall.sh, init Entware - README, FUNDING, полоска автора Made-with: Cursor
This commit is contained in:
52
runner.py
Normal file
52
runner.py
Normal file
@@ -0,0 +1,52 @@
|
||||
"""Выполнение shell-команд на локальном Keenetic (Entware)."""
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import subprocess
|
||||
from datetime import datetime, timezone
|
||||
from typing import Any
|
||||
|
||||
# Entware + системные пути Keenetic
|
||||
_DEFAULT_PATH = "/opt/bin:/opt/sbin:/usr/sbin:/sbin:/bin:/usr/bin"
|
||||
|
||||
|
||||
def run_shell(command: str, timeout: int = 300) -> dict[str, Any]:
|
||||
if not command or not command.strip():
|
||||
return {"ok": False, "output": "", "msg": "Пустая команда"}
|
||||
cur = os.environ.get("PATH", "")
|
||||
extra = ":" + _DEFAULT_PATH if cur else _DEFAULT_PATH
|
||||
if not any(x in cur for x in ("/opt/bin", "/opt/sbin")):
|
||||
cur = (cur + extra) if cur else _DEFAULT_PATH
|
||||
env = {**os.environ, "PATH": cur or _DEFAULT_PATH}
|
||||
try:
|
||||
p = subprocess.run(
|
||||
command,
|
||||
shell=True,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=timeout,
|
||||
env=env,
|
||||
)
|
||||
out = (p.stdout or "") + (("\n--- stderr ---\n" + p.stderr) if p.stderr else "")
|
||||
out = out.strip()[:120_000]
|
||||
ok = p.returncode == 0
|
||||
return {"ok": ok, "output": out, "msg": f"exit {p.returncode}"}
|
||||
except subprocess.TimeoutExpired:
|
||||
return {"ok": False, "output": "", "msg": f"timeout {timeout}s"}
|
||||
except Exception as e:
|
||||
return {"ok": False, "output": "", "msg": str(e)[:500]}
|
||||
|
||||
|
||||
def touch_item(items: list[dict], item_id: str, result: dict[str, Any]) -> None:
|
||||
now = datetime.now(timezone.utc).astimezone().replace(microsecond=0).isoformat()
|
||||
for it in items:
|
||||
if it.get("id") == item_id:
|
||||
it["last_run"] = now
|
||||
it["last_ok"] = result.get("ok")
|
||||
parts = []
|
||||
if result.get("msg"):
|
||||
parts.append(result["msg"])
|
||||
if result.get("output"):
|
||||
parts.append(result["output"])
|
||||
it["last_output"] = "\n".join(parts).strip()[:100_000]
|
||||
break
|
||||
Reference in New Issue
Block a user