benya/tg-admin-bot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Paginate Arcane projects with inline refresh

Browse Source
This commit is contained in:
Alex
2026-02-07 23:40:02 +03:00
parent 08d936ac3e
commit 9d4b9620ee
2 changed files with 89 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files

110
handlers/arcane.py
View File

@@ -7,6 +7,7 @@ from keyboards import docker_kb, arcane_kb
from services.arcane import list_projects, restart_project, set_project_state
from datetime import datetime
from aiogram.types import InlineKeyboardMarkup, InlineKeyboardButton, CallbackQuery
from state import ARCANE_CACHE
def _arcane_cfg():
@@ -14,7 +15,52 @@ def _arcane_cfg():
return arc.get("base_url"), arc.get("api_key"), int(arc.get("env_id", 0))
async def cmd_arcane_projects(msg: Message, *, edit: bool):
def _arcane_kb(page: int, total_pages: int, items: list[dict]) -> InlineKeyboardMarkup:
rows = []
for p in items:
name = p.get("name", "?")
pid = p.get("id", "")
if not pid:
continue
rows.append([
InlineKeyboardButton(text=f"🔄 {name}", callback_data=f"arcane:restart:{pid}"),
InlineKeyboardButton(text="▶️", callback_data=f"arcane:up:{pid}"),
InlineKeyboardButton(text="⏹", callback_data=f"arcane:down:{pid}"),
])
nav = []
if page > 0:
nav.append(InlineKeyboardButton(text="⬅️ Prev", callback_data=f"arcane:page:{page-1}"))
nav.append(InlineKeyboardButton(text="🔄 Refresh", callback_data="arcane:refresh"))
if page < total_pages - 1:
nav.append(InlineKeyboardButton(text="Next ➡️", callback_data=f"arcane:page:{page+1}"))
if nav:
rows.append(nav)
return InlineKeyboardMarkup(inline_keyboard=rows)
def _render_arcane_page(items: list[dict], page: int, page_size: int, ts: str) -> tuple[str, InlineKeyboardMarkup]:
total_pages = max(1, (len(items) + page_size - 1) // page_size)
page = max(0, min(page, total_pages - 1))
start = page * page_size
end = start + page_size
page_items = items[start:end]
lines = [f"🧰 Arcane projects на {ts} (page {page+1}/{total_pages})\n"]
for p in page_items:
status = p.get("status", "unknown")
name = p.get("name", "?")
running = p.get("runningCount", 0)
total = p.get("serviceCount", 0)
icon = "🟢" if status == "running" else "🟡"
lines.append(f"{icon} {name}: {status} ({running}/{total})")
kb = _arcane_kb(page, total_pages, page_items)
return "\n".join(lines), kb
async def cmd_arcane_projects(msg: Message, *, edit: bool, page: int = 0):
base_url, api_key, env_id = _arcane_cfg()
if not base_url or not api_key:
await msg.answer("⚠️ Arcane config missing", reply_markup=docker_kb)
@@ -24,6 +70,12 @@ async def cmd_arcane_projects(msg: Message, *, edit: bool):
try:
await msg.edit_text("⏳ Arcane projects…")
except Exception:
if edit:
try:
await msg.edit_text("⏳ Arcane projects…")
except Exception:
await msg.answer("⏳ Arcane projects…", reply_markup=arcane_kb)
else:
await msg.answer("⏳ Arcane projects…", reply_markup=arcane_kb)
else:
await msg.answer("⏳ Arcane projects…", reply_markup=arcane_kb)
@@ -35,31 +87,19 @@ async def cmd_arcane_projects(msg: Message, *, edit: bool):
return
ts = datetime.now().strftime("%d.%m.%Y %H:%M:%S")
lines = [f"🧰 Arcane projects на {ts}\n"]
rows = []
for p in items:
status = p.get("status", "unknown")
name = p.get("name", "?")
pid = p.get("id", "")
running = p.get("runningCount", 0)
total = p.get("serviceCount", 0)
icon = "🟢" if status == "running" else "🟡"
lines.append(f"{icon} {name}: {status} ({running}/{total})")
if pid:
rows.append([
InlineKeyboardButton(text=f"🔄 {name}", callback_data=f"arcane:restart:{pid}"),
InlineKeyboardButton(text="▶️", callback_data=f"arcane:up:{pid}"),
InlineKeyboardButton(text="⏹", callback_data=f"arcane:down:{pid}"),
])
kb_inline = InlineKeyboardMarkup(inline_keyboard=rows) if rows else None
ARCANE_CACHE[msg.chat.id] = {
"items": items,
"page_size": 4,
"ts": ts,
}
text, kb = _render_arcane_page(items, page, 4, ts)
if edit:
try:
await msg.edit_text("\n".join(lines), reply_markup=kb_inline)
await msg.edit_text(text, reply_markup=kb)
except Exception:
await msg.answer("\n".join(lines), reply_markup=kb_inline or arcane_kb)
await msg.answer(text, reply_markup=kb)
else:
await msg.answer("\n".join(lines), reply_markup=kb_inline or arcane_kb)
await msg.answer(text, reply_markup=kb)
asyncio.create_task(worker())
@@ -76,6 +116,32 @@ async def arcane_refresh(msg: Message):
await cmd_arcane_projects(msg, edit=False)
@dp.callback_query(F.data == "arcane:refresh")
async def arcane_refresh_inline(cb: CallbackQuery):
if cb.from_user.id != cfg["telegram"]["admin_id"]:
return
await cb.answer()
await cmd_arcane_projects(cb.message, edit=True)
@dp.callback_query(F.data.startswith("arcane:page:"))
async def arcane_page(cb: CallbackQuery):
if cb.from_user.id != cfg["telegram"]["admin_id"]:
return
try:
page = int(cb.data.split(":", 2)[2])
except ValueError:
await cb.answer("Bad page")
return
data = ARCANE_CACHE.get(cb.message.chat.id)
if not data:
await cb.answer("No cache")
return
text, kb = _render_arcane_page(data["items"], page, data["page_size"], data["ts"])
await cb.answer()
await cb.message.edit_text(text, reply_markup=kb)
@dp.callback_query(F.data.startswith("arcane:restart:"))
async def arcane_restart(cb: CallbackQuery):
if cb.from_user.id != cfg["telegram"]["admin_id"]:

1
state.py
View File

@@ -3,3 +3,4 @@ from typing import Dict
DOCKER_MAP: Dict[str, str] = {}
LOG_FILTER_PENDING: Dict[int, dict] = {}
UPDATES_CACHE: Dict[int, dict] = {}
ARCANE_CACHE: Dict[int, dict] = {}