admin/kandrusyak_bot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor: remove unreachable legacy command handlers

Browse Source
This commit is contained in:
kandrusyak
2026-07-27 17:41:33 +03:00
parent 3cc60c69e1
commit 79ed741a3b
3 changed files with 0 additions and 340 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
assistant_bot/agent.py
View File

@@ -536,11 +536,3 @@ async def run_agent_prompt(
update.effective_message, update.effective_message,
"⚠️ **Произошла внутренняя ошибка** при запросе к модели.", "⚠️ **Произошла внутренняя ошибка** при запросе к модели.",
) )
async def run_ai_prompt(
update: Update,
context: ContextTypes.DEFAULT_TYPE,
prompt: str,
) -> None:
await run_agent_prompt(update, context, prompt)

1
assistant_bot/application.py
View File

@@ -88,7 +88,6 @@ def create_application() -> Application:
) )
application.bot_data["ai_client"] = ai_client application.bot_data["ai_client"] = ai_client
application.bot_data["assistant_mode"] = mode
application.bot_data["timezone"] = get_local_timezone() application.bot_data["timezone"] = get_local_timezone()
application.bot_data["speech_recognizer"] = speech_recognizer application.bot_data["speech_recognizer"] = speech_recognizer
application.bot_data["voice_max_duration"] = get_voice_max_duration_seconds() application.bot_data["voice_max_duration"] = get_voice_max_duration_seconds()

331
assistant_bot/handlers.py
View File

