#!/usr/bin/env python3 """ Слушает Telegram-бота и добавляет бонусные слова в уровни по команде. Формат сообщения боту (от тебя, в личном чате с ботом): ГЛАВНОЕ_СЛОВО НОВОЕ_СЛОВО Например: желудок рвота ГЛАВНОЕ_СЛОВО используется только для сводки в ответе (чтобы видеть, попало ли слово именно в тот уровень, который ты имел в виду) — но само добавление проходит по ВСЕМ уровням: слово добавляется в бонусные каждого уровня, где его физически можно составить из букв колеса, и где его там ещё нет. ВАЖНО: номер уровня в ответах бота — это номер из ИМЕНИ ФАЙЛА (например, 45.json -> уровень 45), а НЕ поле "id" внутри самого JSON. Так надёжнее: ты можешь свободно добавлять/удалять файлы уровней на сервере, ничего никогда не разъедется и не потребует синхронизации. Буквы "е" и "ё" считаются одной и той же буквой везде — и при поиске уровня по главному слову, и при проверке "можно ли составить слово". БЕЗОПАСНОСТЬ: бот реагирует ТОЛЬКО на сообщения от аккаунтов из ADMIN_CHAT_IDS — сообщения от кого угодно ещё полностью игнорируются. Добавлять слова через само приложение (кнопкой) нельзя в принципе — эта возможность есть только здесь, в этом отдельном скрипте на сервере. ВАЖНО: если токен бота когда-нибудь меняли — проверь, что здесь стоит актуальный (BOT_TOKEN). """ import json import os import time import urllib.request import urllib.parse from collections import Counter BOT_TOKEN = "8794231182:AAHC173mXP51ne99E8VVgfxREShtRCe7OAg" ADMIN_CHAT_IDS = {"741709485", "815047387"} LEVELS_DIR = "/home/WordsGame/levels" API_URL = f"https://api.telegram.org/bot{BOT_TOKEN}" def log(text: str) -> None: print(text, flush=True) def normalize(s: str) -> str: """Приводит к верхнему регистру и считает 'ё' то же самое, что 'е'.""" return s.upper().replace("Ё", "Е") def send_message(chat_id: str, text: str) -> None: url = f"{API_URL}/sendMessage" data = urllib.parse.urlencode({"chat_id": chat_id, "text": text}).encode("utf-8") req = urllib.request.Request(url, data=data, method="POST") try: with urllib.request.urlopen(req, timeout=10) as resp: resp.read() except Exception as e: log(f"Не удалось отправить ответ в Telegram: {e}") def get_updates(offset: int): url = f"{API_URL}/getUpdates?timeout=30&offset={offset}" with urllib.request.urlopen(url, timeout=35) as resp: return json.loads(resp.read().decode("utf-8")) def level_number_from_path(path: str) -> int: """Номер уровня — это имя файла, а не поле id внутри JSON.""" filename = os.path.basename(path) return int(filename.replace(".json", "")) def load_all_levels(): """Возвращает список (путь, словарь-уровень) для всех файлов уровней.""" levels = [] for filename in os.listdir(LEVELS_DIR): if not filename.endswith(".json"): continue path = os.path.join(LEVELS_DIR, filename) try: with open(path, encoding="utf-8") as f: level = json.load(f) levels.append((path, level)) except Exception: continue return levels def can_build(word: str, letters: list) -> bool: word_count = Counter(normalize(word)) letters_count = Counter(normalize(l) for l in letters) for ch, cnt in word_count.items(): if letters_count.get(ch, 0) < cnt: return False return True def try_add_to_level(path: str, level: dict, new_word: str) -> bool: """ Пытается добавить new_word в бонусные этого уровня. Возвращает True, если реально добавил (буквы подошли и слова там не было). """ if not can_build(new_word, level["letters"]): return False existing_grid = {normalize(w["word"]) for w in level["gridWords"]} existing_bonus = {normalize(w) for w in level["bonusWords"]} target = normalize(new_word) if target in existing_grid or target in existing_bonus: return False level["bonusWords"].append(new_word.upper()) with open(path, "w", encoding="utf-8") as f: json.dump(level, f, ensure_ascii=False, indent=2) return True def handle_message(text: str) -> str: parts = text.strip().split() if len(parts) != 2: return ( "Формат: ГЛАВНОЕ_СЛОВО НОВОЕ_СЛОВО\n" "Например: желудок рвота" ) main_word, new_word = parts if not normalize(new_word).isalpha(): return "Слово должно состоять только из букв." all_levels = load_all_levels() if not all_levels: return "На сервере не найдено ни одного файла уровня." added_count = 0 added_numbers = [] main_word_found = False main_word_result = None main_level_number = None target_letters = normalize(main_word) for path, level in all_levels: level_letters_word = normalize("".join(level.get("letters", []))) is_main_level = level_letters_word == target_letters added = try_add_to_level(path, level, new_word) if added: added_count += 1 added_numbers.append(level_number_from_path(path)) if is_main_level: main_word_found = True main_word_result = added main_level_number = level_number_from_path(path) added_numbers.sort() lines = [] if not main_word_found: lines.append(f"(Уровень с главным словом \"{main_word.upper()}\" не найден — это не помешало проверить остальные.)") elif main_word_result: lines.append(f"В том числе добавлено в уровень {main_level_number} (главное слово \"{main_word.upper()}\").") else: lines.append(f"В уровень {main_level_number} (главное слово \"{main_word.upper()}\") НЕ добавлено (буквы не подходят или слово там уже есть).") if added_count == 0: lines.insert(0, f"\"{new_word.upper()}\" не подошло ни к одному уровню (буквы не совпали или оно уже везде есть).") else: lines.insert(0, f"Готово! \"{new_word.upper()}\" добавлено в бонусные {added_count} уровня(ей): {', '.join(map(str, added_numbers))}.") return "\n".join(lines) def main(): log("Бот-слушатель запущен, ждёт сообщений...") offset = 0 while True: try: result = get_updates(offset) for update in result.get("result", []): offset = update["update_id"] + 1 message = update.get("message") if not message: continue chat_id = str(message.get("chat", {}).get("id", "")) text = message.get("text", "") log(f"Сообщение от {chat_id}: {text!r}") if chat_id not in ADMIN_CHAT_IDS: log(" -> проигнорировано (chat_id не в списке разрешённых)") continue if not text: continue reply = handle_message(text) log(f" -> ответ: {reply}") send_message(chat_id, reply) except Exception as e: log(f"Ошибка в цикле опроса: {e}") time.sleep(5) if __name__ == "__main__": main()