SUDOZOVCHIK/bot_tg
1
0
Fork 0
forked from zovos/bot_tg
Code Pull requests Activity

ПОЛНЫЙ РЕФАКТОРИНГ КОДА. Константы вынесены в config.py, упрощены импорты, вроде бы все работает, а там хуй знает))))

Browse source
This commit is contained in:
hitoshi 2026-02-23 10:21:14 +03:00
parent b9c9ddb4bf
commit 07f6efe7b1
4 changed files with 329 additions and 450 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
.gitignore vendored
View file

@ -20,3 +20,5 @@ db/
.mypy_cache/
AI/models/
main_legacy.py

2
Dockerfile
View file

@ -3,7 +3,7 @@ ENV PYTHONUNBUFFERED=1
WORKDIR /app
COPY requirements.txt ./
RUN pip install --no-cache-dir -r requirements.txt
COPY main.py impact.ttf zparser.py ./
COPY main.py config.py impact.ttf zparser.py ./
COPY maket ./maket
COPY AI ./AI
CMD ["python", "main.py"]

90
config.py Normal file
View file

@ -0,0 +1,90 @@
import os
from pathlib import Path
from datetime import timedelta, time
from zoneinfo import ZoneInfo
# --- Базовые пути ---
BASE_DIR = Path(__file__).parent
FONT_PATH = BASE_DIR / "impact.ttf"
DEFAULT_TZ = os.getenv("TZ", "Europe/Moscow")
# --- Настройки /penis ---
PENIS_DB_PATH = "/db/penis_stats.sqlite3"
PENIS_START_LENGTH = 10.0
PENIS_MIN_DELTA = 0.1
PENIS_MAX_DELTA = 7.0
PENIS_COOLDOWN = timedelta(hours=24)
PENIS_COOLDOWN_SECONDS = int(PENIS_COOLDOWN.total_seconds())
PENIS_TOP_LIMIT = 10
# --- Конфигурация макетов ---
TEMPLATES = {
"1": {
"path": BASE_DIR / "maket" / "1.png",
"anchor": (104, 240),
"padding": 20,
"label": "Владимир Владимирович Путин самый лучший президент",
},
}
# --- Расписание ---
def make_time(h: int, m: int) -> time:
return time(hour=h, minute=m)
SCHEDULE = {
0: [ # Понедельник
(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: [ # Вторник
(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: [], # Среда
3: [ # Четверг
(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: [ # Пятница
(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),
],
}
# --- Остальное ---
KEYWORDS = {'крип', 'радиохэд', 'creep', 'radiohead'}
def get_hint_text(templates_str: str) -> str:
return (
"Команды:\n"
f"• /maket «id» «текст» — доступные: {templates_str}.\n"
"• /gen_mem Up:верх Down:низ — отправь с фото.\n"
"• /break — узнать время до перекура.\n"
"• /penis — миниигра 1 раз в 24 часа.\n"
"• /top_penis — топ по длине.\n"
"• /talk [текст] — ИИ базарит на фене.\n"
"• /prices /fetch — цены.\n"
"Подсказка: в группах пиши /команда@username_бота."
)

685
main.py
View file

@ -1,61 +1,36 @@
import asyncio
from AI.talk_handler import handle_talk
import logging
import os
import re
import requests
# Системные импорты
import asyncio, logging, os, re, random, sqlite3
from datetime import datetime, time, timedelta
from io import BytesIO
from pathlib import Path
from textwrap import wrap
import random
import sqlite3
from zoneinfo import ZoneInfo
# Стороние импорты
import requests
from PIL import Image, ImageDraw, ImageFont
from aiogram import Bot, Dispatcher, F, types
from aiogram.enums import ParseMode
from aiogram.filters import Command, CommandStart
from aiogram.enums import ParseMode
from aiogram.types import (
BufferedInputFile,
Message,
BotCommand,
FSInputFile,
BotCommandScopeDefault,
BotCommandScopeAllPrivateChats,
BotCommandScopeAllGroupChats,
BufferedInputFile, Message, BotCommand, FSInputFile,
BotCommandScopeDefault, BotCommandScopeAllPrivateChats, BotCommandScopeAllGroupChats
)
from aiogram.client.default import DefaultBotProperties
from PIL import Image, ImageDraw, ImageFont
from datetime import datetime, time, timedelta
from zoneinfo import ZoneInfo
# Локальные импорты
from AI.talk_handler import handle_talk
from zparser import get_military_data
import config
# Базовые пути/настройки проекта.
BASE_DIR = Path(__file__).parent
FONT_PATH = BASE_DIR / "impact.ttf"
DEFAULT_TZ = os.getenv("TZ", "Europe/Moscow")
# Данные миниигры /penis храним в SQLite, чтобы не держать их в RAM.
PENIS_DB_PATH = "/db/penis_stats.sqlite3"
PENIS_START_LENGTH = 10.0
PENIS_MIN_DELTA = 0.1
PENIS_MAX_DELTA = 7.0
PENIS_COOLDOWN = timedelta(hours=24)
PENIS_COOLDOWN_SECONDS = int(PENIS_COOLDOWN.total_seconds())
PENIS_TOP_LIMIT = 10
# Конфигурация текстовых макетов /maket.
TEMPLATES = {
"1": {
"path": BASE_DIR / "maket" / "1.png",
"anchor": (104, 240),
"padding": 20, # right/bottom padding from edges
"label": "Владимир Владимирович Путин самый лучший президент",
},
}
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
# --- Инициализация и вспомогательные функции ---
def init_penis_db() -> None:
# Инициализация таблицы для миниигры /penis.
with sqlite3.connect(PENIS_DB_PATH) as conn:
"""Инициализация БД для миниигры."""
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 (
@ -66,83 +41,26 @@ def init_penis_db() -> None:
)
"""
)
conn.execute(
"CREATE INDEX IF NOT EXISTS idx_penis_stats_length ON penis_stats(length DESC)"
)
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(TEMPLATES.keys()):
label = TEMPLATES[tid].get("label", "")
if label:
parts.append(f"{tid} — {label}")
else:
parts.append(tid)
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 "нет макетов"
# ---- 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))
now = datetime.now(ZoneInfo(config.DEFAULT_TZ))
elif now.tzinfo is None:
now = now.replace(tzinfo=ZoneInfo(DEFAULT_TZ))
now = now.replace(tzinfo=ZoneInfo(config.DEFAULT_TZ))
day = now.weekday()
today_schedule = SCHEDULE.get(day, [])
today_schedule = config.SCHEDULE.get(day, [])
if not today_schedule:
return "Сегодня кайфуем дома, перекур не нужен."
@ -153,17 +71,10 @@ def minutes_until_break(now: datetime | None = None) -> str:
start_dt = dt_for(start_t)
end_dt = dt_for(end_t)
# If we're before the lesson
if now < start_dt:
if now < start_dt or (start_dt <= now < end_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:
@ -172,78 +83,70 @@ def minutes_until_break(now: datetime | None = None) -> str:
return "Пары закончились, перекур только завтра."
HINT_TEXT = (
# Общая подсказка по командам.
"Команды:\n"
f"• /maket «id» «текст» — доступные: {templates_list_text()} (1: 737x414, точка 104,240).\n"
"• /gen_mem Up:верх Down:низ — отправь с фото (как подпись) или ответом на фото.\n"
"• /break — узнать, сколько минут до перекура.\n"
"• /penis — миниигра: +/− от 0.1 до 7 см, раз в 24 часа.\n"
"• /top_penis — динамический топ по текущей длине.\n"
"• /talk [сообщение] — побазарить с ИИ на фене.\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:
# Грузим Impact нужного размера.
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)
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]:
# Переносим текст по словам так, чтобы каждая строка влезла в max_width.
words = text.split()
if not words:
return []
lines: list[str] = []
current: list[str] = []
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))
if current: lines.append(" ".join(current))
current = [word]
if current:
lines.append(" ".join(current))
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", (max_width, max_height)))
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 = []
for ln in lines:
bbox = draw_dummy.textbbox((0, 0), ln, font=font, stroke_width=stroke_width)
line_heights.append(bbox[3] - bbox[1])
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)
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)
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)
@ -254,33 +157,7 @@ def draw_multiline(draw: ImageDraw.ImageDraw, lines: list[str], position: tuple[
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):
@ -293,15 +170,38 @@ def calculate_font_for_width(text: str, image_width: int, max_ratio: float = 0.0
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
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)
return font, wrap_text_to_width(text, font, int(image_width * 0.95), draw_dummy, stroke_width), stroke_width, max(4, min_size // 6)
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, padding = tpl["anchor"], tpl["padding"]
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:
# Рисуем классический мем-текст сверху/снизу и возвращаем JPEG в памяти.
img = img.convert("RGB")
draw = ImageDraw.Draw(img)
width, height = img.size
@ -309,20 +209,35 @@ def generate_meme(img: Image.Image, top_text: str, bottom_text: str) -> BytesIO:
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))
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 += 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))
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)
@ -330,89 +245,48 @@ def generate_meme(img: Image.Image, top_text: str, bottom_text: str) -> BytesIO:
return output
def parse_up_down(text: str) -> tuple[str, str]:
# Парсим формат /gen_mem Up:... Down:...
# Если меток нет, используем fallback: первое слово -> up, остальное -> down.
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:
# Отдельная "угарная" логика для /zvetok.
import random
if now is None:
now = datetime.now(ZoneInfo(DEFAULT_TZ))
now = datetime.now(ZoneInfo(config.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)
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
)
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
)
return now.replace(hour=lesson_end.hour, minute=lesson_end.minute, second=0, microsecond=0)
# ---- Проверяем: сейчас внутри пары? ----
if day in LESSON_DAYS:
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:
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:
if next_day in lesson_days:
candidate = lesson_start_for(i)
if candidate > now:
delta = candidate - now
@ -422,21 +296,20 @@ def minutes_until_next_lesson(now: datetime | None = None) -> str:
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]:
# Миниигра /penis: изменение длины и кулдаун 24 часа от последнего броска.
if now is None:
now = datetime.now(ZoneInfo(DEFAULT_TZ))
now = datetime.now(ZoneInfo(config.DEFAULT_TZ))
elif now.tzinfo is None:
now = now.replace(tzinfo=ZoneInfo(DEFAULT_TZ))
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(PENIS_DB_PATH) as conn:
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 = ?",
@ -444,9 +317,8 @@ def play_penis(user_id: int, user_name: str | None = None, now: datetime | None
).fetchone()
if row is None:
current_length = PENIS_START_LENGTH
current_length = config.PENIS_START_LENGTH
last_used_ts = None
stored_name = None
else:
current_length = float(row["length"])
last_used_ts = row["last_used_ts"]
@ -455,9 +327,8 @@ def play_penis(user_id: int, user_name: str | None = None, now: datetime | None
display_name = stored_name.strip()
if last_used_ts is not None:
next_ts = int(last_used_ts) + PENIS_COOLDOWN_SECONDS
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 = ?",
@ -469,12 +340,10 @@ def play_penis(user_id: int, user_name: str | None = None, now: datetime | None
conn.commit()
return False, f"Сегодня уже кидал. Попробуй через {hours} ч {minutes} мин."
delta = round(random.uniform(PENIS_MIN_DELTA, PENIS_MAX_DELTA), 1)
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(?, ?, ?, ?)",
@ -498,20 +367,16 @@ def play_penis(user_id: int, user_name: str | None = None, now: datetime | None
"Следующая попытка через 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 = PENIS_TOP_LIMIT) -> str:
# Формируем динамический топ по текущим значениям /penis.
def build_penis_top(limit: int = config.PENIS_TOP_LIMIT) -> str:
try:
with sqlite3.connect(PENIS_DB_PATH) as conn:
with sqlite3.connect(config.PENIS_DB_PATH) as conn:
conn.row_factory = sqlite3.Row
rows = conn.execute(
"""
@ -539,90 +404,30 @@ def build_penis_top(limit: int = PENIS_TOP_LIMIT) -> str:
prefix = medals.get(index, f"{index}.")
lines.append(f"{prefix} {name} — {length:.1f} см")
return "\n".join(lines)
async def handle_start(message: Message):
# Базовый /start.
await message.answer("Соси")
# --- Хендлеры aiogram ---
async def handle_help(message: Message):
# Вывод подсказки по возможностям бота.
await message.answer(HINT_TEXT)
await message.answer(config.get_hint_text(templates_list_text()))
async def handle_break(message: Message):
# Команда расписания перекура.
async def handle_start(message: Message):
await message.answer("Соси")
async def handle_break_cmd(message: Message):
await message.answer(minutes_until_break())
async def handle_maket(message: Message):
# Генерация текста в подготовленном шаблоне /maket.
parts = message.text.split(maxsplit=2)
if len(parts) == 1:
await message.reply(f"Доступные макеты: {templates_list_text()}\nФормат: /maket «id» «текст»")
async def handle_penis_cmd(message: Message):
if not message.from_user:
await message.answer("Команда доступна только пользователям.")
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"))
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_gen_mem(message: Message, bot: Bot):
# Мем можно сделать из фото в этом же сообщении или из reply на фото.
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:
# Берём самое большое фото Telegram и скачиваем в память.
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.
async def handle_size_cmd(message: Message):
size = round(random.uniform(0.1, 35.0), 1)
if size < 5:
status = "СВОи"
elif size < 10:
@ -635,107 +440,53 @@ async def handle_size(message: Message):
status = "Пиздабол"
else:
status = "Говноед"
await message.answer(f"📏 Твой хуек: {size} см\nТы: {status}")
await message.answer(
f"📏 Твой хуек: {size} см\n"
f"Ты: {status}"
)
async def handle_zvetok_cmd(message: Message):
await message.answer(minutes_until_next_lesson())
async def handle_penis(message: Message):
# Хендлер миниигры /penis.
if not message.from_user:
await message.answer("Команда доступна только пользователям.")
return
display_name = format_penis_user_name(message.from_user)
_, result_text = play_penis(message.from_user.id, display_name)
await message.answer(result_text)
async def handle_top_penis(message: Message):
# Хендлер топа по значениям /penis.
await message.answer(build_penis_top())
# Ключевые слова, на которые бот кидает трек Creep.
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):
# Старый источник цен (/prices).
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(message: Message):
# Новый источник цен (/fetch).
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}:")
@ -747,21 +498,12 @@ async def handle_fetch(message: Message):
if not name or price is None:
continue
lines.append(f"{name} — {price} ₽")
text = "\n".join(lines)
await message.answer(text[:4000])
await message.answer("\n".join(lines)[: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)
async def handle_svodka(message: Message):
channel = "creamy_caprice"
data = get_military_data(channel, depth=5)
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']} км² в сутки, "
@ -771,21 +513,93 @@ async def handle_svodka(message: Message):
else:
await message.answer(f"Хохлы ответят за это...: {data['message']}")
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"))
async def handle_keywords(message: Message):
if not message.text or message.text.startswith('/'):
return
if any(re.search(rf'\b{word}\b', message.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...",
)
# --- Запуск ---
def main():
# Точка входа: создаём бота, регистрируем команды и запускаем polling.
token = os.getenv("BOT_TOKEN")
if not token:
raise RuntimeError("BOT_TOKEN env var is required")
if not token: raise RuntimeError("BOT_TOKEN не задан!")
# Готовим SQLite-таблицу для /penis и /top_penis.
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_gen_mem, Command("gen_mem"))
dp.message.register(handle_keywords, F.text)
async def on_startup(bot: Bot):
# Выставляем меню команд в интерфейсе Telegram.
try:
commands = [
BotCommand(command="start", description="Приветствие"),
@ -802,7 +616,6 @@ def main():
BotCommand(command="talk", description="Побазарить"),
BotCommand(command="svodka", description="СВО: итоги"),
]
# Явно задаём команды для default/private/group, чтобы меню через "/" появлялось стабильно.
scopes = (
BotCommandScopeDefault(),
BotCommandScopeAllPrivateChats(),
@ -810,38 +623,12 @@ def main():
)
for scope in scopes:
await bot.set_my_commands(commands, scope=scope)
except Exception: # noqa: WPS440
except Exception:
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_penis, Command("penis"))
dp.message.register(handle_top_penis, Command("top_penis"))
dp.message.register(handle_prices, Command("prices"))
dp.message.register(handle_fetch, Command("fetch"))
dp.message.register(handle_zvetok, Command("zvetok"))
dp.message.register(handle_talk, Command("talk"))
dp.message.register(handle_svodka, Command("svodka"))
async def handle_gen_mem_cmd(message: Message, bot: Bot):
# Обёртка под сигнатуру aiogram для /gen_mem.
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...")
logger.info("Бот запущен...")
asyncio.run(dp.start_polling(bot, allowed_updates=dp.resolve_used_update_types()))
if __name__ == "__main__":
main()