@@ -3,7 +3,6 @@ import logging
import os import os
import tempfile import tempfile
from pathlib import Path from pathlib import Path
from typing import Any
from telegram import Update from telegram import Update
from telegram.constants import ChatAction from telegram.constants import ChatAction
@@ -12,7 +11,6 @@ from telegram.ext import ContextTypes
from .ai import AIClientError from .ai import AIClientError
from .agent import run_agent_prompt from .agent import run_agent_prompt
from .reminders import parse_reminder
from .telegram_utils import ( from .telegram_utils import (
build_inline_results, build_inline_results,
command_text, command_text,
@@ -24,7 +22,6 @@ from .telegram_utils import (
get_tz, get_tz,
get_voice_max_duration, get_voice_max_duration,
markdown_code, markdown_code,
parse_positive_int,
reply_long, reply_long,
reply_markdown, reply_markdown,
require_user_id, require_user_id,
@@ -36,16 +33,6 @@ from .time_utils import format_local_dt
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
def format_numbered_rows(
rows: list[dict[str, Any]],
formatter,
empty_text: str,
) -> str:
if not rows:
return empty_text
return "\n".join(formatter(row) for row in rows)
async def start(update: Update, _context: ContextTypes.DEFAULT_TYPE) -> None: async def start(update: Update, _context: ContextTypes.DEFAULT_TYPE) -> None:
if update.message: if update.message:
await reply_markdown( await reply_markdown(
@@ -285,324 +272,6 @@ async def model_command(update: Update, context: ContextTypes.DEFAULT_TYPE) -> N
) )
async def remember_command(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
if not update.message:
return
user_id = require_user_id(update)
text = command_text(context)
if user_id is None:
await reply_markdown(update.message, "⚠️ **Не могу определить пользователя.**")
return
if not text:
await reply_markdown(
update.message,
"🧠 Напиши факт после команды: "
"`/remember я предпочитаю короткие ответы`",
)
return
memory_id = get_storage(context).add_memory(user_id, text)
await reply_markdown(
update.message,
f"✅ **Запомнил.** ID памяти: {markdown_code(memory_id)}",
)
async def memory_command(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
if not update.message:
return
user_id = require_user_id(update)
if user_id is None:
await reply_markdown(update.message, "⚠️ **Не могу определить пользователя.**")
return
rows = get_storage(context).list_memories(user_id)
text = "**🧠 Долговременная память**\n\n" + format_numbered_rows(
rows,
lambda row: (
f"- {markdown_code('#' + str(row['id']))} "
f"{escape_markdown_text(row['text'])}"
),
"*Пока пусто.*",
)
await reply_long(update.message, text)
async def forget_memory_command(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
if not update.message:
return
user_id = require_user_id(update)
memory_id = parse_positive_int(command_text(context))
if user_id is None:
await reply_markdown(update.message, "⚠️ **Не могу определить пользователя.**")
return
if memory_id is None:
await reply_markdown(
update.message,
"🧠 Укажи ID: `/forget_memory 3`",
)
return
deleted = get_storage(context).delete_memory(user_id, memory_id)
await reply_markdown(
update.message,
"✅ **Запись памяти удалена.**"
if deleted
else "🔎 **Не нашёл такую запись памяти.**",
)
async def note_command(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
if not update.message:
return
user_id = require_user_id(update)
text = command_text(context)
if user_id is None:
await reply_markdown(update.message, "⚠️ **Не могу определить пользователя.**")
return
if not text:
await reply_markdown(
update.message,
"📝 Напиши текст заметки: `/note идея для проекта`",
)
return
note_id = get_storage(context).add_note(user_id, text)
await reply_markdown(
update.message,
f"✅ **Заметка сохранена.** ID: {markdown_code(note_id)}",
)
async def notes_command(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
if not update.message:
return
user_id = require_user_id(update)
if user_id is None:
await reply_markdown(update.message, "⚠️ **Не могу определить пользователя.**")
return
rows = get_storage(context).list_notes(user_id)
text = "**📝 Заметки**\n\n" + format_numbered_rows(
rows,
lambda row: (
f"- {markdown_code('#' + str(row['id']))} "
f"{escape_markdown_text(row['text'])}"
),
"*Пока пусто.*",
)
await reply_long(update.message, text)
async def forget_note_command(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
if not update.message:
return
user_id = require_user_id(update)
note_id = parse_positive_int(command_text(context))
if user_id is None:
await reply_markdown(update.message, "⚠️ **Не могу определить пользователя.**")
return
if note_id is None:
await reply_markdown(update.message, "📝 Укажи ID: `/forget 3`")
return
deleted = get_storage(context).delete_note(user_id, note_id)
await reply_markdown(
update.message,
"✅ **Заметка удалена.**" if deleted else "🔎 **Не нашёл такую заметку.**",
)
async def remind_command(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
if not update.message or not update.effective_chat:
return
user_id = require_user_id(update)
if user_id is None:
await reply_markdown(update.message, "⚠️ **Не могу определить пользователя.**")
return
parsed = parse_reminder(command_text(context), get_tz(context))
if not parsed:
await reply_markdown(
update.message,
"**⏰ Формат напоминания**\n\n"
"- `/remind 30m позвонить`\n"
"- `/remind через 2 часа проверить статус`\n"
"- `/remind 18:30 отправить отчёт`\n"
"- `/remind 2026-07-16 18:30 отправить отчёт`",
)
return
reminder_id = get_storage(context).add_reminder(
user_id=user_id,
chat_id=int(update.message.chat_id),
text=parsed.text,
remind_at_utc=parsed.remind_at_utc,
)
await reply_markdown(
update.message,
f"✅ **Напоминание {markdown_code(f'#{reminder_id}')} поставлено на** "
f"{markdown_code(format_local_dt(parsed.remind_at_utc, get_tz(context)))}.",
)
async def reminders_command(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
if not update.message:
return
user_id = require_user_id(update)
if user_id is None:
await reply_markdown(update.message, "⚠️ **Не могу определить пользователя.**")
return
tz = get_tz(context)
rows = get_storage(context).list_reminders(user_id)
text = "**⏰ Активные напоминания**\n\n" + format_numbered_rows(
rows,
lambda row: (
f"- {markdown_code('#' + str(row['id']))} "
f"{markdown_code(format_local_dt(row['remind_at'], tz))} — "
f"{escape_markdown_text(row['text'])}"
),
"*Активных напоминаний нет.*",
)
await reply_long(update.message, text)
async def cancel_reminder_command(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
if not update.message:
return
user_id = require_user_id(update)
reminder_id = parse_positive_int(command_text(context))
if user_id is None:
await reply_markdown(update.message, "⚠️ **Не могу определить пользователя.**")
return
if reminder_id is None:
await reply_markdown(
update.message,
"⏰ Укажи ID: `/cancelreminder 3`",
)
return
cancelled = get_storage(context).cancel_reminder(user_id, reminder_id)
await reply_markdown(
update.message,
"✅ **Напоминание отменено.**"
if cancelled
else "🔎 **Не нашёл активное напоминание.**",
)
def split_tracking_text(raw: str) -> tuple[str, str]:
if "|" not in raw:
return raw.strip(), "open"
title, status = raw.split("|", 1)
return title.strip(), status.strip() or "open"
async def track_command(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
if not update.message:
return
user_id = require_user_id(update)
raw_text = command_text(context)
if user_id is None:
await reply_markdown(update.message, "⚠️ **Не могу определить пользователя.**")
return
if not raw_text:
await reply_markdown(
update.message,
"📌 Формат: `/track паспорт | жду ответа`",
)
return
title, status = split_tracking_text(raw_text)
if not title:
await reply_markdown(
update.message,
"⚠️ **Название не должно быть пустым.**",
)
return
item_id = get_storage(context).add_tracked_item(user_id, title, status)
await reply_markdown(
update.message,
f"✅ **Добавил отслеживание {markdown_code(f'#{item_id}')}**\n\n"
f"**Название:** {escape_markdown_text(title)}\n"
f"**Статус:** {escape_markdown_text(status)}",
)
async def status_command(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
if not update.message:
return
user_id = require_user_id(update)
if user_id is None:
await reply_markdown(update.message, "⚠️ **Не могу определить пользователя.**")
return
raw_text = command_text(context)
storage = get_storage(context)
if not raw_text:
rows = storage.list_tracked_items(user_id)
text = "**📌 Отслеживаемые статусы**\n\n" + format_numbered_rows(
rows,
lambda row: (
f"- {markdown_code('#' + str(row['id']))} "
f"**{escape_markdown_text(row['title'])}:** "
f"{escape_markdown_text(row['status'])}"
),
"*Нет отслеживаемых статусов.*",
)
await reply_long(update.message, text)
return
parts = raw_text.split(maxsplit=1)
item_id = parse_positive_int(parts[0])
new_status = parts[1].strip() if len(parts) > 1 else ""
if item_id is None or not new_status:
await reply_markdown(
update.message,
"📌 Формат: `/status 3 новый статус`",
)
return
updated = storage.set_tracked_status(user_id, item_id, new_status)
await reply_markdown(
update.message,
"✅ **Статус обновлён.**" if updated else "🔎 **Не нашёл такой объект.**",
)
async def untrack_command(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
if not update.message:
return
user_id = require_user_id(update)
item_id = parse_positive_int(command_text(context))
if user_id is None:
await reply_markdown(update.message, "⚠️ **Не могу определить пользователя.**")
return
if item_id is None:
await reply_markdown(update.message, "📌 Укажи ID: `/untrack 3`")
return
deleted = get_storage(context).delete_tracked_item(user_id, item_id)
await reply_markdown(
update.message,
"✅ **Объект удалён.**" if deleted else "🔎 **Не нашёл такой объект.**",
)
async def inline_query(update: Update, _context: ContextTypes.DEFAULT_TYPE) -> None: async def inline_query(update: Update, _context: ContextTypes.DEFAULT_TYPE) -> None:
if not update.inline_query or not update.inline_query.query: if not update.inline_query or not update.inline_query.query:
return return