import hashlib import hmac import logging import time import telebot from telebot import types from telebot.apihelper import ApiTelegramException import config import database from intradesk import reference from intradesk import api as intradesk_api from intradesk import formatter log = logging.getLogger(__name__) bot = telebot.TeleBot(config.TOKEN, parse_mode="Markdown") # Сотрудники, доступные для назначения исполнителем (ID из Intradesk) EXECUTOR_USER_IDS = config.EXECUTOR_USER_IDS # Разделитель перед секцией "В работе" в тексте поста _RESP_SEP = "\n\n─────────────────────" # ── Текст с футером "В работе" ──────────────────────────────────────────────── def _build_full_text( base_text: str, responsible: list[dict], events_lines: list[str] | None = None, new_count: int = 0, ) -> str: """ Собирает итоговый текст поста: 1. Основной текст (base_text) 2. «📋 Последние события» — все события кроме последних new_count 3. «🆕 Новое» — последние new_count событий (все новые из текущего вебхука) 4. Разделитель + «В работе» — всегда самый последний блок """ clean = base_text.split(_RESP_SEP)[0] result = clean if events_lines: if new_count > 0: history = events_lines[:-new_count] newest = events_lines[-new_count:] else: history = events_lines newest = [] # Показываем последние 6 из истории history = history[-6:] if history: result += "\n\n📋 *Последние события:*\n" + "\n".join(history) if newest: result += "\n\n🆕 *Новое:*\n" + "\n".join(newest) if responsible: names = ", ".join( formatter.escape_markdown(r["user_name"]) or r["user_name"] for r in responsible ) result += _RESP_SEP + f"\n🙋 В работе: {names}" return result # ── Клавиатуры ──────────────────────────────────────────────────────────────── def sign_task_url(task_number: str) -> str: """Генерирует подписанную ссылку на Mini App с токеном (TTL 30 дней).""" expires = int(time.time()) + 30 * 24 * 3600 msg = f"{task_number}:{expires}".encode() sig = hmac.new(config.MINIAPP_SECRET.encode(), msg, hashlib.sha256).hexdigest()[:16] return f"{config.PUBLIC_URL}/miniapp?task={task_number}&exp={expires}&sig={sig}" def make_task_buttons(task_number: str, group_id: int) -> types.InlineKeyboardMarkup: gid = group_id or 0 markup = types.InlineKeyboardMarkup(row_width=1) #markup.add( # types.InlineKeyboardButton("📋 Статус", callback_data=f"st:{task_number}:{gid}"), # types.InlineKeyboardButton("👤 Исполнитель", callback_data=f"ex:{task_number}:{gid}"), # types.InlineKeyboardButton("🙋 Взять", callback_data=f"take:{task_number}"), # types.InlineKeyboardButton("❌ Снять", callback_data=f"drop:{task_number}"), #) # Кнопка Mini App — полная карточка заявки (только если PUBLIC_URL задан) if config.PUBLIC_URL: markup.add( types.InlineKeyboardButton("📄 Предпросмотр", url=sign_task_url(task_number)), types.InlineKeyboardButton("🔗 Открыть в IntraDesk", url=config.BASE_TASK_URL + task_number), ) else: markup.add( types.InlineKeyboardButton("🔗 Открыть заявку", url=config.BASE_TASK_URL + task_number), ) return markup def _make_status_keyboard(task_number: str, group_id: int) -> types.InlineKeyboardMarkup: markup = types.InlineKeyboardMarkup(row_width=1) for sid, name in reference.STATUSES.items(): markup.add(types.InlineKeyboardButton( name, callback_data=f"ss:{task_number}:{group_id}:{sid}" )) markup.add(types.InlineKeyboardButton("↩ Назад", callback_data=f"back:{task_number}:{group_id}")) return markup def _make_executor_keyboard(task_number: str, group_id: int) -> types.InlineKeyboardMarkup: markup = types.InlineKeyboardMarkup(row_width=1) for uid in EXECUTOR_USER_IDS: name = database.resolve("employees", uid) or f"ID {uid}" markup.add(types.InlineKeyboardButton( name, callback_data=f"se:{task_number}:{group_id}:{uid}" )) markup.add(types.InlineKeyboardButton( "➖ Только группа", callback_data=f"su:{task_number}:{group_id}" )) markup.add(types.InlineKeyboardButton("↩ Назад", callback_data=f"back:{task_number}:{group_id}")) return markup # ── Публикация / пересоздание поста ────────────────────────────────────────── def recreate_post( task_id: int, task_number: str, base_text: str, group_id: int = 0, closed: bool = False, new_event_lines: list[str] | None = None, ): """Удаляет старый пост и создаёт новый, накапливая события и сохраняя ответственных.""" # Накапливаем события: # - изменения (• ...) дедуплицируем по тексту — одно и то же изменение не дублируем # - комментарии (💬 ...) и описание (📝 ...) всегда добавляем, даже если текст совпадает stored = database.get_stored_events(task_id) stored_set = set(stored) if new_event_lines: added = [] for line in new_event_lines: if not line: continue is_change = line.startswith("•") if is_change and line in stored_set: continue # дедупликация только для изменений added.append(line) stored_set.add(line) # обновляем set чтобы не дублировать внутри одного вебхука stored = stored + added else: added = [] responsible = database.get_responsible(task_id) full_text = _build_full_text(base_text, responsible, stored, new_count=len(added)) markup = make_task_buttons(task_number, group_id) # Удаляем старый пост old = database.get_message_data(task_id) if old: _delete_message(config.CHANNEL_ID, old["message_id"]) try: msg = bot.send_message( config.CHANNEL_ID, full_text, parse_mode="Markdown", disable_web_page_preview=True, reply_markup=markup, ) # save_message_id сначала создаёт/обновляет строку, потом сохраняем события database.save_message_id(task_id, msg.message_id, task_number, group_id, base_text) if added: database.save_stored_events(task_id, stored) except ApiTelegramException as e: log.error(f"recreate_post failed for task {task_id}: {e}") # ── Callbacks: смена статуса ────────────────────────────────────────────────── @bot.callback_query_handler(func=lambda c: c.data.startswith("st:")) def cb_show_statuses(call): task_number, group_id = _parse2(call.data[3:]) chat_id, msg_id = _current_msg(task_number, call) try: bot.edit_message_reply_markup( chat_id, msg_id, reply_markup=_make_status_keyboard(task_number, group_id), ) bot.answer_callback_query(call.id) except Exception as e: if _is_stale_error(e): bot.answer_callback_query(call.id, "⚠️ Пост обновился, попробуйте ещё раз") else: bot.answer_callback_query(call.id, str(e), show_alert=True) @bot.callback_query_handler(func=lambda c: c.data.startswith("ss:")) def cb_set_status(call): # ss:{task_number}:{group_id}:{status_id} parts = call.data.split(":") task_number = parts[1] group_id = int(parts[2]) status_id = int(parts[3]) status_name = reference.STATUSES.get(status_id, f"#{status_id}") success = intradesk_api.change_status(int(task_number), status_id) if success: bot.answer_callback_query(call.id, f"✅ Статус → {status_name}") _restore_keyboard(call, task_number, group_id) else: bot.answer_callback_query(call.id, "❌ Ошибка изменения статуса", show_alert=True) # ── Callbacks: назначение исполнителя ───────────────────────────────────────── @bot.callback_query_handler(func=lambda c: c.data.startswith("ex:")) def cb_show_executors(call): task_number, group_id = _parse2(call.data[3:]) chat_id, msg_id = _current_msg(task_number, call) try: bot.edit_message_reply_markup( chat_id, msg_id, reply_markup=_make_executor_keyboard(task_number, group_id), ) bot.answer_callback_query(call.id) except Exception as e: if _is_stale_error(e): bot.answer_callback_query(call.id, "⚠️ Пост обновился, попробуйте ещё раз") else: bot.answer_callback_query(call.id, str(e), show_alert=True) @bot.callback_query_handler(func=lambda c: c.data.startswith("se:")) def cb_set_executor(call): # se:{task_number}:{group_id}:{emp_id} parts = call.data.split(":") task_number = parts[1] group_id = int(parts[2]) emp_id = int(parts[3]) emp_name = database.resolve("employees", emp_id) or f"ID {emp_id}" # Сохраняем текущую группу, меняем только исполнителя success = intradesk_api.assign_executor( int(task_number), user_id=emp_id, group_id=group_id or None ) if success: bot.answer_callback_query(call.id, f"✅ Исполнитель → {emp_name}") _restore_keyboard(call, task_number, group_id) else: bot.answer_callback_query(call.id, "❌ Ошибка при назначении", show_alert=True) @bot.callback_query_handler(func=lambda c: c.data.startswith("su:")) def cb_unset_executor(call): """Убрать конкретного исполнителя, оставить только группу.""" task_number, group_id = _parse2(call.data[3:]) if not group_id: bot.answer_callback_query(call.id, "Группа не определена", show_alert=True) return success = intradesk_api.assign_executor(int(task_number), group_id=group_id) if success: bot.answer_callback_query(call.id, "✅ Исполнитель снят, группа сохранена") _restore_keyboard(call, task_number, group_id) else: bot.answer_callback_query(call.id, "❌ Ошибка", show_alert=True) # ── Callbacks: "Взять в работу" / "Снять меня" ──────────────────────────────── @bot.callback_query_handler(func=lambda c: c.data.startswith("take:")) def cb_take(call): task_number = call.data[5:] task_id = database.get_task_id_by_number(task_number) if not task_id: bot.answer_callback_query(call.id, "Заявка не найдена", show_alert=True) return user_id = call.from_user.id user_name = (call.from_user.full_name or call.from_user.username or str(user_id)).strip() database.add_responsible(task_id, user_id, user_name) bot.answer_callback_query(call.id, f"🙋 Вы взяли заявку #{task_number} в работу") _refresh_responsible(call, task_id, task_number) @bot.callback_query_handler(func=lambda c: c.data.startswith("drop:")) def cb_drop(call): task_number = call.data[5:] task_id = database.get_task_id_by_number(task_number) if not task_id: bot.answer_callback_query(call.id, "Заявка не найдена", show_alert=True) return database.remove_responsible(task_id, call.from_user.id) bot.answer_callback_query(call.id, f"✅ Вы сняты с заявки #{task_number}") _refresh_responsible(call, task_id, task_number) # ── Callback: назад (отмена выбора) ────────────────────────────────────────── @bot.callback_query_handler(func=lambda c: c.data.startswith("back:")) def cb_back(call): task_number, group_id = _parse2(call.data[5:]) bot.answer_callback_query(call.id) _restore_keyboard(call, task_number, group_id) # ── Вспомогательные функции ─────────────────────────────────────────────────── def _parse2(s: str) -> tuple[str, int]: """Парсит '{task_number}:{group_id}' → (task_number, group_id).""" parts = s.split(":", 1) return parts[0], int(parts[1]) if len(parts) > 1 and parts[1].isdigit() else 0 def _current_msg(task_number: str, call) -> tuple[int, int]: """ Возвращает (chat_id, message_id) актуального поста из БД. Если поста нет в БД — использует данные из callback (может быть устаревшим). """ task_id = database.get_task_id_by_number(task_number) if task_id: data = database.get_message_data(task_id) if data: return config.CHANNEL_ID, data["message_id"] return call.message.chat.id, call.message.message_id def _is_stale_error(e: Exception) -> bool: """Возвращает True если ошибка означает что сообщение уже удалено или не изменилось.""" s = str(e).lower() return any(x in s for x in ( "message to edit not found", "message is not modified", "message can't be edited", "bad request", )) def _restore_keyboard(call, task_number: str, group_id: int): """Восстанавливает основные кнопки на актуальном посте задачи.""" chat_id, msg_id = _current_msg(task_number, call) try: bot.edit_message_reply_markup( chat_id, msg_id, reply_markup=make_task_buttons(task_number, group_id), ) except Exception as e: if not _is_stale_error(e): log.warning(f"restore_keyboard failed: {e}") def _refresh_responsible(call, task_id: int, task_number: str): """Редактирует текст актуального поста из БД — обновляет список ответственных.""" data = database.get_message_data(task_id) if not data or not data.get("base_text"): return responsible = database.get_responsible(task_id) events_lines = database.get_stored_events(task_id) full_text = _build_full_text(data["base_text"], responsible, events_lines, new_count=0) markup = make_task_buttons(task_number, data.get("group_id") or 0) try: bot.edit_message_text( full_text, config.CHANNEL_ID, data["message_id"], parse_mode="Markdown", disable_web_page_preview=True, reply_markup=markup, ) except ApiTelegramException as e: if not _is_stale_error(e): log.warning(f"_refresh_responsible failed: {e}") def _delete_message(chat_id, message_id): try: bot.delete_message(chat_id, message_id) except Exception: pass # ── Гифка для заявок с ТО ───────────────────────────────────────────────────── def send_gif_for_to(task_id: int): """ Отправляет гифку для заявки с ТО. Предварительно удаляет предыдущую гифку если была. Если GIF_URL пустой — ничего не делает. """ if not config.GIF_URL: return # Удаляем старую гифку old_gif_id = database.get_gif_message_id(task_id) if old_gif_id: _delete_message(config.CHANNEL_ID, old_gif_id) database.delete_gif_message_id(task_id) try: msg = bot.send_animation( config.CHANNEL_ID, config.GIF_URL, disable_notification=True, ) database.save_gif_message_id(task_id, msg.message_id) log.info(f"GIF sent for task {task_id}, message_id={msg.message_id}") except Exception as e: log.error(f"Failed to send GIF for task {task_id}: {e}") # ── Запуск бота ─────────────────────────────────────────────────────────────── def start_bot(): log.info("Telegram bot started (polling)...") bot.infinity_polling(timeout=60)