bot.py - main - rhombe/telegram

bot.py
470 строк21 KB
rhombe
Первый коммит
17 ноя 2025, 10:08
17 ноя 2025, 10:0893e3b15

# bot.py (версия 2.1 — логика обновлена)
import asyncio
import logging
import random
import re
from datetime import datetime
from aiogram import Bot, Dispatcher, types
from aiogram.enums import ParseMode
from apscheduler.schedulers.asyncio import AsyncIOScheduler

from gigachat import GigaChat  # pip install gigachat

logging.basicConfig(level=logging.INFO)

# ========== НАСТРОЙКИ (оставлены как в предыдущем файле) ==========
BOT_TOKEN = "8518030150:AAGM9CM6ycJyBevHJ5ZasNTOxlZfhypkIa8"
GIGA_CREDENTIALS = "MDE5YTgzNjEtN2EzYS03MTg0LWJhY2EtZDI1NmY0ZDU4OWRlOjVkN2I3YjYzLThhYzUtNDU4NC05YTIyLTYxMjY4NTU0OGYxZQ=="
GIGA_SCOPE = "GIGACHAT_API_PERS"

SON_ID = 1732841965
PARENT_ID = 1732841965

MAX_QUESTIONS = 30
BLOCK_KEYS = ["школа", "спорт", "досуг", "планы"]

# ========== Состояние сессии (память в ОЗУ для теста) ==========
daily_messages: list[str] = []
pending_question: str | None = None
questions_asked = 0

blocks_status = {
    "школа": {"closed": False, "clarify_asked": False, "answers": []},
    "спорт": {"closed": False, "clarify_asked": False, "answers": []},
    "досуг": {"closed": False, "clarify_asked": False, "answers": []},
    "планы": {"closed": False, "clarify_asked": False, "answers": []},
}

# ========== Шаблоны / промпты ==========
# Общие правила стиля батя (для включения в промпты)
BATYA_STYLE_BRIEF = (
    "Ты — батя, дружеский, немного саркастичный, без оскорблений. "
    "НЕ используй обращения 'сынок','дочка' и т.п. Не ставь имя каждый раз в начале фразы. "
    "Формат: короткая реакция (одно предложение) и затем один короткий вопрос (если нужен)."
)

GEN_QUESTION_PROMPT = (
    BATYA_STYLE_BRIEF
    + "\n\nКонтекст (коротко):\n{context}\n\n"
    "Нужно уточнить по блоку: {block}\n"
    "Известная информация по этому блоку: {facts}\n\n"
    "Задай ОДИН живой, оригинальный, не шаблонный вопрос в дружеском батином стиле. "
    "Не выдумывай фактов, не используй обращения кроме редкой вставки 'Кирюха'. "
    "Вопрос должен быть коротким (не более 12 слов). Реакция не более 12 слов. "
    "Не повторяй предыдущий вопрос дословно."
)

SUMMARY_PROMPT = (
    "Ты — батя, составляющий отчёт для родителя. ВАЖНО: НЕЛЬЗЯ ВЫДУМЫВАТЬ ФАКТОВ.\n"
    "Используй только то, что прямо сказано в записях (вопросы бота и ответы Кирюхи). "
    "Если по теме нет данных — напиши: 'Нет данных по теме <название>'. "
    "Сделай краткую сводку (3–6 предложений) в дружеском стиле, но без домыслов.\n\n"
    "Записи:\n{context}\n\nСводка:"
)

CLASSIFY_PROMPT = (
    "Классифицируй текст по одной из категорий: школа, спорт, досуг, планы, нет. "
    "Также укажи, содержит ли текст конкретику (specific) или общий ответ (general). "
    "Верни строго: категория|specific  или  категория|general\n\nТекст:\n\"\"\"{answer}\"\"\"\n"
)

START_PHRASES = [
    "Коротко: как день идёт?",
    "Как настроение сейчас?",
    "Коротко: что сегодня интересного было?",
    "Как начался день?",
    "Расскажи — как дела и что делал?"
]

# ========== GigaChat wrapper ==========
async def ask_gigachat(prompt: str, model: str | None = None) -> str:
    def sync_call():
        try:
            giga = GigaChat(credentials=GIGA_CREDENTIALS, scope=GIGA_SCOPE, verify_ssl_certs=False)
            if model:
                r = giga.chat(prompt, model=model)
            else:
                r = giga.chat(prompt)
            try:
                return r.choices[0].message.content
            except Exception:
                return str(r)
        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 error: %s", resp)
    return resp

