beaver/beaver-gateway
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(telegram,conversations): attachments by day with sweep in both modes, merge renames the topic, gone topics unbind, failures visible

Browse Source
This commit is contained in:
hh
2026-09-01 23:13:22 +02:00
parent 7853d61b94
commit b96714338f
3 changed files with 422 additions and 28 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/beaver_gateway/frontends/telegram/frontend.py
+111 -21
View File
@@ -19,8 +19,10 @@ import logging
import tempfile
import time
from dataclasses import dataclass, field
from datetime import datetime
from pathlib import Path
from typing import TYPE_CHECKING, Any, Literal, cast
from zoneinfo import ZoneInfo
from aiogram import Bot
from aiogram.client.default import DefaultBotProperties
@@ -71,14 +73,21 @@ _HELP = "General - мастер, любой другой топик - ветка
class Attachments:
"""Where files from Telegram go.
``ephemeral`` - under the gateway's data dir, swept after ``keep_days``;
``vault`` - into a directory the agent owns. Pulling what is worth
keeping into the vault is the agent's job either way.
Files land in ``<root>/YYYY-MM-DD/<unixts>-<name>`` (the day in ``tz``)
and whatever is older than ``keep_days`` is swept; ``None`` never
sweeps. ``ephemeral`` - under the gateway's data dir; ``vault`` -
``dir`` is an inbox inside the agent's zone: the agent moves keepers
next to the note, the rest is swept after ``keep_days``.
"""
mode: Literal["ephemeral", "vault"] = "ephemeral"
dir: Path | None = None
keep_days: int = 7
keep_days: int | None = 7
tz: str = "UTC"
def day(self, now: float | None = None) -> str:
stamp = datetime.fromtimestamp(now or time.time(), tz=ZoneInfo(self.tz))
return stamp.date().isoformat()
@property
def root(self) -> Path:
@@ -174,7 +183,6 @@ class TelegramFrontend(Frontend):
self.chat_id,
getattr(me, "has_topics_enabled", None),
)
self._sweep_attachments()
await self.bot.set_my_commands(
[BotCommand(command=c, description=d) for c, d in _COMMAND_HELP]
)
@@ -183,6 +191,7 @@ class TelegramFrontend(Frontend):
tg.create_task(self.inbox.run())
tg.create_task(self.outbox.run())
tg.create_task(self._events())
tg.create_task(self._sweep_loop())
finally:
for draft in list(self._drafts.values()):
await draft.stop()
@@ -373,7 +382,11 @@ class TelegramFrontend(Frontend):
return
if message.chat.id != self.chat_id:
return
in_topic = message.is_topic_message or message.forum_topic_created is not None
in_topic = (
message.is_topic_message
or message.forum_topic_created is not None
or message.forum_topic_edited is not None
)
thread_id = message.message_thread_id if in_topic else None
is_master = (
thread_id is None or self._ext(thread_id) == await self._master_ext()
@@ -382,10 +395,12 @@ class TelegramFrontend(Frontend):
return
if message.forum_topic_edited is not None and thread_id is not None:
if message.forum_topic_edited.name:
self._topic_names[thread_id] = message.forum_topic_edited.name
await self._on_topic_renamed(
thread_id, message.forum_topic_edited.name, is_master=is_master
)
return
text = (message.text or message.caption or "").strip()
attachment = await self._save_attachment(message, thread_id)
attachment = await self._save_attachment(message)
if attachment:
text = f"{text}\n\n{attachment}".strip()
if not text:
@@ -414,6 +429,14 @@ class TelegramFrontend(Frontend):
if conv.running_turn or conv.pending_question:
await self._react(message, item.id)
async def _on_topic_renamed(
self, thread_id: int, name: str, *, is_master: bool
) -> None:
self._topic_names[thread_id] = name
conv = None if is_master else await self._live(thread_id)
if conv is not None:
await self.conversations.set_title(conv, name)
async def _live(self, thread_id: int) -> Conversation | None:
conv = await self.conversations.find_bound(
frontend=FRONTEND, external_id=self._ext(thread_id)
@@ -446,13 +469,15 @@ class TelegramFrontend(Frontend):
with contextlib.suppress(TelegramAPIError):
await self.bot.set_message_reaction(target[0], target[1], reaction=[])
async def _save_attachment(
self, message: Message, thread_id: int | None
) -> str | None:
async def _save_attachment(self, message: Message) -> str | None:
file_id: str | None = None
name: str | None = None
kind = ""
size: int | None = None
if message.sticker:
return "[вложение: стикер - не поддерживается]"
if message.animation:
return "[вложение: анимация - не поддерживается]"
if message.photo:
photo = message.photo[-1]
file_id, kind, size = photo.file_id, "фото", photo.file_size
@@ -496,12 +521,13 @@ class TelegramFrontend(Frontend):
f"[вложение: {kind} {name}, {size // 1024 // 1024} МБ - "
"больше 20 МБ, Telegram не отдаёт ботам]"
)
folder = self.attachments.root / (self._ext(thread_id).replace("/", "_"))
now = time.time()
folder = self.attachments.root / self.attachments.day(now)
folder.mkdir(parents=True, exist_ok=True)
with contextlib.suppress(OSError):
self.attachments.root.chmod(0o755)
folder.chmod(0o755)
path = folder / f"{int(time.time())}-{Path(name).name}"
path = folder / f"{int(now)}-{Path(name).name}"
try:
await self.bot.download(file_id, destination=path)
path.chmod(0o644)
@@ -509,19 +535,33 @@ class TelegramFrontend(Frontend):
_log.warning("attachment download failed: %s", exc)
return f"[вложение: {kind} {name} - не скачалось: {exc}]"
shown = f", {size // 1024} КБ" if size else ""
return f"[вложение: {kind} {path}{shown}]"
keep = self.attachments.keep_days
if keep is None:
return f"[вложение: {kind} {path}{shown}]"
return (
f"[вложение: {kind} {path}{shown}; хранится {keep} дн., "
"перенеси в vault, если нужно]"
)
async def _sweep_loop(self, interval: float = 6 * 3600) -> None:
while True:
self._sweep_attachments()
await asyncio.sleep(interval)
def _sweep_attachments(self) -> None:
if self.attachments.mode != "ephemeral":
return
keep = self.attachments.keep_days
root = self.attachments.root
if not root.exists():
if keep is None or not root.exists():
return
cutoff = time.time() - self.attachments.keep_days * 86400
cutoff = time.time() - keep * 86400
for path in root.rglob("*"):
with contextlib.suppress(OSError):
if path.is_file() and path.stat().st_mtime < cutoff:
path.unlink()
for folder in root.iterdir():
with contextlib.suppress(OSError):
if folder.is_dir() and not any(folder.iterdir()):
folder.rmdir()
# ---- commands --------------------------------------------------------
@@ -634,7 +674,10 @@ class TelegramFrontend(Frontend):
if kind == "conversation.bound":
self._targets.pop(str(event.get("conversation_id")), None)
return
if kind in ("delivery.sent", "delivery.failed", "conversation.created"):
if kind == "delivery.failed":
await self._on_delivery_failed(event)
return
if kind in ("delivery.sent", "conversation.created"):
return
key = event.get("conversation_id")
if not isinstance(key, str):
@@ -661,10 +704,13 @@ class TelegramFrontend(Frontend):
draft = self._drafts.get(key)
if draft is not None:
await draft.stop()
if event.get("stop") == "error" and event.get("origin") == "user":
if event.get("stop") == "error":
origin = event.get("origin")
await self._deliver(
conv,
"⚠️ тёрн упал, смотри логи gateway",
"⚠️ тёрн упал, смотри логи gateway"
if origin == "user"
else f"⚠️ фоновый тёрн ({origin}) упал, смотри логи gateway",
turn_id=event.get("turn_id"),
key=f"{event.get('turn_id')}:error",
)
@@ -687,6 +733,38 @@ class TelegramFrontend(Frontend):
case "conversation.merged":
await self._deliver(conv, "✅ слито в мастер", key=f"{key}:merged")
async def _on_delivery_failed(self, event: Event) -> None:
row = event.get("conversation_row")
_log.error(
"delivery to %s/%s (conversation row %s) failed: %s",
event.get("chat_id"),
event.get("thread_id"),
row,
event.get("error"),
)
if not event.get("gone") or not isinstance(row, int):
return
conv = await self.conversations.get_row(row)
if conv is None:
return
bound = next(
(
b
for b in await self.conversations.bindings(conv)
if b.frontend == FRONTEND and b.visible
),
None,
)
if bound is None:
return
_log.warning(
"window %s of %s is gone; unbinding", bound.external_id, conv.external_id
)
await self.conversations.bind(
conv, frontend=FRONTEND, external_id=bound.external_id, visible=False
)
self._targets.pop(conv.external_id, None)
async def _on_reply(self, conv: Conversation, event: Event) -> None:
turn_id = str(event.get("turn_id") or "")
origin = str(event.get("item_origin") or "")
@@ -789,6 +867,7 @@ class TelegramFrontend(Frontend):
ask = self._asks.get(question_id)
if ask is None or not qi_raw.isdigit():
await self._callback_reply(query, "вопрос уже закрыт")
await self._strip_keyboard(query)
return
qi = int(qi_raw)
question = ask.questions[qi]
@@ -831,6 +910,17 @@ class TelegramFrontend(Frontend):
if not self.conversations.answer(question_id, answer):
await self._edit_asks(ask, "⌛ время вышло - ответь текстом")
async def _strip_keyboard(self, query: CallbackQuery) -> None:
message = query.message if isinstance(query.message, Message) else None
if message is None:
return
with contextlib.suppress(TelegramAPIError):
await self.bot.edit_message_reply_markup(
chat_id=message.chat.id,
message_id=message.message_id,
reply_markup=None,
)
async def _callback_reply(self, query: CallbackQuery, text: str | None) -> None:
with contextlib.suppress(TelegramAPIError):
await self.bot.answer_callback_query(query.id, text=text)
src/beaver_gateway/frontends/telegram/outbox.py
+3 -2
View File
@@ -143,7 +143,7 @@ class Outbox:
if "parse" in text and not row.plain:
await self._retry(row, str(exc), delay=0.0, plain=True)
elif any(marker in text for marker in _GONE):
await self._fail(row, str(exc))
await self._fail(row, str(exc), gone=True)
else:
await self._fail(row, str(exc))
except (TelegramNotFound, TelegramForbiddenError) as exc:
@@ -180,7 +180,7 @@ class Outbox:
if delay < 5.0:
self._wake.set()
async def _fail(self, row: Delivery, error: str) -> None:
async def _fail(self, row: Delivery, error: str, *, gone: bool = False) -> None:
_log.error(
"delivery #%s to %s/%s given up: %s",
row.id,
@@ -196,6 +196,7 @@ class Outbox:
chat_id=row.chat_id,
thread_id=row.thread_id,
error=error,
gone=gone,
)
async def _mark(