forked from zovos/bot_tg
490 lines
18 KiB
Python
490 lines
18 KiB
Python
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 — цены с 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
|
||
|
||
|
||
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://твой-домен/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 гандон")
|
||
|
||
|
||
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"),
|
||
]
|
||
)
|
||
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"))
|
||
|
||
|
||
|
||
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()
|