# ========== Утилиты: шум, остановка, санитизация ==========
def is_noise_or_short(text: str) -> bool:
    t = text.strip().lower()
    if not t:
        return True
    letters = re.sub(r'[^A-Za-zА-Яа-яЁё]', '', t)
    if letters:
        cyr = sum(1 for ch in letters if 'а' <= ch <= 'я' or ch == 'ё')
        if len(letters) > 3 and (cyr / len(letters) < 0.5):
            return True
    if re.search(r'\b[^аеёиоуыэюя]{6,}\b', t):
        return True
    if re.fullmatch(r'[\W_]{1,4}', t):
        return True
    if t in ("ау", "ъ", "хз", "нз", "не знаю", "непомню"):
        return True
    if len(re.sub(r'\s+', '', t)) <= 2:
        return True
    return False

STOP_INTENT_PATTERNS = [
    r'\b(хватит|хватит уже|перестань|перестань уже|завязывай|не хочу про|не хочу|отстань|стоп)\b',
    r'\b(надоел|надоело|ты достал|задолбал)\b',
    r'\b(не буду|не собираюсь|не хочу отвечать)\b'
]

def detect_stop_intent(text: str) -> bool:
    t = text.strip().lower()
    for p in STOP_INTENT_PATTERNS:
        if re.search(p, t):
            return True
    return False

FORBIDDEN_ADDR = [
    r'\bсынок\b', r'\bсын\b', r'\bсына\b', r'\bсынуля\b',
    r'\bдочка\b', r'\bдоча\b', r'\bмалыш\b', r'\bпарень\b', r'\bмальчик\b'
]
def sanitize_batya_text(text: str) -> str:
    # Replace forbidden addresses with 'Кирюха' only if necessary, but avoid repeating name
    res = text
    for pat in FORBIDDEN_ADDR:
        res = re.sub(pat, "Кирюха", res, flags=re.I)
    # collapse repeated mentions of name
    res = re.sub(r'(Кирюха[\s,:-]*){2,}', 'Кирюха, ', res, flags=re.I)
    # ensure name not at start of every sentence: if appears more than once, keep only first
    parts = re.split(r'([.!?]\s+)', res)
    name_count = res.lower().count('кирюха')
    if name_count > 1:
        first_kept = False
        new_parts = []
        for part in parts:
            if 'кирюха' in part.lower():
                if not first_kept:
                    new_parts.append(part)
                    first_kept = True
                else:
                    new_parts.append(re.sub(r'(?i)кирюха[,:\s-]*', '', part))
            else:
                new_parts.append(part)
        res = ''.join(new_parts)
    # remove rude words
    res = re.sub(r'\b(ублюдок|пидор|сволочь)\b', '...', res, flags=re.I)
    res = re.sub(r'\s{2,}', ' ', res)
    return res.strip()

# ========== Логика выбора блока с учётом времени и выходных ==========
def is_weekend() -> bool:
    wd = datetime.now().weekday()
    return wd >= 5

def choose_next_block():
    hour = datetime.now().hour
    weekend = is_weekend()
    order = []
    if weekend:
        if hour < 12:
            order = ["досуг", "планы", "спорт", "школа"]
        elif hour < 18:
            order = ["досуг", "спорт", "планы", "школа"]
        else:
            order = ["планы", "досуг", "спорт", "школа"]
    else:
        if 6 <= hour < 9:
            order = ["планы", "досуг", "спорт", "школа"]
        elif 9 <= hour < 15:
            order = ["школа", "спорт", "досуг", "планы"]
        elif 15 <= hour < 21:
            order = ["школа", "спорт", "досуг", "планы"]
        else:
            order = ["досуг", "планы", "спорт", "школа"]

    for b in order:
        if not blocks_status[b]["closed"] and not blocks_status[b]["clarify_asked"]:
            return b
    for b in order:
        if not blocks_status[b]["closed"]:
            return b
    return None

# ========== Генерация вопроса через GigaChat (уникальный, не шаблонный) ==========
async def generate_batya_question(history: list[str], block: str) -> str:
    facts = ""
    if blocks_status.get(block):
        facts = "; ".join(blocks_status[block]["answers"][-3:])
    ctx = "\n".join(history[-12:])
    prompt = GEN_QUESTION_PROMPT.format(context=ctx, block=block, facts=facts)
    # Add explicit instruction to avoid name repetition and keep brevity
    prompt += "\n\nДополнительно: не начинай каждый вопрос с имени; не используй имя в более чем одном предложении; ответ должен быть максимально коротким."
    resp = await ask_gigachat(prompt)
    if not resp or resp.startswith("__GIGACHAT_ERROR__"):
        if block == "школа":
            return "Как прошли занятия сегодня?"
        if block == "спорт":
            return "Занимался сегодня спортом?"
        if block == "досуг":
            return "Чем развлекался сегодня?"
        if block == "планы":
            return "Какие планы на вечер?"
    # sanitize generated text
    resp = sanitize_batya_text(resp)
    # enforce length limits: if longer than ~180 chars, truncate to first sentence + question
    if len(resp) > 200:
        # try to split into sentences and keep first two
        sents = re.split(r'(?<=[.!?])\s+', resp)
        resp = ' '.join(sents[:2])
    return resp

