# Системные импорты import asyncio, json, logging, os, re, random, sqlite3, threading from dataclasses import dataclass from datetime import datetime, time, timedelta from io import BytesIO from pathlib import Path from zoneinfo import ZoneInfo # Стороние импорты import requests from PIL import Image, ImageDraw, ImageFont from aiogram import Bot, Dispatcher, F, types from aiogram.filters import Command, CommandStart from aiogram.enums import ParseMode from aiogram.types import ( BufferedInputFile, Message, BotCommand, FSInputFile, BotCommandScopeDefault, BotCommandScopeAllPrivateChats, BotCommandScopeAllGroupChats ) from aiogram.client.default import DefaultBotProperties # Локальные импорты from AI.talk_handler import handle_talk, generate_autoreply, push_message from games.casino import play_casino from games.fortune import generate_fortune from zparser import get_military_data import config logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) _polychaetsi_lock = threading.Lock() _polychaetsi_word_re = re.compile(r"[а-яё]+", re.IGNORECASE) _no_lesson_re = re.compile(r"(нету\s+пары|нет\s+пары|пары\s+нету|пары\s+нет)", re.IGNORECASE) _weekday_names = { 0: "Понедельник", 1: "Вторник", 2: "Среда", 3: "Четверг", 4: "Пятница", 5: "Суббота", 6: "Воскресенье", } _weekday_aliases = { "mon": 0, "monday": 0, "пн": 0, "пон": 0, "понедельник": 0, "tue": 1, "tues": 1, "tuesday": 1, "вт": 1, "втр": 1, "вторник": 1, "wed": 2, "wednesday": 2, "ср": 2, "срд": 2, "среда": 2, "thu": 3, "thur": 3, "thurs": 3, "thursday": 3, "чт": 3, "чет": 3, "четверг": 3, "fri": 4, "friday": 4, "пт": 4, "пят": 4, "пятница": 4, "sat": 5, "saturday": 5, "сб": 5, "суб": 5, "суббота": 5, "sun": 6, "sunday": 6, "вс": 6, "вск": 6, "воскресенье": 6, } @dataclass(frozen=True) class LessonSlot: pair_number: int | None title: str start: time | None end: time | None is_free: bool # --- Инициализация и вспомогательные функции --- def init_penis_db() -> None: """Инициализация БД для миниигры.""" Path(config.PENIS_DB_PATH).parent.mkdir(parents=True, exist_ok=True) with sqlite3.connect(config.PENIS_DB_PATH) as conn: conn.execute( """ CREATE TABLE IF NOT EXISTS penis_stats ( user_id INTEGER PRIMARY KEY, display_name TEXT NOT NULL, length REAL NOT NULL, last_used_ts INTEGER ) """ ) conn.execute("CREATE INDEX IF NOT EXISTS idx_penis_stats_length ON penis_stats(length DESC)") conn.commit() def init_polychaetsi_storage() -> None: """Инициализация json-хранилища статистики слова 'получается'.""" path = config.POLYCHAETSI_STATS_PATH path.parent.mkdir(parents=True, exist_ok=True) if not path.exists(): path.write_text('{"users": {}}', encoding="utf-8") def templates_list_text() -> str: """Список макетов для подсказки.""" parts = [] for tid in sorted(config.TEMPLATES.keys()): label = config.TEMPLATES[tid].get("label", "") parts.append(f"{tid} — {label}" if label else tid) return ", ".join(parts) if parts else "нет макетов" def _format_delta_ms(delta: timedelta) -> str: """HH:MM:SS.mmm, не допускаем отрицательных значений.""" total_ms = int(max(delta.total_seconds(), 0) * 1000) hours, rem = divmod(total_ms, 3_600_000) minutes, rem = divmod(rem, 60_000) seconds, milliseconds = divmod(rem, 1_000) return f"{hours:02d}:{minutes:02d}:{seconds:02d}.{milliseconds:03d}" def _normalize_now(now: datetime | None = None) -> datetime: if now is None: return datetime.now(ZoneInfo(config.DEFAULT_TZ)) if now.tzinfo is None: return now.replace(tzinfo=ZoneInfo(config.DEFAULT_TZ)) return now def _current_week_key(now: datetime) -> str: week_number = now.isocalendar().week is_upper = (week_number % 2 == 1) == config.UPPER_WEEK_IS_ODD return "upper" if is_upper else "lower" def _resolve_schedule_date(now: datetime, requested_weekday: int | None = None) -> datetime: if requested_weekday is None: return now delta_days = (requested_weekday - now.weekday()) % 7 return now + timedelta(days=delta_days) def _parse_weekday_arg(value: str) -> int | None: normalized = re.sub(r"[^a-zа-яё]", "", value.lower()) if not normalized: return None return _weekday_aliases.get(normalized) def _build_lesson_slot(raw_entry: dict) -> LessonSlot: pair_number = raw_entry.get("pair") start_t, end_t = config.PAIR_TIMES.get(pair_number, (None, None)) title = str(raw_entry.get("title", "")).strip() or "Пара не указана" return LessonSlot( pair_number=pair_number, title=title, start=start_t, end=end_t, is_free=bool(_no_lesson_re.search(title)), ) def get_lessons_for_day(now: datetime | None = None) -> tuple[str, list[LessonSlot]]: now = _normalize_now(now) week_key = _current_week_key(now) raw_day = config.LESSON_SCHEDULE.get(week_key, {}).get(now.weekday(), []) lessons = [_build_lesson_slot(item) for item in raw_day] return week_key, lessons def _lesson_bounds(now: datetime, lesson: LessonSlot) -> tuple[datetime | None, datetime | None]: if lesson.start is None or lesson.end is None: return None, None start_dt = now.replace(hour=lesson.start.hour, minute=lesson.start.minute, second=0, microsecond=0) end_dt = now.replace(hour=lesson.end.hour, minute=lesson.end.minute, second=0, microsecond=0) return start_dt, end_dt def _find_current_lesson(now: datetime, lessons: list[LessonSlot]) -> LessonSlot | None: for lesson in lessons: start_dt, end_dt = _lesson_bounds(now, lesson) if start_dt and end_dt and start_dt <= now < end_dt: return lesson return None def _find_next_lesson(now: datetime, lessons: list[LessonSlot]) -> LessonSlot | None: for lesson in lessons: start_dt, _ = _lesson_bounds(now, lesson) if start_dt and start_dt > now and not lesson.is_free: return lesson return None def _format_lesson_brief(lesson: LessonSlot) -> str: prefix = f"{lesson.pair_number} пара" if lesson.pair_number is not None else "Пара" if lesson.start and lesson.end: return f"{prefix} ({lesson.start:%H:%M}-{lesson.end:%H:%M}) — {lesson.title}" return f"{prefix} — {lesson.title}" def _format_lesson_line(lesson: LessonSlot) -> str: if lesson.pair_number is None: return f"• {lesson.title}" if lesson.start and lesson.end: return f"{lesson.pair_number}. {lesson.start:%H:%M}-{lesson.end:%H:%M} — {lesson.title}" return f"{lesson.pair_number}. {lesson.title}" def next_lesson_text(now: datetime | None = None) -> str: now = _normalize_now(now) _, lessons = get_lessons_for_day(now) next_lesson = _find_next_lesson(now, lessons) if not next_lesson: return "Следующей пары сегодня нет." return f"Следующая пара: {_format_lesson_brief(next_lesson)}" def build_schedule_text(now: datetime | None = None, requested_weekday: int | None = None) -> str: now = _normalize_now(now) target_now = _resolve_schedule_date(now, requested_weekday) week_key, lessons = get_lessons_for_day(target_now) show_live_status = target_now.date() == now.date() current_lesson = _find_current_lesson(now, lessons) if show_live_status else None next_lesson = _find_next_lesson(now, lessons) if show_live_status else None week_label = "Верхняя" if week_key == "upper" else "Нижняя" day_label = _weekday_names.get(target_now.weekday(), "Сегодня") if not show_live_status: day_label = f"{day_label}, {target_now:%d.%m.%Y}" lines = [f"{week_label} неделя", day_label] if lessons: lines.append("") lines.extend(_format_lesson_line(lesson) for lesson in lessons) else: lines.append("") lines.append("На этот день расписание пустое.") if show_live_status: lines.append("") if current_lesson: if current_lesson.is_free: lines.append(f"Сейчас: {current_lesson.pair_number} пара свободна") else: lines.append(f"Сейчас: {_format_lesson_brief(current_lesson)}") else: lines.append("Сейчас: пары нет") if next_lesson: lines.append(f"Следующая: {_format_lesson_brief(next_lesson)}") else: lines.append("Следующая: сегодня нет") return "\n".join(lines) def minutes_until_break(now: datetime | None = None) -> str: """Логика расписания перекуров.""" now = _normalize_now(now) day = now.weekday() today_schedule = config.SCHEDULE.get(day, []) if not today_schedule: return "Сегодня кайфуем дома, перекур не нужен." def dt_for(t: time) -> datetime: return now.replace(hour=t.hour, minute=t.minute, second=0, microsecond=0) for start_t, end_t, br_minutes in today_schedule: start_dt = dt_for(start_t) end_dt = dt_for(end_t) if now < start_dt or (start_dt <= now < end_dt): return f"{_format_delta_ms(end_dt - now)} до перекура" if br_minutes: break_end = end_dt + timedelta(minutes=br_minutes) if end_dt <= now < break_end: return f"Идёт перекур, осталось {_format_delta_ms(break_end - now)}" return "Пары закончились, перекур только завтра." def next_break_window(now: datetime | None = None) -> tuple[datetime | None, datetime | None]: """Возвращает (начало, конец) ближайшего перекура или (None, None).""" now = _normalize_now(now) day = now.weekday() today_schedule = config.SCHEDULE.get(day, []) if not today_schedule: return None, None def dt_for(t: time) -> datetime: return now.replace(hour=t.hour, minute=t.minute, second=0, microsecond=0) for start_t, end_t, br_minutes in today_schedule: if not br_minutes: continue end_dt = dt_for(end_t) br_start = end_dt br_end = end_dt + timedelta(minutes=br_minutes) # Ближайший будущий или текущий перекур if now < br_end: return br_start, br_end return None, None # --- Авто-напоминание о перекуре --- _break_tasks: dict[int, tuple[datetime, asyncio.Task]] = {} _break_notified: dict[int, datetime] = {} async def _send_break_photo(chat_id: int, bot: Bot, start_dt: datetime) -> None: last = _break_notified.get(chat_id) if last and last == start_dt: return # Уже кидали уведомление для этого слота photo_path = config.BASE_DIR / "maket" / "vape.jpg" if not photo_path.exists(): logger.warning("vape.jpg not found, skip break photo") return try: await bot.send_photo(chat_id=chat_id, photo=FSInputFile(photo_path), caption="идем курить") _break_notified[chat_id] = start_dt except Exception: logger.exception("Failed to send break photo") async def _break_notification(chat_id: int, bot: Bot, start_dt: datetime) -> None: try: now = datetime.now(start_dt.tzinfo) delay = (start_dt - now).total_seconds() if delay > 0: await asyncio.sleep(delay) await _send_break_photo(chat_id, bot, start_dt) except asyncio.CancelledError: pass finally: stored = _break_tasks.get(chat_id) if stored and stored[0] == start_dt: _break_tasks.pop(chat_id, None) def schedule_break_notification(chat_id: int, bot: Bot, start_dt: datetime) -> None: existing = _break_tasks.get(chat_id) if existing: scheduled_start, task = existing if scheduled_start == start_dt and not task.done(): return # Уже ждём этот же слот if not task.done(): task.cancel() task = asyncio.create_task(_break_notification(chat_id, bot, start_dt)) _break_tasks[chat_id] = (start_dt, task) # --- Работа с изображениями (Макеты и Мемы) --- def load_font(size: int) -> ImageFont.FreeTypeFont: if not config.FONT_PATH.exists(): raise FileNotFoundError(f"Шрифт не найден: {config.FONT_PATH}") return ImageFont.truetype(str(config.FONT_PATH), size=size) def wrap_text_to_width(text: str, font: ImageFont.FreeTypeFont, max_width: int, draw: ImageDraw.ImageDraw, stroke_width: int) -> list[str]: words = text.split() lines, current = [], [] for word in words: tentative = " ".join(current + [word]) if current else word bbox = draw.textbbox((0, 0), tentative, font=font, stroke_width=stroke_width) if bbox[2] - bbox[0] <= max_width: current.append(word) else: if current: lines.append(" ".join(current)) current = [word] if current: lines.append(" ".join(current)) return lines def parse_up_down(text: str) -> tuple[str, str]: """Парсит формат /gen_mem Up:текст Down:текст""" up = "" down = "" # Очищаем текст от самой команды text_normalized = re.sub(r'^/\w+\s*', '', text, flags=re.IGNORECASE).replace("\n", " ") up_match = re.search(r"(?is)up:(.*?)(?=\s+down:|$)", text_normalized) down_match = re.search(r"(?is)down:(.*)$", text_normalized) if up_match: up = up_match.group(1).strip() if down_match: down = down_match.group(1).strip() # Если меток нет, по старинке: первое слово вверх, остальное вниз if not up and not down: parts = text_normalized.split(None, 1) if parts: up = parts[0].strip() if len(parts) > 1: down = parts[1].strip() return up, down def fit_text_in_box(text: str, max_width: int, max_height: int, start_size: int = 72, min_size: int = 14): draw_dummy = ImageDraw.Draw(Image.new("RGB", (1, 1))) for size in range(start_size, min_size - 1, -2): font = load_font(size) stroke_width = max(2, size // 12) lines = wrap_text_to_width(text, font, max_width, draw_dummy, stroke_width) if not lines: continue line_heights = [draw_dummy.textbbox((0, 0), ln, font=font, stroke_width=stroke_width)[3] - draw_dummy.textbbox((0, 0), ln, font=font, stroke_width=stroke_width)[1] for ln in lines] total_height = sum(line_heights) + max(0, len(lines) - 1) * max(6, size // 6) if total_height <= max_height: return font, lines, stroke_width, max(6, size // 6) font = load_font(min_size) return font, [text], max(1, min_size // 12), max(4, min_size // 6) def draw_multiline(draw: ImageDraw.ImageDraw, lines: list[str], position: tuple[int, int], font: ImageFont.FreeTypeFont, stroke_width: int, line_spacing: int, align_center: bool = False, max_width: int | None = None): x, y = position for line in lines: bbox = draw.textbbox((0, 0), line, font=font, stroke_width=stroke_width) line_width = bbox[2] - bbox[0] draw_x = x if align_center and max_width is not None: draw_x = x + (max_width - line_width) // 2 draw.text((draw_x, y), line, font=font, fill="white", stroke_width=stroke_width, stroke_fill="black") y += (bbox[3] - bbox[1]) + line_spacing def calculate_font_for_width(text: str, image_width: int, max_ratio: float = 0.09, min_size: int = 18): start_size = max(int(image_width * max_ratio), min_size) draw_dummy = ImageDraw.Draw(Image.new("RGB", (image_width, image_width))) for size in range(start_size, min_size - 1, -2): font = load_font(size) stroke_width = max(2, size // 12) wrapped = wrap_text_to_width(text, font, int(image_width * 0.95), draw_dummy, stroke_width) if wrapped: height = 0 for ln in wrapped: bbox = draw_dummy.textbbox((0, 0), ln, font=font, stroke_width=stroke_width) height += bbox[3] - bbox[1] height += max(0, len(wrapped) - 1) * max(6, size // 6) if height <= image_width * 0.35: return font, wrapped, stroke_width, max(6, size // 6) font = load_font(min_size) stroke_width = max(1, min_size // 12) wrapped = wrap_text_to_width(text, font, int(image_width * 0.95), draw_dummy, stroke_width) return font, wrapped or [text], stroke_width, max(4, min_size // 6) def generate_maket(text: str, template_id: str = "1") -> BytesIO: tpl = config.TEMPLATES.get(template_id) if not tpl or not tpl["path"].exists(): raise FileNotFoundError(f"Шаблон {template_id} не найден") with Image.open(tpl["path"]).convert("RGB") as img: anchor = tpl["anchor"] if "box_size" in tpl: max_w, max_h = tpl["box_size"] else: padding = tpl.get("padding", 0) max_w, max_h = img.width - anchor[0] - padding, img.height - anchor[1] - padding font, lines, stroke, spacing = fit_text_in_box(text, max_w, max_h) draw = ImageDraw.Draw(img) y_offset = anchor[1] for line in lines: draw.text((anchor[0], y_offset), line, font=font, fill="white", stroke_width=stroke, stroke_fill="black") bbox = draw.textbbox((0, 0), line, font=font, stroke_width=stroke) y_offset += (bbox[3] - bbox[1]) + spacing output = BytesIO() img.save(output, format="PNG") output.seek(0) return output def generate_meme(img: Image.Image, top_text: str, bottom_text: str) -> BytesIO: img = img.convert("RGB") draw = ImageDraw.Draw(img) width, height = img.size margin_y = int(height * 0.04) if top_text: font, lines, stroke_width, spacing = calculate_font_for_width(top_text, width) draw_multiline( draw, lines, (int(width * 0.025), margin_y), font, stroke_width, spacing, align_center=True, max_width=int(width * 0.95), ) if bottom_text: font, lines, stroke_width, spacing = calculate_font_for_width(bottom_text, width) block_height = 0 for ln in lines: bbox = draw.textbbox((0, 0), ln, font=font, stroke_width=stroke_width) block_height += bbox[3] - bbox[1] block_height += max(0, len(lines) - 1) * spacing start_y = height - margin_y - block_height draw_multiline( draw, lines, (int(width * 0.025), start_y), font, stroke_width, spacing, align_center=True, max_width=int(width * 0.95), ) output = BytesIO() img.save(output, format="JPEG", quality=95) output.seek(0) return output def minutes_until_next_lesson(now: datetime | None = None) -> str: if now is None: now = datetime.now(ZoneInfo(config.DEFAULT_TZ)) elif now.tzinfo is None: now = now.replace(tzinfo=ZoneInfo(config.DEFAULT_TZ)) lesson_days = [1, 3] # вторник и четверг lesson_start = time(9, 50) lesson_end = time(11, 25) day = now.weekday() def lesson_start_for(day_offset: int) -> datetime: target = now + timedelta(days=day_offset) return target.replace(hour=lesson_start.hour, minute=lesson_start.minute, second=0, microsecond=0) def lesson_end_today() -> datetime: return now.replace(hour=lesson_end.hour, minute=lesson_end.minute, second=0, microsecond=0) if day in lesson_days: start_today = lesson_start_for(0) end_today = lesson_end_today() if start_today <= now < end_today: for i in range(1, 8): next_day = (day + i) % 7 if next_day in lesson_days: next_dt = lesson_start_for(i) break delta = next_dt - now total_seconds = int(delta.total_seconds()) days = total_seconds // 86400 hours = (total_seconds % 86400) // 3600 minutes = (total_seconds % 3600) // 60 percent = random.randint(0, 100) return ( f"Сегодня твой мозг сгнил на {percent}%\n" f"Мозг начнет догнивать через {days} дн {hours} ч {minutes} мин" ) for i in range(0, 8): next_day = (day + i) % 7 if next_day in lesson_days: candidate = lesson_start_for(i) if candidate > now: delta = candidate - now break total_seconds = int(delta.total_seconds()) days = total_seconds // 86400 hours = (total_seconds % 86400) // 3600 minutes = (total_seconds % 3600) // 60 return f"Твой мозг окончательно сгниет, а очко порвется через: {days} дн {hours} ч {minutes} мин" def _prepare_magnit_query(value: str) -> str: value = re.sub(r"(?<=[a-zа-яё])(?=[A-ZА-ЯЁ])", " ", value) value = re.sub(r"[_-]+", " ", value) return re.sub(r"\s+", " ", value).strip() def _normalize_magnit_product(entry) -> dict[str, str]: if isinstance(entry, str): raw_query = entry.strip() if not raw_query: raise ValueError("Пустой query в MAGNIT_PRODUCTS.") return { "query": _prepare_magnit_query(raw_query), "label": raw_query, "store_code": config.MAGNIT_STORE_CODE, } if isinstance(entry, dict): raw_query = str(entry.get("query", "")).strip() if not raw_query: raise ValueError("У товара в MAGNIT_PRODUCTS отсутствует query.") return { "query": _prepare_magnit_query(raw_query), "label": str(entry.get("label") or raw_query).strip() or raw_query, "store_code": str(entry.get("store_code") or config.MAGNIT_STORE_CODE).strip(), } raise ValueError("MAGNIT_PRODUCTS должен содержать строки или словари.") def _pick_magnit_items(data: dict) -> list[dict]: items = data.get("items") if isinstance(items, list): return [item for item in items if isinstance(item, dict)] best_match = data.get("best_match") if isinstance(best_match, dict): return [best_match] return [] def _format_money(value) -> str | None: if value is None: return None amount = f"{float(value):.2f}".rstrip("0").rstrip(".") return f"{amount} ₽" def _format_magnit_price(item: dict) -> str: new_price = _format_money(item.get("price")) old_price = _format_money(item.get("old_price")) if old_price and new_price and item.get("old_price") and item.get("old_price") != item.get("price"): return f"{old_price} -> {new_price}" return new_price or "нет цены" def _format_magnit_quantity(value) -> str: if value is None: return "нет данных" return f"{value} шт." async def _send_lines_in_chunks(message: Message, lines: list[str], limit: int = 4000) -> None: if not lines: await message.answer("Товары не найдены.") return chunk: list[str] = [] chunk_len = 0 for line in lines: extra_len = len(line) + (1 if chunk else 0) if chunk and chunk_len + extra_len > limit: await message.answer("\n".join(chunk)) chunk = [line] chunk_len = len(line) continue chunk.append(line) chunk_len += extra_len if chunk: await message.answer("\n".join(chunk)) # --- Игровая логика (/penis) --- def play_penis(user_id: int, user_name: str | None = None, now: datetime | None = None) -> tuple[bool, str]: if now is None: now = datetime.now(ZoneInfo(config.DEFAULT_TZ)) elif now.tzinfo is None: now = now.replace(tzinfo=ZoneInfo(config.DEFAULT_TZ)) now_ts = int(now.timestamp()) display_name = user_name.strip() if user_name and user_name.strip() else f"user_{user_id}" try: with sqlite3.connect(config.PENIS_DB_PATH) as conn: conn.row_factory = sqlite3.Row row = conn.execute( "SELECT length, last_used_ts, display_name FROM penis_stats WHERE user_id = ?", (user_id,), ).fetchone() if row is None: current_length = config.PENIS_START_LENGTH last_used_ts = None else: current_length = float(row["length"]) last_used_ts = row["last_used_ts"] stored_name = row["display_name"] if not user_name and isinstance(stored_name, str) and stored_name.strip(): display_name = stored_name.strip() if last_used_ts is not None: next_ts = int(last_used_ts) + config.PENIS_COOLDOWN_SECONDS if now_ts < next_ts: if row is not None and row["display_name"] != display_name: conn.execute( "UPDATE penis_stats SET display_name = ? WHERE user_id = ?", (display_name, user_id), ) left_seconds = next_ts - now_ts hours = left_seconds // 3600 minutes = (left_seconds % 3600) // 60 conn.commit() return False, f"Сегодня уже кидал. Попробуй через {hours} ч {minutes} мин." delta = round(random.uniform(config.PENIS_MIN_DELTA, config.PENIS_MAX_DELTA), 1) sign = random.choice([-1, 1]) new_length = round(max(0.1, current_length + sign * delta), 1) if row is None: conn.execute( "INSERT INTO penis_stats(user_id, display_name, length, last_used_ts) VALUES(?, ?, ?, ?)", (user_id, display_name, new_length, now_ts), ) else: conn.execute( "UPDATE penis_stats SET display_name = ?, length = ?, last_used_ts = ? WHERE user_id = ?", (display_name, new_length, now_ts, user_id), ) conn.commit() except sqlite3.Error: logger.exception("Failed to process /penis") return False, "Ошибка БД. Попробуй позже." sign_text = "+" if sign > 0 else "-" return ( True, f"📏 Изменение: {sign_text}{delta:.1f} см\n" f"Текущая длина: {new_length:.1f} см\n" "Следующая попытка через 24 часа.", ) def format_penis_user_name(user: types.User) -> str: if user.username: return f"@{user.username}" if user.full_name: return user.full_name return f"user_{user.id}" def build_penis_top(limit: int = config.PENIS_TOP_LIMIT) -> str: try: with sqlite3.connect(config.PENIS_DB_PATH) as conn: conn.row_factory = sqlite3.Row rows = conn.execute( """ SELECT user_id, display_name, length FROM penis_stats ORDER BY length DESC, user_id ASC LIMIT ? """, (limit,), ).fetchall() except sqlite3.Error: logger.exception("Failed to build /top_penis") return "Не удалось загрузить топ. Ошибка БД." if not rows: return "🏆 Топ пуст. Сначала пусть кто-то крутнет /penis." medals = {1: "🥇", 2: "🥈", 3: "🥉"} lines = ["🏆 Топ по длине:"] for index, row in enumerate(rows, start=1): user_id = int(row["user_id"]) length = float(row["length"]) name = row["display_name"] if isinstance(row["display_name"], str) else f"user_{user_id}" name = name.strip() if name.strip() else f"user_{user_id}" prefix = medals.get(index, f"{index}.") lines.append(f"{prefix} {name} — {length:.1f} см") return "\n".join(lines) def _load_polychaetsi_stats() -> dict: path = config.POLYCHAETSI_STATS_PATH if not path.exists(): return {"users": {}} try: with path.open("r", encoding="utf-8") as fh: data = json.load(fh) except (OSError, json.JSONDecodeError): logger.exception("Failed to load polychaetsi stats") return {"users": {}} if not isinstance(data, dict): return {"users": {}} users = data.get("users") if not isinstance(users, dict): data["users"] = {} return data def _save_polychaetsi_stats(data: dict) -> None: path = config.POLYCHAETSI_STATS_PATH path.parent.mkdir(parents=True, exist_ok=True) tmp_path = path.with_name(f"{path.name}.tmp") with tmp_path.open("w", encoding="utf-8") as fh: json.dump(data, fh, ensure_ascii=False, indent=2) tmp_path.replace(path) def _damerau_levenshtein_distance(source: str, target: str, max_distance: int = 2) -> int: if source == target: return 0 if abs(len(source) - len(target)) > max_distance: return max_distance + 1 previous_previous: list[int] | None = None previous = list(range(len(target) + 1)) for i, source_char in enumerate(source, start=1): current = [i] + [0] * len(target) row_min = current[0] for j, target_char in enumerate(target, start=1): cost = 0 if source_char == target_char else 1 current[j] = min( previous[j] + 1, current[j - 1] + 1, previous[j - 1] + cost, ) if ( previous_previous is not None and i > 1 and j > 1 and source_char == target[j - 2] and source[i - 2] == target_char ): current[j] = min(current[j], previous_previous[j - 2] + 1) row_min = min(row_min, current[j]) if row_min > max_distance: return max_distance + 1 previous_previous, previous = previous, current return previous[-1] def _normalize_polychaetsi_word(word: str) -> str: normalized = word.lower().replace("ё", "е") # Подтягиваем самые частые ошибки в возвратном окончании. return re.sub(r"(ть?ся|ть?са|цца|ца)$", "тся", normalized) def is_polychaetsi_variant(word: str) -> bool: normalized = _normalize_polychaetsi_word(word) target = config.POLYCHAETSI_TARGET if normalized == target: return True if not normalized or normalized[0] != "п": return False if len(normalized) < len(target) - 1 or len(normalized) > len(target) + 1: return False if "ч" not in normalized or "а" not in normalized or "е" not in normalized: return False if not normalized.endswith(("я", "а")): return False return _damerau_levenshtein_distance(normalized, target, max_distance=2) <= 2 def count_polychaetsi_mentions(text: str) -> int: return sum(1 for word in _polychaetsi_word_re.findall(text) if is_polychaetsi_variant(word)) def register_polychaetsi_hits(user: types.User, hits: int) -> int: display_name = format_penis_user_name(user) with _polychaetsi_lock: data = _load_polychaetsi_stats() users = data.setdefault("users", {}) user_key = str(user.id) entry = users.get(user_key) if not isinstance(entry, dict): entry = {"display_name": display_name, "count": 0} stored_count = entry.get("count", 0) try: count = int(stored_count) except (TypeError, ValueError): count = 0 count += hits entry["display_name"] = display_name entry["count"] = count users[user_key] = entry _save_polychaetsi_stats(data) return count def build_polychaetsi_top(limit: int = config.POLYCHAETSI_TOP_LIMIT) -> str: with _polychaetsi_lock: data = _load_polychaetsi_stats() users = data.get("users", {}) if not isinstance(users, dict): return "ТОП получается:\nпока пусто." entries: list[tuple[int, str, int]] = [] for user_key, raw_entry in users.items(): if not isinstance(raw_entry, dict): continue try: count = int(raw_entry.get("count", 0)) except (TypeError, ValueError): continue if count <= 0: continue try: numeric_user_id = int(user_key) except (TypeError, ValueError): numeric_user_id = 0 name = str(raw_entry.get("display_name") or f"user_{user_key}").strip() or f"user_{user_key}" entries.append((count, name, numeric_user_id)) if not entries: return "ТОП получается:\nпока пусто." entries.sort(key=lambda item: (-item[0], item[1].lower(), item[2])) medals = {1: "🥇", 2: "🥈", 3: "🥉"} lines = ["ТОП получается:"] for index, (count, name, _) in enumerate(entries[:limit], start=1): prefix = medals.get(index, f"{index}.") lines.append(f"{prefix} {name} — {count}") return "\n".join(lines) # --- Хендлеры aiogram --- async def handle_help(message: Message): await message.answer(config.get_hint_text(templates_list_text())) async def handle_start(message: Message): await message.answer("Соси") async def handle_schedule_cmd(message: Message): now = datetime.now(ZoneInfo(config.DEFAULT_TZ)) raw_text = message.text or "" parts = raw_text.split(maxsplit=1) requested_weekday = None if len(parts) > 1: requested_weekday = _parse_weekday_arg(parts[1].split(maxsplit=1)[0]) if requested_weekday is None: await message.answer( "Не понял день недели. Примеры: /schedule mon, /schedule fri, /schedule пн, /schedule чт." ) return await message.answer(build_schedule_text(now, requested_weekday)) async def handle_break_cmd(message: Message): now = datetime.now(ZoneInfo(config.DEFAULT_TZ)) await message.answer(f"{minutes_until_break(now)}\n{next_lesson_text(now)}") # Ставим будильник на ближайший перекур для этого чата start_dt, _ = next_break_window(now) if start_dt: schedule_break_notification(message.chat.id, message.bot, start_dt) async def handle_penis_cmd(message: Message): if not message.from_user: await message.answer("Команда доступна только пользователям.") return display_name = format_penis_user_name(message.from_user) await message.answer(play_penis(message.from_user.id, display_name)[1]) async def handle_top_penis_cmd(message: Message): await message.answer(build_penis_top()) async def handle_polychaetsi_cmd(message: Message): await message.answer(build_polychaetsi_top()) async def handle_size_cmd(message: Message): size = round(random.uniform(0.1, 35.0), 1) if size < 5: status = "СВОи" elif size < 10: status = "Неплохой" elif size < 18: status = "Шлюхи ахуевают при виде тебя" elif size < 25: status = "Хуек с коробок" elif size < 32: status = "Пиздабол" else: status = "Говноед" await message.answer(f"📏 Твой хуек: {size} см\nТы: {status}") async def handle_zvetok_cmd(message: Message): await message.answer(minutes_until_next_lesson()) async def handle_prices_cmd(message: Message): try: products = [_normalize_magnit_product(entry) for entry in config.MAGNIT_PRODUCTS] except ValueError as exc: await message.answer(f"Проверь MAGNIT_PRODUCTS в config.py: {exc}") return if not products: await message.answer("Список MAGNIT_PRODUCTS пуст. Добавь товары в config.py.") return async def fetch_product(product_cfg: dict[str, str]) -> list[str]: def fetch(): return requests.get( f"{config.MAGNIT_API_BASE_URL.rstrip('/')}/price", params={ "store_code": product_cfg["store_code"], "query": product_cfg["query"], }, timeout=config.MAGNIT_REQUEST_TIMEOUT, ) label = product_cfg["label"] or product_cfg["query"] or "товар" try: resp = await asyncio.to_thread(fetch) if resp.status_code != 200: logger.warning("Magnit API returned %s for %s", resp.status_code, product_cfg["query"]) return [f"{label}, ошибка API ({resp.status_code}), -"] data = resp.json() if not isinstance(data, dict): return [f"{label}, битый ответ API, -"] items = _pick_magnit_items(data) if not items: return [f"{label}, товар не найден, -"] lines = [] for item in items: name = str(item.get("name") or label).strip() or label price = _format_magnit_price(item) quantity = _format_magnit_quantity(item.get("quantity")) lines.append(f"{name}, {price}, {quantity}") return lines except requests.RequestException: logger.exception("Failed to fetch Magnit product: %s", product_cfg["query"]) return [f"{label}, ошибка запроса, -"] except ValueError: logger.exception("Magnit API returned invalid JSON for %s", product_cfg["query"]) return [f"{label}, битый JSON, -"] result_groups = await asyncio.gather(*(fetch_product(product) for product in products)) lines = [line for group in result_groups for line in group] await _send_lines_in_chunks(message, lines) async def handle_fetch_cmd(message: Message): await handle_prices_cmd(message) async def handle_svodka_cmd(message: Message): data = await asyncio.to_thread(get_military_data, "creamy_caprice", 5) if data["status"] == "success": response = ( f"в период с {data['dates']} хохлы в среднем проебывали по {data['pace']} км² в сутки, " f"а за день ВС РФ лишала чубатых в среднем по {data['total']} км²" ) await message.answer(response) else: await message.answer(f"Хохлы ответят за это...: {data['message']}") async def handle_ebalnik_cmd(message: Message): chat_id = message.chat.id if chat_id in _autoreply_disabled_chats: _autoreply_disabled_chats.remove(chat_id) await message.reply("Автоответы вернул, бот снова влезает в чат.") else: _autoreply_disabled_chats.add(chat_id) await message.reply("Автоответы срезал, бот теперь молчит в этом чате.") async def handle_maket_cmd(message: Message): parts = message.text.split(maxsplit=2) if len(parts) == 1: await message.reply(f"Доступные макеты: {templates_list_text()}\nФормат: /maket «id» «текст»") return if len(parts) < 3: await message.reply( f"Формат: /maket «id» «текст»\nДоступные макеты: {templates_list_text()}\n\n" + config.get_hint_text(templates_list_text()) ) return template_id = parts[1] if template_id not in config.TEMPLATES: await message.reply(f"Такого макета нет. Доступные: {templates_list_text()}") return text = parts[2].strip() if not text: await message.reply("Нужен текст после номера макета.") return try: output = await asyncio.to_thread(generate_maket, text, template_id) await message.reply_photo(BufferedInputFile(output.getvalue(), filename="maket.png")) except Exception as exc: logger.exception("Failed to generate maket") await message.reply(f"Не получилось: {exc}") async def handle_gen_mem(message: Message, bot: Bot): photo_msg = message if message.photo else (message.reply_to_message if message.reply_to_message and message.reply_to_message.photo else None) if not photo_msg: await message.reply("Еблан, /gen_mem Up: Down:") return full_text = message.text or "" top_text, bottom_text = parse_up_down(full_text) if not top_text and not bottom_text: await message.reply("Еблан, /gen_mem Up: Down:") return try: file = await bot.get_file(photo_msg.photo[-1].file_id) buf = BytesIO() await bot.download_file(file.file_path, destination=buf) buf.seek(0) img = Image.open(buf) output = await asyncio.to_thread(generate_meme, img, top_text, bottom_text) except Exception as exc: logger.exception("Failed to render meme") await message.reply(f"Не получилось сделать мем: {exc}") return await message.reply_photo(BufferedInputFile(output.getvalue(), filename="meme.jpg")) _last_autoreply_ts = 0 _autoreply_disabled_chats: set[int] = set() async def handle_keywords(message: Message): global _last_autoreply_ts if not message.text or message.text.startswith('/'): return text_lower = message.text.lower() chat_id = message.chat.id user_name = (message.from_user.first_name or "кент") if message.from_user else "кент" if any(re.search(rf'\b{word}\b', text_lower) for word in config.KEYWORDS): audio_path = config.BASE_DIR / "maket" / "Radiohead - Creep.mp3" if audio_path.exists(): await message.answer_audio( audio=FSInputFile(audio_path), caption="I'm a creep, I'm a weirdo...", ) return polychaetsi_hits = count_polychaetsi_mentions(text_lower) if polychaetsi_hits > 0 and message.from_user: total_hits = await asyncio.to_thread(register_polychaetsi_hits, message.from_user, polychaetsi_hits) photo_path = config.POLYCHAETSI_IMAGE_PATH if photo_path.exists(): await message.answer_photo( photo=FSInputFile(photo_path), caption=f"x{total_hits}", ) else: logger.warning("cvetok.jpg not found, fallback to text response") await message.answer(f"x{total_hits}") return # Записываем каждое обычное сообщение в историю чата await asyncio.to_thread(push_message, chat_id, "user", user_name, message.text) if chat_id in _autoreply_disabled_chats: return import time as _time now = _time.time() triggered = any(trigger in text_lower for trigger in config.AUTOREPLY_TRIGGERS) should_reply = triggered or ( random.random() < config.AUTOREPLY_CHANCE and (now - _last_autoreply_ts) > config.AUTOREPLY_COOLDOWN ) if should_reply: try: await message.bot.send_chat_action(chat_id=chat_id, action="typing") response = await generate_autoreply(chat_id, message.text, user_name) if response: await asyncio.to_thread(push_message, chat_id, "assistant", "бот", response) await message.reply(response, parse_mode=None) _last_autoreply_ts = now except Exception: logger.exception("Autoreply failed") async def handle_penis_casino_cmd(message: Message): if not message.from_user: await message.answer("Команда доступна только пользователям.") return parts = message.text.split(maxsplit=1) if len(parts) < 2: await message.reply("Формат: /penis_casino <ставка>\nПример: /penis_casino 0.5") return try: bet = round(float(parts[1].replace(",", ".")), 1) except ValueError: await message.reply("Ставка должна быть числом, фраер. Пример: /penis_casino 0.5") return result = await asyncio.to_thread(play_casino, message.from_user.id, bet) await message.reply(result, parse_mode=None) async def handle_gadanie_cmd(message: Message): parts = message.text.split(maxsplit=1) if len(parts) < 2 or not parts[1].strip(): await message.reply("Напиши на что гадать, кент. Пример: /gadanie любовь") return topic = parts[1].strip() await message.bot.send_chat_action(chat_id=message.chat.id, action="typing") try: fortune_text, up_text, down_text = await generate_fortune(topic) await message.reply(f"🔮 Гадание на «{topic}»:\n\n{fortune_text}", parse_mode=None) photo_path = config.BASE_DIR / "maket" / "vape.jpg" if photo_path.exists(): img = Image.open(photo_path).convert("RGB") meme_output = await asyncio.to_thread(generate_meme, img, up_text, down_text) await message.reply_photo( BufferedInputFile(meme_output.getvalue(), filename="gadanie.jpg") ) except Exception: logger.exception("Gadanie failed") await message.reply("Бля, карты рассыпались, попробуй позже.") # --- Запуск --- def main(): token = os.getenv("BOT_TOKEN") if not token: raise RuntimeError("BOT_TOKEN не задан!") init_penis_db() init_polychaetsi_storage() bot = Bot(token=token, default=DefaultBotProperties(parse_mode=ParseMode.HTML)) dp = Dispatcher() # Регистрация хендлеров dp.message.register(handle_start, CommandStart()) dp.message.register(handle_help, Command("help")) dp.message.register(handle_schedule_cmd, Command("schedule")) dp.message.register(handle_break_cmd, Command("break")) dp.message.register(handle_maket_cmd, Command("maket")) dp.message.register(handle_size_cmd, Command("size")) dp.message.register(handle_penis_cmd, Command("penis")) dp.message.register(handle_top_penis_cmd, Command("top_penis")) dp.message.register(handle_polychaetsi_cmd, Command("polychaetsi")) dp.message.register(handle_prices_cmd, Command("prices")) dp.message.register(handle_fetch_cmd, Command("fetch")) dp.message.register(handle_zvetok_cmd, Command("zvetok")) dp.message.register(handle_talk, Command("talk")) dp.message.register(handle_svodka_cmd, Command("svodka")) dp.message.register(handle_ebalnik_cmd, Command("ebalnik")) dp.message.register(handle_penis_casino_cmd, Command("penis_casino")) dp.message.register(handle_gadanie_cmd, Command("gadanie")) dp.message.register(handle_gen_mem, Command("gen_mem")) dp.message.register(handle_keywords, F.text) async def on_startup(bot: Bot): try: commands = [ BotCommand(command="start", description="Приветствие"), BotCommand(command="help", description="Подсказки по командам"), BotCommand(command="schedule", description="расписание пар"), BotCommand(command="maket", description="Вписать текст в макет 1"), BotCommand(command="gen_mem", description="Мем из фото (ответом)"), BotCommand(command="break", description="Сколько минут до перекура"), BotCommand(command="size", description="хуй"), BotCommand(command="penis", description="миниигра 1 раз в 24ч"), BotCommand(command="top_penis", description="топ по длине"), BotCommand(command="polychaetsi", description="топ по получается"), BotCommand(command="prices", description="цены и остатки Магнита"), BotCommand(command="fetch", description="alias для /prices"), BotCommand(command="zvetok", description="Цветянский бля"), BotCommand(command="talk", description="Побазарить"), BotCommand(command="ebalnik", description="вкл/выкл автответов"), BotCommand(command="penis_casino", description="казино на размер"), BotCommand(command="gadanie", description="гадание на фене"), BotCommand(command="svodka", description="СВО: итоги"), ] scopes = ( BotCommandScopeDefault(), BotCommandScopeAllPrivateChats(), BotCommandScopeAllGroupChats(), ) for scope in scopes: await bot.set_my_commands(commands, scope=scope) except Exception: logger.warning("Не удалось выставить команды, но бот продолжит работу.") dp.startup.register(on_startup) logger.info("Бот запущен...") asyncio.run(dp.start_polling(bot, allowed_updates=dp.resolve_used_update_types())) if __name__ == "__main__": main()