forked from zovos/bot_tg
768 lines
31 KiB
Python
768 lines
31 KiB
Python
# Системные импорты
|
||
import asyncio, logging, os, re, random, sqlite3
|
||
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 zparser import get_military_data
|
||
import config
|
||
|
||
logging.basicConfig(level=logging.INFO)
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# --- Инициализация и вспомогательные функции ---
|
||
|
||
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 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 minutes_until_break(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))
|
||
|
||
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):
|
||
minutes = int((end_dt - now).total_seconds() // 60)
|
||
return f"{minutes} минут до перекура"
|
||
|
||
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 "Пары закончились, перекур только завтра."
|
||
|
||
|
||
def next_break_window(now: datetime | None = None) -> tuple[datetime | None, datetime | None]:
|
||
"""Возвращает (начало, конец) ближайшего перекура или (None, None)."""
|
||
if now is None:
|
||
now = datetime.now(ZoneInfo(config.DEFAULT_TZ))
|
||
elif now.tzinfo is None:
|
||
now = now.replace(tzinfo=ZoneInfo(config.DEFAULT_TZ))
|
||
|
||
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} мин"
|
||
|
||
# --- Игровая логика (/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)
|
||
|
||
# --- Хендлеры 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_break_cmd(message: Message):
|
||
now = datetime.now(ZoneInfo(config.DEFAULT_TZ))
|
||
await message.answer(minutes_until_break(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_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):
|
||
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_cmd(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} ₽")
|
||
await message.answer("\n".join(lines)[:4000])
|
||
except Exception:
|
||
await message.answer("unable to connect x-server гандон")
|
||
|
||
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
|
||
|
||
# Записываем каждое обычное сообщение в историю чата
|
||
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")
|
||
|
||
# --- Запуск ---
|
||
|
||
def main():
|
||
token = os.getenv("BOT_TOKEN")
|
||
if not token: raise RuntimeError("BOT_TOKEN не задан!")
|
||
|
||
init_penis_db()
|
||
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_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_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_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="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="prices", description="цены x-server"),
|
||
BotCommand(command="fetch", description="цены (новый источник)"),
|
||
BotCommand(command="zvetok", description="Цветянский бля"),
|
||
BotCommand(command="talk", description="Побазарить"),
|
||
BotCommand(command="ebalnik", 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()
|