oldbot.py - main - rhombe/telegram

oldbot.py
216 строк11 KB
rhombe
Первый коммит
17 ноя 2025, 10:08
17 ноя 2025, 10:0893e3b15
# bot.py
import asyncio
import logging
from aiogram import Bot, Dispatcher, types
from aiogram.enums import ParseMode
from apscheduler.schedulers.asyncio import AsyncIOScheduler
from datetime import datetime
import aiohttp

# внешняя библиотека для GigaChat
from gigachat import GigaChat

logging.basicConfig(level=logging.INFO)

# ========== НАСТРОЙКИ ==========
BOT_TOKEN = "8518030150:AAGM9CM6ycJyBevHJ5ZasNTOxlZfhypkIa8"

# Вставь сюда Authorization Key (ключ, который ты получил в личном кабинете,
# выглядит как base64-строка). НЕ публикуй этот ключ где попало.
GIGA_CREDENTIALS = "MDE5YTgzNjEtN2EzYS03MTg0LWJhY2EtZDI1NmY0ZDU4OWRlOjVkN2I3YjYzLThhYzUtNDU4NC05YTIyLTYxMjY4NTU0OGYxZQ=="

# По умолчанию версия для физлиц; не трогай, если использовал personal scope
GIGA_SCOPE = "GIGACHAT_API_PERS"

SON_ID = 1732841965   # тестируем с собой
PARENT_ID = 1732841965

MAX_QUESTIONS = 6     # максимально допустимое количество вопросов в цепочке

# хранение структуры диалога
daily_messages = []         # список строк "Вопрос: ... | Ответ: ..."
pending_question = None     # текущий открытый вопрос (если есть)


# ========== Вспомогательная функция: запрос к GigaChat ==========
# Реализация использует синхронный клиент gigachat внутри потока,
# чтобы не блокировать событийный цикл asyncio.
async def ask_gigachat(prompt: str, model: str | None = None) -> str:
    """
    Отправляет prompt в GigaChat и возвращает текст ответа.
    В случае ошибки возвращает понятное сообщение и логирует полный ответ.
    """
    def sync_call():
        try:
            # создаём клиент, передаём credentials; отключаем проверку сертификатов
            # только если у тебя возникнут ошибки с корневым CA; безопаснее убрать verify_ssl_certs=False
            giga = GigaChat(credentials=GIGA_CREDENTIALS, scope=GIGA_SCOPE, verify_ssl_certs=False)
            # если нужно, можно явно указать модель: giga.chat(prompt, model="...") (см. docs)
            if model:
                response = giga.chat(prompt, model=model)
            else:
                response = giga.chat(prompt)
            # response должен иметь структуру .choices[0].message.content
            try:
                return response.choices[0].message.content
            except Exception:
                # если структура неожиданна — вернём текст представления
                return str(response)
        except Exception as e:
            # возвращаем текст ошибки для обработки в асинхронном коде
            return f"__GIGACHAT_ERROR__:{repr(e)}"

    resp = await asyncio.to_thread(sync_call)
    # обработка ошибки
    if isinstance(resp, str) and resp.startswith("__GIGACHAT_ERROR__:"):
        logging.error("GigaChat sync error: %s", resp)
        return "К сожалению, сервис генерации ответа недоступен. Попробуй ещё раз позже."
    return resp


# ========== логика формирования следующего вопроса ==========
async def get_next_question_from_giga(prev_answer: str) -> str:
    prompt = (
        "На основе этого ответа сформулируй понятный уточняющий вопрос, "
        "который поможет собрать полезную информацию. "
        "Ответ должен быть коротким, вежливым и прямо задавать уточнение.\n\n"
        f"Ответ: {prev_answer}"
    )
    return await ask_gigachat(prompt)


# ========== логика определения, достаточно ли информации ==========
async def should_finish_giga(prev_answer: str) -> bool:
    """
    Спрашиваем у GigaChat, достаточно ли накопленных ответов для итоговой сводки.
    Возвращаем True/False.
    """
    prompt = (
        "Исходя из этого последнего ответа, достаточно ли информации, чтобы "
        "сформировать итоговую сводку (коротко описать день)? "
        "Ответь одним словом: да или нет. Причину не указывай.\n\n"
        f"Ответ: {prev_answer}"
    )
    resp = await ask_gigachat(prompt)
    # упрощённая логика: смотрим на слово "да" в ответе
    if isinstance(resp, str) and "да" in resp.lower():
        return True
    return False


