import os import sys import time import json import uuid import secrets import logging import logging.handlers import random import shutil import threading from datetime import datetime, timedelta from typing import Dict, Optional, Any, List, Tuple import base64 from dotenv import load_dotenv import requests import vk_api import re from vk_api.bot_longpoll import VkBotLongPoll, VkBotEventType from vk_api.keyboard import VkKeyboard, VkKeyboardColor from vk_api.utils import get_random_id from vk_api.exceptions import ApiError from vk_api import upload from collections import Counter from PIL import Image, ImageDraw, ImageFont import io import textwrap import db # Лекции # ========== КОНСТАНТЫ ========== ADMIN_ID = 603557110 DATA_FILE = "bot_data.json" CACHE_FILE_PREFIX = "rasp_cache" # Папка для кэша расписаний (по группам) RASP_DIR = "rasp" # Папка для логов LOG_DIR = "logs" os.makedirs(RASP_DIR, exist_ok=True) os.makedirs(LOG_DIR, exist_ok=True) CACHE_TTL = 3600 NOTIFIED_EXPIRED_TOKENS = set() START_TIME = time.time() IDEAS_FILE = "ideas.json" # ========== ТИПЫ УВЕДОМЛЕНИЙ ========== # Добавляй новые типы сюда — кнопки и статус появятся автоматически NOTIFY_TYPES = [ {"key": "reply", "label": "Ответы администратора", "desc": "Когда админ отвечает на ваши баги и идеи", "emoji": "📬"}, ] def get_user_notify_status(user_id, key): if key == "reply": return db.get_user_notify_reply(user_id) return True def set_user_notify_status(user_id, key, value): if key == "reply": db.set_user_notify_reply(user_id, value) def build_notify_menu(user_id): msg = "🔔 **Управление уведомлениями**\n\n" for nt in NOTIFY_TYPES: val = get_user_notify_status(user_id, nt['key']) icon = "✅" if val else "❌" msg += f"{nt['emoji']} **{nt['label']}**\n{nt['desc']}\n▸ Статус: {icon}\n\n" msg += "Выберите уведомление для изменения:" kb = VkKeyboard(one_time=True) for nt in NOTIFY_TYPES: val = get_user_notify_status(user_id, nt['key']) btn_text = f"{'❌ Выкл' if val else '✅ Вкл'} {nt['emoji']} {nt['label']}" kb.add_button(btn_text, color=VkKeyboardColor.NEGATIVE if val else VkKeyboardColor.POSITIVE) kb.add_line() kb.add_button("🔙 Назад", color=VkKeyboardColor.SECONDARY) return msg, kb # ========== СООБЩЕНИЕ ПРИ ЛЮБОЙ ОБНОВЕ ============================== BOT_VERSION = "1.06" UPDATE_MESSAGE = ( f"🔔 Бот обновлён до версии {BOT_VERSION}!\n\n" "Кнопки «Следующая / Предыдущая» заменили на 1️⃣2️⃣3️⃣ — меню переключается быстрее!\n\n" "🆕 Что нового:\n" "— Улучшено переключение страниц клавиатуры (1️⃣2️⃣3️⃣)\n" "— В личном кабинете появилась история обновлений\n" "— Исправлены ошибки предыдущей версии\n\n" "Спасибо, что пользуетесь ботом!" ) #=========== ДЕВАЙСЫ ============= CLIENT_VERSION = "2026-05-20T18:59:45.284Z" LATEST_CLIENT_VERSION = "2026-05-20T18:59:45.284Z" DEVICE_PROFILES = { "windows": { "user_agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/148.0.0.0 Safari/537.36", "sec_ch_ua": '"Chromium";v="148", "Brave";v="148", "Not/A)Brand";v="99"', "sec_ch_ua_mobile": "?0", "sec_ch_ua_platform": '"Windows"', }, "linux": { "user_agent": "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/148.0.0.0 Safari/537.36", "sec_ch_ua": '"Chromium";v="148", "Brave";v="148", "Not/A)Brand";v="99"', "sec_ch_ua_mobile": "?0", "sec_ch_ua_platform": '"Linux"', } } DEFAULT_PLATFORM = "windows" # ========== НАСТРОЙКА ЛОГИРОВАНИЯ ========== logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', handlers=[ logging.handlers.RotatingFileHandler(os.path.join(LOG_DIR, "bot.log"), maxBytes=5*1024*1024, backupCount=2, encoding="utf-8"), logging.StreamHandler() ] ) logger = logging.getLogger("ZoLiryzikBot") # ========== ЛОГИРОВАНИЕ ЗАПРОСОВ К API ДЕКАНАТА ========== api_logger = logging.getLogger("DecanatAPI") api_logger.setLevel(logging.INFO) api_file_handler = logging.handlers.RotatingFileHandler(os.path.join(LOG_DIR, "api_requests.log"), maxBytes=5*1024*1024, backupCount=2, encoding="utf-8") api_file_handler.setFormatter(logging.Formatter('%(asctime)s - %(levelname)s - %(message)s', datefmt='%Y-%m-%d %H:%M:%S')) api_logger.addHandler(api_file_handler) api_console_handler = logging.StreamHandler() api_console_handler.setFormatter(logging.Formatter('%(asctime)s - %(message)s', datefmt='%H:%M:%S')) api_logger.addHandler(api_console_handler) # ========== ЗАГРУЗКА .env ========== load_dotenv() VK_TOKEN = os.getenv("VK_TOKEN") GROUP_ID_VK = os.getenv("GROUP_ID_VK") if not VK_TOKEN or not GROUP_ID_VK: logger.critical("❌ Нет VK_TOKEN или GROUP_ID_VK в .env") sys.exit(1) try: GROUP_ID_VK = int(GROUP_ID_VK) except ValueError: logger.critical("❌ GROUP_ID_VK должен быть числом") sys.exit(1) # ========== ЗАГРУЗКА ГРУПП ДЕКАНАТА (с ID студента) ========== GROUPS: Dict[int, Dict[str, Any]] = {} for i in range(1, 11): group_id_str = os.getenv(f"GROUP{i}_ID") token = os.getenv(f"GROUP{i}_TOKEN") user_id_str = os.getenv(f"GROUP{i}_USER_ID") # ID студента (владельца токена) peer_id_str = os.getenv(f"GROUP{i}_PEER_ID") login = os.getenv(f"GROUP{i}_LOGIN") password = os.getenv(f"GROUP{i}_PASSWORD") platform = os.getenv(f"GROUP{i}_PLATFORM", DEFAULT_PLATFORM) if group_id_str and token and user_id_str and peer_id_str: try: group_id = int(group_id_str) user_id = int(user_id_str) peer_id = int(peer_id_str) GROUPS[group_id] = { "token": token, "user_id": user_id, "peer_id": peer_id, "login": login, "password": password, "platform": platform } logger.info(f"✅ Загружена группа {group_id} (user_id={user_id}, peer_id={peer_id})") except ValueError: logger.error(f"❌ GROUP{i}_ID, GROUP{i}_USER_ID или GROUP{i}_PEER_ID должны быть числами, пропускаю") elif group_id_str or token or user_id_str or peer_id_str: logger.warning(f"⚠️ Для группы {i} не хватает параметров, пропускаю") if not GROUPS: logger.critical("❌ Не загружено ни одной группы деканата. Проверьте .env") sys.exit(1) logger.info(f"✅ Всего загружено групп: {len(GROUPS)}") # ---------- ВСТАВИТЬ ЗДЕСЬ ---------- def get_token_exp(token: str): """Извлекает timestamp истечения JWT-токена.""" try: parts = token.strip().split('.') if len(parts) != 3: return None payload = parts[1] payload += '=' * (4 - len(payload) % 4) decoded = json.loads(base64.urlsafe_b64decode(payload)) return decoded.get('exp') except: return None # ----- конец вставки ----- db.init_db() # Сохраняем текущую версию бота в историю обновлений db.save_bot_version(BOT_VERSION, UPDATE_MESSAGE) user_names_cache = {} admin_state = {} accounts = db.get_all_group_accounts() accounts_lock = threading.Lock() env_lock = threading.Lock() def save_accounts(acc: Dict[int, Dict[str, Any]]) -> None: with accounts_lock: for gid, data in acc.items(): db.save_group_account(gid, data) for gid in list(accounts.keys()): if accounts[gid].get("blocked") and not accounts[gid].get("revoked"): accounts[gid]["blocked"] = False db.save_group_account(gid, accounts[gid]) logger.info(f"🔄 Сброшена блокировка группы {gid} при перезапуске") # Заполняем token_expires_at для групп, у которых его ещё нет with accounts_lock: for gid, ginfo in GROUPS.items(): token = ginfo.get("token") if token and (gid not in accounts or not accounts[gid].get("token_expires_at")): exp_ts = get_token_exp(token) if exp_ts: if gid not in accounts: accounts[gid] = {} accounts[gid]["token_expires_at"] = exp_ts # Сохраняем под тем же замком for gid, data in accounts.items(): db.save_group_account(gid, data) def parse_api_log(real_requests_only=True): stats = {"groups": {}, "users": {}} log_file = os.path.join(LOG_DIR, "api_requests.log") if not os.path.exists(log_file): return stats with open(log_file, "r", encoding="utf-8") as f: for line in f: if real_requests_only and "🌐 ЗАПРОС" not in line: continue group_match = re.search(r"группа=(\d+)", line) user_match = re.search(r"user_id=(\d+)", line) if group_match: gid = int(group_match.group(1)) stats["groups"][gid] = stats["groups"].get(gid, 0) + 1 if user_match: uid = int(user_match.group(1)) stats["users"][uid] = stats["users"].get(uid, 0) + 1 return stats def parse_commands_log(): """Читает commands.log и возвращает: - all_msgs: общее количество сообщений по пользователям, - commands: количество команд (только распознанные) по пользователям, - command_names: количество по типам команд. """ all_msgs = Counter() commands = Counter() command_names = Counter() log_file = os.path.join(LOG_DIR, "commands.log") if not os.path.exists(log_file): return all_msgs, commands, command_names base_commands = {"сегодня", "завтра", "день", "неделя", "след", "след.", "следующая", "обновить", "сброс", "помощь", "меню"} with open(log_file, "r", encoding="utf-8") as f: for line in f: line = line.strip() if not line: continue parts = line.split(" - ", 2) if len(parts) != 3: continue _, user_id_str, text = parts try: user_id = int(user_id_str) except ValueError: continue all_msgs[user_id] += 1 if re.fullmatch(r'[0-9a-f]{16}', text): continue clean = re.sub(r'^[^\w\s]+', '', text).strip().lower() words = clean.split() if not words: continue cmd = words[0] # Пропускаем команды навигации по страницам if "страница" in clean: continue normalized_cmd = cmd if cmd in ("след", "след.", "следующая", "следующая_неделя", "след_неделя"): normalized_cmd = "след" is_command = False if cmd in base_commands: is_command = True elif cmd.startswith('!') or cmd in ["отозвать"]: is_command = True if is_command: commands[user_id] += 1 command_names[normalized_cmd] += 1 return all_msgs, commands, command_names def get_top_users_by_commands(n=5, vk=None): all_msgs, commands, command_names = parse_commands_log() if not commands: return "📭 Нет данных о командах." top = commands.most_common(n) lines = [f"🏆 Топ-{n} пользователей по количеству команд:\n"] for i, (uid, count) in enumerate(top, 1): name = get_user_name(vk, uid) if vk else f"id{uid}" lines.append(f"{i}. {name} (ID {uid}) — {count}") return "\n".join(lines) def get_top_commands(n=5): all_msgs, commands, command_names = parse_commands_log() if not command_names: return "📭 Нет данных о командах." top = command_names.most_common(n) lines = [f"🏆 Топ-{n} команд:\n"] for i, (cmd, count) in enumerate(top, 1): lines.append(f"{i}. {cmd} — {count}") return "\n".join(lines) def get_top_users_by_commands_in_group(group_id: int, n: int = 5, vk=None): """Возвращает топ-N пользователей группы по количеству команд (только распознанные).""" # Получаем всех зарегистрированных пользователей группы users_in_group = [uid for uid, _ in get_users_by_group(group_id)] if not users_in_group: return f"📭 В группе {group_id} нет зарегистрированных пользователей." # Получаем счётчик команд (только распознанные) _, commands, _ = parse_commands_log() # Фильтруем только тех, кто есть в группе group_commands = {uid: commands[uid] for uid in users_in_group if uid in commands} if not group_commands: return f"📭 Нет данных о командах для группы {group_id}." top = sorted(group_commands.items(), key=lambda x: x[1], reverse=True)[:n] lines = [f"🏆 Топ-{n} пользователей в группе {group_id} по количеству команд:\n"] for i, (uid, count) in enumerate(top, 1): name = get_user_name(vk, uid) if vk else f"id{uid}" lines.append(f"{i}. {name} (ID {uid}) — {count}") return "\n".join(lines) def revoke_consent(group_id: int, owner_id: int, vk) -> bool: """Отзывает согласие: блокирует группу, удаляет consent и сессионные данные.""" with accounts_lock: group_data = accounts.get(group_id) if not group_data or group_data.get("owner_id") != owner_id: return False owner_name = group_data.get("owner_name", f"id{owner_id}") # Помечаем группу как заблокированную и отозванную group_data["blocked"] = True group_data["revoked"] = True # Удаляем consent и сессионные данные (куки) if "consent" in group_data: del group_data["consent"] if "session" in group_data: del group_data["session"] # Сохраняем изменения в БД (всё ещё под замком) db.save_group_account(group_id, group_data) # Дальше без блокировки — получение списка пользователей и рассылка уведомлений users_in_group = get_users_by_group(group_id) # Уведомляем админа admin_msg = f"❌ Владелец {owner_name} (ID {owner_id}) отозвал согласие для группы {group_id}. Доступ отключён." safe_vk_call(vk.messages.send, peer_id=ADMIN_ID, message=admin_msg, random_id=get_random_id()) # Уведомляем всех пользователей группы for uid, _ in users_in_group: safe_vk_call(vk.messages.send, peer_id=uid, message=f"❌ Владелец аккаунта группы {group_id} отозвал согласие. Расписание для этой группы больше недоступно.", random_id=get_random_id()) time.sleep(0.5) # защита от превышения лимита VK API return True def mask_headers(headers: dict) -> dict: """Возвращает копию заголовков с замаскированными токенами.""" masked = headers.copy() # Маскируем Authorization if 'Authorization' in masked and masked['Authorization'].startswith('Bearer '): token = masked['Authorization'][7:] # после 'Bearer ' if token: masked['Authorization'] = f"Bearer {token[:10]}... (длина {len(token)})" # Маскируем Cookie (только значение authToken) if 'Cookie' in masked: cookie_str = masked['Cookie'] # Заменяем authToken=xxxx... на authToken=перв10... остальное masked['Cookie'] = re.sub(r'(authToken=)([^;]+)', lambda m: f"{m.group(1)}{m.group(2)[:10]}...", cookie_str) return masked # ========== ЛОГИРОВАНИЕ ЗАПРОСОВ К API ПО ГРУППАМ ========== def log_api_request(group_id: int, user_id: Optional[int], user_name: Optional[str], date_str: str, status_code: int, headers: Optional[dict] = None): log_dir = "api_logs" os.makedirs(log_dir, exist_ok=True) log_txt = os.path.join(log_dir, f"group_{group_id}_api.log") log_line = (f"{datetime.now().strftime('%Y-%m-%d %H:%M:%S')} - " f"group={group_id}, date={date_str}, user_id={user_id}, user={user_name}, status={status_code}\n") with open(log_txt, "a", encoding="utf-8") as f: f.write(log_line) if headers: f.write(f"\n==*~30\n📤 Отправляемые заголовки:\n{json.dumps(headers, indent=2, ensure_ascii=False)}\n==*~30\n\n") # JSON-лог (оставляем как есть) log_json = os.path.join(log_dir, f"group_{group_id}_api.json") entry = { "timestamp": datetime.now().isoformat(), "user_id": user_id, "user_name": user_name, "date_str": date_str, "status_code": status_code } if os.path.exists(log_json): with open(log_json, "r", encoding="utf-8") as f: try: data = json.load(f) except: data = [] else: data = [] data.append(entry) with open(log_json, "w", encoding="utf-8") as f: json.dump(data, f, ensure_ascii=False, indent=2) # ========== ХРАНИЛИЩЕ ОЖИДАЮЩИХ ОТВЕТОВ ========== PENDING_CONSENT_FILE = "pending_consent.json" def load_pending_consent(): """Загружает pending_consent из файла.""" if os.path.exists(PENDING_CONSENT_FILE): try: with open(PENDING_CONSENT_FILE, "r", encoding="utf-8") as f: data = json.load(f) return {int(k): v for k, v in data.items()} except Exception as e: logger.error(f"Ошибка загрузки pending_consent: {e}") return {} def save_pending_consent(data): """Сохраняет pending_consent в файл.""" try: with open(PENDING_CONSENT_FILE, "w", encoding="utf-8") as f: json.dump(data, f, ensure_ascii=False, indent=2) except Exception as e: logger.error(f"Ошибка сохранения pending_consent: {e}") pending_consent = load_pending_consent() # key: owner_id, value: {"group_id": int, "admin_id": int, "timestamp": float} # ========== ХРАНИЛИЩЕ СОСТОЯНИЙ ОТЗЫВА ========== user_revoke_state = {} # key: user_id, value: {"step": "confirm" или "select", "groups": list или "group_id": int} # ========== ХРАНИЛИЩЕ СОСТОЯНИЙ СТАТИСТИКИ ========== user_stats_state = {} # key: user_id, value: {"action": "group_stats" или "api_stats", "groups": list} # ========== СОСТОЯНИЯ ДЛЯ ЗАМЕТОК ========== user_notes_state = {} # key: user_id, value: {"step": "add_date"|"add_subject"|"add_text", "temp_data": {}} # ========== ДЛЯ ИДЕЙ ========== user_idea_state = {} # статус user_idea_status_state = {} # key: user_id, value: {"idea_id": int} #delete user_idea_delete_state = {} # key: user_id, value: {"idea_id": int} #================ ДЛЯ ОТЗЫВОВ ================ user_reviews_page = {} # key: user_id, value: текущая страница (int) REVIEWS_PER_PAGE = 10 # ========================= user_review_state = {} # Для хранения временных состояний пользователей user_bug_state = {} # key: user_id, value: {"step": "title"|"description"|"severity"|"confirm", "title": str, "description": str, "severity": str, "reply_text": str, "reply_cmd": str} # ============== Для хранение страницы пользователя ============= user_main_page = {} # key: user_id, value: 1, 2 или 3 (текущая страница главного меню) # ========== ХРАНИЛИЩЕ ОЖИДАЮЩИХ КОМАНД ========== pending_commands = {} # {group_id: {"user_id": int, "user_name": str, "command": str, "date_str": str}} def save_pending_command(group_id: int, user_id: int, user_name: str, command: str, date_str: str): """Сохраняет команду ожидающую обновления.""" pending_commands[group_id] = { "user_id": user_id, "user_name": user_name, "command": command, "date_str": date_str } def get_pending_command(group_id: int) -> dict: """Возвращает данные ожидающей команды.""" return pending_commands.get(group_id, {}) def clear_pending_command(group_id: int): pending_commands.pop(group_id, None) def generate_key(group_id: int) -> str: key = str(uuid.uuid4()).replace("-", "")[:16] db.add_key(key, group_id) return key def register_user(user_id: int, key: str) -> Optional[int]: return db.register_user(user_id, key) def get_user_group(user_id: int) -> Optional[int]: return db.get_user_group(user_id) def delete_user(user_id: int) -> bool: return db.delete_user(user_id) def get_user_name(vk, user_id): if user_id in user_names_cache: return user_names_cache[user_id] try: user_info = safe_vk_call(vk.users.get, user_ids=user_id) if user_info: name = f"{user_info[0]['first_name']} {user_info[0]['last_name']}" else: name = f"id{user_id}" except: name = f"id{user_id}" user_names_cache[user_id] = name return name def get_all_users() -> List[Tuple[int, int]]: return db.get_all_users() def get_users_by_group(group_id: int) -> List[Tuple[int, int]]: return db.get_users_by_group(group_id) def get_user_info(user_id: int) -> dict: info = { "user_id": user_id, "name": None, "registered": False, "group_id": None, "registered_at": None, "owned_groups": [], "consent_given": False, } user_data = db.get_user(user_id) if user_data: info["registered"] = True info["group_id"] = user_data["group_id"] info["registered_at"] = user_data.get("registered_at") for gid, acc in accounts.items(): if acc.get("owner_id") == user_id: info["owned_groups"].append({ "group_id": gid, "has_credentials": acc.get("has_credentials", False), "consent": acc.get("consent") is not None }) if acc.get("consent"): info["consent_given"] = True return info # ========== RATE LIMITER ========== class RateLimiter: def __init__(self, max_per_second: float = 2.5): self.min_interval = 1.0 / max_per_second self.last_call = 0.0 def wait(self) -> None: elapsed = time.time() - self.last_call if elapsed < self.min_interval: time.sleep(self.min_interval - elapsed) self.last_call = time.time() rate_limiter = RateLimiter() # ========== БЕЗОПАСНЫЙ ВЫЗОВ VK API ========== def safe_vk_call(vk_method, max_retries: int = 5, **kwargs): rate_limiter.wait() retry_delay = 1 for attempt in range(max_retries): try: return vk_method(**kwargs) except ApiError as e: if e.code in (15, 912): logger.error(f"❌ Критическая ошибка VK {e.code}: {e}") return None elif e.code in (6, 9): wait = retry_delay * (2 ** attempt) logger.warning(f"⚠️ Флуд-контроль. Повтор через {wait:.1f} сек") time.sleep(wait) else: logger.error(f"❌ VK API ошибка {e.code}: {e}") return None except Exception as e: logger.exception(f"❌ Неизвестная ошибка: {e}") return None return None def notify_admin(text: str, vk_api_obj=None): if not vk_api_obj: logger.error("❌ ОТПРАВКА АДМИНУ НЕ УДАЛАСЬ: vk объект не передан!") return try: rate_limiter.wait() result = vk_api_obj.messages.send(peer_id=ADMIN_ID, message=text, random_id=get_random_id()) logger.info(f"✅ Уведомление админу отправлено (msg_id={result})") except Exception as e: logger.error(f"❌ ОТПРАВКА АДМИНУ НЕ УДАЛАСЬ: {e}") # ========== КЛАВИАТУРЫ ========== def get_main_keyboard() -> str: kb = VkKeyboard(one_time=False) kb.add_button('📅 Сегодня', color=VkKeyboardColor.PRIMARY) kb.add_button('📆 Завтра', color=VkKeyboardColor.PRIMARY) kb.add_line() kb.add_button('📌 Неделя', color=VkKeyboardColor.SECONDARY) kb.add_button('📌 След. неделя', color=VkKeyboardColor.SECONDARY) kb.add_line() kb.add_button('1️⃣', color=VkKeyboardColor.PRIMARY) # текущая страница kb.add_button('2️⃣', color=VkKeyboardColor.SECONDARY) kb.add_button('3️⃣', color=VkKeyboardColor.SECONDARY) return kb.get_keyboard() def get_main_keyboard_page2(user_id=None) -> str: kb = VkKeyboard(one_time=False) kb.add_button('📝 Заметки', color=VkKeyboardColor.PRIMARY) kb.add_button('⭐ Отзывы', color=VkKeyboardColor.PRIMARY) kb.add_line() kb.add_button('✍️ Оставить отзыв', color=VkKeyboardColor.SECONDARY) kb.add_button('🐛 Баг', color=VkKeyboardColor.SECONDARY) kb.add_line() kb.add_button('💡 Предложить идею', color=VkKeyboardColor.SECONDARY) kb.add_line() kb.add_button('1️⃣', color=VkKeyboardColor.SECONDARY) kb.add_button('2️⃣', color=VkKeyboardColor.PRIMARY) # текущая страница kb.add_button('3️⃣', color=VkKeyboardColor.SECONDARY) return kb.get_keyboard() def get_main_keyboard_page3(user_id=None) -> str: kb = VkKeyboard(one_time=False) kb.add_button('👤 Личный кабинет', color=VkKeyboardColor.PRIMARY) kb.add_line() kb.add_button('🐛 Мои баги', color=VkKeyboardColor.SECONDARY) kb.add_button('💡 Мои идеи', color=VkKeyboardColor.SECONDARY) kb.add_line() kb.add_button('🔔 Уведомления', color=VkKeyboardColor.SECONDARY) fmt = db.get_user_format(user_id) if user_id else "text" fmt_label = "Картинка 🖼" if fmt == "image" else "Текст 📄" kb.add_button(f'🖼 Формат: {fmt_label}', color=VkKeyboardColor.SECONDARY) kb.add_line() kb.add_button('❓ Помощь', color=VkKeyboardColor.SECONDARY) kb.add_line() kb.add_button('1️⃣', color=VkKeyboardColor.SECONDARY) kb.add_button('2️⃣', color=VkKeyboardColor.SECONDARY) kb.add_button('3️⃣', color=VkKeyboardColor.PRIMARY) # текущая страница return kb.get_keyboard() def get_keyboard_for_page(page: int, user_id=None) -> str: if page == 1: return get_main_keyboard() elif page == 2: return get_main_keyboard_page2(user_id) else: return get_main_keyboard_page3(user_id) def get_notes_menu_keyboard() -> str: kb = VkKeyboard(one_time=True) kb.add_button('📋 Все заметки', color=VkKeyboardColor.PRIMARY) kb.add_line() kb.add_button('➕ Добавить заметку', color=VkKeyboardColor.POSITIVE) kb.add_button('🗑 Удалить заметку', color=VkKeyboardColor.NEGATIVE) kb.add_line() kb.add_button('📅 Заметки за дату', color=VkKeyboardColor.SECONDARY) kb.add_button('🔙 Назад', color=VkKeyboardColor.SECONDARY) return kb.get_keyboard() def get_group_selection_keyboard(only_with_owner: bool = False) -> str: kb = VkKeyboard(one_time=True) groups = [gid for gid in GROUPS.keys() if not (only_with_owner and gid not in accounts)] for i in range(0, len(groups), 3): for gid in groups[i:i+3]: kb.add_button(f"Группа {gid}", color=VkKeyboardColor.PRIMARY) if i + 3 < len(groups): kb.add_line() if groups: kb.add_line() kb.add_button("❌ Отмена", color=VkKeyboardColor.NEGATIVE) return kb.get_keyboard() def get_remkey_keyboard(): kb = VkKeyboard(one_time=True) groups = sorted(GROUPS.keys()) for i in range(0, len(groups), 3): for gid in groups[i:i+3]: kb.add_button(f"Группа {gid}", color=VkKeyboardColor.PRIMARY) if i + 3 < len(groups): kb.add_line() kb.add_line() kb.add_button("❌ Все группы", color=VkKeyboardColor.NEGATIVE) kb.add_button("❌ Отмена", color=VkKeyboardColor.NEGATIVE) return kb.get_keyboard() def get_confirm_keyboard(): kb = VkKeyboard(one_time=True) kb.add_button("✅ Да", color=VkKeyboardColor.POSITIVE) kb.add_button("❌ Нет", color=VkKeyboardColor.NEGATIVE) return kb.get_keyboard() def get_group_selection_keyboard_for_revoke(group_ids): kb = VkKeyboard(one_time=True) for i in range(0, len(group_ids), 3): for gid in group_ids[i:i+3]: kb.add_button(f"Группа {gid}", color=VkKeyboardColor.PRIMARY) if i + 3 < len(group_ids): kb.add_line() if group_ids: kb.add_line() kb.add_button("❌ Отмена", color=VkKeyboardColor.NEGATIVE) return kb.get_keyboard() def get_group_selection_keyboard_for_stats(group_ids): kb = VkKeyboard(one_time=True) for i in range(0, len(group_ids), 3): for gid in group_ids[i:i+3]: kb.add_button(f"Группа {gid}", color=VkKeyboardColor.PRIMARY) if i + 3 < len(group_ids): kb.add_line() if group_ids: kb.add_line() kb.add_button("❌ Отмена", color=VkKeyboardColor.NEGATIVE) return kb.get_keyboard() def get_cancel_keyboard() -> str: kb = VkKeyboard(one_time=True) kb.add_button("❌ Отмена", color=VkKeyboardColor.NEGATIVE) return kb.get_keyboard() def get_bug_severity_keyboard() -> str: kb = VkKeyboard(one_time=True) kb.add_button('🟢 Мелкая', color=VkKeyboardColor.SECONDARY) kb.add_button('🟡 Средняя', color=VkKeyboardColor.PRIMARY) kb.add_line() kb.add_button('🔴 Критичная', color=VkKeyboardColor.NEGATIVE) kb.add_line() kb.add_button("❌ Отмена", color=VkKeyboardColor.NEGATIVE) return kb.get_keyboard() def get_bug_confirm_keyboard() -> str: kb = VkKeyboard(one_time=True) kb.add_button('✅ Отправить', color=VkKeyboardColor.POSITIVE) kb.add_button('✏️ Изменить', color=VkKeyboardColor.SECONDARY) kb.add_line() kb.add_button("❌ Отмена", color=VkKeyboardColor.NEGATIVE) return kb.get_keyboard() # ========== ВСПОМОГАТЕЛЬНЫЕ ФУНКЦИИ ДЛЯ ЗАПРОСОВ К API ДЕКАНАТА ========== def generate_fingerprint() -> str: """Генерирует случайный 32-символьный hex fingerprint.""" return ''.join(random.choices('0123456789abcdef', k=32)) def generate_ddg1() -> str: """Генерирует случайную строку для куки __ddg1_ (16-32 символа).""" return ''.join(random.choices('0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ', k=16)) def get_monday(date_str: str = None) -> str: """Возвращает дату понедельника недели для указанной даты (или сегодня).""" if date_str: dt = datetime.strptime(date_str, "%Y-%m-%d") else: dt = datetime.now() monday = dt - timedelta(days=dt.weekday()) return monday.strftime("%Y-%m-%d") def get_spa_headers(token: str, group_id: int, user_id: int, session_id: Optional[str] = None, xsrf_token: Optional[str] = None, fp: Optional[str] = None, platform: Optional[str] = None) -> dict: if platform is None: platform = GROUPS.get(group_id, {}).get("platform", DEFAULT_PLATFORM) profile = DEVICE_PROFILES.get(platform, DEVICE_PROFILES[DEFAULT_PLATFORM]) cookie = f"authToken={token}" if session_id: cookie += f"; ASP.NET_SessionId={session_id}" if xsrf_token: cookie += f"; __AntiXsrfToken={xsrf_token}" ddg1 = accounts.get(group_id, {}).get('ddg1') if ddg1: cookie += f"; __ddg1_={ddg1}" if fp is None: fp = generate_fingerprint() headers = { "Accept": "application/json", "Accept-Encoding": "gzip, deflate, br, zstd", "Accept-Language": "ru-RU,ru;q=0.8", "Authorization": f"Bearer {token}", "Cache-Control": "no-cache", "Client-Version": CLIENT_VERSION, "Cookie": cookie, "Current-Path": f"https://dec.mgutm.ru/WebApp/#/Rasp/Group/{group_id}", "fp": fp, "Priority": "u=1, i", "Referer": "https://dec.mgutm.ru/WebApp/", "Sec-Ch-Ua": profile["sec_ch_ua"], "Sec-Ch-Ua-Mobile": profile["sec_ch_ua_mobile"], "Sec-Ch-Ua-Platform": profile["sec_ch_ua_platform"], "Sec-Fetch-Dest": "empty", "Sec-Fetch-Mode": "cors", "Sec-Fetch-Site": "same-origin", "Sec-Gpc": "1", "User-Agent": profile["user_agent"], "User-Id": str(user_id), "X-User-Id": str(user_id), "latest-client-version": LATEST_CLIENT_VERSION, } return headers def parse_cookies(set_cookie_header: str) -> dict: """ Парсит заголовок Set-Cookie и возвращает словарь с именем и значением куки. Поддерживает несколько кук в одном заголовке (разделённых запятой). """ cookies = {} parts = set_cookie_header.split(',') for part in parts: part = part.strip() if '=' in part: name, rest = part.split('=', 1) value = rest.split(';')[0].strip() cookies[name.strip()] = value return cookies def should_auto_refresh() -> bool: """Проверяет, нужно ли автообновить токен. Обновляем в пятницу, субботу или воскресенье — чтобы выглядело как студент.""" import calendar weekday = datetime.now().weekday() return weekday in (calendar.FRIDAY, calendar.SATURDAY, calendar.SUNDAY) def refresh_token(group_id: int, force: bool = False) -> Optional[str]: """Обновляет токен через /api/tokenauth. Возвращает новый токен или None. force=True — принудительно, без проверки дня недели.""" global CLIENT_VERSION, LATEST_CLIENT_VERSION group_info = GROUPS.get(group_id) login = group_info.get("login") password = group_info.get("password") if not login or not password: api_logger.warning(f"Нет логина/пароля для группы {group_id}, не могу обновить токен") return None # Проверка дня недели (только если не force) if not force and not should_auto_refresh(): api_logger.info(f"⏭ Пропускаем автообновление — не пятница/выходные") return None # Рандомная задержка 30 сек - 3 мин (как будто человек думает) # НЕ делаем задержку при force=True (когда пользователь ждёт ответа) if not force: delay = random.randint(30, 180) api_logger.info(f"⏳ Задержка {delay} сек перед обновлением...") time.sleep(delay) try: api_logger.info(f"🔄 Попытка обновить токен для группы {group_id}...") new_fp = generate_fingerprint() platform = GROUPS.get(group_id, {}).get("platform", DEFAULT_PLATFORM) profile = DEVICE_PROFILES.get(platform, DEVICE_PROFILES[DEFAULT_PLATFORM]) # Берём сохранённые сессионные куки (если есть) with accounts_lock: saved_session = accounts.get(group_id, {}).get("session") saved_xsrf = accounts.get(group_id, {}).get("xsrf") # Если нет сессионных кук — сначала получаем их через GET на сайт if not saved_session or not saved_xsrf: api_logger.info(f"🔄 Получаем сессионные куки для группы {group_id}...") pre_headers = { "Accept": "application/json", "Accept-Encoding": "gzip, deflate, br, zstd", "Accept-Language": "ru-RU,ru;q=0.8", "Cache-Control": "no-cache", "Client-Version": CLIENT_VERSION, "Cookie": "authToken=", "Current-Path": "https://dec.mgutm.ru/WebApp/#/login", "fp": new_fp, "Priority": "u=1, i", "Referer": "https://dec.mgutm.ru/WebApp/", "Sec-Ch-Ua": profile["sec_ch_ua"], "Sec-Ch-Ua-Mobile": profile["sec_ch_ua_mobile"], "Sec-Ch-Ua-Platform": profile["sec_ch_ua_platform"], "Sec-Fetch-Dest": "empty", "Sec-Fetch-Mode": "cors", "Sec-Fetch-Site": "same-origin", "Sec-Gpc": "1", "User-Agent": profile["user_agent"], } pre_resp = requests.get( "https://dec.mgutm.ru/api/tokenauth", headers=pre_headers, timeout=30 ) set_cookie = pre_resp.headers.get("Set-Cookie") if set_cookie: parsed = parse_cookies(set_cookie) saved_session = parsed.get("ASP.NET_SessionId") or saved_session saved_xsrf = parsed.get("__AntiXsrfToken") or saved_xsrf api_logger.info(f"✅ Получены сессионные куки для группы {group_id}") time.sleep(random.uniform(1, 2)) cookie = "authToken=" if saved_session: cookie += f"; ASP.NET_SessionId={saved_session}" if saved_xsrf: cookie += f"; __AntiXsrfToken={saved_xsrf}" browser_headers = { "Accept": "application/json", "Accept-Encoding": "gzip, deflate, br, zstd", "Accept-Language": "ru-RU,ru;q=0.8", "Cache-Control": "no-cache", "Client-Version": CLIENT_VERSION, "Content-Type": "application/json", "Cookie": cookie, "Current-Path": "https://dec.mgutm.ru/WebApp/#/login", "fp": new_fp, "Origin": "https://dec.mgutm.ru", "Priority": "u=1, i", "Referer": "https://dec.mgutm.ru/WebApp/", "Sec-Ch-Ua": profile["sec_ch_ua"], "Sec-Ch-Ua-Mobile": profile["sec_ch_ua_mobile"], "Sec-Ch-Ua-Platform": profile["sec_ch_ua_platform"], "Sec-Fetch-Dest": "empty", "Sec-Fetch-Mode": "cors", "Sec-Fetch-Site": "same-origin", "Sec-Gpc": "1", "User-Agent": profile["user_agent"], } resp = requests.post( "https://dec.mgutm.ru/api/tokenauth", json={"username": login, "password": password}, headers=browser_headers, timeout=30 ) # Рандомная задержка после запроса (1-2 сек) time.sleep(random.uniform(1, 2)) # Обновляем latest-client-version из ответа сервера (если прислали) server_latest = resp.headers.get("latest-client-version") if server_latest: CLIENT_VERSION = server_latest LATEST_CLIENT_VERSION = server_latest api_logger.info(f"📦 Client-Version обновлён до {server_latest}") # Парсим Set-Cookie из ответа (authToken, ASP.NET_SessionId, __AntiXsrfToken) set_cookie = resp.headers.get("Set-Cookie") session_id = None xsrf_token = None if set_cookie: parsed = parse_cookies(set_cookie) session_id = parsed.get("ASP.NET_SessionId") xsrf_token = parsed.get("__AntiXsrfToken") if resp.status_code == 200: data = resp.json() new_token = data.get("data", {}).get("accessToken") if new_token: # Сохраняем новый fp и сессионные куки под замком with accounts_lock: if group_id not in accounts: accounts[group_id] = {} accounts[group_id]["fp"] = new_fp accounts[group_id]["ddg1"] = generate_ddg1() if session_id: accounts[group_id]["session"] = session_id if xsrf_token: accounts[group_id]["xsrf"] = xsrf_token exp_ts = get_token_exp(new_token) if exp_ts: accounts[group_id]["token_expires_at"] = exp_ts db.save_group_account(group_id, accounts[group_id]) api_logger.info(f"✅ Токен + fp + сессия обновлены для группы {group_id}") return new_token api_logger.error(f"❌ Не удалось обновить токен: код {resp.status_code}") return None except Exception as e: api_logger.error(f"❌ Ошибка обновления токена: {e}") return None def refresh_token_background(group_id: int, date_str: str, vk, user_id: int, user_name: str): """Запускает обновление токена в фоне (отдельный поток).""" api_logger.info(f"🧵 ЗАПУСК ФОНОВОГО ОБНОЛЕНИЯ токена для группы {group_id}...") def _run(): try: # Очищаем session и fp перед обновлением with accounts_lock: if group_id in accounts: accounts[group_id].pop("session", None) accounts[group_id].pop("fp", None) db.save_group_account(group_id, accounts[group_id]) new_token = refresh_token(group_id, force=True) if new_token: GROUPS[group_id]["token"] = new_token with accounts_lock: if group_id not in accounts: accounts[group_id] = {} accounts[group_id]["token"] = new_token exp_ts = get_token_exp(new_token) if exp_ts: accounts[group_id]["token_expires_at"] = exp_ts accounts[group_id]["fp"] = generate_fingerprint() accounts[group_id]["ddg1"] = generate_ddg1() db.save_group_account(group_id, accounts[group_id]) update_env_token(group_id, new_token) api_logger.info(f"✅ Токен + fp обновлены в фоне для группы {group_id}") # Чистим updating with accounts_lock: accounts[group_id].pop("updating", None) accounts[group_id]["blocked"] = False db.save_group_account(group_id, accounts[group_id]) # Пробуем получить расписание time.sleep(1) data = fetch_schedule_from_dec_with_vk(group_id, date_str, vk, user_id, user_name) if data and not data.get("_error"): save_week_to_cache(group_id, data, date_str) save_week_images(group_id, data, date_str) user_fmt = db.get_user_format(user_id) if user_fmt == "image" and data.get("data", {}).get("rasp"): rasp = data["data"]["rasp"] date_lessons = [l for l in rasp if l.get("дата", "").split("T")[0] == date_str] if date_lessons: info = data.get("data", {}).get("info", {}) img_bytes = generate_schedule_image(date_lessons, group_id, date_str, info) try: uploader = upload.VkUpload(vk) photo = uploader.photo_messages(photos=img_bytes)[0] attachment = f"photo{photo['owner_id']}_{photo['id']}" safe_vk_call(vk.messages.send, peer_id=user_id, message="✅ Токен обновлён. Вот расписание:", attachment=attachment, random_id=get_random_id()) except Exception: result = format_schedule(data, group_id, date_str) if result: safe_vk_call(vk.messages.send, peer_id=user_id, message=result, random_id=get_random_id()) else: result = format_schedule(data, group_id, date_str) if result: safe_vk_call(vk.messages.send, peer_id=user_id, message=result, random_id=get_random_id()) else: result = format_schedule(data, group_id, date_str) if result: try: safe_vk_call(vk.messages.send, peer_id=user_id, message=result, random_id=get_random_id()) except ApiError as e: if e.code == 917: api_logger.warning(f"⚠️ Не могу написать пользователю {user_id} напрямую (917). Пропускаю.") else: raise # Владельцу owner_peer = GROUPS.get(group_id, {}).get("peer_id") if owner_peer and owner_peer != user_id: safe_vk_call(vk.messages.send, peer_id=owner_peer, message=f"✅ Токен группы {group_id} обновлён", random_id=get_random_id()) # Админу if ADMIN_ID != user_id: safe_vk_call(vk.messages.send, peer_id=ADMIN_ID, message=f"✅ Токен группы {group_id} обновлён (фоново, инициатор: {user_name})", random_id=get_random_id()) else: try: safe_vk_call(vk.messages.send, peer_id=user_id, message=f"⚠️ Не удалось получить расписание после обновления токена", random_id=get_random_id()) except ApiError as e: if e.code == 917: api_logger.warning(f"⚠️ Не могу написать пользователю {user_id} (917).") else: raise else: try: safe_vk_call(vk.messages.send, peer_id=user_id, message=f"❌ Не удалось обновить токен для группы {group_id}", random_id=get_random_id()) except ApiError as e: if e.code == 917: api_logger.warning(f"⚠️ Не могу написать пользователю {user_id} (917).") else: raise safe_vk_call(vk.messages.send, peer_id=ADMIN_ID, message=f"❌ Не удалось обновить токен группы {group_id}. Проверьте логин/пароль.", random_id=get_random_id()) with accounts_lock: accounts[group_id].pop("updating", None) accounts[group_id]["blocked"] = True db.save_group_account(group_id, accounts[group_id]) except Exception as e: api_logger.error(f"🔥 Ошибка в фоновом потоке: {e}") try: with accounts_lock: accounts[group_id].pop("updating", None) db.save_group_account(group_id, accounts[group_id]) except: pass thread = threading.Thread(target=_run, daemon=True) thread.start() api_logger.info(f"🧵 Поток запущен для группы {group_id}") def fetch_schedule_from_dec_with_vk(group_id: int, date_str: str, vk, user_id: int, user_name: str): """Обёртка для fetch_schedule_from_dec с vk доступным.""" return fetch_schedule_from_dec(group_id, date_str, vk=vk, user_id=user_id, user_name=user_name) def update_env_token(group_id: int, new_token: str) -> bool: """Обновляет токен в .env файле.""" env_path = ".env" if not os.path.exists(env_path): return False with env_lock: # Находим какой GROUP{i}_ID соответствует group_id group_num = None for i in range(1, 11): gid = os.getenv(f"GROUP{i}_ID") if gid and int(gid) == group_id: group_num = i break if not group_num: return False token_key = f"GROUP{group_num}_TOKEN" # Читаем .env lines = [] try: with open(env_path, "r", encoding="utf-8") as f: lines = f.readlines() except: return False # Находим и заменяем токен new_lines = [] for line in lines: if line.strip().startswith(token_key + "="): new_lines.append(f'{token_key}="{new_token}"\n') else: new_lines.append(line) try: with open(env_path, "w", encoding="utf-8") as f: f.writelines(new_lines) logger.info(f"✅ Токен обновлён в .env для группы {group_id}") return True except Exception as e: logger.error(f"❌ Не удалось обновить .env: {e}") return False # ========== ПОЛУЧЕНИЕ АКТУАЛЬНЫХ ФЛАГОВ ГРУППЫ ИЗ БД ========== def get_account_flags(group_id: int): """Возвращает словарь с blocked, revoked, consent из БД. Не использует кэш accounts — всегда актуально для веб-монитора.""" acc = db.get_group_account(group_id) if acc is None: return {"blocked": False, "revoked": False, "consent": None} return { "blocked": acc.get("blocked", False), "revoked": acc.get("revoked", False), "consent": acc.get("consent", None), } # ========== РАБОТА С РАСПИСАНИЕМ (ОБНОВЛЁННАЯ ВЕРСИЯ) ========== def fetch_schedule_from_dec(group_id: int, date_str: str, vk=None, user_id=None, user_name=None) -> Optional[dict]: group_info = GROUPS.get(group_id) user_suffix = f", user_id={user_id}, user={user_name}" if user_id and user_name else "" # Проверяем флаги из БД (веб-монитор мог их изменить) flags = get_account_flags(group_id) # Флаг updating берём из кэша (его ставит только бот) with accounts_lock: updating = accounts.get(group_id, {}).get("updating", False) if flags["blocked"] or flags["revoked"] or updating or not flags["consent"]: api_logger.info(f"⛔ Группа {group_id} заблокирована/обновляется/нет согласия. Пропускаем запрос.") error_info = {"_error": True, "status_code": 401, "blocked": True} if updating: error_info["waiting"] = True if flags["revoked"]: error_info["revoked"] = True if not flags["consent"]: error_info["no_consent"] = True return error_info if not group_info: api_logger.warning(f"Группа {group_id} не найдена в конфиге") return None api_logger.info("="*60) url = f"https://dec.mgutm.ru/api/Rasp?idGroup={group_id}&sdate={date_str}" # ===== Управление fingerprint'ом для группы ===== with accounts_lock: if "fp" not in accounts.get(group_id, {}): if group_id not in accounts: accounts[group_id] = {} accounts[group_id]["fp"] = generate_fingerprint() db.save_group_account(group_id, accounts[group_id]) logger.info(f"Сгенерирован новый fingerprint для группы {group_id}: {accounts[group_id]['fp']}") fp = accounts[group_id]["fp"] # ===== Получение сессионных кук (если есть) ===== session_data = accounts.get(group_id, {}).get("session", {}) session_id = session_data.get("session_id") xsrf_token = session_data.get("xsrf_token") # ===== Берём ID студента из настроек группы ===== decanat_user_id = group_info.get('user_id', -40446) # если не указан, подставляем -40446 headers = get_spa_headers( token=group_info['token'], group_id=group_id, user_id=decanat_user_id, session_id=session_id, xsrf_token=xsrf_token, fp=fp ) log_msg = f"🌐 ЗАПРОС: группа={group_id}, дата={date_str}" if user_id: log_msg += f", user_id={user_id}" if user_name: log_msg += f", user={user_name}" api_logger.info(log_msg) masked_headers = mask_headers(headers) log_api_request(group_id, user_id, user_name, date_str, status_code=0, headers=masked_headers) # Небольшая случайная задержка перед запросом time.sleep(random.uniform(0.8, 2.2)) try: resp = requests.get(url, headers=headers, timeout=30) # Обработка успешного ответа (код 200) if resp.status_code == 200: data = resp.json() # 1. Фикс бага с авторизацией: явная проверка state=0 или data=null if data.get("state") == 0 or not data.get("data"): api_logger.warning(f"⚠️ АВТОРИЗАЦИЯ СЛЕТЕЛА (state=0 или data=null){user_suffix}") with accounts_lock: if group_id not in accounts: accounts[group_id] = {} accounts[group_id]["updating"] = True db.save_group_account(group_id, accounts[group_id]) if user_id and vk: try: safe_vk_call(vk.messages.send, peer_id=user_id, message=f"⚠️ Токен устарел. Обновляю, ждите...", random_id=get_random_id()) except: pass if vk: try: safe_vk_call(vk.messages.send, peer_id=ADMIN_ID, message=f"⚠️ ТОКЕН УСТАРЕЛ (state=0/data=null)\n📌 Группа: {group_id}\n👤 Инициатор: {user_name}\n📅 Дата: {date_str}\n🔄 Обновляю...", random_id=get_random_id()) except: pass critical_log_path = os.path.join(LOG_DIR, "critical_errors.log") with open(critical_log_path, "a", encoding="utf-8") as f_err: f_err.write(f"{datetime.now().isoformat()} - AUTH_FAIL: group={group_id}, user={user_id}, date={date_str} - state={data.get('state')}, data={data.get('data')}\n") refresh_token_background(group_id, date_str, vk, user_id, user_name) return {"_error": True, "waiting": True, "status_code": 401} # Проверяем наличие info (признак успешной авторизации) if data["data"].get("info"): # Нормальный ответ (даже если rasp пустой) api_logger.info(f"✅ УСПЕХ: группа={group_id}, дата={date_str}, код=200 (есть info){user_suffix}") log_api_request(group_id, user_id, user_name, date_str, resp.status_code) with accounts_lock: if accounts.get(group_id, {}).get("blocked"): del accounts[group_id]["blocked"] logger.info(f"✅ Блокировка группы {group_id} снята (токен работает)") if 'Set-Cookie' in resp.headers: set_cookie = resp.headers['Set-Cookie'] api_logger.info(f"🍪 Set-Cookie: {set_cookie[:100]}...") cookies = parse_cookies(set_cookie) session_id = cookies.get('ASP.NET_SessionId') xsrf_token = cookies.get('__AntiXsrfToken') if session_id or xsrf_token: if group_id not in accounts: accounts[group_id] = {} if 'session' not in accounts[group_id]: accounts[group_id]['session'] = {} if session_id: accounts[group_id]['session']['session_id'] = session_id if xsrf_token: accounts[group_id]['session']['xsrf_token'] = xsrf_token api_logger.info(f"💾 Сохранены сессионные куки для группы {group_id}") info = data["data"]["info"] if group_id not in accounts: accounts[group_id] = {} accounts[group_id]["week_info"] = { "curWeekNumber": info.get("curWeekNumber"), "curNumNed": info.get("curNumNed"), "selectedNumNed": info.get("selectedNumNed"), "dateUploadingRasp": info.get("dateUploadingRasp"), "lastDate": info.get("lastDate") } db.save_group_account(group_id, accounts[group_id]) return data else: # Подозрительный ответ: скорее всего, токен не работает api_logger.warning(f"⚠️ ПОДОЗРИТЕЛЬНЫЙ ОТВЕТ: группа={group_id}, дата={date_str}, код=200, но отсутствует info. Возможно, токен недействителен.{user_suffix}") log_api_request(group_id, user_id, user_name, date_str, resp.status_code) # Флаг что идёт обновление with accounts_lock: if group_id not in accounts: accounts[group_id] = {} accounts[group_id]["updating"] = True db.save_group_account(group_id, accounts[group_id]) # Пишем участнику if user_id and vk: try: safe_vk_call(vk.messages.send, peer_id=user_id, message=f"⚠️ Токен устарел. Обновляю, ждите...", random_id=get_random_id()) except: pass # Админу if vk: try: safe_vk_call(vk.messages.send, peer_id=ADMIN_ID, message=f"⚠️ ТОКЕН УСТАРЕЛ\n📌 Группа: {group_id}\n👤 Инициатор: {user_name}\n📅 Дата: {date_str}\n🔄 Обновляю...", random_id=get_random_id()) except: pass # Запись в критический лог critical_log_path = os.path.join(LOG_DIR, "critical_errors.log") with open(critical_log_path, "a", encoding="utf-8") as f_err: f_err.write(f"{datetime.now().isoformat()} - SUSPECT: group={group_id}, user={user_id}, date={date_str} - ответ без info. Токен возможно протух.\n") # Запускаем обновление в фоне, не блокируя бота refresh_token_background(group_id, date_str, vk, user_id, user_name) return {"_error": True, "waiting": True, "status_code": 200} # Ошибки авторизации (401, 403) — НЕ блокируем, обновляем elif resp.status_code in (401, 403): api_logger.error(f"❌ ОШИБКА АВТОРИЗАЦИИ: группа={group_id}, дата={date_str}, код={resp.status_code}{user_suffix}") log_api_request(group_id, user_id, user_name, date_str, resp.status_code) # Только флаг что идёт обновление with accounts_lock: if group_id not in accounts: accounts[group_id] = {} accounts[group_id]["updating"] = True if group_id in accounts: accounts[group_id].pop("session", None) db.save_group_account(group_id, accounts[group_id]) # Пишем участнику if user_id and vk: try: safe_vk_call(vk.messages.send, peer_id=user_id, message=f"⚠️ Токен устарел. Обновляю, ждите...", random_id=get_random_id()) except: pass # Админу тоже if vk: try: safe_vk_call(vk.messages.send, peer_id=ADMIN_ID, message=f"⚠️ ТОКЕН УСТАРЕЛ\n📌 Группа: {group_id}\n👤 Инициатор: {user_name}\n📅 Дата: {date_str}\n🔄 Обновляю...", random_id=get_random_id()) except: pass # Чистим кэш (файловая операция, блокировка не нужна) clear_cache_for_group(group_id) # Запускаем обновление в фоне refresh_token_background(group_id, date_str, vk, user_id, user_name) return {"_error": True, "waiting": True, "status_code": resp.status_code} # Другие ошибки (4xx, 5xx) else: api_logger.warning(f"⚠️ НЕУДАЧА: группа={group_id}, дата={date_str}, код={resp.status_code}{user_suffix}") log_api_request(group_id, user_id, user_name, date_str, resp.status_code) return {"_error": True, "status_code": resp.status_code} except Exception as e: api_logger.error(f"🔥 ИСКЛЮЧЕНИЕ: группа={group_id}, дата={date_str}, ошибка={e}{user_suffix}") log_api_request(group_id, user_id, user_name, date_str, 0) # При сетевых ошибках тоже возвращаем ошибку для кэширования return {"_error": True, "exception": str(e)} def get_cache_path(group_id: int, date_str: str) -> str: group_dir = os.path.join(RASP_DIR, str(group_id)) os.makedirs(group_dir, exist_ok=True) return os.path.join(group_dir, f"{date_str}.json") def load_cached_schedule(group_id: int, date_str: str) -> Optional[dict]: cache_file = get_cache_path(group_id, date_str) if os.path.exists(cache_file): try: with open(cache_file, "r", encoding="utf-8") as f: cache = json.load(f) if time.time() - cache["timestamp"] < CACHE_TTL: return cache["data"] except Exception as e: logger.warning(f"Ошибка чтения кэша: {e}") return None def save_schedule_to_cache(group_id: int, date_str: str, data: dict) -> None: cache_file = get_cache_path(group_id, date_str) try: with open(cache_file, "w", encoding="utf-8") as f: json.dump({"timestamp": time.time(), "data": data}, f, ensure_ascii=False) except Exception as e: logger.error(f"Ошибка записи кэша: {e}") def save_week_to_cache(group_id: int, week_data: dict, requested_date: str) -> None: """ Сохраняет каждый день из недельного ответа в отдельный файл кэша. Также обновляет пустые кэши для всех дней недели. """ if week_data is None: return import copy rasp_list = week_data.get("data", {}).get("rasp", []) days_with_lessons = set() if rasp_list: from collections import defaultdict days = defaultdict(list) for lesson in rasp_list: day_str = lesson["дата"].split("T")[0] days[day_str].append(lesson) days_with_lessons.add(day_str) # Сохраняем дни, по которым есть занятия for day_str, lessons in days.items(): day_data = copy.deepcopy(week_data) day_data["data"]["rasp"] = lessons save_schedule_to_cache(group_id, day_str, day_data) # Если запрошенная дата отсутствует в ответе, сохраняем для неё пустой кэш if requested_date not in days_with_lessons: empty_data = copy.deepcopy(week_data) empty_data["data"]["rasp"] = [] save_schedule_to_cache(group_id, requested_date, empty_data) days_with_lessons.add(requested_date) else: # Нет занятий на неделю – сохраняем пустой кэш для запрошенной даты save_schedule_to_cache(group_id, requested_date, week_data) days_with_lessons.add(requested_date) # ========== ОБНОВЛЯЕМ ПУСТЫЕ КЭШИ ДЛЯ ВСЕХ ДНЕЙ НЕДЕЛИ ========== monday = get_monday(requested_date) dt = datetime.strptime(monday, "%Y-%m-%d") for i in range(7): date_str = (dt + timedelta(days=i)).strftime("%Y-%m-%d") if date_str in days_with_lessons: continue # уже сохранён (с занятиями или пустой) # Создаём/обновляем пустой кэш (перезаписываем, даже если файл уже есть) empty_data = copy.deepcopy(week_data) empty_data["data"]["rasp"] = [] save_schedule_to_cache(group_id, date_str, empty_data) def get_schedule_for_date(group_id: int, date_str: str, vk=None, user_id=None, user_name=None): # Читаем флаги из БД (веб-монитор мог их изменить) flags = get_account_flags(group_id) if flags["blocked"] or flags["revoked"] or not flags["consent"]: error_info = {"_error": True, "status_code": 401, "blocked": True} if flags["revoked"]: error_info["revoked"] = True if not flags["consent"]: error_info["no_consent"] = True return error_info, False # Инициализация fp и ddg1 (технические поля, меняются только ботом) with accounts_lock: if group_id not in accounts: accounts[group_id] = {} if 'fp' not in accounts[group_id]: accounts[group_id]['fp'] = generate_fingerprint() if 'ddg1' not in accounts[group_id]: accounts[group_id]['ddg1'] = generate_ddg1() db.save_group_account(group_id, accounts[group_id]) # Сначала пробуем загрузить из кэша конкретный день data = load_cached_schedule(group_id, date_str) if data is not None: # Сохраняем week_info, если ещё не сохранена if data.get("data") and data["data"].get("info"): info = data["data"]["info"] with accounts_lock: acc = accounts.get(group_id) if acc is None or "week_info" not in acc: if group_id not in accounts: accounts[group_id] = {} accounts[group_id]["week_info"] = { "curWeekNumber": info.get("curWeekNumber"), "curNumNed": info.get("curNumNed"), "selectedNumNed": info.get("selectedNumNed"), "dateUploadingRasp": info.get("dateUploadingRasp"), "lastDate": info.get("lastDate") } db.save_group_account(group_id, accounts[group_id]) return data, True # Нет кэша – делаем запрос к API data = fetch_schedule_from_dec(group_id, date_str, vk, user_id, user_name) if data is None: data = {"_error": True, "details": "fetch returned None"} save_schedule_to_cache(group_id, date_str, data) return data, False if not data.get("_error"): save_week_to_cache(group_id, data, date_str) save_week_images(group_id, data, date_str) else: save_schedule_to_cache(group_id, date_str, data) return data, False IMG_DIR = "img" os.makedirs(IMG_DIR, exist_ok=True) def get_img_cache_path(group_id: int, date_str: str) -> str: group_dir = os.path.join(IMG_DIR, str(group_id)) os.makedirs(group_dir, exist_ok=True) return os.path.join(group_dir, f"{date_str}.png") def load_cached_image(group_id: int, date_str: str) -> Optional[io.BytesIO]: img_file = get_img_cache_path(group_id, date_str) if os.path.exists(img_file): try: with open(img_file, "rb") as f: data = f.read() if time.time() - os.path.getmtime(img_file) < CACHE_TTL: buf = io.BytesIO(data) return buf except Exception as e: logger.warning(f"Ошибка чтения кэша картинки: {e}") return None def save_image_to_cache(group_id: int, date_str: str, img_bytes: io.BytesIO) -> None: img_file = get_img_cache_path(group_id, date_str) try: with open(img_file, "wb") as f: f.write(img_bytes.getvalue()) except Exception as e: logger.error(f"Ошибка записи картинки в кэш: {e}") def clear_image_cache_for_group(group_id: int) -> int: group_dir = os.path.join(IMG_DIR, str(group_id)) if not os.path.exists(group_dir): return 0 removed = 0 for filename in os.listdir(group_dir): if filename.endswith('.png'): try: os.remove(os.path.join(group_dir, filename)) removed += 1 except Exception as e: logger.warning(f"Не удалось удалить {filename}: {e}") return removed # --- Оптимизация генерации изображений (асинхронная) --- _week_img_lock = threading.Lock() _img_gen_lock = threading.Lock() # Блокировка на сам процесс рисования, чтобы не перегружать CPU _bg_generating_weeks = set() def _generate_single_img(day_str, lessons, group_id, week_data): """Генерирует одну картинку (проверяет наличие перед генерацией).""" img_path = get_img_cache_path(group_id, day_str) # Быстрая проверка без блокировки (если файл уже точно есть) if os.path.exists(img_path): return with _img_gen_lock: # Повторная проверка внутри блокировки (кто-то мог успеть создать файл пока мы ждали) if os.path.exists(img_path): return info = week_data.get("data", {}).get("info", {}) try: img_bytes = generate_schedule_image(lessons, group_id, day_str, info) save_image_to_cache(group_id, day_str, img_bytes) except Exception as e: logger.error(f"Ошибка генерации картинки {day_str}: {e}") def _bg_generate_rest(group_id, week_data, requested_date): """Фоновая генерация остальных дней недели.""" from collections import defaultdict rasp_list = week_data.get("data", {}).get("rasp", []) days = defaultdict(list) if rasp_list: for lesson in rasp_list: d = lesson["дата"].split("T")[0] days[d].append(lesson) monday = get_monday(requested_date) dt = datetime.strptime(monday, "%Y-%m-%d") try: for i in range(7): current_date = (dt + timedelta(days=i)).strftime("%Y-%m-%d") if current_date == requested_date: continue # Генерируем, только если файла еще нет _generate_single_img(current_date, days.get(current_date, []), group_id, week_data) finally: with _week_img_lock: _bg_generating_weeks.discard((group_id, monday)) def save_week_images(group_id: int, week_data: dict, requested_date: str) -> None: """Сохраняет картинку для запрошенного дня, остальные - в фоне.""" if week_data is None: return from collections import defaultdict rasp_list = week_data.get("data", {}).get("rasp", []) days = defaultdict(list) if rasp_list: for lesson in rasp_list: d = lesson["дата"].split("T")[0] days[d].append(lesson) # 1. Синхронно генерируем запрошенную картинку (бот сразу ее отправит) _generate_single_img(requested_date, days.get(requested_date, []), group_id, week_data) # 2. Запускаем фон для остальных дней, если задача еще не запущена monday = get_monday(requested_date) key = (group_id, monday) with _week_img_lock: if key not in _bg_generating_weeks: _bg_generating_weeks.add(key) t = threading.Thread(target=_bg_generate_rest, args=(group_id, week_data, requested_date), daemon=True) t.start() # --- Конец оптимизации --- def generate_schedule_image(rasp_list, group_id, date_str, info=None): """Генерирует картинку с расписанием. rasp_list - это массив data['data']['rasp']""" bg_color = (26, 26, 46) card_color = (22, 33, 62) text_color = (224, 224, 224) accent_color = (96, 165, 250) secondary_text = (156, 163, 175) divider_color = (31, 41, 55) width = 800 margin = 30 card_width = width - 2 * margin date_obj = datetime.strptime(date_str, "%Y-%m-%d") weekdays = ["Понедельник", "Вторник", "Среда", "Четверг", "Пятница", "Суббота", "Воскресенье"] weekday_ru = weekdays[date_obj.weekday()] header_text = f"Группа {group_id} • {date_str} ({weekday_ru})" # --- ШРИФТЫ --- script_dir = os.path.dirname(os.path.abspath(__file__)) font_paths = { "bold": os.path.join(script_dir, "DejaVuSans-Bold.ttf"), "reg": os.path.join(script_dir, "DejaVuSans.ttf"), } sys_paths = { "bold": "/usr/share/fonts/truetype/dejavu/DejaVuSans-Bold.ttf", "reg": "/usr/share/fonts/truetype/dejavu/DejaVuSans.ttf", } def _load_font(name, size): for paths in [font_paths, sys_paths]: p = paths.get(name) if p and os.path.exists(p): try: return ImageFont.truetype(p, size) except: pass return ImageFont.load_default() font_title = _load_font("bold", 28) font_date = _load_font("reg", 18) font_time = _load_font("bold", 20) font_main = _load_font("reg", 20) font_small = _load_font("reg", 15) font_tiny = _load_font("reg", 13) # --- ОПТИМИЗИРОВАННЫЙ ЗАМЕР ТЕКСТА (БЕЗ СОЗДАНИЯ КАРТИНОК В ЦИКЛЕ) --- measure_img = Image.new('RGB', (1, 1)) measure_draw = ImageDraw.Draw(measure_img) def _text_w(text, font): return measure_draw.textlength(text, font=font) def _wrap_text_smart(text, font, max_width): words = text.split() if not words: return [""] lines = [] current_line = words[0] for word in words[1:]: test = current_line + " " + word if _text_w(test, font) <= max_width: current_line = test else: lines.append(current_line) current_line = word if current_line: lines.append(current_line) return lines time_col_width = 150 time_line_gap = 20 content_x = margin + time_col_width + time_line_gap # Ширина контента с учётом обоих отступов: от content_x + margin до width - margin content_width = width - margin - (content_x + margin) line_height_main = 28 line_height_small = 20 # 1. Считаем высоту card_gap = 24 def _measure_card_height(lesson): subject = lesson.get('дисциплина', lesson.get('discipline', 'Неизвестный предмет')) lesson_type = lesson.get('видЗанятия', '').strip() type_prefix = "" if lesson_type: lt = lesson_type.lower() if "лекц" in lt: type_prefix = "📖 " elif "лаб" in lt: type_prefix = "🔧 " elif "прак" in lt or "семин" in lt: type_prefix = "📝 " else: type_prefix = "📚 " subject_lines = _wrap_text_smart(type_prefix + subject, font_main, content_width) subject_h = len(subject_lines) * line_height_main teacher = lesson.get('фиоПреподавателя', '') auditoria = lesson.get('аудитория', lesson.get('auditoria', '')) detail_parts = [] if teacher: detail_parts.append(f"👤 {teacher}") if auditoria: detail_parts.append(f"📍 ауд. {auditoria}") detail_text = " | ".join(detail_parts) detail_lines = _wrap_text_smart(detail_text, font_small, content_width) if detail_text else [] detail_h = len(detail_lines) * line_height_small top_pad = 22 bottom_pad = 22 gap = 10 return max(top_pad + subject_h + gap + detail_h + bottom_pad, 80) header_height = 110 total_h = header_height card_h_list = [] for i, lesson in enumerate(rasp_list): ch = _measure_card_height(lesson) card_h_list.append(ch) total_h += ch if i < len(rasp_list) - 1: total_h += card_gap total_h += 20 img = Image.new('RGB', (width, total_h), color=bg_color) draw = ImageDraw.Draw(img) draw.text((margin, 20), header_text, font=font_title, fill=text_color) if info and info.get("curWeekNumber") and info.get("curNumNed"): week_type = "чётная" if info.get("curNumNed") == 2 else "нечётная" week_text = f"Неделя {info['curWeekNumber']} ({week_type})" draw.text((margin, 65), week_text, font=font_date, fill=secondary_text) draw.line([(margin, 95), (width - margin, 95)], fill=accent_color, width=3) y_offset = header_height for idx, lesson in enumerate(rasp_list): card_h = card_h_list[idx] draw.rounded_rectangle( [(margin, y_offset), (width - margin, y_offset + card_h)], radius=12, fill=card_color ) time_start = lesson.get('начало', lesson.get('timeStart', '')) time_end = lesson.get('конец', lesson.get('timeEnd', '')) t_start_w = _text_w(time_start, font_time) t_end_w = _text_w(time_end, font_time) max_tw = max(t_start_w, t_end_w) tx = (time_col_width - max_tw) // 2 th = 28 total_th = th * 2 ty_start = y_offset + (card_h - total_th) // 2 ty_end = ty_start + th + 2 draw.text((margin + tx, ty_start), time_start, font=font_time, fill=accent_color) draw.text((margin + tx, ty_end), time_end, font=font_time, fill=secondary_text) line_x = margin + time_col_width + time_line_gap - 1 draw.line([(line_x, y_offset + 14), (line_x, y_offset + card_h - 14)], fill=divider_color, width=2) content_y = y_offset + 22 subject = lesson.get('дисциплина', lesson.get('discipline', 'Неизвестный предмет')) lesson_type = lesson.get('видЗанятия', '').strip() type_prefix = "" if lesson_type: lt = lesson_type.lower() if "лекц" in lt: type_prefix = "📖 " elif "лаб" in lt: type_prefix = "🔧 " elif "прак" in lt or "семин" in lt: type_prefix = "📝 " else: type_prefix = "📚 " subject_lines = _wrap_text_smart(type_prefix + subject, font_main, content_width) for line in subject_lines[:10]: draw.text((content_x + margin, content_y), line, font=font_main, fill=text_color) content_y += line_height_main teacher = lesson.get('фиоПреподавателя', '') auditoria = lesson.get('аудитория', lesson.get('auditoria', '')) detail_parts = [] if teacher: detail_parts.append(f"👤 {teacher}") if auditoria: detail_parts.append(f"📍 ауд. {auditoria}") detail_text = " | ".join(detail_parts) if detail_text: detail_lines = _wrap_text_smart(detail_text, font_small, content_width) for line in detail_lines: draw.text((content_x + margin, content_y + 6), line, font=font_small, fill=secondary_text) content_y += line_height_small y_offset += card_h + card_gap img_byte_arr = io.BytesIO() img.save(img_byte_arr, format='PNG') img_byte_arr.seek(0) return img_byte_arr def format_schedule(data: Optional[dict], group_id: int, date_str: str) -> str: if data is None: return f"❌ Группа {group_id}: временная ошибка получения расписания." # Обработка ошибок if data.get("_error"): # Нет согласия if data.get("no_consent"): return f"❌ Группа {group_id}: нет согласия владельца на использование аккаунта. Обратитесь к администратору." # Отзыв согласия if data.get("revoked"): return f"❌ Владелец аккаунта группы {group_id} отозвал своё согласие. Доступ к расписанию невозмοжен." # Группа заблокирована или идёт обновление if data.get("blocked"): if data.get("waiting"): return (f"⏳ Группа {group_id}: идёт обновление токена. Попробуйте через 10-20 секунд.") return (f"⏳ Группа {group_id}: идёт обновление токена. Скоро будет расписание.") # Таймаут или другая сетевая ошибка if data.get("exception"): exc = data["exception"] if "Read timed out" in exc: return f"❌ Группа {group_id}: сервер деканата не отвечает (таймаут). Попробуйте позже." elif "Connection refused" in exc: return f"❌ Группа {group_id}: не удаётся подключиться к серверу деканата. Проверьте интернет-соединение." else: return f"❌ Группа {group_id}: сетевая ошибка. Попробуйте позже." # Коды HTTP status_code = data.get("status_code") if status_code == 401: return f"❌ Группа {group_id}: токен устарел. Бот обновит автоматически." elif status_code == 403: return f"❌ Группа {group_id}: доступ запрещён." elif status_code == 200: return f"❌ Группа {group_id}: проблема с токеном. Бот обновит." elif status_code and 500 <= status_code < 600: return f"❌ Группа {group_id}: временная ошибка сервера. Попробуйте позже." else: return f"❌ Группа {group_id}: ошибка. Попробуйте позже." # Получаем мета-информацию из ответа API info = data.get("data", {}).get("info", {}) # Если нет ошибки, но state не 1 или rasp пуст if not data or data.get("state") != 1 or not data["data"].get("rasp"): extra = [] if info.get("lastDate"): extra.append(f"📅 Доступно расписание до {info['lastDate'][:10]}") if info.get("curWeekNumber"): week_type = "чётная" if info.get("curNumNed") == 2 else "нечётная" extra.append(f"📆 {info['curWeekNumber']}-я неделя ({week_type})") if info.get("dateUploadingRasp"): try: dt = datetime.fromisoformat(info["dateUploadingRasp"].replace('Z', '+00:00')) extra.append(f"🕒 Обновлено: {dt.strftime('%d.%m.%Y %H:%M')}") except: extra.append(f"🕒 Обновлено: {info['dateUploadingRasp']}") if extra: return f"❌ Группа {group_id}: на {date_str} пар нет.\n" + "\n".join(extra) else: return f"❌ Группа {group_id}: на {date_str} пар нет" # Формирование расписания (если есть занятия) date_obj = datetime.strptime(date_str, "%Y-%m-%d") weekdays = ["ПН", "ВТ", "СР", "ЧТ", "ПТ", "СБ", "ВС"] text = f"📅 Группа {group_id}, {date_str} ({weekdays[date_obj.weekday()]})\n\n" # Добавляем информацию о неделе из ответа API if info.get("curWeekNumber") and info.get("curNumNed"): week_type = "чётная" if info.get("curNumNed") == 2 else "нечётная" text += f"📆 Неделя {info['curWeekNumber']} ({week_type})\n\n" found = False for lesson in data["data"]["rasp"]: # Сравниваем только дату без времени if lesson["дата"].split("T")[0] == date_str: found = True lesson_type = lesson.get("видЗанятия", "").strip() if lesson_type: type_emoji = "📖" if "лекц" in lesson_type.lower() else "🔧" if "лаб" in lesson_type.lower() else "📝" text += f"{type_emoji} {lesson_type}\n" text += f"⏰ {lesson['начало']}—{lesson['конец']}\n" text += f"📘 {lesson['дисциплина']}\n" text += f"👤 {lesson['фиоПреподавателя']}\n" text += f"📍 ауд. {lesson['аудитория']}\n\n" return text if found else f"❌ Группа {group_id}: на {date_str} пар нет" def format_user_info(vk, info: dict) -> str: name = get_user_name(vk, info["user_id"]) lines = [f"👤 Информация о пользователе {name} (ID: {info['user_id']})\n"] if info["registered"]: reg_date = "неизвестно" if info["registered_at"]: reg_date = datetime.fromtimestamp(info["registered_at"]).strftime("%Y-%m-%d %H:%M:%S") lines.append(f"✅ Зарегистрирован в боте: группа {info['group_id']} (с {reg_date})") else: lines.append("❌ Не зарегистрирован в боте (нет активного ключа)") if info["owned_groups"]: lines.append("\n📌 Является владельцем групп:") for g in info["owned_groups"]: cred = "✅" if g["has_credentials"] else "❌" consent = "✅" if g["consent"] else "❌" lines.append(f" • Группа {g['group_id']} – данные: {cred}, согласие: {consent}") else: lines.append("\n📭 Не является владельцем ни одной группы.") return "\n".join(lines) def show_users_of_group(vk, peer_id: int, group_id: int) -> None: users = get_users_by_group(group_id) if not users: safe_vk_call(vk.messages.send, peer_id=peer_id, message=f"📭 В группе {group_id} нет зарегистрированных пользователей.", random_id=get_random_id()) return lines = [f"📋 Пользователи группы {group_id}:\n"] for uid, _ in users: name = get_user_name(vk, uid) user_data = db.get_user(uid) reg_time = user_data.get("registered_at") if user_data else None reg_date = datetime.fromtimestamp(reg_time).strftime("%d.%m.%Y %H:%M") if reg_time else "?" lines.append(f"• {name} (ID: {uid}) – зарегистрирован {reg_date}") instruction = "\n\n🔹 Чтобы отвязать пользователя, отправьте: отвязать ID" safe_vk_call(vk.messages.send, peer_id=peer_id, message="\n".join(lines) + instruction, random_id=get_random_id()) def clear_cache_for_week(group_id: Optional[int] = None) -> int: cleared = 0 groups_to_clear = [group_id] if group_id else GROUPS.keys() for gid in groups_to_clear: for i in range(7): date_str = (datetime.now() + timedelta(days=i)).strftime("%Y-%m-%d") cache_file = get_cache_path(gid, date_str) if os.path.exists(cache_file): try: os.remove(cache_file) cleared += 1 except: pass return cleared def clear_cache_for_group(group_id: int) -> int: """Удаляет все файлы кэша для указанной группы.""" group_dir = os.path.join(RASP_DIR, str(group_id)) if not os.path.exists(group_dir): return 0 removed = 0 for filename in os.listdir(group_dir): if filename.endswith('.json'): try: os.remove(os.path.join(group_dir, filename)) removed += 1 except Exception as e: logger.warning(f"Не удалось удалить {filename}: {e}") removed += clear_image_cache_for_group(group_id) return removed def auto_clean_images(days_old: int = 1) -> int: removed = 0 if not os.path.exists(IMG_DIR): return 0 for group_id_str in os.listdir(IMG_DIR): group_dir = os.path.join(IMG_DIR, group_id_str) if not os.path.isdir(group_dir): continue for filename in os.listdir(group_dir): if filename.endswith('.png'): filepath = os.path.join(group_dir, filename) date_str = filename.replace('.png', '') try: file_date = datetime.strptime(date_str, "%Y-%m-%d") if (datetime.now() - file_date).days > days_old: os.remove(filepath) removed += 1 except: if time.time() - os.path.getmtime(filepath) > days_old * 24 * 3600: os.remove(filepath) removed += 1 return removed def auto_clean_cache(days_old: int = 7, vk=None) -> int: removed = 0 now = time.time() try: if not os.path.exists(RASP_DIR): return 0 for group_id_str in os.listdir(RASP_DIR): group_dir = os.path.join(RASP_DIR, group_id_str) if not os.path.isdir(group_dir): continue for filename in os.listdir(group_dir): if not filename.endswith('.json'): continue filepath = os.path.join(group_dir, filename) try: date_str = filename.replace('.json', '') file_date = datetime.strptime(date_str, "%Y-%m-%d") if (datetime.now() - file_date).days > days_old: os.remove(filepath) removed += 1 logger.info(f"🧹 Автоочистка: удалён {filepath}") except (ValueError, IndexError): file_time = os.path.getmtime(filepath) if now - file_time > days_old * 24 * 3600: os.remove(filepath) removed += 1 logger.info(f"🧹 Автоочистка: удалён {filepath} (по времени)") except Exception as e: logger.error(f"Ошибка автоочистки: {e}") if removed > 0 and vk: try: msg = f"🧹 Автоочистка кэша: удалено {removed} старых файлов" safe_vk_call(vk.messages.send, peer_id=ADMIN_ID, message=msg, random_id=get_random_id()) logger.info(f"📬 Уведомление отправлено админу") except Exception as notify_error: logger.error(f"Не удалось отправить уведомление: {notify_error}") return removed def get_stats(vk): stats_data = parse_api_log() all_users_list = db.get_all_users() total_users = len(all_users_list) total_groups_in_accounts = len(accounts) groups_with_consent = sum(1 for g in accounts.values() if g.get("consent")) groups_blocked = sum(1 for g in accounts.values() if g.get("blocked")) users_by_group = {} for uid, gid in all_users_list: users_by_group.setdefault(gid, 0) users_by_group[gid] += 1 # Кэш файлы # Подсчёт файлов кэша в папке rasp (рекурсивно) cache_count = 0 cache_size = 0 if os.path.exists(RASP_DIR): for root, dirs, files in os.walk(RASP_DIR): for file in files: if file.endswith('.json'): filepath = os.path.join(root, file) cache_count += 1 cache_size += os.path.getsize(filepath) cache_size /= (1024*1024) # Файлы согласий agreement_count = 0 agreements_path = "agreements/разрешенные" # или "/root/mgutmbot/agreements/разрешенные" if os.path.exists(agreements_path): try: agreement_count = len([f for f in os.listdir(agreements_path) if f.endswith('.docx')]) except Exception as e: logger.error(f"Ошибка при подсчёте соглашений: {e}") # Время работы uptime_seconds = time.time() - START_TIME uptime_str = str(timedelta(seconds=int(uptime_seconds))) lines = [] lines.append("📊 **Глобальная статистика бота**") lines.append(f"🕒 Время работы: {uptime_str}") lines.append(f"👥 Зарегистрировано пользователей: {total_users}") lines.append(f"📁 Групп в accounts: {total_groups_in_accounts}") lines.append(f"✅ Групп с согласием: {groups_with_consent}") lines.append(f"⛔ Заблокированных групп: {groups_blocked}") lines.append(f"📈 Всего запросов к API: {sum(stats_data['groups'].values())}") lines.append(f"🗂️ Файлов кэша: {cache_count} ({cache_size:.2f} МБ)") lines.append(f"📄 Соглашений (docx): {agreement_count}") lines.append("") lines.append("👥 Пользователи по группам:") if users_by_group: for gid, cnt in sorted(users_by_group.items()): lines.append(f" Группа {gid}: {cnt}") else: lines.append(" Нет пользователей") lines.append("") lines.append("📋 Группы в accounts:") if accounts: for gid, info in sorted(accounts.items()): owner_name = info.get("owner_name", "?") consent = "✅" if info.get("consent") else "❌" blocked = "⛔" if info.get("blocked") else "" lines.append(f" {gid}: {owner_name} {consent} {blocked}") else: lines.append(" Нет групп") return "\n".join(lines) def get_group_stats(group_id: int): lines = [f"📊 Статистика группы {group_id}\n"] acc_info = accounts.get(group_id) if acc_info: owner_name = acc_info.get("owner_name", "?") consent = "✅" if acc_info.get("consent") else "❌" blocked = "⛔" if acc_info.get("blocked") else "" lines.append(f"👤 Владелец: {owner_name} {consent} {blocked}") else: lines.append("❌ Группа не имеет владельца в accounts") users = get_users_by_group(group_id) lines.append(f"👥 Пользователей в группе: {len(users)}") if users: lines.append(" Список:") for uid, _ in users: name = user_names_cache.get(uid, f"id{uid}") lines.append(f" • {name} (ID: {uid})") # Подсчёт файлов кэша для данной группы cache_count = 0 cache_size = 0 group_cache_dir = os.path.join(RASP_DIR, str(group_id)) if os.path.exists(group_cache_dir): for filename in os.listdir(group_cache_dir): if filename.endswith('.json'): filepath = os.path.join(group_cache_dir, filename) cache_count += 1 cache_size += os.path.getsize(filepath) cache_size_kb = cache_size / 1024 lines.append(f"🗂️ Кэш: {cache_count} файлов ({cache_size_kb:.1f} КБ)") lines.append(get_api_stats_for_group(group_id)) return "\n".join(lines) def get_api_stats_for_group(group_id: int): lines = [] log_file = os.path.join(LOG_DIR, "api_requests.log") if not os.path.exists(log_file): return "❌ Лог API не найден" requests_total = 0 requests_last_24h = 0 successful = 0 auth_errors = 0 # 401, 403, подозрительный 200 server_errors = 0 # 503, другие 5xx connection_errors = 0 # таймауты, исключения other_errors = 0 one_day_ago = time.time() - 24*3600 with open(log_file, "r", encoding="utf-8") as f: for line in f: if f"группа={group_id}" not in line: continue if "🌐 ЗАПРОС" in line: requests_total += 1 match = re.search(r"(\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2})", line) if match: try: dt = datetime.strptime(match.group(1), "%Y-%m-%d %H:%M:%S") if dt.timestamp() >= one_day_ago: requests_last_24h += 1 except: pass elif "✅ УСПЕХ" in line: successful += 1 elif "❌ ОШИБКА" in line or "⚠️ ПОДОЗРИТЕЛЬНЫЙ" in line: auth_errors += 1 elif "⚠️ НЕУДАЧА" in line: # пытаемся извлечь код ответа code_match = re.search(r"код=(\d+)", line) if code_match: code = int(code_match.group(1)) if 500 <= code <= 599: server_errors += 1 else: other_errors += 1 else: other_errors += 1 elif "🔥 ИСКЛЮЧЕНИЕ" in line: connection_errors += 1 total_responses = successful + auth_errors + server_errors + connection_errors + other_errors success_rate = (successful / total_responses * 100) if total_responses else 0 lines.append(f"📈 Статистика API для группы {group_id}") lines.append(f" Всего запросов: {requests_total}") lines.append(f" За 24 часа: {requests_last_24h}") lines.append(f" ✅ Успешных ответов: {successful}") lines.append(f" 🔐 Ошибок авторизации: {auth_errors} (401,403,токен)") lines.append(f" 🖥️ Ошибок сервера: {server_errors} (503 и др. 5xx)") lines.append(f" 🌐 Сбоев соединения: {connection_errors} (таймауты, сеть)") if other_errors: lines.append(f" ❓ Прочих ошибок: {other_errors}") lines.append(f" 📊 Успешность: {success_rate:.1f}%") return "\n".join(lines) def get_last_commands(user_id: int, limit: int = 5) -> List[str]: """Возвращает последние limit команд пользователя из commands.log.""" commands = [] log_file = os.path.join(LOG_DIR, "commands.log") if not os.path.exists(log_file): return commands try: with open(log_file, "r", encoding="utf-8") as f: for line in f: line = line.strip() if not line: continue parts = line.split(" - ", 2) if len(parts) != 3: continue _, uid_str, text = parts if uid_str == str(user_id): commands.append(text) except Exception as e: logger.error(f"Ошибка чтения commands.log: {e}") return commands[-limit:] def get_user_stats(user_id: int, vk): info = get_user_info(user_id) name = get_user_name(vk, user_id) lines = [f"👤 **{name}** (ID: {user_id})"] lines.append("─────────────────────────────") if user_id == ADMIN_ID: lines.append("🛡️ **Администратор бота**") if info["registered"]: reg_date = datetime.fromtimestamp(info["registered_at"]).strftime("%d.%m.%Y") if info["registered_at"] else "?" lines.append(f"✅ Зарегистрирован: {reg_date}, группа {info['group_id']}") else: lines.append("❌ Не зарегистрирован") return "\n".join(lines) # Статистика по API из лога log_file = os.path.join(LOG_DIR, "api_requests.log") total_entries = 0 requests = 0 successful = 0 errors = 0 today_requests = 0 week_requests = 0 today_start = datetime.now().replace(hour=0, minute=0, second=0, microsecond=0) week_start = today_start - timedelta(days=7) last_date = None if os.path.exists(log_file): with open(log_file, "r", encoding="utf-8") as f: for line in f: if f"user_id={user_id}" not in line: continue total_entries += 1 # Извлекаем дату из строки (для любых типов записей) date_match = re.search(r"(\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2})", line) if date_match: dt = datetime.strptime(date_match.group(1), "%Y-%m-%d %H:%M:%S") last_date = dt # последняя активность (любая) # Подсчёт запросов (только строки с "🌐 ЗАПРОС") if "🌐 ЗАПРОС" in line: requests += 1 if date_match: # используем уже извлечённую дату if dt >= today_start: today_requests += 1 if dt >= week_start: week_requests += 1 # Подсчёт успешных ответов и ошибок if "✅ УСПЕХ" in line: successful += 1 if "❌ ОШИБКА" in line or "⚠️ ПОДОЗРИТЕЛЬНЫЙ" in line: errors += 1 if "⚠️ НЕУДАЧА" in line: # добавляем errors += 1 if "🔥 ИСКЛЮЧЕНИЕ" in line: # добавляем errors += 1 # Успешность total_responses = successful + errors if total_responses > 0: success_rate = (successful / total_responses) * 100 filled = int(success_rate / 10) bar = "▰" * filled + "▱" * (10 - filled) success_line = f" {bar} {success_rate:.1f}% ({successful}/{total_responses}) — успешность" else: success_line = " Нет данных об ответах" lines.append("\n📊 **Активность**") lines.append(success_line) lines.append(f" 📅 Запросов сегодня: {today_requests}") lines.append(f" 📅 За эту неделю: {week_requests}") if last_date: last_date_str = last_date.strftime("%d.%m.%Y %H:%M") lines.append(f" 🕒 Последний запрос: {last_date_str}") # Подписки на группы subscribed_groups = [gid for (uid, gid) in get_all_users() if uid == user_id] if subscribed_groups: groups_str = ", ".join(map(str, subscribed_groups)) lines.append(f"\n📌 **Подписки на группы:** {groups_str}") # Владелец групп if info["owned_groups"]: lines.append("\n📌 **Владелец групп:**") for g in info["owned_groups"]: consent = "✅" if g["consent"] else "❌" blocked = "⛔" if accounts.get(g['group_id'], {}).get("blocked") else "" lines.append(f" • Группа {g['group_id']} {consent} {blocked}") # Заметки user_notes = db.get_notes_by_user(user_id) notes_count = len(user_notes) lines.append(f"\n📝 **Заметки:** {notes_count} шт.") # Идеи ideas = db.get_all_ideas() user_ideas = [idea for idea in ideas if idea.get("user_id") == user_id] ideas_count = len(user_ideas) if ideas_count > 0: statuses = {} for idea in user_ideas: status = idea.get("status", "?") statuses[status] = statuses.get(status, 0) + 1 status_str = ", ".join([f"{s}: {c}" for s, c in statuses.items()]) lines.append(f"💡 **Идеи:** {ideas_count} шт. ({status_str})") else: lines.append(f"💡 **Идеи:** 0 шт.") # Последние команды last_commands = get_last_commands(user_id, 5) if last_commands: lines.append("\n🕹️ **Последние команды:**") for cmd in last_commands: if len(cmd) > 30: cmd = cmd[:27] + "..." lines.append(f" • {cmd}") # Рейтинг по API all_stats = parse_api_log() user_counts = all_stats["users"] registered_users = [uid for uid, _ in get_all_users()] ranking_api = [(uid, user_counts.get(uid, 0)) for uid in registered_users] ranking_api.sort(key=lambda x: x[1], reverse=True) position_api = next((i+1 for i, (uid, _) in enumerate(ranking_api) if uid == user_id), None) total_users = len(ranking_api) if position_api and total_users > 0: lines.append(f"\n🏆 **Рейтинг (API):** #{position_api} из {total_users}") # Рейтинг по командам _, commands_counter, _ = parse_commands_log() ranking_cmd = [(uid, commands_counter.get(uid, 0)) for uid in registered_users] ranking_cmd.sort(key=lambda x: x[1], reverse=True) position_cmd = next((i+1 for i, (uid, _) in enumerate(ranking_cmd) if uid == user_id), None) if position_cmd and total_users > 0: lines.append(f"🏆 **Рейтинг (команды):** #{position_cmd} из {total_users}") # Рейтинг в группе по API if info["registered"] and info["group_id"]: group_id = info["group_id"] users_in_group = get_users_by_group(group_id) group_user_ids = [uid for uid, _ in users_in_group] group_stats = {uid: user_counts.get(uid, 0) for uid in group_user_ids} sorted_group = sorted(group_stats.items(), key=lambda x: x[1], reverse=True) group_position = next((i+1 for i, (uid, _) in enumerate(sorted_group) if uid == user_id), None) group_total = len(sorted_group) if group_position and group_total > 0: lines.append(f"🏆 **Рейтинг в группе {group_id}:** #{group_position} из {group_total}") # Рейтинг в группе по командам if info["registered"] and info["group_id"]: group_cmd_stats = {uid: commands_counter.get(uid, 0) for uid in group_user_ids} sorted_group_cmd = sorted(group_cmd_stats.items(), key=lambda x: x[1], reverse=True) group_cmd_position = next((i+1 for i, (uid, _) in enumerate(sorted_group_cmd) if uid == user_id), None) group_cmd_total = len(sorted_group_cmd) if group_cmd_position and group_cmd_total > 0: lines.append(f"🏆 **Рейтинг команд в группе {group_id}:** #{group_cmd_position} из {group_cmd_total}") lines.append("─────────────────────────────") return "\n".join(lines) # =================== АРХИВ ЛОГОВ ==================== def archive_logs(vk=None) -> int: """Переносит содержимое logs/ и api_logs/ в system_log/дата/имя_папки/ и очищает исходные папки.""" moved = 0 archive_root = "system_log" os.makedirs(archive_root, exist_ok=True) date_str = datetime.now().strftime("%Y-%m-%d") target_base = os.path.join(archive_root, date_str) for src in [LOG_DIR, "api_logs"]: if not os.path.exists(src): continue # Создаём подпапку с именем исходной папки target_dir = os.path.join(target_base, src) os.makedirs(target_dir, exist_ok=True) for filename in os.listdir(src): src_path = os.path.join(src, filename) if os.path.isfile(src_path): dst_path = os.path.join(target_dir, filename) if os.path.exists(dst_path): base, ext = os.path.splitext(filename) new_name = f"{base}_{datetime.now().strftime('%H%M%S')}{ext}" dst_path = os.path.join(target_dir, new_name) shutil.move(src_path, dst_path) moved += 1 logger.info(f"Архивация логов: перемещено {moved} файлов в {target_base}") return moved def get_top_groups_by_requests(limit=5): stats = parse_api_log() sorted_groups = sorted(stats["groups"].items(), key=lambda x: x[1], reverse=True)[:limit] lines = [f"🏆 Топ-{limit} групп по запросам к API\n"] if not sorted_groups: lines.append(" Нет данных") else: for i, (gid, count) in enumerate(sorted_groups, 1): lines.append(f"{i}. Группа {gid}: {count} запросов") return "\n".join(lines) def get_top_users_by_requests(limit=5, vk=None): stats = parse_api_log() sorted_users = sorted(stats["users"].items(), key=lambda x: x[1], reverse=True)[:limit] lines = [f"🏆 Топ-{limit} пользователей по запросам к API\n"] if not sorted_users: lines.append(" Нет данных") else: for i, (uid, count) in enumerate(sorted_users, 1): if vk: name = get_user_name(vk, uid) else: name = f"id{uid}" lines.append(f"{i}. {name} (ID {uid}): {count} запросов") return "\n".join(lines) def get_top_users_in_group(group_id: int, limit: int = 5, vk=None): """Возвращает топ-N пользователей по количеству запросов к API для указанной группы.""" stats = {} log_file = os.path.join(LOG_DIR, "api_requests.log") if not os.path.exists(log_file): return "❌ Лог API не найден" with open(log_file, "r", encoding="utf-8") as f: for line in f: if f"группа={group_id}" not in line: continue user_match = re.search(r"user_id=(\d+)", line) if user_match: uid = int(user_match.group(1)) stats[uid] = stats.get(uid, 0) + 1 if not stats: return f"📭 Для группы {group_id} нет запросов в логе." sorted_users = sorted(stats.items(), key=lambda x: x[1], reverse=True)[:limit] lines = [f"🏆 Топ-{limit} пользователей в группе {group_id} по запросам к API\n"] for i, (uid, count) in enumerate(sorted_users, 1): if vk: name = get_user_name(vk, uid) else: name = f"id{uid}" lines.append(f"{i}. {name} (ID {uid}): {count} запросов") return "\n".join(lines) def get_archive_stats(date_str: str, log_type: str) -> str: """Возвращает статистику из архива за указанную дату и тип логов.""" base_path = os.path.join("system_log", date_str, log_type) if not os.path.exists(base_path): return f"❌ Архив не найден: {base_path}" if log_type == "api_logs": log_file = os.path.join(base_path, "api_requests.log") if not os.path.exists(log_file): return "❌ Файл api_requests.log не найден в архиве." stats = {"groups": {}, "users": {}} successful = 0 errors = 0 total_requests = 0 with open(log_file, "r", encoding="utf-8") as f: for line in f: if "🌐 ЗАПРОС" in line: total_requests += 1 group_match = re.search(r"группа=(\d+)", line) user_match = re.search(r"user_id=(\d+)", line) if group_match: gid = int(group_match.group(1)) stats["groups"][gid] = stats["groups"].get(gid, 0) + 1 if user_match: uid = int(user_match.group(1)) stats["users"][uid] = stats["users"].get(uid, 0) + 1 elif "✅ УСПЕХ" in line: successful += 1 elif "❌ ОШИБКА" in line or "⚠️ ПОДОЗРИТЕЛЬНЫЙ" in line: errors += 1 total_responses = successful + errors success_rate = (successful / total_responses * 100) if total_responses else 0 lines = [f"📊 Статистика API за {date_str}"] lines.append(f"📁 Архив: {log_type}") lines.append(f"🌐 Всего запросов: {total_requests}") lines.append(f"✅ Успешных ответов: {successful}") lines.append(f"❌ Ошибок: {errors}") lines.append(f"📈 Успешность: {success_rate:.1f}%") top_groups = sorted(stats["groups"].items(), key=lambda x: x[1], reverse=True)[:5] if top_groups: lines.append("\n🏆 Топ групп по запросам:") for i, (gid, cnt) in enumerate(top_groups, 1): lines.append(f" {i}. Группа {gid}: {cnt}") top_users = sorted(stats["users"].items(), key=lambda x: x[1], reverse=True)[:5] if top_users: lines.append("\n🏆 Топ пользователей по запросам:") for i, (uid, cnt) in enumerate(top_users, 1): lines.append(f" {i}. ID {uid}: {cnt}") return "\n".join(lines) elif log_type == "logs": log_file = os.path.join(base_path, "commands.log") if not os.path.exists(log_file): return "❌ Файл commands.log не найден в архиве." all_msgs = Counter() commands = Counter() command_names = Counter() base_commands = {"сегодня", "завтра", "день", "неделя", "след", "след.", "следующая", "обновить", "сброс", "помощь", "меню"} with open(log_file, "r", encoding="utf-8") as f: for line in f: line = line.strip() if not line: continue parts = line.split(" - ", 2) if len(parts) != 3: continue _, user_id_str, text = parts try: user_id = int(user_id_str) except ValueError: continue all_msgs[user_id] += 1 if re.fullmatch(r'[0-9a-f]{16}', text): continue clean = re.sub(r'^[^\w\s]+', '', text).strip().lower() words = clean.split() if not words: continue cmd = words[0] # Пропускаем команды навигации по страницам if "страница" in clean: continue normalized_cmd = cmd if cmd in ("след", "след.", "следующая", "следующая_неделя", "след_неделя"): normalized_cmd = "след" if cmd in base_commands or cmd.startswith('!') or cmd in ["отозвать"]: commands[user_id] += 1 command_names[normalized_cmd] += 1 lines = [f"📊 Статистика команд за {date_str}"] lines.append(f"📁 Архив: {log_type}") lines.append(f"👥 Всего сообщений: {sum(all_msgs.values())}") lines.append(f"🎮 Всего команд: {sum(commands.values())}") top_cmds = command_names.most_common(5) if top_cmds: lines.append("\n🏆 Топ команд:") for i, (cmd, cnt) in enumerate(top_cmds, 1): lines.append(f" {i}. {cmd}: {cnt}") top_users = commands.most_common(5) if top_users: lines.append("\n🏆 Топ пользователей по командам:") for i, (uid, cnt) in enumerate(top_users, 1): lines.append(f" {i}. ID {uid}: {cnt}") return "\n".join(lines) else: return f"❌ Неизвестный тип логов: {log_type}" # ========== форматирования страницы отзывов ============= def get_reviews_page(page: int = 1, per_page: int = REVIEWS_PER_PAGE) -> Tuple[str, int, int]: reviews, total_pages = db.get_reviews_page(page, per_page) if not reviews: return "📭 Отзывов пока нет.", page, total_pages lines = [f"📢 Отзывы (страница {page}/{total_pages}):\n"] for rev in reviews: user_name = rev.get('user_name', 'Аноним') rating = rev.get('rating') stars = f"{rating}★ " if rating else "" timestamp = rev.get('timestamp', '') date_str = timestamp[:10] if timestamp else 'дата не указана' text = rev.get('text', '') lines.append(f"{user_name} ({date_str}) {stars}: {text}") return "\n".join(lines), page, total_pages def _get_date_from_command(text: str) -> str: """Извлекает дату из команды расписания.""" text_lower = text.lower() if "сегодня" in text_lower: return datetime.now().strftime("%Y-%m-%d") if "завтра" in text_lower: return (datetime.now() + timedelta(days=1)).strftime("%Y-%m-%d") if text_lower.startswith("день "): parts = text.split(maxsplit=1) if len(parts) >= 2: arg = parts[1].strip() if '-' in arg: try: return datetime.strptime(arg, "%Y-%m-%d").strftime("%Y-%m-%d") except ValueError: pass else: try: days = int(arg) return (datetime.now() + timedelta(days=days)).strftime("%Y-%m-%d") except ValueError: pass if "неделя" in text_lower: return datetime.now().strftime("%Y-%m-%d") return datetime.now().strftime("%Y-%m-%d") # ========== ОБРАБОТКА КОМАНД РАСПИСАНИЯ ========== def handle_schedule_command(text: str, user_group: int, vk=None, user_id=None, user_name=None) -> Optional[str]: text_lower = text.lower() if "сегодня" in text_lower: d = datetime.now().strftime("%Y-%m-%d") data, _ = get_schedule_for_date(user_group, d, vk, user_id, user_name) return format_schedule(data, user_group, d) if data else "❌ Ошибка" if "завтра" in text_lower: d = (datetime.now() + timedelta(days=1)).strftime("%Y-%m-%d") data, _ = get_schedule_for_date(user_group, d, vk, user_id, user_name) return format_schedule(data, user_group, d) if data else "❌ Ошибка" if text_lower.startswith("день "): # Получаем второй аргумент (всё, что после "день ") parts = text.split(maxsplit=1) if len(parts) < 2: return "❌ Укажите дату или количество дней. Пример: день 3 или день 2025-04-15" arg = parts[1].strip() # Пробуем распарсить как дату в формате ГГГГ-ММ-ДД try: # Проверяем, похоже ли на дату (содержит дефисы) if '-' in arg: target_date = datetime.strptime(arg, "%Y-%m-%d") d = target_date.strftime("%Y-%m-%d") else: # Если нет дефиса, считаем числом дней days = int(arg) d = (datetime.now() + timedelta(days=days)).strftime("%Y-%m-%d") except ValueError: # Если не удалось, пробуем как число дней (на случай, если ввели просто число) try: days = int(arg) d = (datetime.now() + timedelta(days=days)).strftime("%Y-%m-%d") except ValueError: return "❌ Неверный формат. Используйте: день 3 (через N дней) или день 2025-04-15 (конкретная дата)." # Получаем и возвращаем расписание data, _ = get_schedule_for_date(user_group, d, vk, user_id, user_name) return format_schedule(data, user_group, d) if data else "❌ Ошибка получения расписания" if "неделя" in text_lower: is_next_week = "след" in text_lower today = datetime.now() if is_next_week: # Следующая неделя: находим понедельник следующей недели days_until_next_monday = (7 - today.weekday()) % 7 if days_until_next_monday == 0: days_until_next_monday = 7 base = today + timedelta(days=days_until_next_monday) week_dates = [(base - timedelta(days=base.weekday()) + timedelta(days=i)).strftime("%Y-%m-%d") for i in range(7)] header = "📋 СЛЕДУЮЩАЯ НЕДЕЛЯ" else: # Текущая неделя if today.weekday() == 6: base = today + timedelta(days=1) else: base = today monday = base - timedelta(days=base.weekday()) week_dates = [(monday + timedelta(days=i)).strftime("%Y-%m-%d") for i in range(7)] header = "📋 НЕДЕЛЯ" result = f"{header} (группа {user_group})\n" result += f"⏱️ {datetime.now().strftime('%d.%m.%Y %H:%M')}\n\n" all_data = [] info = None error = False for date_str in week_dates: data, _ = get_schedule_for_date(user_group, date_str, vk, user_id, user_name) if data is None or data.get("_error"): result += format_schedule(data, user_group, date_str) + "\n" error = True break if info is None: info = data.get("data", {}).get("info", {}) lessons = [l for l in data.get("data", {}).get("rasp", []) if l["дата"].startswith(date_str)] all_data.append((date_str, data, lessons)) if error: return result if info.get("curWeekNumber") and info.get("curNumNed"): week_type = "чётная" if info.get("curNumNed") == 2 else "нечётная" result += f"📆 Неделя {info['curWeekNumber']} ({week_type})\n\n" weekdays_ru = ["ПН", "ВТ", "СР", "ЧТ", "ПТ", "СБ", "ВС"] for i, (date_str, data, lessons) in enumerate(all_data): dt = datetime.strptime(date_str, "%Y-%m-%d") weekday_ru = weekdays_ru[dt.weekday()] result += f"📅 {date_str} ({weekday_ru})\n" if lessons: for lesson in sorted(lessons, key=lambda x: x["начало"]): result += f"⏰ {lesson['начало']}—{lesson['конец']}\n" result += f"📘 {lesson['дисциплина']}\n" result += f"👤 {lesson['фиоПреподавателя']}\n" result += f"📍 ауд. {lesson['аудитория']}\n\n" else: result += "Пар нет\n\n" if i < len(all_data) - 1: result += "-" * 30 + "\n\n" return result if any(w in text_lower for w in ["обновить", "сброс"]): cleared = clear_cache_for_group(user_group) return f"✅ Кэш очищен. Удалено файлов: {cleared}." # ---------- ОБРАБОТКА КОМАНДЫ ПОМОЩИ ---------- # ========== ПРОВЕРКА ТОКЕНА VK ========== def check_vk_token(vk) -> bool: try: groups = vk.groups.getById(group_id=GROUP_ID_VK) if groups: logger.info(f"✅ Сообщество: {groups[0]['name']}") return True except ApiError as e: logger.critical(f"❌ Ошибка токена VK: {e}") return False # ========== ОБРАБОТЧИК СООБЩЕНИЙ ========== def process_message(vk, event): msg = event.obj.message peer_id = msg['peer_id'] from_id = msg['from_id'] text = msg.get('text', '').strip() reply_msg = msg.get('reply_message') reply_text = "" reply_cmd = "" if reply_msg: reply_text = reply_msg.get('text', '') reply_cmd = reply_msg.get('text', '') if len(reply_text) > 500: reply_text = reply_text[:500] + "..." logger.info(f"‼️ Сообщение от {from_id} в чат {peer_id}: {text}") # Запись всех личных сообщений в лог команд if peer_id == from_id: try: commands_log_path = os.path.join(LOG_DIR, "commands.log") with open(commands_log_path, "a", encoding="utf-8") as f_cmd: f_cmd.write(f"{datetime.now().isoformat()} - {from_id} - {text}\n") except Exception as e: logger.error(f"Не удалось записать в commands.log: {e}") # ---------- ОБРАБОТКА ОТВЕТОВ НА ЗАПРОСЫ СОГЛАСИЯ (В САМОМ НАЧАЛЕ) ---------- if from_id in pending_consent: consent_data = pending_consent[from_id] group_id = consent_data["group_id"] admin_id = consent_data["admin_id"] text_lower = text.lower().strip() group = accounts.get(group_id) if not group or group.get("owner_id") != from_id: del pending_consent[from_id] save_pending_consent(pending_consent) safe_vk_call(vk.messages.send, peer_id=peer_id, message="❌ Ошибка: данные устарели.", random_id=get_random_id()) return owner_name = group.get("owner_name", f"id{from_id}") # Положительный ответ if any(word in text_lower for word in ["да", "разрешить", "✅", "yes", "ага", "ок"]): if group.get("revoked"): group["revoked"] = False if group.get("blocked"): group["blocked"] = False if not group.get("has_credentials"): msg_to_owner = (f"{owner_name}, для обновления кук необходимы ваш логин и пароль от аккаунта деканата.\n" f"Пожалуйста, отправьте их администратору в личные сообщения: https://vk.com/id{admin_id}\n" f"После получения данных администратор сможет обновить куки и расписание продолжит работу.") safe_vk_call(vk.messages.send, peer_id=from_id, message=msg_to_owner, random_id=get_random_id()) groups_of_owner = [str(g) for g, v in accounts.items() if v.get("owner_id") == from_id] groups_str = ", ".join(groups_of_owner) msg_to_admin = (f"⚠️ Пользователь {owner_name} (ID {from_id}) согласился на обновление кук для группы {group_id}.\n" f"У него отсутствуют сохранённые логин/пароль. Попросите его прислать данные вам в ЛС.\n" f"Группы владельца: {groups_str}") safe_vk_call(vk.messages.send, peer_id=admin_id, message=msg_to_admin, random_id=get_random_id()) del pending_consent[from_id] save_pending_consent(pending_consent) return # Есть данные – создаём документ Word os.makedirs("agreements", exist_ok=True) timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") filename = f"agreements/consent_{group_id}_{timestamp}.docx" groups_of_owner = [str(g) for g, v in accounts.items() if v.get("owner_id") == from_id] groups_str = ", ".join(groups_of_owner) from docx import Document from docx.shared import Pt, Cm from docx.enum.text import WD_ALIGN_PARAGRAPH doc = Document() # Заголовок p = doc.add_paragraph("СОГЛАШЕНИЕ НА ИСПОЛЬЗОВАНИЕ АККАУНТА") p.alignment = WD_ALIGN_PARAGRAPH.CENTER run = p.runs[0] run.font.name = 'Times New Roman' run.font.size = Pt(14) run.bold = True p.paragraph_format.space_after = Pt(0) p.paragraph_format.space_before = Pt(0) p.paragraph_format.line_spacing = 1.5 p.paragraph_format.left_indent = Cm(0) p.paragraph_format.right_indent = Cm(0) p.paragraph_format.first_line_indent = Cm(0) # Функция для создания обычного абзаца (без отступа) def add_normal_paragraph(text): p = doc.add_paragraph(text) p.alignment = WD_ALIGN_PARAGRAPH.JUSTIFY run = p.runs[0] run.font.name = 'Times New Roman' run.font.size = Pt(14) p.paragraph_format.space_after = Pt(0) p.paragraph_format.space_before = Pt(0) p.paragraph_format.line_spacing = 1.5 p.paragraph_format.left_indent = Cm(0) p.paragraph_format.right_indent = Cm(0) p.paragraph_format.first_line_indent = Cm(0) return p # Дата add_normal_paragraph(f"Дата: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}") # Владелец add_normal_paragraph(f"Владелец аккаунта: {owner_name} (ID: {from_id})") # Группы add_normal_paragraph(f"Группы: {groups_str}") # Администратор add_normal_paragraph(f"Администратор: ID {admin_id}") # Основной текст (с отступом первой строки) p = doc.add_paragraph() p.alignment = WD_ALIGN_PARAGRAPH.JUSTIFY run = p.add_run( f"Настоящим я, {owner_name}, даю согласие администратору бота " f"на использование моего аккаунта на сайте деканата для получения расписания указанных групп. " f"Я понимаю, что для этого администратору потребуется войти в мой аккаунт и обновить куки." ) run.font.name = 'Times New Roman' run.font.size = Pt(14) p.paragraph_format.space_after = Pt(0) p.paragraph_format.space_before = Pt(0) p.paragraph_format.line_spacing = 1.5 p.paragraph_format.left_indent = Cm(0) p.paragraph_format.right_indent = Cm(0) p.paragraph_format.first_line_indent = Cm(1.25) # отступ первой строки 1.25 см # Подпись add_normal_paragraph("Подпись: (согласие выражено через бота ВК, отправлен ответ «Да»)") # Информация о файле add_normal_paragraph(f"Это соглашение сохранено в файле: {filename}") # Сохраняем документ doc.save(filename) # Отправляем документ админу try: uploader = upload.VkUpload(vk) doc = uploader.document_message( filename, peer_id=admin_id, title=f"Согласие_группа{group_id}.docx" ) attachment = f"doc{doc['doc']['owner_id']}_{doc['doc']['id']}" safe_vk_call(vk.messages.send, peer_id=admin_id, message=f"✅ Получено согласие от {owner_name} для группы {group_id}.", attachment=attachment, random_id=get_random_id()) except Exception as e: logger.exception("Ошибка при отправке документа через VkUpload") safe_vk_call(vk.messages.send, peer_id=admin_id, message=f"⚠️ Согласие получено, но не удалось отправить документ. Файл сохранён локально: {filename}", random_id=get_random_id()) with accounts_lock: accounts[group_id]["consent"] = { "given": True, "file": filename, "given_at": time.time(), "admin_id": admin_id } db.save_group_account(group_id, accounts[group_id]) safe_vk_call(vk.messages.send, peer_id=from_id, message="✅ Ваше согласие записано. Спасибо!", random_id=get_random_id()) del pending_consent[from_id] save_pending_consent(pending_consent) return # Отрицательный ответ elif any(word in text_lower for word in ["нет", "запретить", "❌", "no", "не"]): safe_vk_call(vk.messages.send, peer_id=admin_id, message=f"❌ Владелец отклонил запрос для группы {group_id}.", random_id=get_random_id()) safe_vk_call(vk.messages.send, peer_id=from_id, message="❌ Вы отклонили запрос. Доступ к расписанию ваших групп будет отключён.", random_id=get_random_id()) del pending_consent[from_id] save_pending_consent(pending_consent) return else: safe_vk_call(vk.messages.send, peer_id=peer_id, message="Пожалуйста, ответьте 'Да' или 'Нет'.", random_id=get_random_id()) return # ---------- РЕГИСТРАЦИЯ ПО КЛЮЧУ ---------- user_str = str(from_id) user_data = db.get_user(from_id) if user_data and user_data.get("notified_version") != BOT_VERSION: if peer_id == from_id: safe_vk_call(vk.messages.send, peer_id=peer_id, message=UPDATE_MESSAGE, random_id=get_random_id(), keyboard=get_main_keyboard()) db.update_user_notified_version(from_id, BOT_VERSION) # ---------- РЕГИСТРАЦИЯ ПО КЛЮЧУ ---------- key_info = db.get_key(text) if text and key_info: group_id = register_user(from_id, text) if group_id: user_info = safe_vk_call(vk.users.get, user_ids=from_id) user_name = f"id{from_id}" if user_info: user_name = f"{user_info[0]['first_name']} {user_info[0]['last_name']}" user_main_page[from_id] = 1 # сразу сохраняем первую страницу # Улучшенное приветствие welcome_text = ( f"✅ Привет, {user_name}!\n" f"Ты привязан к группе {group_id}.\n\n" "📌 **Что я умею:**\n" "• 📅 Расписание: сегодня, завтра, неделя\n" "• 📝 Личные заметки с привязкой к датам\n" "• 💡 Предложить идею для бота\n" "• ✍️ Оставить отзыв или пожелание\n\n" "🔹 **Быстрые команды:**\n" "`сегодня`, `завтра`, `неделя`, `день 3`, `заметки`\n\n" "Используй кнопки меню для быстрого доступа.\n" "Удачи!" ) safe_vk_call(vk.messages.send, peer_id=peer_id, message=welcome_text, random_id=get_random_id(), keyboard=get_main_keyboard()) safe_vk_call(vk.messages.send, peer_id=ADMIN_ID, message=f"🔔 Новый пользователь: {user_name} (ID: {from_id}) группа {group_id}\n" f"👉 Чтобы отвязать, отправь: отвязать {from_id}", random_id=get_random_id()) logger.info(f"Пользователь {from_id} зарегистрирован в группе {group_id}") return # ---------- ОТОЗВАТЬ СОГЛАСИЕ (ВЛАДЕЛЕЦ) ---------- if peer_id == from_id and text.lower() in ["отозвать", "!revoke"]: owned_groups = [gid for gid, acc in accounts.items() if acc.get("owner_id") == from_id] if not owned_groups: safe_vk_call(vk.messages.send, peer_id=peer_id, message="❌ Вы не являетесь владельцем ни одной группы.", random_id=get_random_id()) return if len(owned_groups) == 1: # Одна группа – сразу запрос подтверждения user_revoke_state[from_id] = {"step": "confirm", "group_id": owned_groups[0]} safe_vk_call(vk.messages.send, peer_id=peer_id, message=f"Вы действительно хотите отозвать согласие для группы {owned_groups[0]}?", random_id=get_random_id(), keyboard=get_confirm_keyboard()) else: # Несколько групп – показываем выбор user_revoke_state[from_id] = {"step": "select", "groups": owned_groups} safe_vk_call(vk.messages.send, peer_id=peer_id, message="Выберите группу, для которой хотите отозвать согласие:", random_id=get_random_id(), keyboard=get_group_selection_keyboard_for_revoke(owned_groups)) return # ---------- ВЫБОР ГРУППЫ ДЛЯ ОТЗЫВА ---------- if from_id in user_revoke_state and user_revoke_state[from_id]["step"] == "select" and text.startswith("Группа "): try: group_id = int(text.replace("Группа ", "")) except ValueError: safe_vk_call(vk.messages.send, peer_id=peer_id, message="❌ Неверный формат группы.", random_id=get_random_id()) del user_revoke_state[from_id] return if group_id in user_revoke_state[from_id]["groups"]: # Переходим к подтверждению user_revoke_state[from_id] = {"step": "confirm", "group_id": group_id} safe_vk_call(vk.messages.send, peer_id=peer_id, message=f"Вы действительно хотите отозвать согласие для группы {group_id}?", random_id=get_random_id(), keyboard=get_confirm_keyboard()) else: safe_vk_call(vk.messages.send, peer_id=peer_id, message="❌ Эта группа вам не принадлежит.", random_id=get_random_id()) del user_revoke_state[from_id] return # ---------- ПОДТВЕРЖДЕНИЕ ОТЗЫВА ---------- if from_id in user_revoke_state and user_revoke_state[from_id]["step"] == "confirm": if text == "✅ Да": group_id = user_revoke_state[from_id]["group_id"] if revoke_consent(group_id, from_id, vk): safe_vk_call(vk.messages.send, peer_id=peer_id, message=f"✅ Согласие для группы {group_id} отозвано. Доступ к расписанию отключён.", random_id=get_random_id(), keyboard=VkKeyboard.get_empty_keyboard()) else: safe_vk_call(vk.messages.send, peer_id=peer_id, message="❌ Не удалось отозвать согласие.", random_id=get_random_id()) del user_revoke_state[from_id] elif text == "❌ Нет": safe_vk_call(vk.messages.send, peer_id=peer_id, message="❌ Отменено.", random_id=get_random_id(), keyboard=VkKeyboard.get_empty_keyboard()) del user_revoke_state[from_id] else: # Если ответ не "Да" и не "Нет", остаёмся в состоянии – ничего не делаем pass return # ---------- ОТМЕНА В СОСТОЯНИИ ВЫБОРА ---------- if from_id in user_revoke_state and user_revoke_state[from_id]["step"] == "select" and text == "❌ Отмена": del user_revoke_state[from_id] safe_vk_call(vk.messages.send, peer_id=peer_id, message="❌ Отменено.", random_id=get_random_id(), keyboard=VkKeyboard.get_empty_keyboard()) return # ---------- СПИСОК ГРУПП ВЛАДЕЛЬЦА ---------- if peer_id == from_id and text.lower() in ["мои группы", "!mygroups"]: owned = [gid for gid, acc in accounts.items() if acc.get("owner_id") == from_id] if not owned: safe_vk_call(vk.messages.send, peer_id=peer_id, message="❌ Вы не являетесь владельцем ни одной группы.", random_id=get_random_id()) return lines = ["📋 **Ваши группы:**\n"] for gid in owned: acc = accounts[gid] consent = "✅" if acc.get("consent") else "❌" blocked = "⛔" if acc.get("blocked") else "" lines.append(f"• Группа {gid}: согласие {consent} {blocked}") lines.append("\n💡 Чтобы отозвать согласие для группы, отправьте: **отозвать**") safe_vk_call(vk.messages.send, peer_id=peer_id, message="\n".join(lines), random_id=get_random_id()) return # ---------- АДМИН-КОМАНДЫ ---------- is_admin = (from_id == ADMIN_ID) is_private = (peer_id == from_id) if is_admin and is_private: # ========== СТАТИСТИКА (В ПЕРВУЮ ОЧЕРЕДЬ) ========== if text.lower() == "!stats": stats_text = get_stats(vk) safe_vk_call(vk.messages.send, peer_id=peer_id, message=stats_text, random_id=get_random_id()) return if text.lower() == "!stats group": group_ids = sorted(accounts.keys()) if not group_ids: safe_vk_call(vk.messages.send, peer_id=peer_id, message="❌ Нет групп в accounts.", random_id=get_random_id()) return user_stats_state[from_id] = {"action": "group_stats", "groups": group_ids} safe_vk_call(vk.messages.send, peer_id=peer_id, message="Выберите группу для просмотра статистики:", random_id=get_random_id(), keyboard=get_group_selection_keyboard_for_stats(group_ids)) return if text.lower() == "!stats api": group_ids = sorted(accounts.keys()) if not group_ids: safe_vk_call(vk.messages.send, peer_id=peer_id, message="❌ Нет групп в accounts.", random_id=get_random_id()) return user_stats_state[from_id] = {"action": "api_stats", "groups": group_ids} safe_vk_call(vk.messages.send, peer_id=peer_id, message="Выберите группу для просмотра статистики API:", random_id=get_random_id(), keyboard=get_group_selection_keyboard_for_stats(group_ids)) return if text.lower() == "!stats archive": # Поиск доступных дат в system_log archive_root = "system_log" if not os.path.exists(archive_root): safe_vk_call(vk.messages.send, peer_id=peer_id, message="❌ Архив логов не найден.", random_id=get_random_id()) return dates = [] for item in os.listdir(archive_root): if os.path.isdir(os.path.join(archive_root, item)): # Проверяем, есть ли внутри папки logs или api_logs has_logs = os.path.exists(os.path.join(archive_root, item, "logs")) or \ os.path.exists(os.path.join(archive_root, item, "api_logs")) if has_logs: dates.append(item) if not dates: safe_vk_call(vk.messages.send, peer_id=peer_id, message="❌ Нет доступных архивов.", random_id=get_random_id()) return dates.sort(reverse=True) # Выводим список дат msg = "📅 Доступные даты:\n" + "\n".join(f"• {d}" for d in dates[:20]) msg += "\n\nВведите дату в формате ГГГГ-ММ-ДД, чтобы выбрать." admin_state[from_id] = "archive" safe_vk_call(vk.messages.send, peer_id=peer_id, message=msg, random_id=get_random_id(), keyboard=get_cancel_keyboard()) return if text.lower().startswith("!stats user "): parts = text.split() if len(parts) == 3: try: target_id = int(parts[2]) stats_text = get_user_stats(target_id, vk) safe_vk_call(vk.messages.send, peer_id=peer_id, message=stats_text, random_id=get_random_id()) except ValueError: safe_vk_call(vk.messages.send, peer_id=peer_id, message="❌ Неверный ID пользователя.", random_id=get_random_id()) else: safe_vk_call(vk.messages.send, peer_id=peer_id, message="❌ Использование: !stats user ", random_id=get_random_id()) return # ---------- БАГ-РЕПОРТЫ (АДМИН) ---------- if text.lower() == "!bugs": bugs = db.get_all_bug_reports() if not bugs: safe_vk_call(vk.messages.send, peer_id=peer_id, message="📭 Баг-репортов пока нет.", random_id=get_random_id()) return status_emojis = {"новая": "🆕", "в работе": "⚙️", "выполнена": "✅", "отклонена": "❌"} sev_emojis = {"мелкая": "🟢", "средняя": "🟡", "критичная": "🔴"} new_count = sum(1 for b in bugs if b.get("status") == "новая") work_count = sum(1 for b in bugs if b.get("status") == "в работе") done_count = sum(1 for b in bugs if b.get("status") == "выполнена") lines = [f"🐛 **Баг-репорты** ({len(bugs)} всего)\n" f"🆕 {new_count} новых | ⚙️ {work_count} в работе | ✅ {done_count} выполнено\n"] for bug in bugs[:20]: status = bug.get("status", "новая") emoji = status_emojis.get(status, "❓") full_text = bug.get("text", "") title = "" severity = "" desc = full_text if "📌 " in full_text: parts = full_text.split("\n") # Правильная обрезка с учётом длины префикса title = parts[0][len("📌 "):] sev_line = next((l for l in parts if "🔴 Важность" in l), "") if sev_line: severity = sev_line[len("🔴 Важность: "):] # Собираем описание (всё между заголовком и важностью) desc_lines = [] in_desc = False for p in parts: if p.startswith("📋 "): in_desc = True desc_lines.append(p[len("📋 "):]) elif p.startswith("🔴 Важность"): in_desc = False elif in_desc: desc_lines.append(p) desc = "\n".join(desc_lines).strip() sev = sev_emojis.get(severity, "🟡") preview = desc[:40] + ("…" if len(desc) > 40 else "") line = f"{emoji} #{bug['id']} {sev} {bug.get('user_name', '?')}: {title or preview}" if bug.get("admin_reply"): line += f"\n 💬 {bug['admin_reply'][:50]}" line += f"\n 📅 {bug['timestamp'][:10]}" lines.append(line) if len(bugs) > 20: lines.append(f"... и ещё {len(bugs)-20}") lines.append("\n💬 `!bug reply <текст>` — ответить") lines.append("✅ `!bug status <статус>` — статус") lines.append("🗑 `!bug delete ` — удалить") safe_vk_call(vk.messages.send, peer_id=peer_id, message="\n".join(lines), random_id=get_random_id()) return if text.lower().startswith("!bug reply"): parts = text.split(maxsplit=3) if len(parts) < 4: safe_vk_call(vk.messages.send, peer_id=peer_id, message="❌ Использование: !bug reply <текст>", random_id=get_random_id()) return try: bug_id = int(parts[2]) reply_text = parts[3] except ValueError: safe_vk_call(vk.messages.send, peer_id=peer_id, message="❌ ID должен быть числом.", random_id=get_random_id()) return bug = db.get_bug_report(bug_id) if not bug: safe_vk_call(vk.messages.send, peer_id=peer_id, message=f"❌ Баг #{bug_id} не найден.", random_id=get_random_id()) return old_status = bug.get("status", "новая") if old_status == "новая": db.update_bug_status(bug_id, "в работе", reply_text) else: db.update_bug_status(bug_id, old_status, reply_text) author_id = bug.get("user_id") if author_id: safe_vk_call(vk.messages.send, peer_id=author_id, message=f"🐛 **Ответ на ваш баг #{bug_id}**\n\n" f"💬 {reply_text}\n\n" f"📌 {bug['text'].split(chr(10))[0].replace('📌 ', '') if chr(10) in bug['text'] else bug['text'][:50]}", random_id=get_random_id()) safe_vk_call(vk.messages.send, peer_id=peer_id, message=f"✅ Ответ на баг #{bug_id} отправлен.", random_id=get_random_id()) return if text.lower().startswith("!bug delete"): parts = text.split(maxsplit=2) if len(parts) < 3: safe_vk_call(vk.messages.send, peer_id=peer_id, message="❌ Использование: !bug delete ", random_id=get_random_id()) return try: bug_id = int(parts[2]) except ValueError: safe_vk_call(vk.messages.send, peer_id=peer_id, message="❌ ID должен быть числом.", random_id=get_random_id()) return if db.delete_bug_report(bug_id): safe_vk_call(vk.messages.send, peer_id=peer_id, message=f"🗑 Баг #{bug_id} удалён.", random_id=get_random_id()) else: safe_vk_call(vk.messages.send, peer_id=peer_id, message=f"❌ Баг #{bug_id} не найден.", random_id=get_random_id()) return if text.lower().startswith("!bug status"): parts = text.split() if len(parts) < 4: safe_vk_call(vk.messages.send, peer_id=peer_id, message="❌ Использование: !bug status <статус>\nСтатусы: новая, в работе, выполнена, отклонена", random_id=get_random_id()) return try: bug_id = int(parts[2]) new_status = " ".join(parts[3:]) except ValueError: safe_vk_call(vk.messages.send, peer_id=peer_id, message="❌ ID должен быть числом.", random_id=get_random_id()) return valid_statuses = ["новая", "в работе", "выполнена", "отклонена"] if new_status not in valid_statuses: safe_vk_call(vk.messages.send, peer_id=peer_id, message=f"❌ Статусы: {', '.join(valid_statuses)}", random_id=get_random_id()) return bug = db.get_bug_report(bug_id) if not bug: safe_vk_call(vk.messages.send, peer_id=peer_id, message=f"❌ Баг #{bug_id} не найден.", random_id=get_random_id()) return old_status = bug.get("status", "?") db.update_bug_status(bug_id, new_status) author_id = bug.get("user_id") if author_id: full_text = bug.get("text", "") title = "" if "📌 " in full_text: title = full_text.split("\n")[0].replace("📌 ", "") status_emoji = {"новая": "🆕", "в работе": "⚙️", "выполнена": "✅", "отклонена": "❌"} emoji = status_emoji.get(new_status, "📋") user_msg = f"🐛 **Баг #{bug_id}** — {emoji} {new_status.upper()}\n\n" if title: user_msg += f"📌 {title}\n\n" if new_status == "выполнена": user_msg += "✅ Баг исправлен! Спасибо за ваш репорт 🎉" elif new_status == "отклонена": user_msg += "❌ К сожалению, этот баг не удалось воспроизвести или он не является ошибкой." elif new_status == "в работе": user_msg += "⚙️ Баг взят в работу. Мы работаем над исправлением." else: user_msg += f"📋 Статус изменён: '{old_status}' → '{new_status}'" safe_vk_call(vk.messages.send, peer_id=author_id, message=user_msg, random_id=get_random_id()) safe_vk_call(vk.messages.send, peer_id=peer_id, message=f"✅ Статус бага #{bug_id}: '{old_status}' → '{new_status}'", random_id=get_random_id()) return if text.lower().startswith("!bug view"): parts = text.split(maxsplit=2) if len(parts) < 3: safe_vk_call(vk.messages.send, peer_id=peer_id, message="❌ Использование: !bug view ", random_id=get_random_id()) return try: bug_id = int(parts[2]) except ValueError: safe_vk_call(vk.messages.send, peer_id=peer_id, message="❌ ID должен быть числом.", random_id=get_random_id()) return bug = db.get_bug_report(bug_id) if not bug: safe_vk_call(vk.messages.send, peer_id=peer_id, message=f"❌ Баг #{bug_id} не найден.", random_id=get_random_id()) return status_emoji = {"новая": "🆕", "в работе": "⚙️", "выполнена": "✅", "отклонена": "❌"}.get(bug.get("status", "?"), "❓") msg = (f"🐛 **Баг #{bug_id}** {status_emoji} {bug.get('status', '?')}\n" f"👤 {bug.get('user_name', '?')} (ID: {bug.get('user_id', '?')})\n" f"📅 {bug.get('timestamp', '?')}\n\n" f"📋 {bug.get('text', '?')}") if bug.get("reply_to_msg"): msg += f"\n\n📎 **Прикреплённое сообщение:**\n{bug['reply_to_msg']}" if bug.get("admin_reply"): msg += f"\n\n💬 **Ответ админа:**\n{bug['admin_reply']}" msg += f"\n\n💬 `!bug reply {bug_id} <текст>`\n✅ `!bug status {bug_id} <статус>`\n🗑 `!bug delete {bug_id}`" safe_vk_call(vk.messages.send, peer_id=peer_id, message=msg, random_id=get_random_id()) return if text.lower() == "!stats top groups": safe_vk_call(vk.messages.send, peer_id=peer_id, message=get_top_groups_by_requests(), random_id=get_random_id()) return if text.lower().startswith("!stats top users commands"): parts = text.split() n = 5 if len(parts) >= 5: # !stats top users commands 10 try: n = int(parts[4]) except ValueError: n = 5 safe_vk_call(vk.messages.send, peer_id=peer_id, message=get_top_users_by_commands(n, vk), random_id=get_random_id()) return if text.lower().startswith("!stats top commands"): parts = text.split() n = 5 if len(parts) >= 4: # !stats top commands 10 try: n = int(parts[3]) except ValueError: n = 5 safe_vk_call(vk.messages.send, peer_id=peer_id, message=get_top_commands(n), random_id=get_random_id()) return if text.lower() == "!stats top users": safe_vk_call(vk.messages.send, peer_id=peer_id, message=get_top_users_by_requests(vk=vk), random_id=get_random_id()) return if text.lower().startswith("!stats top group") and not text.lower().startswith("!stats top group commands"): parts = text.split() if len(parts) == 4: try: group_id = int(parts[3]) stats_text = get_top_users_in_group(group_id, vk=vk) safe_vk_call(vk.messages.send, peer_id=peer_id, message=stats_text, random_id=get_random_id()) except ValueError: safe_vk_call(vk.messages.send, peer_id=peer_id, message="❌ Неверный ID группы.", random_id=get_random_id()) else: safe_vk_call(vk.messages.send, peer_id=peer_id, message="❌ Использование: !stats top group ", random_id=get_random_id()) return if text.lower().startswith("!stats top group commands"): parts = text.split() n = 5 group_id = None # Форматы: !stats top group commands [n] if len(parts) >= 5: try: group_id = int(parts[4]) if len(parts) >= 6: n = int(parts[5]) except ValueError: safe_vk_call(vk.messages.send, peer_id=peer_id, message="❌ Неверный ID группы или число.", random_id=get_random_id()) return else: safe_vk_call(vk.messages.send, peer_id=peer_id, message="❌ Использование: !stats top group commands [количество]", random_id=get_random_id()) return stats_text = get_top_users_by_commands_in_group(group_id, n, vk) safe_vk_call(vk.messages.send, peer_id=peer_id, message=stats_text, random_id=get_random_id()) return # ---------- ВЫБОР ГРУППЫ ДЛЯ СТАТИСТИКИ ---------- if from_id in user_stats_state and text.startswith("Группа "): try: group_id = int(text.replace("Группа ", "")) except ValueError: safe_vk_call(vk.messages.send, peer_id=peer_id, message="❌ Неверный формат группы.", random_id=get_random_id()) del user_stats_state[from_id] return action = user_stats_state[from_id]["action"] if group_id not in user_stats_state[from_id]["groups"]: safe_vk_call(vk.messages.send, peer_id=peer_id, message="❌ Эта группа не доступна.", random_id=get_random_id()) del user_stats_state[from_id] return if action == "group_stats": stats_text = get_group_stats(group_id) elif action == "api_stats": stats_text = get_api_stats_for_group(group_id) else: stats_text = "❌ Неизвестное действие." safe_vk_call(vk.messages.send, peer_id=peer_id, message=stats_text, random_id=get_random_id(), keyboard=VkKeyboard.get_empty_keyboard()) del user_stats_state[from_id] return # ---------- ОТМЕНА В СОСТОЯНИИ СТАТИСТИКИ ---------- if from_id in user_stats_state and text == "❌ Отмена": del user_stats_state[from_id] safe_vk_call(vk.messages.send, peer_id=peer_id, message="❌ Отменено.", random_id=get_random_id(), keyboard=VkKeyboard.get_empty_keyboard()) return # ========= Админская команда !ideas =========== if text.lower().startswith("!ideas"): parts = text.split() page = 1 if len(parts) > 1: try: page = int(parts[1]) except ValueError: safe_vk_call(vk.messages.send, peer_id=peer_id, message="❌ Номер страницы должен быть числом.", random_id=get_random_id()) return ideas = db.get_all_ideas() total = len(ideas) per_page = 10 if total == 0: safe_vk_call(vk.messages.send, peer_id=peer_id, message="📭 Идей пока нет.", random_id=get_random_id()) return pages = (total + per_page - 1) // per_page if page < 1 or page > pages: safe_vk_call(vk.messages.send, peer_id=peer_id, message=f"❌ Страница {page} не существует. Всего страниц: {pages}", random_id=get_random_id()) return start = (page - 1) * per_page end = start + per_page msg = f"📋 Идеи (страница {page}/{pages}):\n" for idea in ideas[start:end]: status = idea.get("status", "?") reply_flag = "📨" if idea.get("reply") else "" msg += f"[{idea.get('id', '?')}] {idea['timestamp']} {idea['user_name']}: {idea['idea']} | Статус: {status} {reply_flag}\n" safe_vk_call(vk.messages.send, peer_id=peer_id, message=msg, random_id=get_random_id()) return # ---------- ЗАПРОС НА УДАЛЕНИЕ ИДЕИ ---------- if text.lower().startswith("!idea delete"): parts = text.split() if len(parts) != 3: safe_vk_call(vk.messages.send, peer_id=peer_id, message="❌ Использование: !idea delete ", random_id=get_random_id()) return try: idea_id = int(parts[2]) except ValueError: safe_vk_call(vk.messages.send, peer_id=peer_id, message="❌ ID должен быть числом.", random_id=get_random_id()) return ideas = db.get_all_ideas() idea_exists = any(idea.get("id") == idea_id for idea in ideas) if not idea_exists: safe_vk_call(vk.messages.send, peer_id=peer_id, message=f"❌ Идея с ID {idea_id} не найдена.", random_id=get_random_id()) return user_idea_delete_state[from_id] = {"idea_id": idea_id} safe_vk_call(vk.messages.send, peer_id=peer_id, message=f"❓ Вы действительно хотите удалить идею #{idea_id}?", random_id=get_random_id(), keyboard=get_confirm_keyboard()) return # ← добавь эту строку # ---------- ПОДТВЕРЖДЕНИЕ УДАЛЕНИЯ ИДЕИ ---------- if from_id in user_idea_delete_state: if text.lower() in ["отмена", "❌ отмена", "отменить"]: del user_idea_delete_state[from_id] page = user_main_page.get(from_id, 1) keyboard = get_keyboard_for_page(page, from_id) safe_vk_call(vk.messages.send, peer_id=peer_id, message="❌ Удаление отменено.", random_id=get_random_id(), keyboard=keyboard) return idea_id = user_idea_delete_state[from_id]["idea_id"] answer = text.strip().lower() if any(word in answer for word in ["да", "yes", "ok", "ага"]) or "✅" in answer: db.delete_idea(idea_id) page = user_main_page.get(from_id, 1) keyboard = get_keyboard_for_page(page, from_id) safe_vk_call(vk.messages.send, peer_id=peer_id, message=f"✅ Идея #{idea_id} удалена.", random_id=get_random_id(), keyboard=keyboard) else: page = user_main_page.get(from_id, 1) keyboard = get_keyboard_for_page(page, from_id) safe_vk_call(vk.messages.send, peer_id=peer_id, message="❌ Удаление отменено.", random_id=get_random_id(), keyboard=keyboard) del user_idea_delete_state[from_id] return # ---------- ИЗМЕНЕНИЕ СТАТУСА ИДЕИ (ЗАПРОС НОВОГО СТАТУСА) ---------- if text.lower().startswith("!idea status"): parts = text.split() if len(parts) != 3: safe_vk_call(vk.messages.send, peer_id=peer_id, message="❌ Использование: !idea status ", random_id=get_random_id()) return try: idea_id = int(parts[2]) except ValueError: safe_vk_call(vk.messages.send, peer_id=peer_id, message="❌ ID должен быть числом.", random_id=get_random_id()) return ideas = db.get_all_ideas() idea_exists = any(idea.get("id") == idea_id for idea in ideas) if not idea_exists: safe_vk_call(vk.messages.send, peer_id=peer_id, message=f"❌ Идея с ID {idea_id} не найдена.", random_id=get_random_id()) return user_idea_status_state[from_id] = {"idea_id": idea_id} safe_vk_call(vk.messages.send, peer_id=peer_id, message=f"Введите новый статус для идеи #{idea_id} (например: новая, в работе, выполнена, отклонена):", random_id=get_random_id(), keyboard=get_cancel_keyboard()) return # ========== АРХИВАЦИЯ ЛОГОВ ========== if text.lower() == "!wipe logs": moved = archive_logs(vk) safe_vk_call(vk.messages.send, peer_id=peer_id, message=f"✅ Архивация выполнена. Перемещено {moved} файлов в system_log/{datetime.now().strftime('%Y-%m-%d')}", random_id=get_random_id()) return # ========== ОСТАЛЬНЫЕ АДМИН-КОМАНДЫ ========== if text.lower() in ["сгенерировать", "!genkey"]: admin_state[from_id] = "genkey" safe_vk_call(vk.messages.send, peer_id=peer_id, message="Выберите группу:", random_id=get_random_id(), keyboard=get_group_selection_keyboard()) return if text.lower() == "!remkey": admin_state[from_id] = "remkey" safe_vk_call(vk.messages.send, peer_id=peer_id, message="Выберите группу для удаления ключей:", random_id=get_random_id(), keyboard=get_group_selection_keyboard()) return if text.startswith("Группа "): try: group_id = int(text.replace("Группа ", "")) except ValueError: return mode = admin_state.get(from_id) if mode == "token": if group_id not in accounts: safe_vk_call(vk.messages.send, peer_id=peer_id, message=f"❌ Для группы {group_id} не назначен владелец.", random_id=get_random_id()) admin_state.pop(from_id, None) return # Проверяем, есть ли owner_id в записи if "owner_id" not in accounts[group_id]: safe_vk_call(vk.messages.send, peer_id=peer_id, message=f"❌ Для группы {group_id} отсутствует информация о владельце. Пересоздайте запись через !addacc.", random_id=get_random_id()) admin_state.pop(from_id, None) return owner_id = accounts[group_id]["owner_id"] owner_name = accounts[group_id]["owner_name"] groups_of_owner = [str(g) for g, v in accounts.items() if v.get("owner_id") == owner_id] groups_str = ", ".join(groups_of_owner) msg = (f"{owner_name}, ваш аккаунт используется для доступа к расписанию групп: {groups_str}.\n\n" f"Необходимо обновить куки. Разрешаете ли вы администратору войти в ваш аккаунт и выполнить обновление?\n\n" f"✅ Разрешить — администратор обновит куки, расписание всех групп продолжит работу.\n" f"❌ Запретить — доступ к расписанию для всех ваших групп будет отключён.\n\n" f"Ответьте 'Да' или 'Нет' (можно с эмодзи).") pending_consent[owner_id] = { "group_id": group_id, "admin_id": from_id, "timestamp": time.time() } save_pending_consent(pending_consent) safe_vk_call(vk.messages.send, peer_id=owner_id, message=msg, random_id=get_random_id()) safe_vk_call(vk.messages.send, peer_id=peer_id, message=f"✅ Запрос согласия отправлен пользователю {owner_name} (ID {owner_id}).", random_id=get_random_id()) admin_state.pop(from_id, None) return elif mode == "users_group": # Режим просмотра пользователей группы show_users_of_group(vk, peer_id, group_id) admin_state.pop(from_id, None) return elif mode == "reloadgroup": # Выполняем перезагрузку выбранной группы try: load_dotenv(override=True) for i in range(1, 11): gid_str = os.getenv(f"GROUP{i}_ID") if gid_str and int(gid_str) == group_id: new_token = os.getenv(f"GROUP{i}_TOKEN") new_user_id = int(os.getenv(f"GROUP{i}_USER_ID")) new_peer_id = int(os.getenv(f"GROUP{i}_PEER_ID")) new_login = os.getenv(f"GROUP{i}_LOGIN") new_password = os.getenv(f"GROUP{i}_PASSWORD") new_platform = os.getenv(f"GROUP{i}_PLATFORM", DEFAULT_PLATFORM) GROUPS[group_id] = { "token": new_token, "user_id": new_user_id, "peer_id": new_peer_id, "login": new_login, "password": new_password, "platform": new_platform } with accounts_lock: if group_id in accounts: accounts[group_id].pop("session", None) accounts[group_id].pop("fp", None) accounts[group_id].pop("ddg1", None) accounts[group_id].pop("blocked", None) db.save_group_account(group_id, accounts[group_id]) cleared = clear_cache_for_group(group_id) logger.info(f"Кэш группы {group_id} очищен, удалено {cleared} файлов") safe_vk_call(vk.messages.send, peer_id=peer_id, message=f"✅ Группа {group_id} перезагружена.", random_id=get_random_id(), keyboard=VkKeyboard.get_empty_keyboard()) break else: safe_vk_call(vk.messages.send, peer_id=peer_id, message="❌ Группа не найдена в .env.", random_id=get_random_id()) except Exception as e: safe_vk_call(vk.messages.send, peer_id=peer_id, message=f"❌ Ошибка: {e}", random_id=get_random_id()) admin_state.pop(from_id, None) return elif mode == "remkey": removed = db.delete_keys_by_group(group_id) if removed: safe_vk_call(vk.messages.send, peer_id=peer_id, message=f"✅ Удалено {removed} ключей для группы {group_id}.", random_id=get_random_id()) else: safe_vk_call(vk.messages.send, peer_id=peer_id, message=f"❌ Ключей для группы {group_id} не найдено.", random_id=get_random_id()) # Скрываем клавиатуру и очищаем состояние safe_vk_call(vk.messages.send, peer_id=peer_id, message="✅ Клавиатура скрыта.", random_id=get_random_id(), keyboard=VkKeyboard.get_empty_keyboard()) admin_state.pop(from_id, None) return else: # Обычный режим генерации ключа (без состояния) if group_id in GROUPS: key = generate_key(group_id) safe_vk_call(vk.messages.send, peer_id=peer_id, message=f"🔑 Ключ для группы {group_id}:\n`{key}`", random_id=get_random_id()) safe_vk_call(vk.messages.send, peer_id=peer_id, message="✅ Клавиатура скрыта.", random_id=get_random_id(), keyboard=VkKeyboard.get_empty_keyboard()) else: safe_vk_call(vk.messages.send, peer_id=peer_id, message="❌ Неизвестная группа.", random_id=get_random_id()) return # ---------- ОБРАБОТКА ВЫБОРА ДАТЫ ДЛЯ АРХИВА ---------- if from_id in admin_state and admin_state[from_id] == "archive": if text.lower() in ["отмена", "❌ отмена"]: del admin_state[from_id] safe_vk_call(vk.messages.send, peer_id=peer_id, message="❌ Отменено.", random_id=get_random_id(), keyboard=VkKeyboard.get_empty_keyboard()) return if re.match(r'^\d{4}-\d{2}-\d{2}$', text): # Сохраняем выбранную дату и переходим к выбору типа admin_state[from_id] = {"state": "archive_type", "date": text} kb = VkKeyboard(one_time=True) kb.add_button("logs", color=VkKeyboardColor.PRIMARY) kb.add_button("api_logs", color=VkKeyboardColor.PRIMARY) kb.add_line() kb.add_button("❌ Отмена", color=VkKeyboardColor.NEGATIVE) safe_vk_call(vk.messages.send, peer_id=peer_id, message="Выберите тип логов:", random_id=get_random_id(), keyboard=kb.get_keyboard()) return else: safe_vk_call(vk.messages.send, peer_id=peer_id, message="❌ Неверный формат даты. Введите ГГГГ-ММ-ДД", random_id=get_random_id()) return # ---------- ОБРАБОТКА ВЫБОРА ТИПА ЛОГОВ ---------- if from_id in admin_state and isinstance(admin_state[from_id], dict) and admin_state[from_id].get("state") == "archive_type": if text.lower() in ["отмена", "❌ отмена"]: del admin_state[from_id] safe_vk_call(vk.messages.send, peer_id=peer_id, message="❌ Отменено.", random_id=get_random_id(), keyboard=VkKeyboard.get_empty_keyboard()) return if text in ["logs", "api_logs"]: date = admin_state[from_id]["date"] stats = get_archive_stats(date, text) safe_vk_call(vk.messages.send, peer_id=peer_id, message=stats, random_id=get_random_id(), keyboard=VkKeyboard.get_empty_keyboard()) del admin_state[from_id] elif text == "❌ Отмена": del admin_state[from_id] safe_vk_call(vk.messages.send, peer_id=peer_id, message="❌ Отменено.", random_id=get_random_id(), keyboard=VkKeyboard.get_empty_keyboard()) else: safe_vk_call(vk.messages.send, peer_id=peer_id, message="❌ Введите 'logs' или 'api_logs'", random_id=get_random_id()) return # ---------- ГЛОБАЛЬНАЯ ОТМЕНА ---------- # ---------- ГЛОБАЛЬНАЯ ОТМЕНА ---------- if text.lower() in ["отмена", "❌ отмена"]: # Очищаем состояние админа, если оно есть if from_id in admin_state: admin_state.pop(from_id, None) safe_vk_call(vk.messages.send, peer_id=peer_id, message="❌ Действие отменено.", random_id=get_random_id(), keyboard=VkKeyboard.get_empty_keyboard()) return if text.lower() == "!users": users = get_all_users() if not users: safe_vk_call(vk.messages.send, peer_id=peer_id, message="📭 Нет пользователей.", random_id=get_random_id()) return lines = ["📋 Список всех пользователей:\n"] for uid, gid in users: name = get_user_name(vk, uid) lines.append(f"• {name} (ID: {uid}) — группа {gid}") instruction = "\n\n🔹 Чтобы отвязать пользователя, отправьте: отвязать ID\n🔹 Чтобы посмотреть пользователей группы: !users group " safe_vk_call(vk.messages.send, peer_id=peer_id, message="\n".join(lines) + instruction, random_id=get_random_id()) return if text.lower().startswith("!users group"): parts = text.split() if len(parts) == 3: try: group_id = int(parts[2]) show_users_of_group(vk, peer_id, group_id) except ValueError: safe_vk_call(vk.messages.send, peer_id=peer_id, message="❌ ID группы должен быть числом.", random_id=get_random_id()) else: all_users = get_all_users() groups_with_users = set(gid for _, gid in all_users) if not groups_with_users: safe_vk_call(vk.messages.send, peer_id=peer_id, message="📭 Нет зарегистрированных пользователей ни в одной группе.", random_id=get_random_id()) return kb = VkKeyboard(one_time=True) for gid in sorted(groups_with_users): kb.add_button(f"Группа {gid}", color=VkKeyboardColor.PRIMARY) kb.add_line() kb.add_button("❌ Отмена", color=VkKeyboardColor.NEGATIVE) admin_state[from_id] = "users_group" safe_vk_call(vk.messages.send, peer_id=peer_id, message="Выберите группу:", random_id=get_random_id(), keyboard=kb.get_keyboard()) return if text.lower().startswith("отвязать "): parts = text.split() if len(parts) == 2: try: target_id = int(parts[1]) if delete_user(target_id): safe_vk_call(vk.messages.send, peer_id=peer_id, message=f"✅ Пользователь {target_id} отвязан.", random_id=get_random_id()) safe_vk_call(vk.messages.send, peer_id=target_id, message="❌ Ваш доступ к боту был отозван администратором.", random_id=get_random_id()) logger.info(f"Админ {from_id} отвязал пользователя {target_id}") else: safe_vk_call(vk.messages.send, peer_id=peer_id, message=f"❌ Пользователь {target_id} не найден.", random_id=get_random_id()) except ValueError: safe_vk_call(vk.messages.send, peer_id=peer_id, message="❌ ID должен быть числом.", random_id=get_random_id()) else: safe_vk_call(vk.messages.send, peer_id=peer_id, message="❌ Использование: отвязать ID", random_id=get_random_id()) return # ===== Управление IP===== if text.lower().startswith("!ip "): parts = text.split() if len(parts) >= 2: action = parts[1].lower() if action == "list": try: resp = requests.get("http://127.0.0.1:5000/api/ip/list") if resp.status_code == 200: ips = resp.json() if not ips: safe_vk_call(vk.messages.send, peer_id=peer_id, message="📭 Белый список пуст", random_id=get_random_id()) else: lines = ["🛡️ Белый список IP:\n"] for ip, info in ips.items(): exp = info.get("expires_at") if exp: left = int((exp - time.time()) / 60) if left > 0: if left >= 1440: exp_str = f"ещё {left // 1440} дн. {left % 1440 // 60} ч." elif left >= 60: exp_str = f"ещё {left // 60} ч. {left % 60} мин." else: exp_str = f"ещё {left} мин." else: exp_str = "истекает" else: exp_str = "навсегда" lines.append(f"• {ip} — {exp_str}") safe_vk_call(vk.messages.send, peer_id=peer_id, message="\n".join(lines), random_id=get_random_id()) else: safe_vk_call(vk.messages.send, peer_id=peer_id, message="❌ Ошибка получения списка", random_id=get_random_id()) except Exception as e: safe_vk_call(vk.messages.send, peer_id=peer_id, message=f"❌ Ошибка: {e}", random_id=get_random_id()) return if action == "allow" and len(parts) >= 4: ip = parts[2] duration = parts[3].lower() try: resp = requests.post("http://127.0.0.1:5000/api/ip/manage", json={ "action": "allow", "ip": ip, "duration": duration }) if resp.status_code == 200: data = resp.json() safe_vk_call(vk.messages.send, peer_id=peer_id, message=data.get("msg", "✅ Добавлено"), random_id=get_random_id()) else: safe_vk_call(vk.messages.send, peer_id=peer_id, message="❌ Ошибка при добавлении IP", random_id=get_random_id()) except Exception as e: safe_vk_call(vk.messages.send, peer_id=peer_id, message=f"❌ Ошибка: {e}", random_id=get_random_id()) return elif action == "revoke" and len(parts) == 3: ip = parts[2] try: resp = requests.post("http://127.0.0.1:5000/api/ip/manage", json={ "action": "revoke", "ip": ip }) if resp.status_code == 200: data = resp.json() safe_vk_call(vk.messages.send, peer_id=peer_id, message=data.get("msg", "✅ Удалено"), random_id=get_random_id()) else: safe_vk_call(vk.messages.send, peer_id=peer_id, message="❌ Ошибка при удалении IP", random_id=get_random_id()) except Exception as e: safe_vk_call(vk.messages.send, peer_id=peer_id, message=f"❌ Ошибка: {e}", random_id=get_random_id()) return safe_vk_call(vk.messages.send, peer_id=peer_id, message="❌ Использование:\n" "!ip allow <время> — 30m, 1h, 2d, perm\n" "!ip revoke \n" "!ip list — показать все IP", random_id=get_random_id()) return # ===== Управление IP для пропуска логов ===== if text.lower().startswith("!logip "): parts = text.split() if len(parts) >= 2: action = parts[1].lower() if action == "list": try: resp = requests.get("http://127.0.0.1:5000/api/logip/list") if resp.status_code == 200: ips = resp.json() if not ips: safe_vk_call(vk.messages.send, peer_id=peer_id, message="📭 Список IP для пропуска логов пуст", random_id=get_random_id()) else: lines = ["📝 IP, исключённые из логов:\n"] for ip in ips: lines.append(f"• {ip}") safe_vk_call(vk.messages.send, peer_id=peer_id, message="\n".join(lines), random_id=get_random_id()) else: safe_vk_call(vk.messages.send, peer_id=peer_id, message="❌ Ошибка получения списка", random_id=get_random_id()) except Exception as e: safe_vk_call(vk.messages.send, peer_id=peer_id, message=f"❌ Ошибка: {e}", random_id=get_random_id()) return if action == "allow" and len(parts) == 3: ip = parts[2] try: resp = requests.post("http://127.0.0.1:5000/api/logip/manage", json={ "action": "allow", "ip": ip }) if resp.status_code == 200: data = resp.json() safe_vk_call(vk.messages.send, peer_id=peer_id, message=data.get("msg", "✅ IP добавлен в исключения логов"), random_id=get_random_id()) else: safe_vk_call(vk.messages.send, peer_id=peer_id, message="❌ Ошибка при добавлении IP", random_id=get_random_id()) except Exception as e: safe_vk_call(vk.messages.send, peer_id=peer_id, message=f"❌ Ошибка: {e}", random_id=get_random_id()) return elif action == "revoke" and len(parts) == 3: ip = parts[2] try: resp = requests.post("http://127.0.0.1:5000/api/logip/manage", json={ "action": "revoke", "ip": ip }) if resp.status_code == 200: data = resp.json() safe_vk_call(vk.messages.send, peer_id=peer_id, message=data.get("msg", "✅ IP удалён из исключений логов"), random_id=get_random_id()) else: safe_vk_call(vk.messages.send, peer_id=peer_id, message="❌ Ошибка при удалении IP", random_id=get_random_id()) except Exception as e: safe_vk_call(vk.messages.send, peer_id=peer_id, message=f"❌ Ошибка: {e}", random_id=get_random_id()) return safe_vk_call(vk.messages.send, peer_id=peer_id, message="❌ Использование:\n" "!logip allow — добавить IP в исключения логов\n" "!logip revoke — удалить IP из исключений\n" "!logip list — показать все исключённые IP", random_id=get_random_id()) return # ===== Управление владельцами ===== if text.lower().startswith("!addacc"): parts = text.split() if len(parts) == 3: try: gid = int(parts[1]) owner = int(parts[2]) if gid not in GROUPS: safe_vk_call(vk.messages.send, peer_id=peer_id, message=f"❌ Группа {gid} не найдена в .env.", random_id=get_random_id()) return user_info = safe_vk_call(vk.users.get, user_ids=owner) if not user_info: safe_vk_call(vk.messages.send, peer_id=peer_id, message="❌ Пользователь с таким ID не найден в VK.", random_id=get_random_id()) return owner_name = f"{user_info[0]['first_name']} {user_info[0]['last_name']}" with accounts_lock: if gid not in accounts: accounts[gid] = {} accounts[gid]["owner_id"] = owner accounts[gid]["owner_name"] = owner_name db.save_group_account(gid, accounts[gid]) safe_vk_call(vk.messages.send, peer_id=peer_id, message=f"✅ Аккаунт группы {gid} успешно привязан к владельцу {owner_name} (ID {owner}).", random_id=get_random_id()) except ValueError: safe_vk_call(vk.messages.send, peer_id=peer_id, message="❌ Ошибка: ID группы и владельца должны быть числами.", random_id=get_random_id()) else: safe_vk_call(vk.messages.send, peer_id=peer_id, message="❌ Использование: !addacc ", random_id=get_random_id()) return if text.lower().startswith("!rmacc"): parts = text.split() if len(parts) == 2: try: gid = int(parts[1]) if gid in accounts: del accounts[gid] save_accounts(accounts) safe_vk_call(vk.messages.send, peer_id=peer_id, message=f"✅ Запись о владельце для группы {gid} удалена.", random_id=get_random_id()) else: safe_vk_call(vk.messages.send, peer_id=peer_id, message=f"❌ Группа {gid} не найдена в accounts.", random_id=get_random_id()) except ValueError: safe_vk_call(vk.messages.send, peer_id=peer_id, message="❌ group_id должен быть числом.", random_id=get_random_id()) else: safe_vk_call(vk.messages.send, peer_id=peer_id, message="❌ Используйте: !rmacc ", random_id=get_random_id()) return if text.lower() == "!listacc": if not accounts: safe_vk_call(vk.messages.send, peer_id=peer_id, message="📭 Нет записей о владельцах.", random_id=get_random_id()) else: lines = ["📋 Список групп с владельцами:\n"] for gid, info in accounts.items(): owner = info.get("owner_name") or f"id{info['owner_id']}" creds = "✅" if info.get("has_credentials") else "❌" consent = "✅" if info.get("consent") else "⏳" lines.append(f"• Группа {gid}: {owner}, данные {creds}, согласие {consent}") safe_vk_call(vk.messages.send, peer_id=peer_id, message="\n".join(lines), random_id=get_random_id()) return if text.lower().startswith("!reloadgroup"): parts = text.split() # Вариант без аргументов – клавиатура if len(parts) == 1: group_ids = sorted(GROUPS.keys()) if not group_ids: safe_vk_call(vk.messages.send, peer_id=peer_id, message="❌ Нет доступных групп.", random_id=get_random_id()) return admin_state[from_id] = "reloadgroup" safe_vk_call(vk.messages.send, peer_id=peer_id, message="Выберите группу для перезагрузки:", random_id=get_random_id(), keyboard=get_group_selection_keyboard()) return # Вариант "all" if len(parts) == 2 and parts[1].lower() == "all": success_count = 0 error_count = 0 for group_id in list(GROUPS.keys()): try: load_dotenv(override=True) for i in range(1, 11): gid_str = os.getenv(f"GROUP{i}_ID") if gid_str and int(gid_str) == group_id: new_token = os.getenv(f"GROUP{i}_TOKEN") new_user_id = int(os.getenv(f"GROUP{i}_USER_ID")) new_peer_id = int(os.getenv(f"GROUP{i}_PEER_ID")) new_login = os.getenv(f"GROUP{i}_LOGIN") new_password = os.getenv(f"GROUP{i}_PASSWORD") new_platform = os.getenv(f"GROUP{i}_PLATFORM", DEFAULT_PLATFORM) GROUPS[group_id] = { "token": new_token, "user_id": new_user_id, "peer_id": new_peer_id, "login": new_login, "password": new_password, "platform": new_platform } with accounts_lock: if group_id in accounts: accounts[group_id].pop("session", None) accounts[group_id].pop("fp", None) accounts[group_id].pop("ddg1", None) accounts[group_id].pop("blocked", None) db.save_group_account(group_id, accounts[group_id]) success_count += 1 break else: error_count += 1 except Exception as e: logger.error(f"Ошибка при перезагрузке группы {group_id}: {e}") error_count += 1 for group_id in list(GROUPS.keys()): clear_cache_for_group(group_id) safe_vk_call(vk.messages.send, peer_id=peer_id, message=f"✅ Перезагружено групп: {success_count}, ошибок: {error_count}.", random_id=get_random_id()) return # Вариант с указанием ID if len(parts) == 2: try: group_id = int(parts[1]) load_dotenv(override=True) found = False for i in range(1, 11): gid_str = os.getenv(f"GROUP{i}_ID") if gid_str and int(gid_str) == group_id: new_token = os.getenv(f"GROUP{i}_TOKEN") new_user_id = int(os.getenv(f"GROUP{i}_USER_ID")) new_peer_id = int(os.getenv(f"GROUP{i}_PEER_ID")) new_login = os.getenv(f"GROUP{i}_LOGIN") new_password = os.getenv(f"GROUP{i}_PASSWORD") new_platform = os.getenv(f"GROUP{i}_PLATFORM", DEFAULT_PLATFORM) GROUPS[group_id] = { "token": new_token, "user_id": new_user_id, "peer_id": new_peer_id, "login": new_login, "password": new_password, "platform": new_platform } with accounts_lock: if group_id in accounts: accounts[group_id].pop("session", None) accounts[group_id].pop("fp", None) accounts[group_id].pop("ddg1", None) accounts[group_id].pop("blocked", None) db.save_group_account(group_id, accounts[group_id]) # ========== ДОБАВИТЬ ЭТИ СТРОКИ ========== cleared = clear_cache_for_group(group_id) logger.info(f"Кэш группы {group_id} очищен, удалено {cleared} файлов") # ========================================= safe_vk_call(vk.messages.send, peer_id=peer_id, message=f"✅ Группа {group_id} перезагружена.", random_id=get_random_id(), keyboard=VkKeyboard.get_empty_keyboard()) found = True break if not found: safe_vk_call(vk.messages.send, peer_id=peer_id, message="❌ Группа не найдена в .env.", random_id=get_random_id()) except Exception as e: safe_vk_call(vk.messages.send, peer_id=peer_id, message=f"❌ Ошибка: {e}", random_id=get_random_id()) return # ===== Команды лекций для админа ===== if text.lower().startswith("!setcreds"): parts = text.split() if len(parts) == 2: try: gid = int(parts[1]) if gid not in accounts: safe_vk_call(vk.messages.send, peer_id=peer_id, message=f"❌ Группа {gid} не найдена в accounts.", random_id=get_random_id()) return with accounts_lock: accounts[gid]["has_credentials"] = True db.save_group_account(gid, accounts[gid]) safe_vk_call(vk.messages.send, peer_id=peer_id, message=f"✅ Для группы {gid} отмечено наличие логина/пароля.", random_id=get_random_id()) except ValueError: safe_vk_call(vk.messages.send, peer_id=peer_id, message="❌ group_id должен быть числом.", random_id=get_random_id()) else: safe_vk_call(vk.messages.send, peer_id=peer_id, message="❌ Используйте: !setcreds ", random_id=get_random_id()) return if text.lower().startswith("!unsetcreds"): parts = text.split() if len(parts) == 2: try: gid = int(parts[1]) if gid not in accounts: safe_vk_call(vk.messages.send, peer_id=peer_id, message=f"❌ Группа {gid} не найдена в accounts.", random_id=get_random_id()) return with accounts_lock: accounts[gid]["has_credentials"] = False db.save_group_account(gid, accounts[gid]) safe_vk_call(vk.messages.send, peer_id=peer_id, message=f"✅ Для группы {gid} пометка о наличии логина/пароля снята.", random_id=get_random_id()) except ValueError: safe_vk_call(vk.messages.send, peer_id=peer_id, message="❌ group_id должен быть числом.", random_id=get_random_id()) else: safe_vk_call(vk.messages.send, peer_id=peer_id, message="❌ Используйте: !unsetcreds ", random_id=get_random_id()) return if text.lower() == "!token": admin_state[from_id] = "token" safe_vk_call(vk.messages.send, peer_id=peer_id, message="Выберите группу для запроса согласия:", random_id=get_random_id(), keyboard=get_group_selection_keyboard(only_with_owner=True)) return # Команда !settoken — обновить ВСЕ токены + archive + reloadgroup (в фоне) if text.lower() == "!settoken": def background_settoken(): vk2 = vk safe_vk_call(vk2.messages.send, peer_id=peer_id, message="🔄 Обновляю токены...", random_id=get_random_id()) results = [] for group_id in list(GROUPS.keys()): new_token = refresh_token(group_id, force=True) if new_token: GROUPS[group_id]["token"] = new_token with accounts_lock: if group_id not in accounts: accounts[group_id] = {} accounts[group_id]["token"] = new_token exp_ts = get_token_exp(new_token) if exp_ts: accounts[group_id]["token_expires_at"] = exp_ts db.save_group_account(group_id, accounts[group_id]) update_env_token(group_id, new_token) results.append(f"✅ {group_id}") time.sleep(random.randint(30, 90)) else: results.append(f"❌ {group_id}") msg = "\n".join(results) safe_vk_call(vk2.messages.send, peer_id=peer_id, message=msg, random_id=get_random_id()) # archive logs moved = archive_logs() logger.info(f"Архивировано: {moved}") # reloadgroup all with accounts_lock: for g in list(GROUPS.keys()): clear_cache_for_group(g) if g in accounts: accounts[g].pop("blocked", None) accounts[g].pop("session", None) db.save_group_account(g, accounts[g]) safe_vk_call(vk2.messages.send, peer_id=peer_id, message="✅ Готово!", random_id=get_random_id()) threading.Thread(target=background_settoken, daemon=True).start() return # Команда !settoken <группа> <токен> — вручную прописать токен if text.lower().startswith("!settoken"): parts = text.split() if len(parts) >= 3: # Первый аргумент — группа, остальное — токен try: gid_or_num = int(parts[1]) token = " ".join(parts[2:]) # Токен может содержать пробелы # Определяем group_id group_id = None for gid in GROUPS.keys(): if gid == gid_or_num or (1 <= gid_or_num <= 10 and list(GROUPS.keys()).index(gid) + 1 == gid_or_num): group_id = gid break if not group_id: safe_vk_call(vk.messages.send, peer_id=peer_id, message=f"❌ Группа не найдена.", random_id=get_random_id()) return # Сохраняем токен GROUPS[group_id]["token"] = token with accounts_lock: if group_id not in accounts: accounts[group_id] = {} accounts[group_id]["token"] = token exp_ts = get_token_exp(token) if exp_ts: accounts[group_id]["token_expires_at"] = exp_ts if group_id in accounts: accounts[group_id].pop("blocked", None) accounts[group_id].pop("updating", None) db.save_group_account(group_id, accounts[group_id]) update_env_token(group_id, token) safe_vk_call(vk.messages.send, peer_id=peer_id, message=f"✅ Токен для группы {group_id} обновлён вручную.", random_id=get_random_id()) except ValueError: safe_vk_call(vk.messages.send, peer_id=peer_id, message="❌ Используйте: !settoken <группа> <токен>", random_id=get_random_id()) else: safe_vk_call(vk.messages.send, peer_id=peer_id, message="❌ Используйте: !settoken <группа> <токен>\nПример: !settoken 2 eyJhbGci...", random_id=get_random_id()) return if text.lower().startswith("!updatetoken"): parts = text.split() if len(parts) == 2: try: # Может быть group_num (1-4) или group_id gid_or_num = int(parts[1]) # Определяем group_id group_id = None for gid in GROUPS.keys(): # Проверяем как group_id или как порядковый номер if gid == gid_or_num or (1 <= gid_or_num <= 10 and list(GROUPS.keys()).index(gid) + 1 == gid_or_num): group_id = gid break if not group_id: safe_vk_call(vk.messages.send, peer_id=peer_id, message=f"❌ Группа не найдена.", random_id=get_random_id()) return # Пробуем обновить токен new_token = refresh_token(group_id, force=True) if new_token: GROUPS[group_id]["token"] = new_token with accounts_lock: if group_id not in accounts: accounts[group_id] = {} accounts[group_id]["token"] = new_token exp_ts = get_token_exp(new_token) if exp_ts: accounts[group_id]["token_expires_at"] = exp_ts if group_id in accounts: accounts[group_id].pop("blocked", None) accounts[group_id].pop("updating", None) db.save_group_account(group_id, accounts[group_id]) update_env_token(group_id, new_token) safe_vk_call(vk.messages.send, peer_id=peer_id, message=f"✅ Токен для группы {group_id} обновлён и сохранён.", random_id=get_random_id()) else: safe_vk_call(vk.messages.send, peer_id=peer_id, message=f"❌ Не удалось обновить токен. Проверьте логин/пароль в .env", random_id=get_random_id()) except ValueError: safe_vk_call(vk.messages.send, peer_id=peer_id, message="❌ Используйте: !updatetoken <номер_группы>", random_id=get_random_id()) else: safe_vk_call(vk.messages.send, peer_id=peer_id, message="❌ Используйте: !updatetoken <номер_группы>\nПример: !updatetoken 2", random_id=get_random_id()) return # ---------- ОБЫЧНЫЕ ПОЛЬЗОВАТЕЛИ ---------- user_group = get_user_group(from_id) if user_group is None: return user_name = get_user_name(vk, from_id) # ---------- ГЛОБАЛЬНАЯ ОТМЕНА ДЛЯ ОБЫЧНЫХ ПОЛЬЗОВАТЕЛЕЙ ---------- if text.lower() in ["отмена", "❌ отмена", "отменить"]: # Очищаем все состояния, которые могут быть у пользователя if from_id in user_notes_state: del user_notes_state[from_id] if from_id in user_idea_state: del user_idea_state[from_id] if from_id in user_idea_status_state: del user_idea_status_state[from_id] if from_id in user_idea_delete_state: del user_idea_delete_state[from_id] if from_id in user_review_state: del user_review_state[from_id] if from_id in user_bug_state: del user_bug_state[from_id] if from_id in user_revoke_state: del user_revoke_state[from_id] if from_id in user_stats_state: del user_stats_state[from_id] if from_id in admin_state: del admin_state[from_id] if from_id in user_reviews_page: del user_reviews_page[from_id] page = user_main_page.get(from_id, 1) safe_vk_call(vk.messages.send, peer_id=peer_id, message="❌ Действие отменено.", random_id=get_random_id(), keyboard=get_keyboard_for_page(page, from_id)) return # ---------- КОМАНДА !user (для всех) ---------- if text.lower().startswith("!user"): # Разрешаем только в личных сообщениях if peer_id != from_id: return parts = text.split() if len(parts) == 1: # Без параметров - показываем статистику отправителя # Проверяем, зарегистрирован ли пользователь (кроме админа) if from_id != ADMIN_ID and get_user_group(from_id) is None: # Незарегистрированный не-админ просто игнорируется return target_id = from_id elif len(parts) == 2: # С параметром - только для админа if from_id != ADMIN_ID: safe_vk_call(vk.messages.send, peer_id=peer_id, message="❌ У вас нет прав для просмотра статистики других пользователей.", random_id=get_random_id()) return try: target_id = int(parts[1]) except ValueError: safe_vk_call(vk.messages.send, peer_id=peer_id, message="❌ ID должен быть числом.", random_id=get_random_id()) return else: safe_vk_call(vk.messages.send, peer_id=peer_id, message="❌ Использование: !user [ID]", random_id=get_random_id()) return stats_text = get_user_stats(target_id, vk) safe_vk_call(vk.messages.send, peer_id=peer_id, message=stats_text, random_id=get_random_id()) return # ---------- ЛИЧНЫЙ КАБИНЕТ (ЗАМЕТКИ) ---------- # Обработка состояний добавления заметки if from_id in user_notes_state: state = user_notes_state[from_id] step = state["step"] # Проверка отмены if text.lower() in ["отмена", "❌ отмена", "отменить"]: del user_notes_state[from_id] page = user_main_page.get(from_id, 1) safe_vk_call(vk.messages.send, peer_id=peer_id, message="❌ Действие отменено.", random_id=get_random_id(), keyboard=get_keyboard_for_page(page, from_id)) return # Шаг 1: ввод даты if step == "add_date": try: datetime.strptime(text, "%Y-%m-%d") state["temp_data"]["date"] = text state["step"] = "add_subject" safe_vk_call(vk.messages.send, peer_id=peer_id, message="Введите предмет:", random_id=get_random_id(), keyboard=get_cancel_keyboard()) except: safe_vk_call(vk.messages.send, peer_id=peer_id, message="❌ Неверная дата. Введите дату в формате ГГГГ-ММ-ДД", random_id=get_random_id(), keyboard=get_cancel_keyboard()) return # Шаг 2: ввод предмета if step == "add_subject": state["temp_data"]["subject"] = text state["step"] = "add_text" safe_vk_call(vk.messages.send, peer_id=peer_id, message="Введите текст заметки:", random_id=get_random_id(), keyboard=get_cancel_keyboard()) return # Шаг 3: ввод текста и сохранение if step == "add_text": date_str = state["temp_data"]["date"] subject = state["temp_data"]["subject"] note_text = text note_id = db.add_note(from_id, date_str, subject, note_text) del user_notes_state[from_id] safe_vk_call(vk.messages.send, peer_id=peer_id, message=f"✅ Заметка #{note_id} добавлена на {date_str} по предмету {subject}", random_id=get_random_id(), keyboard=get_notes_menu_keyboard()) return # Обработка команд заметок (кнопки меню) if text == "📝 Заметки" or text.lower() == "заметки": safe_vk_call(vk.messages.send, peer_id=peer_id, message="📋 Меню заметок:", random_id=get_random_id(), keyboard=get_notes_menu_keyboard()) return if text == "📋 Все заметки": notes = db.get_notes_by_user(from_id) if not notes: safe_vk_call(vk.messages.send, peer_id=peer_id, message="📭 У вас пока нет заметок.", random_id=get_random_id(), keyboard=get_notes_menu_keyboard()) return notes_sorted = sorted(notes, key=lambda x: x['date'], reverse=True) msg = "📋 Все ваши заметки (последние 10):\n" for n in notes_sorted[:10]: msg += f"#{n['id']} {n['date']} {n['subject']}: {n['text']}\n" if len(notes) > 10: msg += f"... и ещё {len(notes)-10}" safe_vk_call(vk.messages.send, peer_id=peer_id, message=msg, random_id=get_random_id(), keyboard=get_notes_menu_keyboard()) return if text == "➕ Добавить заметку": user_notes_state[from_id] = {"step": "add_date", "temp_data": {}} safe_vk_call(vk.messages.send, peer_id=peer_id, message="Введите дату в формате ГГГГ-ММ-ДД:", random_id=get_random_id()) return if text == "🗑 Удалить заметку": notes = db.get_notes_by_user(from_id) if not notes: safe_vk_call(vk.messages.send, peer_id=peer_id, message="📭 У вас нет заметок для удаления.", random_id=get_random_id(), keyboard=get_notes_menu_keyboard()) return notes_sorted = sorted(notes, key=lambda x: x['date'], reverse=True) msg = "🗑 Ваши заметки (введите ID для удаления):\n" for n in notes_sorted: msg += f"#{n['id']} {n['date']} {n['subject']}: {n['text']}\n" user_notes_state[from_id] = {"step": "delete_wait_id", "temp_data": {}} safe_vk_call(vk.messages.send, peer_id=peer_id, message=msg + "\nВведите ID заметки для удаления:", random_id=get_random_id(), keyboard=get_cancel_keyboard()) return if text == "📅 Заметки за дату": user_notes_state[from_id] = {"step": "view_by_date", "temp_data": {}} safe_vk_call(vk.messages.send, peer_id=peer_id, message="Введите дату в формате ГГГГ-ММ-ДД:", random_id=get_random_id(), keyboard=get_cancel_keyboard()) return if text == "🔙 Назад": # Возвращаем пользователя на ту страницу, с которой он зашёл page = user_main_page.get(from_id, 1) keyboard = get_keyboard_for_page(page, from_id) safe_vk_call(vk.messages.send, peer_id=peer_id, message="Главное меню:", random_id=get_random_id(), keyboard=keyboard) return # ---------- ПЕРЕКЛЮЧЕНИЕ СТРАНИЦ МЕНЮ ---------- if text == "➡️ Следующая страница": current = user_main_page.get(from_id, 1) new_page = min(current + 1, 3) user_main_page[from_id] = new_page safe_vk_call(vk.messages.send, peer_id=peer_id, message=f"Главное меню (страница {new_page}):", random_id=get_random_id(), keyboard=get_keyboard_for_page(new_page, from_id)) return if text == "⬅️ Предыдущая страница": current = user_main_page.get(from_id, 1) new_page = max(current - 1, 1) user_main_page[from_id] = new_page safe_vk_call(vk.messages.send, peer_id=peer_id, message=f"Главное меню (страница {new_page}):", random_id=get_random_id(), keyboard=get_keyboard_for_page(new_page, from_id)) return # ---------- ПЕРЕКЛЮЧЕНИЕ СТРАНИЦ МЕНЮ (цифровые эмодзи) ---------- stripped = text.strip() if stripped in ('1️⃣', '2️⃣', '3️⃣'): target_page = int(stripped[0]) user_main_page[from_id] = target_page safe_vk_call(vk.messages.send, peer_id=peer_id, message=f"Главное меню (страница {target_page}):", random_id=get_random_id(), keyboard=get_keyboard_for_page(target_page, from_id)) return # ---------- ПЕРЕКЛЮЧЕНИЕ ФОРМАТА РАСПИСАНИЯ ---------- if text.startswith("🖼 Формат:"): current_fmt = db.get_user_format(from_id) new_fmt = "image" if current_fmt == "text" else "text" db.set_user_format(from_id, new_fmt) fmt_label = "Картинка 🖼" if new_fmt == "image" else "Текст 📄" page = user_main_page.get(from_id, 1) safe_vk_call(vk.messages.send, peer_id=peer_id, message=f"✅ Формат расписания изменён на: {fmt_label}", random_id=get_random_id(), keyboard=get_keyboard_for_page(page, from_id)) return # ---------- ОТЗЫВЫ ---------- if text == "✍️ Оставить отзыв" or text.lower() in ["отзыв", "оставить отзыв", "отзывы"]: safe_vk_call(vk.messages.send, peer_id=peer_id, message="Напишите ваш отзыв или пожелание.\n" "Если хотите оценить работу бота, поставьте оценку от 1 до 5 звёзд (например, «4» или «4★»).", random_id=get_random_id(), keyboard=get_cancel_keyboard()) user_review_state[from_id] = True return # ---------- БАГ-РЕПОРТ ---------- if text == "🐛 Баг" or text.lower() in ["баг", "ошибка", "bug"]: user_bug_state[from_id] = { "step": "title", "reply_text": reply_text, "reply_cmd": reply_cmd } safe_vk_call(vk.messages.send, peer_id=peer_id, message="🐛 **Баг-репорт**\n\n" "Кратко опишите суть проблемы (1 строка).\n" "Например: «Не показывает расписание на завтра»", random_id=get_random_id(), keyboard=get_cancel_keyboard()) return # ---------- ОТМЕНА В СОСТОЯНИИ БАГ-РЕПОРТА ---------- if from_id in user_bug_state and text.lower() in ["отмена", "❌ отмена", "отменить"]: del user_bug_state[from_id] page = user_main_page.get(from_id, 1) keyboard = get_keyboard_for_page(page, from_id) safe_vk_call(vk.messages.send, peer_id=peer_id, message="❌ Отмена.", random_id=get_random_id(), keyboard=keyboard) return # ---------- ПОШАГОВЫЙ БАГ-РЕПОРТ ---------- if from_id in user_bug_state: bug = user_bug_state[from_id] step = bug.get("step", "title") if step == "title": if len(text.strip()) < 3: safe_vk_call(vk.messages.send, peer_id=peer_id, message="⚠️ Описание слишком короткое. Опишите баг подробнее:", random_id=get_random_id(), keyboard=get_cancel_keyboard()) return bug["title"] = text.strip() bug["step"] = "description" safe_vk_call(vk.messages.send, peer_id=peer_id, message="📋 Теперь опишите подробнее:\n\n" "• Что вы делали?\n" "• Что ожидали увидеть?\n" "• Что произошло на самом деле?\n\n" "💡 Можно ответить на сообщение с ошибкой, чтобы прикрепить его.", random_id=get_random_id(), keyboard=get_cancel_keyboard()) return if step == "description": # Команда завершения ввода описания if text.lower().strip() in ["готово", "далее", "дальше", "next"]: if not bug.get("description") or len(bug["description"].strip()) < 3: safe_vk_call(vk.messages.send, peer_id=peer_id, message="⚠️ Описание пустое. Напишите подробнее, затем «готово»:", random_id=get_random_id(), keyboard=get_cancel_keyboard()) return bug["step"] = "severity" safe_vk_call(vk.messages.send, peer_id=peer_id, message="🔴 Насколько серьёзная проблема?", random_id=get_random_id(), keyboard=get_bug_severity_keyboard()) return if len(text.strip()) < 3: safe_vk_call(vk.messages.send, peer_id=peer_id, message="⚠️ Описание слишком короткое. Расскажите подробнее (или «готово» для завершения):", random_id=get_random_id(), keyboard=get_cancel_keyboard()) return # Накапливаем текст if bug.get("description"): bug["description"] += "\n" + text.strip() else: bug["description"] = text.strip() safe_vk_call(vk.messages.send, peer_id=peer_id, message=f"✍️ Принято. Можете продолжить или напишите «готово»", random_id=get_random_id(), keyboard=get_cancel_keyboard()) return if step == "severity": severity_map = { "🟢 мелкая": "мелкая", "🟡 средняя": "средняя", "🔴 критичная": "критичная" } severity = severity_map.get(text.lower(), "средняя") bug["severity"] = severity bug["step"] = "confirm" reply_info = "" if bug.get("reply_text"): reply_info = f"\n📎 Прикреплено: «{bug['reply_text'][:80]}»" preview = (f"🐛 **Предпросмотр баг-репорта:**\n\n" f"📌 {bug['title']}\n" f"📋 {bug['description']}\n" f"🔴 Важность: {severity}" f"{reply_info}\n\n" f"Всё верно?") safe_vk_call(vk.messages.send, peer_id=peer_id, message=preview, random_id=get_random_id(), keyboard=get_bug_confirm_keyboard()) return if step == "confirm": text_clean = text.lower().strip() if "отправить" in text_clean: user_name = get_user_name(vk, from_id) bug_id = db.add_bug_report(from_id, user_name, bug['title'], bug['description'], bug.get('severity', 'средняя'), bug.get("reply_text") if bug.get("reply_text") else None, bug.get("reply_cmd") if bug.get("reply_cmd") else None) sev_emoji = {"мелкая": "🟢", "средняя": "🟡", "критичная": "🔴"}.get(bug['severity'], "🟡") admin_msg = (f"🐛 **Баг #{bug_id}** {sev_emoji} **{bug['severity'].upper()}**\n" f"👤 {user_name} (ID: {from_id})\n\n" f"📌 **{bug['title']}**\n\n" f"📋 {bug['description']}") if bug.get("reply_text"): admin_msg += f"\n\n📎 **Прикреплённое сообщение:**\n{bug['reply_text']}" logger.info(f"📤 Отправка бага #{bug_id} админу (ADMIN_ID={ADMIN_ID})") notify_admin(admin_msg, vk) page = user_main_page.get(from_id, 1) keyboard = get_keyboard_for_page(page, from_id) safe_vk_call(vk.messages.send, peer_id=peer_id, message=f"✅ Баг #{bug_id} отправлен! Спасибо, что помогаете сделать бота лучше 🙏", random_id=get_random_id(), keyboard=keyboard) del user_bug_state[from_id] return if "изменить" in text_clean: bug["step"] = "description" safe_vk_call(vk.messages.send, peer_id=peer_id, message="✏️ Напишите новое описание:", random_id=get_random_id(), keyboard=get_cancel_keyboard()) return del user_bug_state[from_id] page = user_main_page.get(from_id, 1) keyboard = get_keyboard_for_page(page, from_id) safe_vk_call(vk.messages.send, peer_id=peer_id, message="❌ Баг-репорт не отправлен.", random_id=get_random_id(), keyboard=keyboard) return if text == "⭐ Отзывы": user_reviews_page[from_id] = 1 page = 1 reviews, total_pages = db.get_reviews_page(page, REVIEWS_PER_PAGE) if not reviews: page = user_main_page.get(from_id, 1) keyboard = get_keyboard_for_page(page, from_id) safe_vk_call(vk.messages.send, peer_id=peer_id, message="📭 Отзывов пока нет.", random_id=get_random_id(), keyboard=keyboard) return msg = f"📢 Отзывы (страница {page}/{total_pages}):\n" for rev in reviews: user_name = rev.get('user_name', 'Аноним') rating = rev.get('rating') stars = f"{rating}★ " if rating else "" timestamp = rev.get('timestamp', '') date_str = timestamp[:10] if timestamp else '' text_preview = rev.get('text', '')[:100] msg += f"\n{user_name} ({date_str}) {stars}: {text_preview}\n" if total_pages > 1: msg += "\nИспользуйте кнопки ◀️ Назад / ▶️ Вперёд для навигации." kb = VkKeyboard(one_time=False) kb.add_button("◀️ Назад", color=VkKeyboardColor.SECONDARY) kb.add_button("▶️ Вперёд", color=VkKeyboardColor.SECONDARY) kb.add_line() kb.add_button("🔙 Назад", color=VkKeyboardColor.SECONDARY) safe_vk_call(vk.messages.send, peer_id=peer_id, message=msg, random_id=get_random_id(), keyboard=kb.get_keyboard()) else: page = user_main_page.get(from_id, 1) keyboard = get_keyboard_for_page(page, from_id) safe_vk_call(vk.messages.send, peer_id=peer_id, message=msg, random_id=get_random_id(), keyboard=keyboard) return if text == "📝 Мои отзывы": my_reviews = db.get_user_reviews(from_id) page = user_main_page.get(from_id, 1) keyboard = get_keyboard_for_page(page, from_id) if not my_reviews: safe_vk_call(vk.messages.send, peer_id=peer_id, message="📭 Вы ещё не оставляли отзывов.", random_id=get_random_id(), keyboard=keyboard) return msg = "📋 **Ваши отзывы:**\n" for i, r in enumerate(my_reviews[:5], 1): msg += f"{i}. {r['timestamp'][:10]} " if r.get("rating"): msg += f"{r['rating']}★ " text_preview = r['text'][:50] + "…" if len(r['text']) > 50 else r['text'] msg += f"{text_preview}\n" if len(my_reviews) > 5: msg += f"... и ещё {len(my_reviews)-5}." safe_vk_call(vk.messages.send, peer_id=peer_id, message=msg, random_id=get_random_id(), keyboard=keyboard) return if text == "👤 Личный кабинет": user_name = get_user_name(vk, from_id) user_data = db.get_user(from_id) group_id = user_data.get("group_id") if user_data else "?" reg_date = datetime.fromtimestamp(user_data.get("registered_at", 0)).strftime("%d.%m.%Y") if user_data and user_data.get("registered_at") else "?" dashboard_base = os.getenv("DASHBOARD_URL", "https://api.zoliryzik.ru") kb_dashboard = VkKeyboard(one_time=True) kb_dashboard.add_button('🔄 Пересоздать ссылку', color=VkKeyboardColor.POSITIVE) kb_dashboard.add_line() kb_dashboard.add_button('🗑 Удалить ссылку', color=VkKeyboardColor.NEGATIVE) kb_dashboard.add_line() kb_dashboard.add_button('🔙 Назад', color=VkKeyboardColor.SECONDARY) existing_token = db.get_dashboard_token_by_user(from_id) now_iso = datetime.now().isoformat() if existing_token and existing_token.get('expires_at', '') > now_iso: expires_at = datetime.fromisoformat(existing_token['expires_at']) expires_label = expires_at.strftime("%d.%m.%Y %H:%M") web_url = f"{dashboard_base}/dashboard/{existing_token['token']}" msg = (f"👤 **Личный кабинет**\n\n" f"👤 {user_name}\n" f"📅 Регистрация: {reg_date}\n" f"🏫 Группа: {group_id}\n\n" f"🔗 Ваша ссылка:\n" f"{web_url}\n\n" f"⏰ Действует до {expires_label}") elif existing_token: expires_label = datetime.fromisoformat(existing_token['expires_at']).strftime("%d.%m.%Y %H:%M") msg = (f"👤 **Личный кабинет**\n\n" f"👤 {user_name}\n" f"📅 Регистрация: {reg_date}\n" f"🏫 Группа: {group_id}\n\n" f"⚠️ Ссылка истекла ({expires_label})\n" f"Нажмите 🔄 Пересоздать ссылку чтобы получить новую.") else: token = secrets.token_urlsafe(32) expires_at = datetime.now() + timedelta(hours=24) expires_str = expires_at.isoformat() expires_label = expires_at.strftime("%d.%m.%Y %H:%M") db.create_dashboard_token(from_id, "web", token, expires_str) web_url = f"{dashboard_base}/dashboard/{token}" msg = (f"👤 **Личный кабинет**\n\n" f"👤 {user_name}\n" f"📅 Регистрация: {reg_date}\n" f"🏫 Группа: {group_id}\n\n" f"🔗 Ваша ссылка (24 часа):\n" f"{web_url}\n\n" f"⚠️ Привязана к вашему IP. Действует до {expires_label}") safe_vk_call(vk.messages.send, peer_id=peer_id, message=msg, random_id=get_random_id(), keyboard=kb_dashboard.get_keyboard()) return if text == "🔄 Пересоздать ссылку": user_name = get_user_name(vk, from_id) user_data = db.get_user(from_id) group_id = user_data.get("group_id") if user_data else "?" reg_date = datetime.fromtimestamp(user_data.get("registered_at", 0)).strftime("%d.%m.%Y") if user_data and user_data.get("registered_at") else "?" db.delete_dashboard_token(from_id) token = secrets.token_urlsafe(32) expires_at = datetime.now() + timedelta(hours=24) expires_str = expires_at.isoformat() expires_label = expires_at.strftime("%d.%m.%Y %H:%M") db.create_dashboard_token(from_id, "web", token, expires_str) dashboard_base = os.getenv("DASHBOARD_URL", "https://api.zoliryzik.ru") web_url = f"{dashboard_base}/dashboard/{token}" safe_vk_call(vk.messages.send, peer_id=peer_id, message=f"🔄 **Ссылка пересоздана!**\n\n" f"👤 {user_name}\n" f"📅 Регистрация: {reg_date}\n" f"🏫 Группа: {group_id}\n\n" f"🔗 Новая ссылка (действует 24 часа):\n" f"{web_url}\n\n" f"⚠️ Старая ссылка больше не работает. " f"Новая привязана к вашему IP и действует до {expires_label}.", random_id=get_random_id(), keyboard=get_keyboard_for_page(user_main_page.get(from_id, 1), from_id)) return if text == "🗑 Удалить ссылку": db.delete_dashboard_token(from_id) page = user_main_page.get(from_id, 1) safe_vk_call(vk.messages.send, peer_id=peer_id, message="🗑 Ссылка личного кабинета удалена.\n\n" "Нажмите «👤 Личный кабинет» чтобы создать новую.", random_id=get_random_id(), keyboard=get_keyboard_for_page(page, from_id)) return if text == "🐛 Мои баги": my_bugs = db.get_user_bug_reports(from_id) page = user_main_page.get(from_id, 1) keyboard = get_keyboard_for_page(page, from_id) if not my_bugs: safe_vk_call(vk.messages.send, peer_id=peer_id, message="📭 Вы ещё не отправляли баг-репортов.\n\n" "🐛 Нажмите «Баг» на странице 2, чтобы сообщить об ошибке.", random_id=get_random_id(), keyboard=keyboard) return status_map = {"новая": "🆕", "в работе": "⚙️", "выполнена": "✅", "отклонена": "❌"} msg = "🐛 **Ваши баг-репорты:**\n\n" for i, b in enumerate(my_bugs[:10], 1): status = b.get("status", "новая") emoji = status_map.get(status, "❓") lines = b['text'].split("\n") title = lines[0].replace("📌 ", "") if lines[0].startswith("📌") else lines[0] sev_line = next((l for l in lines if l.startswith("🔴")), "") sev = sev_line.replace("🔴 Важность: ", "") if sev_line else "?" msg += f"{i}. {emoji} #{b['id']} — {title}\n" msg += f" 🔴 {sev} | 📅 {b['timestamp'][:10]}\n" if b.get("admin_reply"): msg += f" 💬 {b['admin_reply'][:60]}\n" msg += "\n" if len(my_bugs) > 10: msg += f"... и ещё {len(my_bugs)-10}.\n" msg += "\n💡 Статусы: 🆕 новая → ⚙️ в работе → ✅ выполнена / ❌ отклонена" safe_vk_call(vk.messages.send, peer_id=peer_id, message=msg, random_id=get_random_id(), keyboard=keyboard) return # ---------- ПРОСМОТР ВСЕХ ОТЗЫВОВ ---------- if text == "📢 Все отзывы": page = user_reviews_page.get(from_id, 1) reviews_text, current_page, total_pages = get_reviews_page(page, REVIEWS_PER_PAGE) kb = VkKeyboard(one_time=False) if total_pages > 1: if current_page > 1: kb.add_button("◀️ Назад", color=VkKeyboardColor.SECONDARY) if current_page < total_pages: kb.add_button("▶️ Вперёд", color=VkKeyboardColor.SECONDARY) # Добавляем разделитель только если есть кнопки пагинации if current_page > 1 or current_page < total_pages: kb.add_line() kb.add_button("🔙 В меню", color=VkKeyboardColor.SECONDARY) safe_vk_call(vk.messages.send, peer_id=peer_id, message=reviews_text, random_id=get_random_id(), keyboard=kb.get_keyboard()) user_reviews_page[from_id] = current_page return # ---------- ПАГИНАЦИЯ ОТЗЫВОВ ---------- if text in ["◀️ Назад", "▶️ Вперёд"] and from_id in user_reviews_page: current_page = user_reviews_page[from_id] if text == "◀️ Назад": new_page = current_page - 1 else: new_page = current_page + 1 reviews_text, new_page, total_pages = get_reviews_page(new_page, REVIEWS_PER_PAGE) user_reviews_page[from_id] = new_page kb = VkKeyboard(one_time=False) if total_pages > 1: if new_page > 1: kb.add_button("◀️ Назад", color=VkKeyboardColor.SECONDARY) if new_page < total_pages: kb.add_button("▶️ Вперёд", color=VkKeyboardColor.SECONDARY) # Добавляем разделитель только если есть кнопки пагинации if new_page > 1 or new_page < total_pages: kb.add_line() kb.add_button("🔙 В меню", color=VkKeyboardColor.SECONDARY) safe_vk_call(vk.messages.send, peer_id=peer_id, message=reviews_text, random_id=get_random_id(), keyboard=kb.get_keyboard()) return # ---------- ВОЗВРАТ В ГЛАВНОЕ МЕНЮ ИЗ ОТЗЫВОВ ---------- if text == "🔙 В меню": if from_id in user_reviews_page: del user_reviews_page[from_id] page = user_main_page.get(from_id, 1) keyboard = get_keyboard_for_page(page, from_id) safe_vk_call(vk.messages.send, peer_id=peer_id, message="Главное меню:", random_id=get_random_id(), keyboard=keyboard) return # Обработка ввода ID для удаления if from_id in user_notes_state and user_notes_state[from_id]["step"] == "delete_wait_id": try: note_id = int(text) except: safe_vk_call(vk.messages.send, peer_id=peer_id, message="❌ Введите число (ID заметки).", random_id=get_random_id()) return deleted = db.delete_note(note_id) if deleted: safe_vk_call(vk.messages.send, peer_id=peer_id, message=f"✅ Заметка #{note_id} удалена.", random_id=get_random_id(), keyboard=get_notes_menu_keyboard()) else: safe_vk_call(vk.messages.send, peer_id=peer_id, message=f"❌ Заметка #{note_id} не найдена.", random_id=get_random_id(), keyboard=get_notes_menu_keyboard()) del user_notes_state[from_id] return # Обработка ввода даты для просмотра if from_id in user_notes_state and user_notes_state[from_id]["step"] == "view_by_date": date_str = text try: datetime.strptime(date_str, "%Y-%m-%d") except: safe_vk_call(vk.messages.send, peer_id=peer_id, message="❌ Неверная дата. Введите дату в формате ГГГГ-ММ-ДД", random_id=get_random_id(), keyboard=get_cancel_keyboard()) return notes = db.get_notes_by_user(from_id) filtered = [n for n in notes if n["date"] == date_str] if not filtered: safe_vk_call(vk.messages.send, peer_id=peer_id, message=f"📭 На {date_str} заметок нет", random_id=get_random_id(), keyboard=get_notes_menu_keyboard()) else: msg = f"📋 Заметки на {date_str}:\n" for n in filtered: msg += f"#{n['id']} {n['subject']}: {n['text']}\n" safe_vk_call(vk.messages.send, peer_id=peer_id, message=msg, random_id=get_random_id(), keyboard=get_notes_menu_keyboard()) del user_notes_state[from_id] return # ---------- ОБРАБОТКА КОМАНДЫ ПОМОЩИ ---------- if text.lower() in ["помощь", "❓ помощь", "меню", "help"]: if from_id == ADMIN_ID: help_text = ( "🛠 **Команды администратора**\n\n" "📊 **Статистика**\n" "• `!stats` – общая статистика бота\n" "• `!stats group` – статистика выбранной группы\n" "• `!stats api` – статистика API группы\n" "• `!stats user ` – информация о пользователе\n" "• `!stats top groups` – топ групп по запросам\n" "• `!stats top users` – топ пользователей по запросам\n" "• `!stats top users commands` – топ по командам\n" "• `!stats top commands` – топ команд\n" "• `!stats top group commands [N]` – топ в группе\n" "• `!stats archive` – просмотр архивных логов\n\n" "💡 **Идеи**\n" "• `!ideas [страница]` – список идей\n" "• `!idea status ` – изменить статус идеи\n" "• `!idea delete ` – удалить идею\n\n" "👥 **Пользователи**\n" "• `!genkey` / `сгенерировать` – создать ключ для группы\n" "• `!users` – список всех пользователей\n" "• `!users group [ID]` – пользователи группы\n" "• `!user ` – статистика пользователя\n" "• `отвязать ` – отвязать пользователя\n\n" "🔐 **Управление владельцами**\n" "• `!addacc ` – назначить владельца\n" "• `!rmacc ` – удалить запись о владельце\n" "• `!listacc` – список групп с владельцами\n" "• `!setcreds ` – отметить наличие логина/пароля\n" "• `!unsetcreds ` – снять пометку\n" "• `!token` – запросить согласие у владельца\n" "• `!reloadgroup [ID/all]` – перезагрузить группу из .env\n\n" "📦 **Другое**\n" "• `!wipe logs` – архивировать логи\n\n" "🛡️ **Управление IP**\n" "• `!ip allow <время>` — добавить IP (30m, 1h, 2d, perm)\n" "• `!ip revoke ` — удалить IP из белого списка\n" "• `!ip list` — показать все IP-адреса\n\n" "🐛 **Баг-репорты**\n" "• `!bugs` — список всех багов\n" "• `!bug view ` — подробный просмотр бага\n" "• `!bug reply <текст>` — ответить автору бага\n" "• `!bug status <статус>` — изменить статус\n" "• `!bug delete ` — удалить баг-репорт\n\n" "📚 **Доступные команды для пользователей**\n\n" "📅 **Расписание**\n" "• `сегодня` – расписание на сегодня\n" "• `завтра` – расписание на завтра\n" "• `день N` – расписание через N дней (например, день 3)\n" "• `день ГГГГ-ММ-ДД` – расписание на конкретную дату\n" "• `неделя` – расписание на ближайшие 7 дней\n" "• `обновить` / `сброс` – очистить кэш расписания\n\n" "📝 **Заметки**\n" "• Нажми кнопку `📝 Заметки` для входа в меню\n" " – `📋 Все заметки` – показать последние заметки\n" " – `➕ Добавить заметку` – создать новую заметку\n" " – `🗑 Удалить заметку` – удалить по ID\n" " – `📅 Заметки за дату` – просмотр за конкретную дату\n" " – `🔙 Назад` – вернуться в главное меню\n\n" "💡 **Обратная связь**\n" "• `💡 Предложить идею` – отправить идею администратору\n" "• `✍️ Оставить отзыв` – оставить отзыв\n" "• `📝 Мои отзывы` – посмотреть свои отзывы\n" "• `📢 Все отзывы` – посмотреть отзывы всех пользователей\n\n" "🐛 **Баг-репорты**\n" "• `🐛 Баг` – сообщить об ошибке (пошагово: заголовок → описание → важность → подтверждение)\n" "• `🐛 Мои баги` – посмотреть свои баг-репорты и их статус\n\n" "🔔 **Уведомления**\n" "• `!notify` – управление уведомлениями (ответы админа)\n\n" "❓ **Помощь** – это сообщение" ) else: # Проверяем, является ли пользователь владельцем хотя бы одной группы is_owner = any(acc.get("owner_id") == from_id for acc in accounts.values()) if is_owner: help_text = ( "📚 **Доступные команды**\n\n" "📅 **Расписание**\n" "• `сегодня` – расписание на сегодня\n" "• `завтра` – расписание на завтра\n" "• `день N` – расписание через N дней (например, день 3)\n" "• `день ГГГГ-ММ-ДД` – расписание на конкретную дату\n" "• `неделя` – расписание на ближайшие 7 дней\n" "• `обновить` / `сброс` – очистить кэш расписания\n\n" "📝 **Заметки**\n" "• Нажми кнопку `📝 Заметки` для входа в меню\n" " – `📋 Все заметки` – показать последние заметки\n" " – `➕ Добавить заметку` – создать новую заметку\n" " – `🗑 Удалить заметку` – удалить по ID\n" " – `📅 Заметки за дату` – просмотр за конкретную дату\n" " – `🔙 Назад` – вернуться в главное меню\n\n" "💡 **Обратная связь**\n" "• `💡 Предложить идею` – отправить идею администратору\n" "• `✍️ Оставить отзыв` – оставить отзыв\n" "• `📝 Мои отзывы` – посмотреть свои отзывы\n" "• `📢 Все отзывы` – посмотреть отзывы всех пользователей\n\n" "🐛 **Баг-репорты**\n" "• `🐛 Баг` – сообщить об ошибке\n" "• `🐛 Мои баги` – посмотреть свои баг-репорты и их статус\n\n" "🔐 **Для владельцев групп**\n" "• `отозвать` – отозвать согласие на использование аккаунта\n\n" "🔔 **Уведомления**\n" "• `!notify` – управление уведомлениями (ответы админа)\n\n" "❓ **Помощь** – это сообщение" ) else: help_text = ( "📚 **Доступные команды**\n\n" "📅 **Расписание**\n" "• `сегодня` – расписание на сегодня\n" "• `завтра` – расписание на завтра\n" "• `день N` – расписание через N дней (например, день 3)\n" "• `день ГГГГ-ММ-ДД` – расписание на конкретную дату\n" "• `неделя` – расписание на ближайшие 7 дней\n" "• `обновить` / `сброс` – очистить кэш расписания\n\n" "📝 **Заметки**\n" "• Нажми кнопку `📝 Заметки` для входа в меню\n" " – `📋 Все заметки` – показать последние заметки\n" " – `➕ Добавить заметку` – создать новую заметку\n" " – `🗑 Удалить заметку` – удалить по ID\n" " – `📅 Заметки за дату` – просмотр за конкретную дату\n" " – `🔙 Назад` – вернуться в главное меню\n\n" "💡 **Обратная связь**\n" "• `💡 Предложить идею` – отправить идею администратору\n" "• `✍️ Оставить отзыв` – оставить отзыв\n" "• `📝 Мои отзывы` – посмотреть свои отзывы\n" "• `📢 Все отзывы` – посмотреть отзывы всех пользователей\n\n" "🐛 **Баг-репорты**\n" "• `🐛 Баг` – сообщить об ошибке\n" "• `🐛 Мои баги` – посмотреть свои баг-репорты и их статус\n\n" "🔔 **Уведомления**\n" "• `!notify` – управление уведомлениями (ответы админа)\n\n" "❓ **Помощь** – это сообщение" ) page = user_main_page.get(from_id, 1) keyboard = get_keyboard_for_page(page, from_id) safe_vk_call(vk.messages.send, peer_id=peer_id, message=help_text, random_id=get_random_id(), keyboard=keyboard) return # ---------- УПРАВЛЕНИЕ УВЕДОМЛЕНИЯМИ ---------- if text.lower() in ["!notify", "🔔 уведомления", "уведомления"]: msg, kb = build_notify_menu(from_id) safe_vk_call(vk.messages.send, peer_id=peer_id, message=msg, random_id=get_random_id(), keyboard=kb.get_keyboard()) return # Обработка нажатий кнопок уведомлений notify_match = False for nt in NOTIFY_TYPES: for prefix in ["❌ Выкл", "✅ Вкл"]: if text.lower().startswith(f"{prefix.lower()} {nt['emoji']} {nt['label'].lower()}"): enabled = prefix != "❌ Выкл" set_user_notify_status(from_id, nt['key'], enabled) msg, kb = build_notify_menu(from_id) safe_vk_call(vk.messages.send, peer_id=peer_id, message=msg, random_id=get_random_id(), keyboard=kb.get_keyboard()) notify_match = True break if notify_match: break if notify_match: return # =========== ПРЕДЛОЖИТЬ ИДЕЮ ============ # ---------- ПРЕДЛОЖИТЬ ИДЕЮ ---------- if from_id in user_idea_state: if text.lower() in ["отмена", "❌ отмена", "❌ Отмена", "отменить"]: del user_idea_state[from_id] page = user_main_page.get(from_id, 1) keyboard = get_keyboard_for_page(page, from_id) safe_vk_call(vk.messages.send, peer_id=peer_id, message="❌ Отправка идеи отменена.", random_id=get_random_id(), keyboard=keyboard) return user_name = get_user_name(vk, from_id) idea_id = db.add_idea(from_id, user_name, text) notify_admin(f"💡 Новая идея #{idea_id} от {user_name} (ID {from_id}):\n{text}", vk) page = user_main_page.get(from_id, 1) safe_vk_call(vk.messages.send, peer_id=peer_id, message="Спасибо! Ваша идея отправлена администратору.", random_id=get_random_id(), keyboard=get_keyboard_for_page(page, from_id)) del user_idea_state[from_id] return if text == "💡 Мои идеи": ideas = db.get_all_ideas() my_ideas = [i for i in ideas if i.get('user_id') == from_id] page = user_main_page.get(from_id, 1) keyboard = get_keyboard_for_page(page, from_id) if not my_ideas: safe_vk_call(vk.messages.send, peer_id=peer_id, message="💡 У вас пока нет идей.", random_id=get_random_id(), keyboard=keyboard) return msg = "💡 **Ваши идеи:**\n" for i, idea in enumerate(my_ideas[:5], 1): status_emoji = {"новая": "🆕", "в работе": "⚙️", "выполнена": "✅", "отклонена": "❌"}.get(idea.get('status', 'новая'), '❓') msg += f"{i}. #{idea['id']} {status_emoji} {idea.get('idea', '')[:50]}\n" if len(my_ideas) > 5: msg += f"... и ещё {len(my_ideas)-5}." safe_vk_call(vk.messages.send, peer_id=peer_id, message=msg, random_id=get_random_id(), keyboard=keyboard) return if text == "💡 Предложить идею" or text.lower() in ["идея", "предложить идею", "предложения"]: safe_vk_call(vk.messages.send, peer_id=peer_id, message="Напишите вашу идею, и она будет отправлена администратору.", random_id=get_random_id(), keyboard=get_cancel_keyboard()) user_idea_state[from_id] = True return # ---------- ВВОД НОВОГО СТАТУСА ДЛЯ ИДЕИ ---------- if from_id in user_idea_status_state: if text.lower() in ["отмена", "❌ отмена", "отменить"]: del user_idea_status_state[from_id] page = user_main_page.get(from_id, 1) keyboard = get_keyboard_for_page(page, from_id) safe_vk_call(vk.messages.send, peer_id=peer_id, message="❌ Изменение статуса отменено.", random_id=get_random_id(), keyboard=keyboard) return idea_id = user_idea_status_state[from_id]["idea_id"] new_status = text.strip() if not new_status: safe_vk_call(vk.messages.send, peer_id=peer_id, message="❌ Статус не может быть пустым. Попробуйте ещё раз.", random_id=get_random_id()) return ideas = db.get_all_ideas() for idea in ideas: if idea.get("id") == idea_id: old_status = idea.get("status", "?") db.update_idea_status(idea_id, new_status) author_id = idea.get("user_id") if author_id: idea_text = idea.get("idea", "") preview = idea_text[:80] + ("…" if len(idea_text) > 80 else "") status_emoji = {"новая": "🆕", "в работе": "⚙️", "выполнена": "✅", "отклонена": "❌"} emoji = status_emoji.get(new_status, "📋") user_msg = f"💡 **Идея #{idea_id}** — {emoji} {new_status.upper()}\n\n" user_msg += f"📋 {preview}\n\n" if new_status == "выполнена": user_msg += "✅ Ваша идея реализована! Спасибо за предложение 🎉" elif new_status == "отклонена": user_msg += "❌ К сожалению, эта идея не может быть реализована." elif new_status == "в работе": user_msg += "⚙️ Идея взята в работу. Мы работаем над её реализацией." else: user_msg += f"📋 Статус изменён: '{old_status}' → '{new_status}'" safe_vk_call(vk.messages.send, peer_id=author_id, message=user_msg, random_id=get_random_id()) page = user_main_page.get(from_id, 1) keyboard = get_keyboard_for_page(page, from_id) status_emoji = {"новая": "🆕", "в работе": "⚙️", "выполнена": "✅", "отклонена": "❌"} emoji = status_emoji.get(new_status, "📋") safe_vk_call(vk.messages.send, peer_id=peer_id, message=f"✅ {emoji} Статус идеи #{idea_id} изменён: '{old_status}' → '{new_status}'", random_id=get_random_id(), keyboard=keyboard) break else: page = user_main_page.get(from_id, 1) safe_vk_call(vk.messages.send, peer_id=peer_id, message=f"❌ Ошибка: идея #{idea_id} не найдена.", random_id=get_random_id(), keyboard=get_keyboard_for_page(page, from_id)) del user_idea_status_state[from_id] return # ---------- ОБРАБОТКА ВВОДА ОТЗЫВА ---------- if from_id in user_review_state: # Отмена if text.lower() in ["отмена", "❌ отмена", "❌ Отмена", "отменить"]: del user_review_state[from_id] page = user_main_page.get(from_id, 1) keyboard = get_keyboard_for_page(page, from_id) safe_vk_call(vk.messages.send, peer_id=peer_id, message="❌ Отправка отзыва отменена.", random_id=get_random_id(), keyboard=keyboard) return user_name = get_user_name(vk, from_id) review_text = text rating = None match = re.match(r'^(\d)(?:★|\s*★)?', review_text) if match: rating = int(match.group(1)) review_text = re.sub(r'^\d(?:★|\s*★)?\s*', '', review_text).strip() if not review_text: review_text = "Без текста" admin_msg = f"✍️ **Отзыв** от {user_name} (ID {from_id})\n" if rating: admin_msg += f"Оценка: {rating}★\n" admin_msg += f"Текст: {review_text}" notify_admin(admin_msg, vk) try: db.add_review(from_id, user_name, review_text, rating) except Exception as e: logger.error(f"Ошибка сохранения отзыва: {e}") page = user_main_page.get(from_id, 1) keyboard = get_keyboard_for_page(page, from_id) safe_vk_call(vk.messages.send, peer_id=peer_id, message="✅ Спасибо за ваш отзыв! Он передан администратору.", random_id=get_random_id(), keyboard=keyboard) del user_review_state[from_id] return # ---------- ОБРАБОТКА КОМАНД РАСПИСАНИЯ ---------- answer = handle_schedule_command(text, user_group, vk, user_id=from_id, user_name=user_name) if answer: user_format = db.get_user_format(from_id) page = user_main_page.get(from_id, 1) keyboard = get_keyboard_for_page(page, from_id) is_error = answer.startswith("❌") or answer.startswith("⏳") is_reset = answer.startswith("✅ Кэш очищен") is_week = "📋" in answer and ("НЕДЕЛЯ" in answer or "СЛЕДУЮЩАЯ" in answer) if is_reset: safe_vk_call(vk.messages.send, peer_id=peer_id, message=answer, random_id=get_random_id(), keyboard=keyboard) elif user_format == "image" and not is_error and not is_week: target_date = _get_date_from_command(text) def _send_schedule_image(): """Генерирует и отправляет картинку расписания.""" data_for_image, _ = get_schedule_for_date(user_group, target_date, vk, user_id=from_id, user_name=user_name) if data_for_image and not data_for_image.get("_error") and data_for_image.get("data", {}).get("rasp"): rasp = data_for_image["data"]["rasp"] date_lessons = [l for l in rasp if l.get("дата", "").split("T")[0] == target_date] if date_lessons: info = data_for_image.get("data", {}).get("info", {}) img_bytes = generate_schedule_image(date_lessons, user_group, target_date, info) save_image_to_cache(user_group, target_date, img_bytes) try: uploader = upload.VkUpload(vk) photo = uploader.photo_messages(photos=img_bytes)[0] attachment = f"photo{photo['owner_id']}_{photo['id']}" safe_vk_call(vk.messages.send, peer_id=peer_id, message="📅 Вот твоё расписание:", attachment=attachment, random_id=get_random_id(), keyboard=keyboard) return True except Exception as e: logger.warning(f"Не удалось отправить картинку: {e}") return False cached_img = load_cached_image(user_group, target_date) if cached_img: try: uploader = upload.VkUpload(vk) photo = uploader.photo_messages(photos=cached_img)[0] attachment = f"photo{photo['owner_id']}_{photo['id']}" safe_vk_call(vk.messages.send, peer_id=peer_id, message="📅 Вот твоё расписание:", attachment=attachment, random_id=get_random_id(), keyboard=keyboard) except Exception as e: logger.warning(f"Кэш картинка не отправлена, генерирую заново: {e}") if not _send_schedule_image(): safe_vk_call(vk.messages.send, peer_id=peer_id, message=answer, random_id=get_random_id(), keyboard=keyboard) else: if not _send_schedule_image(): safe_vk_call(vk.messages.send, peer_id=peer_id, message=answer, random_id=get_random_id(), keyboard=keyboard) else: safe_vk_call(vk.messages.send, peer_id=peer_id, message=answer, random_id=get_random_id(), keyboard=keyboard) # ========== ОСНОВНОЙ ЦИКЛ ========== def run_bot(): last_clean = time.time() CLEAN_INTERVAL = 24 * 3600 last_heartbeat = time.time() HEARTBEAT_INTERVAL = 24 * 3600 last_token_check = 0 last_token_date = "" # Чтобы не обновлять 2 раза в день TOKEN_CHECK_INTERVAL = 7 * 24 * 3600 # Раз в неделю проверяем токены while True: try: vk_session = vk_api.VkApi(token=VK_TOKEN) vk = vk_session.get_api() if not check_vk_token(vk): sys.exit(1) longpoll = VkBotLongPoll(vk_session, GROUP_ID_VK) logger.info("✅ Бот запущен. Ожидаю команды...") for event in longpoll.listen(): # Обработка входящих сообщений if event.type == VkBotEventType.MESSAGE_NEW: process_message(vk, event) # Обработка выхода пользователя из сообщества if event.type == VkBotEventType.GROUP_LEAVE: try: user_id = event.obj['user_id'] # Получаем имя пользователя (если не получится – будет просто ID) try: user_name = get_user_name(vk, user_id) except: user_name = f"id{user_id}" # Проверяем, был ли зарегистрирован user_data = db.get_user(user_id) was_registered = user_data is not None user_group = user_data["group_id"] if was_registered else None # Удаляем пользователя из базы (если был) if was_registered: delete_user(user_id) logger.info(f"Пользователь {user_name} (ID {user_id}) удалён из бота") else: logger.info(f"Пользователь {user_name} (ID {user_id}) не был зарегистрирован") # Проверяем, не был ли он владельцем групп owned_groups = [] for gid, data in accounts.items(): if data.get("owner_id") == user_id: owned_groups.append(gid) # Сообщение админу msg_parts = [f"👋 Пользователь {user_name} (ID {user_id}) покинул сообщество."] if was_registered: msg_parts.append(f" Зарегистрирован в группе: {user_group}") else: msg_parts.append(" Не был зарегистрирован.") if owned_groups: groups_str = ", ".join(str(g) for g in owned_groups) msg_parts.append(f" Владел группами: {groups_str}. Доступ отключён, согласия в архиве.") with accounts_lock: for gid in owned_groups: group_data = accounts.get(gid) if not group_data: continue # Сохраняем информацию о согласии до удаления consent_info = group_data.pop("consent", None) group_data["blocked"] = True group_data["revoked"] = True group_data.pop("session", None) db.save_group_account(gid, group_data) # Архивируем файл согласия, если он был if consent_info and "file" in consent_info: old_path = consent_info["file"] if old_path and os.path.exists(old_path): try: archive_dir = os.path.join("agreements", "разрешенные", "архив") os.makedirs(archive_dir, exist_ok=True) base = os.path.basename(old_path) name, ext = os.path.splitext(base) new_name = f"{name}_revoked_{datetime.now().strftime('%Y%m%d')}{ext}" new_path = os.path.join(archive_dir, new_name) os.rename(old_path, new_path) logger.info(f"Файл согласия группы {gid} перемещён в архив") except Exception as e: logger.error(f"Ошибка при архивации файла {old_path}: {e}") else: msg_parts.append(" Не владел группами.") safe_vk_call(vk.messages.send, peer_id=ADMIN_ID, message="\n".join(msg_parts), random_id=get_random_id()) except Exception as e: logger.exception(f"Ошибка при обработке GROUP_LEAVE: {e}") continue # Автоочистка кэша раз в сутки if time.time() - last_clean > CLEAN_INTERVAL: removed_json = auto_clean_cache(days_old=1, vk=vk) removed_img = auto_clean_images(days_old=1) total_removed = removed_json + removed_img if total_removed > 0: logger.info(f"🧹 Автоочистка: удалено {removed_json} json и {removed_img} картинок") last_clean = time.time() # Heartbeat раз в сутки if time.time() - last_heartbeat > HEARTBEAT_INTERVAL: api_logger.info("🔁 Ежедневная проверка: бот работает, логи активны") last_heartbeat = time.time() except (requests.exceptions.RequestException, vk_api.exceptions.ApiError) as e: logger.error(f"Ошибка соединения или API VK: {e}. Переподключение через 5с...") time.sleep(5) except Exception as e: logger.exception(f"Критическая ошибка: {e}. Перезапуск через 30с...") time.sleep(30) if __name__ == "__main__": run_bot()