# ========== Классификация ответа ==========
async def classify_answer(answer: str) -> tuple[str, str]:
    prompt = CLASSIFY_PROMPT.format(answer=answer)
    resp = await ask_gigachat(prompt)
    if not resp or resp.startswith("__GIGACHAT_ERROR__"):
        return "нет", "general"
    text = resp.strip().lower()
    if "|" in text:
        cat, spec = [p.strip() for p in text.split("|", 1)]
        if cat not in BLOCK_KEYS and cat != "нет":
            return "нет", "general"
        return cat, spec if spec in ("specific", "general") else "general"
    else:
        if text in BLOCK_KEYS:
            return text, "specific"
        return "нет", "general"

# ========== Итоговая сводка (строго без домыслов) ==========
async def generate_summary_strict(history: list[str]) -> str:
    ctx = "\n".join(history)
    prompt = SUMMARY_PROMPT.format(context=ctx)
    resp = await ask_gigachat(prompt)
    if not resp or resp.startswith("__GIGACHAT_ERROR__"):
        return "Не удалось сформировать сводку — сервис недоступен."
    return sanitize_batya_text(resp)

# ========== Первый вопрос (тестовый режим) ==========
async def first_question(bot: Bot):
    global pending_question, questions_asked
    q = random.choice(START_PHRASES)
    try:
        await bot.send_message(SON_ID, q)
        pending_question = q
        questions_asked += 1
    except Exception as e:
        logging.exception("Ошибка отправки первого вопроса: %s", e)

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

    if message.chat.id != SON_ID or not message.text:
        return

    user_text = message.text.strip()
    now_hour = datetime.now().hour

    # 1) шум/набор букв -> мягкая отмашка
    if is_noise_or_short(user_text):
        reply = random.choice([
            "Не понял, скажи проще, ладно?",
            "Слишком коротко — скажи пару слов понятных.",
            "Хмм, не врубился. Может, опиши одним простым предложением."
        ])
        await message.bot.send_message(SON_ID, reply)
        pending_question = reply
        questions_asked += 1
        daily_messages.append(f"Вопрос: {pending_question}\nОтвет: {user_text}")
        return

    # 2) stop intent
    if detect_stop_intent(user_text):
        target = choose_next_block()
        if target:
            blocks_status[target]["answers"].append("отказ: пользователь попросил не продолжать")
            blocks_status[target]["closed"] = True
        await message.bot.send_message(SON_ID, random.choice([
            "Окей, понял — не буду дальше тянуть эту тему.",
            "Хорошо, хватит. Переходим к другому.",
            "Ладно, не буду давить, расскажи про что-то другое."
        ]))
        pending_question = None
        questions_asked += 1
        daily_messages.append(f"Ответ: {user_text} (стоп-интенция)")
        return

    # 3) save Q/A pair
    if pending_question:
        daily_messages.append(f"Вопрос: {pending_question}\nОтвет: {user_text}")
    else:
        daily_messages.append(f"Ответ: {user_text}")

    # 4) classify
    category, specificity = await classify_answer(user_text)
    logging.info("Классификация: %s | %s", category, specificity)

    # 5) refusal phrases
    if re.search(r'\b(не буду|не хочу|не отвечаю|не скажу)\b', user_text.lower()):
        target = choose_next_block()
        if target:
            blocks_status[target]["answers"].append("отказ: не хотел отвечать")
            blocks_status[target]["closed"] = True
        await message.bot.send_message(SON_ID, "Понял, не настаиваю.")
        pending_question = None
        questions_asked += 1
        return

    # 6) handle categorized content
    if category in BLOCK_KEYS:
        # save answer
        blocks_status[category]["answers"].append(user_text)
        # enhanced closure: if specific OR answer reasonably informative -> close
        if specificity == "specific" or len(re.sub(r'\s+','',user_text)) > 20:
            blocks_status[category]["closed"] = True
        else:
            if not blocks_status[category]["clarify_asked"]:
                # time rule: avoid school questions early
                if category == "школа" and (is_weekend() or now_hour < 9):
                    blocks_status[category]["clarify_asked"] = True
                else:
                    clar_q = await generate_batya_question(daily_messages, category)
                    # ensure clar_q doesn't start with name repeatedly
                    clar_q = sanitize_batya_text(clar_q)
                    await message.bot.send_message(SON_ID, clar_q)
                    pending_question = clar_q
                    blocks_status[category]["clarify_asked"] = True
                    questions_asked += 1
                    return
            else:
                blocks_status[category]["closed"] = True
    else:
        # pick next block respecting time of day
        target = choose_next_block()
        if target:
            if target == "школа" and (is_weekend() or now_hour < 9):
                alt = "досуг" if not blocks_status["досуг"]["closed"] else "планы"
                next_q = await generate_batya_question(daily_messages, alt)
                next_q = sanitize_batya_text(next_q)
                await message.bot.send_message(SON_ID, next_q)
                pending_question = next_q
                blocks_status[alt]["clarify_asked"] = True
                questions_asked += 1
                return
            else:
                if not blocks_status[target]["clarify_asked"]:
                    next_q = await generate_batya_question(daily_messages, target)
                    next_q = sanitize_batya_text(next_q)
                    await message.bot.send_message(SON_ID, next_q)
                    pending_question = next_q
                    blocks_status[target]["clarify_asked"] = True
                    questions_asked += 1
                    return
                else:
                    next_q = await generate_batya_question(daily_messages, target)
                    next_q = sanitize_batya_text(next_q)
                    await message.bot.send_message(SON_ID, next_q)
                    pending_question = next_q
                    questions_asked += 1
                    return

    # 7) finish condition
    all_closed = all(blocks_status[b]["closed"] for b in BLOCK_KEYS)
    if all_closed or questions_asked >= MAX_QUESTIONS:
        h = now_hour
        if h >= 22:
            bye = "Ладно, отдыхай. Спокойной ночи."
        elif 18 <= h < 22:
            bye = "Хорошего вечера."
        else:
            bye = "Ладно, давай дальше по делам."
        try:
            await message.answer(bye)
        except Exception:
            pass

        summary = await generate_summary_strict(daily_messages)
        try:
            await message.bot.send_message(SON_ID, summary)
        except Exception:
            logging.exception("Не удалось отправить сводку сыну.")
        try:
            await message.bot.send_message(PARENT_ID, f"📘 *Ежедневный отчёт:*\n\n{summary}", parse_mode=ParseMode.MARKDOWN)
        except Exception:
            logging.exception("Не удалось отправить сводку родителю.")

        # reset
        daily_messages = []
        pending_question = None
        questions_asked = 0
        for b in BLOCK_KEYS:
            blocks_status[b] = {"closed": False, "clarify_asked": False, "answers": []}
        return

    # 8) ask next block question
    target = choose_next_block()
    if target:
        if not blocks_status[target]["clarify_asked"]:
            next_q = await generate_batya_question(daily_messages, target)
            next_q = sanitize_batya_text(next_q)
            await message.bot.send_message(SON_ID, next_q)
            pending_question = next_q
            blocks_status[target]["clarify_asked"] = True
            questions_asked += 1
            return
        else:
            next_q = await generate_batya_question(daily_messages, target)
            next_q = sanitize_batya_text(next_q)
            await message.bot.send_message(SON_ID, next_q)
            pending_question = next_q
            questions_asked += 1
            return

    fallback = "Давай коротко: что самое важное сегодня?"
    await message.bot.send_message(SON_ID, fallback)
    pending_question = fallback
    questions_asked += 1
    return

# ========== Вечерняя сводка (cron-ready) ==========
async def evening_summary(bot: Bot):
    global daily_messages, pending_question, questions_asked
    if daily_messages:
        history = daily_messages + ([f"Вопрос без ответа: {pending_question}"] if pending_question else [])
        summary = await generate_summary_strict(history)
    else:
        summary = "Сегодня нет данных."
    try:
        await bot.send_message(PARENT_ID, f"📘 *Ежедневный отчёт:*\n\n{summary}", parse_mode=ParseMode.MARKDOWN)
    except Exception:
        logging.exception("Ошибка отправки вечерней сводки.")
    # reset
    daily_messages = []
    pending_question = None
    questions_asked = 0
    for b in BLOCK_KEYS:
        blocks_status[b] = {"closed": False, "clarify_asked": False, "answers": []}

# ========== MAIN ==========
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())