# ========== генерация итоговой сводки ==========
async def summarize_messages(messages: list[str]) -> str:
    if not messages:
        return "Сегодня не было сообщений."
    prompt = "Сделай короткую, спокойную и человеческую сводку по этим записям:\n\n" + "\n".join(messages)
    return await ask_gigachat(prompt)


# ========== первый вопрос (тестовый режим: сразу после старта) ==========
async def first_question(bot: Bot):
    global pending_question
    text = "Привет! 😊 Давай начнём дневник. Как у тебя дела сегодня?"
    try:
        await bot.send_message(SON_ID, text)
        pending_question = text
    except Exception as e:
        logging.error("Не удалось отправить первый вопрос: %s", e)


# ========== обработка входящих сообщений ==========
async def on_message(message: types.Message):
    global daily_messages, pending_question

    # принимаем ответы только от ожидаемого чата (для теста — твой ID)
    if message.chat.id != SON_ID or not message.text:
        return

    # сохраняем: сохраняем связку вопрос/ответ если был вопрос
    if pending_question:
        daily_messages.append(f"Вопрос: {pending_question}\nОтвет: {message.text}")
    else:
        daily_messages.append(f"Ответ: {message.text}")

    # решить, закончить ли опрос
    # сперва пробуем интеллектуальную оценку через GigaChat (с защитой от ошибок)
    finish = False
    try:
        # если накопилось слишком много — обязательно завершить
        if len(daily_messages) >= MAX_QUESTIONS:
            finish = True
        else:
            # ask whether to finish (может вернуть текст ошибки, обработаем спокойно)
            finish = await should_finish_giga(message.text)
    except Exception as e:
        logging.exception("Ошибка при should_finish_giga: %s", e)
        finish = False  # на ошибке не завершаем

    if finish:
        # отправляем пользователю подтверждение и формируем сводку
        try:
            await message.answer("Спасибо — формирую итоговую сводку по ответам...")
            summary = await summarize_messages(daily_messages)
            await message.answer(summary)
            # отправляем родителю
            await message.bot.send_message(PARENT_ID, f"📘 *Ежедневный отчёт:*\n\n{summary}", parse_mode=ParseMode.MARKDOWN)
        except Exception as e:
            logging.exception("Ошибка при отправке сводки: %s", e)
        finally:
            daily_messages = []
            pending_question = None
        return

    # иначе генерируем следующий вопрос
    try:
        next_q = await get_next_question_from_giga(message.text)
        # если GigaChat вернул текст с признаком ошибки, заменим на fallback
        if next_q.startswith("__GIGACHAT_ERROR__") or next_q.strip() == "":
            next_q = "Расскажи об одной небольшой задаче, которую ты сегодня решил."
        await message.bot.send_message(SON_ID, next_q)
        pending_question = next_q
    except Exception as e:
        logging.exception("Ошибка при get_next_question_from_giga: %s", e)
        # fallback-вопрос
        fallback = "Можешь рассказать подробнее об одной ситуации сегодня?"
        await message.bot.send_message(SON_ID, fallback)
        pending_question = fallback


# ========== вечерняя сводка, вызывается по расписанию ==========
async def evening_summary(bot: Bot):
    global daily_messages, pending_question
    try:
        if daily_messages:
            summary_lines = list(daily_messages)
            if pending_question:
                summary_lines.append(f"Вопрос без ответа: {pending_question}")
            summary = await summarize_messages(summary_lines)
        else:
            summary = "Сегодня не было сообщений."
        await bot.send_message(PARENT_ID, f"📘 *Ежедневный отчёт:*\n\n{summary}", parse_mode=ParseMode.MARKDOWN)
    except Exception as e:
        logging.exception("Ошибка при формировании вечерней сводки: %s", e)
    finally:
        daily_messages = []
        pending_question = None


# ========== Главная функция: запуск бота и планировщик ==========
async def main():
    bot = Bot(token=BOT_TOKEN)
    dp = Dispatcher()
    dp.message.register(on_message)

    scheduler = AsyncIOScheduler(timezone="Europe/Moscow")
    # первый вопрос сразу после старта (тестовый режим)
    scheduler.add_job(first_question, "date", run_date=datetime.now(), args=[bot])
    # вечерняя сводка в 21:00
    scheduler.add_job(evening_summary, "cron", hour=21, minute=0, args=[bot])
    scheduler.start()

    print("Бот запущен — ожидаем сообщений...")
    await dp.start_polling(bot)


if __name__ == "__main__":
    asyncio.run(main())