import asyncio import logging import os import re import requests from io import BytesIO from pathlib import Path from textwrap import wrap import random from aiogram import Bot, Dispatcher, F, types from aiogram.enums import ParseMode from aiogram.filters import Command, CommandStart from aiogram.types import BufferedInputFile, Message, BotCommand, FSInputFile from aiogram.client.default import DefaultBotProperties from PIL import Image, ImageDraw, ImageFont from datetime import datetime, time, timedelta from zoneinfo import ZoneInfo BASE_DIR = Path(__file__).parent FONT_PATH = BASE_DIR / "impact.ttf" DEFAULT_TZ = os.getenv("TZ", "Europe/Moscow") TEMPLATES = { "1": { "path": BASE_DIR / "maket" / "1.png", "anchor": (104, 240), "padding": 20, # right/bottom padding from edges "label": "Владимир Владимирович Путин самый лучший президент", }, } def templates_list_text() -> str: parts = [] for tid in sorted(TEMPLATES.keys()): label = TEMPLATES[tid].get("label", "") if label: parts.append(f"{tid} — {label}") else: parts.append(tid) return ", ".join(parts) if parts else "нет макетов" # ---- Schedule & breaks ---- def make_time(h: int, m: int) -> time: return time(hour=h, minute=m) # Schedule format: list of (lesson_start, lesson_end, break_minutes_after) SCHEDULE: dict[int, list[tuple[time, time, int | None]]] = { 0: [ # Monday (make_time(9, 50), make_time(10, 35), 5), (make_time(10, 40), make_time(11, 25), 30), (make_time(11, 55), make_time(12, 40), 5), (make_time(12, 45), make_time(13, 30), 15), (make_time(13, 45), make_time(14, 30), 5), (make_time(14, 35), make_time(15, 20), 30), (make_time(15, 50), make_time(16, 35), 5), (make_time(16, 40), make_time(17, 25), None), ], 1: [ # Tuesday (make_time(9, 50), make_time(10, 35), 5), (make_time(10, 40), make_time(11, 25), 30), (make_time(11, 55), make_time(12, 40), 5), (make_time(12, 45), make_time(13, 30), 15), (make_time(13, 45), make_time(14, 30), 5), (make_time(14, 35), make_time(15, 20), None), ], 2: [], # Wednesday — кайфуем дома 3: [ # Thursday (make_time(8, 0), make_time(8, 45), 5), (make_time(8, 50), make_time(9, 35), 15), (make_time(9, 50), make_time(10, 35), 5), (make_time(10, 40), make_time(11, 25), 30), (make_time(11, 55), make_time(12, 40), 5), (make_time(12, 45), make_time(13, 30), 15), (make_time(13, 45), make_time(14, 30), 5), (make_time(14, 35), make_time(15, 20), 30), (make_time(15, 50), make_time(16, 35), 5), (make_time(16, 40), make_time(17, 25), None), ], 4: [ # Friday (make_time(9, 50), make_time(10, 35), 5), (make_time(10, 40), make_time(11, 25), 30), (make_time(11, 55), make_time(12, 40), 5), (make_time(12, 45), make_time(13, 30), 15), (make_time(13, 45), make_time(14, 30), 5), (make_time(14, 35), make_time(15, 20), None), ], 5: [], # Saturday — выходной 6: [], # Sunday — выходной } def minutes_until_break(now: datetime | None = None) -> str: if now is None: now = datetime.now(ZoneInfo(DEFAULT_TZ)) elif now.tzinfo is None: now = now.replace(tzinfo=ZoneInfo(DEFAULT_TZ)) day = now.weekday() today_schedule = 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 we're before the lesson if now < start_dt: minutes = int((end_dt - now).total_seconds() // 60) return f"{minutes} минут до перекура" # If we're inside the lesson if start_dt <= now < end_dt: minutes = int((end_dt - now).total_seconds() // 60) return f"{minutes} минут до перекура" # If we're inside the break after this lesson if br_minutes: break_end = end_dt + timedelta(minutes=br_minutes) if end_dt <= now < break_end: remaining = int((break_end - now).total_seconds() // 60) return f"Идёт перекур, осталось {remaining} минут" return "Пары закончились, перекур только завтра." HINT_TEXT = ( "Команды:\n" f"• /maket «id» «текст» — доступные: {templates_list_text()} (1: 737x414, точка 104,240).\n" "• /gen_mem Up:верх Down:низ — отправь с фото (как подпись) или ответом на фото.\n" "• /break — узнать, сколько минут до перекура.\n" "• /prices — старый способ цен.\n" "• /fetch — свежие цены с x-server.\n" "Подсказка: в группах команда должна быть слитно с ботом (/gen_mem@username)." ) logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) def load_font(size: int) -> ImageFont.FreeTypeFont: if not FONT_PATH.exists(): raise FileNotFoundError("impact.ttf not found. Place the font in the project root.") return ImageFont.truetype(str(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() if not words: return [] lines: list[str] = [] current: list[str] = [] 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 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", (max_width, max_height))) 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 = [] for ln in lines: bbox = draw_dummy.textbbox((0, 0), ln, font=font, stroke_width=stroke_width) line_heights.append(bbox[3] - bbox[1]) 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) # Fallback to minimal size font = load_font(min_size) stroke_width = max(1, min_size // 12) lines = wrap_text_to_width(text, font, max_width, draw_dummy, stroke_width) return font, lines or [text], stroke_width, 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 generate_maket(text: str, template_id: str = "1") -> BytesIO: tpl = TEMPLATES.get(template_id) if not tpl: raise FileNotFoundError(f"Template {template_id} is missing") template_path = tpl["path"] anchor = tpl["anchor"] padding = tpl["padding"] if not template_path.exists(): raise FileNotFoundError(f"Template file missing: {template_path}") with Image.open(template_path).convert("RGB") as img: max_width = img.width - anchor[0] - padding max_height = img.height - anchor[1] - padding font, lines, stroke_width, spacing = fit_text_in_box(text, max_width, max_height, start_size=72) draw = ImageDraw.Draw(img) draw_multiline(draw, lines, anchor, font, stroke_width, spacing, align_center=False) output = BytesIO() img.save(output, format="PNG") output.seek(0) return output 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: # heuristic to keep blocks from overlapping return font, wrapped, stroke_width, max(6, size // 6) font = load_font(min_size) stroke_width = max(1, min_size // 12) return font, wrap_text_to_width(text, font, int(image_width * 0.95), draw_dummy, stroke_width), stroke_width, max(4, min_size // 6) 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 line_heights = [] for ln in lines: bbox = draw.textbbox((0, 0), ln, font=font, stroke_width=stroke_width) h = bbox[3] - bbox[1] line_heights.append(h) block_height += h 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 parse_up_down(text: str) -> tuple[str, str]: up = "" down = "" # Normalize line breaks text_normalized = text.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 minutes_until_next_lesson(now: datetime | None = None) -> str: import random if now is None: now = datetime.now(ZoneInfo(DEFAULT_TZ)) elif now.tzinfo is None: now = now.replace(tzinfo=ZoneInfo(DEFAULT_TZ)) LESSON_DAYS = [1, 3] # 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} мин" async def handle_start(message: Message): await message.answer("Соси") async def handle_help(message: Message): await message.answer(HINT_TEXT) async def handle_break(message: Message): await message.answer(minutes_until_break()) async def handle_maket(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" + HINT_TEXT) return template_id = parts[1] if template_id not in 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) except Exception as exc: # pragma: no cover - user feedback path logger.exception("Failed to generate maket") await message.reply(f"Не получилось: {exc}") return await message.reply_photo(BufferedInputFile(output.getvalue(), filename="maket.png")) async def handle_gen_mem(message: Message, bot: Bot): photo_message = None if message.photo: photo_message = message elif message.reply_to_message and message.reply_to_message.photo: photo_message = message.reply_to_message if not photo_message: await message.reply("Еблан, /gen_mem Up: Down:") return top_text, bottom_text = parse_up_down(message.text or "") if not top_text and not bottom_text: await message.reply("Еблан, /gen_mem Up: Down:") return photo = photo_message.photo[-1] try: file = await bot.get_file(photo.file_id) buf = BytesIO() await bot.download_file(file.file_path, destination=buf) buf.seek(0) img = Image.open(buf) except Exception as exc: # pragma: no cover logger.exception("Failed to download photo") await message.reply(f"Не могу скачать фото: {exc}") return try: output = await asyncio.to_thread(generate_meme, img, top_text, bottom_text) except Exception as exc: # pragma: no cover logger.exception("Failed to render meme") await message.reply(f"Не получилось сделать мем: {exc}") return await message.reply_photo(BufferedInputFile(output.getvalue(), filename="meme.jpg")) async def handle_size(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" f"Ты: {status}" ) KEYWORDS = {'крип', 'радиохэд', 'creep', 'radiohead'} # I'M CREEP async def handle_keywords(message: types.Message): if message.text and message.text.startswith('/'): return text = message.text.lower() if any(re.search(rf'\b{word}\b', text) for word in KEYWORDS): audio_path = 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..." ) else: logger.warning(f"Файл не найден по пути: {audio_path}") async def handle_prices(message: Message): def fetch(): return requests.get( "https://bebrik.xyz/magnit/products", headers={"X-Token": "sisi"}, timeout=10, ) try: resp = await asyncio.to_thread(fetch) if resp.status_code != 200: await message.answer("пошел нахуй") return data = resp.json() if not data: await message.answer("иди нахуй") return text = "смерть в 25:\n\n" for item in data: name = item.get("name") price = item.get("price") text += f"{name} — {price} ₽\n" await message.answer(text[:4000]) except Exception: await message.answer("unable to connect x-server гандон") async def handle_fetch(message: Message): def fetch(): return requests.get( "https://mirror.porno4free.ru/magnit/magnit/prices/all", headers={"x-token": "secret_token"}, timeout=10, ) try: resp = await asyncio.to_thread(fetch) if resp.status_code != 200: await message.reply("пошел нахуй") return data = resp.json() if not isinstance(data, dict) or not data: await message.reply("иди нахуй") return lines = ["смерть в 25:"] # сортируем категории для стабильного вывода for group in sorted(data.keys()): items = data.get(group) lines.append(f"\n{group}:") if not isinstance(items, list): continue for item in items: name = item.get("name") price = item.get("price") if not name or price is None: continue lines.append(f"{name} — {price} ₽") text = "\n".join(lines) await message.answer(text[:4000]) except Exception: await message.answer("unable to connect x-server гандон") async def handle_zvetok(message: Message): result = minutes_until_next_lesson() await message.answer(result) def main(): token = os.getenv("BOT_TOKEN") if not token: raise RuntimeError("BOT_TOKEN env var is required") bot = Bot(token=token, default=DefaultBotProperties(parse_mode=ParseMode.HTML)) dp = Dispatcher() async def on_startup(bot: Bot): try: await bot.set_my_commands( [ BotCommand(command="start", description="Приветствие"), BotCommand(command="help", description="Подсказки по командам"), BotCommand(command="maket", description="Вписать текст в макет 1"), BotCommand(command="gen_mem", description="Мем из фото (ответом)"), BotCommand(command="break", description="Сколько минут до перекура"), BotCommand(command="size", description="хуй"), BotCommand(command="prices", description="цены x-server"), BotCommand(command="fetch", description="цены (новый источник)"), BotCommand(command="zvetok", description="Цветянский бля"), ] ) except Exception: # noqa: WPS440 logger.warning("Не удалось выставить команды, но бот продолжит работу.") dp.startup.register(on_startup) dp.message.register(handle_start, CommandStart()) dp.message.register(handle_help, Command("help")) dp.message.register(handle_break, Command("break")) dp.message.register(handle_maket, Command("maket")) dp.message.register(handle_size, Command("size")) dp.message.register(handle_prices, Command("prices")) dp.message.register(handle_fetch, Command("fetch")) dp.message.register(handle_zvetok, Command("zvetok")) async def handle_gen_mem_cmd(message: Message, bot: Bot): await handle_gen_mem(message, bot) dp.message.register(handle_gen_mem_cmd, Command("gen_mem")) dp.message.register(handle_keywords, F.text) logger.info("Bot is starting...") asyncio.run(dp.start_polling(bot, allowed_updates=dp.resolve_used_update_types())) if __name__ == "__main__": main()