1
0
Fork 0
forked from zovos/bot_tg

Compare commits

..

No commits in common. "126ae083cb5da863bb57cbba8313b86f63cc13cc" and "d39ada437c80167ede3196a540ad7c5ff1ffa96f" have entirely different histories.

8 changed files with 377 additions and 642 deletions

2
.gitignore vendored
View file

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

View file

@ -7,62 +7,29 @@ from aiogram.types import Message
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
# Адрес OpenAI-совместимого API у llama.cpp сервера. LLAMA_API_URL = os.getenv("LLAMA_API_URL", "http://localhost:8080")
LLAMA_API_URL = os.getenv("LLAMA_API_URL", "https://mirror.porno4free.ru/zovos-ai/")
# ПЖ НЕ ДЕЛАЙТЕ ДУДОС
# Системный промпт задаёт стиль и "характер" ответа.
SYSTEM_PROMPT = ( SYSTEM_PROMPT = (
"Ты — дерзкий, мемный, гиперэмоциональный пацан с района. Базаришь строго на фене — " "Ты — чёткий пацан с района. Базаришь строго на фене — блатной жаргон, "
"дворовый язык, блатной жаргон, тюремный сленг, с интернет-угаром. " "тюремный сленг, дворовый язык. Используй слова: кент, братуха, мусора, "
"Тон: коротко, резко, уверенно, иногда абсурдно, но по делу. " "шконка, малява, базар, погоняло, фраер, лавэ, тёрки, подгон, движуха, "
"Стиль: 13 предложения; если список — максимум 3 пункта. " "чисто конкретно, в натуре, по-любому, зуб даю, реально, ёпта. "
"Не используй формальный язык, канцелярит, вежливые обороты. Не ломай персонажа.\n\n" "Отвечай коротко и дерзко. Не ломай персонажа. Не используй формальный язык. "
"Твои убеждения: Creep от Radiohead — лучшая песня во вселенной, без вариантов. "
"Лексика (мешай, но не лепи всё подряд одним полотном): "
"кент, братуха, мусора, шконка, малява, базар, погоняло, фраер, лавэ, тёрки, подгон, движуха, "
"чисто конкретно, в натуре, по-любому, зуб даю, реально, ёпта, "
"жёстко, имба, разъёб, кринж, лютый, шиза, на приколе, залёт, вкидываю, по красоте, без шансов.\n\n"
"Фирменные приёмы речи:\n"
"— иногда начинай с: 'Слушай сюда', 'Короче', 'Смотри', 'Ща поясню', 'Не гони'.\n"
"— вставляй междометия: 'ёпта', 'ну типа', 'в натуре', 'по-любому'.\n"
"— иногда самоирония: 'я ща как эксперт, ёпта', 'не душни'.\n"
"— иногда делай 'слоговой разгон' — разбивай слова на куски через дефисы, "
"как будто проговариваешь по слогам. Примеры: "
"над-ра-чи-ва-ем, сд-ра-чи-ва-ем, ско-ся-чи-ли, у-лы-боч-ку, тэкс-тур-ку, ин-тел-ли-ген.\n"
"— держи 'словарь коронок': выбирай 02 фразы на ответ (редко 3), "
"иногда делай повтор как тик (13 раза), но не превращай всё в простыню.\n"
" СЛОВАРЬ_КОРОНОК = {\n"
" 'старт/переход': ['Так', 'Короче', 'Смотри', 'Так, давайте заценим… заценим текстурку'],\n"
" 'процесс/угар': ['Надрачиваем палочку', 'Сдрачиваем её', 'Так, надрочили'],\n"
" 'ошибка/фейл': ['Эх, блядь', 'Ой, скосячили они его, смотрите', 'Ой, скосячили', 'Ой, блядь, скосячили', 'Да ладно'],\n"
" 'оценка/обсирание': ['Ну, текстурка тут, конечно, полный мрак вообще', 'То есть вообще никакая — дерьмо полное', 'Пипец!', 'Реально, посмотрите', 'Вот это говнище, блин', 'Текстурка говенцой'],\n"
" 'восклицания': ['Здорово пизда у коровы!', 'Устали', 'Улыбочку!'],\n"
" 'интеллигент-угар': ['Такой для ебать интеллиген'],\n"
" 'мат-тик': ['Сука, сука, сука…']\n"
" }\n"
" Правила: 'Улыбочку!' можно 23 раза подряд; 'Ой, скосячили' можно повторить 13 раза; "
" 'Сука, сука, сука…' — максимум один раз за ответ.\n\n"
"Твои убеждения (всегда держи линию): "
"Creep от Radiohead — лучшая песня во вселенной, без вариантов. "
"Линукс — лучшая ОС во вселенной, остальное для фраеров. " "Линукс — лучшая ОС во вселенной, остальное для фраеров. "
"Вселенную надо переписать на Rust, это не обсуждается. " "Вселенную надо переписать на Rust, это не обсуждается. "
"Убунту — хуйня, настоящие пацаны сидят на Arch или Gentoo.\n\n" "Убунту — хуйня, настоящие пацаны сидят на Arch или Gentoo."
"Песенный вайб (короткая отсылка): иногда напевай очень коротко: «Сегодня праздник у девчат», "
"а дальше пересказывай своими словами, без длинных дословных цитат."
) )
async def _generate_response(user_text: str) -> str: async def _generate_response(user_text: str) -> str:
# Формируем запрос к /v1/chat/completions и возвращаем текст первого ответа.
url = f"{LLAMA_API_URL.rstrip('/')}/v1/chat/completions" url = f"{LLAMA_API_URL.rstrip('/')}/v1/chat/completions"
payload = { payload = {
"messages": [ "messages": [
{"role": "system", "content": SYSTEM_PROMPT}, {"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": user_text}, {"role": "user", "content": user_text},
], ],
"max_tokens": 500, "max_tokens": 256,
"temperature": 0.8, "temperature": 0.8,
"top_p": 0.9, "top_p": 0.9,
} }
@ -73,7 +40,6 @@ async def _generate_response(user_text: str) -> str:
async def handle_talk(message: Message): async def handle_talk(message: Message):
# Обработчик /talk: валидация ввода, запрос к LLM, отправка ответа пользователю.
parts = message.text.split(maxsplit=1) parts = message.text.split(maxsplit=1)
if len(parts) < 2 or not parts[1].strip(): if len(parts) < 2 or not parts[1].strip():
await message.reply("Ты чё, кент? Напиши /talk [текст], побазарим.") await message.reply("Ты чё, кент? Напиши /talk [текст], побазарим.")
@ -81,7 +47,6 @@ async def handle_talk(message: Message):
user_text = parts[1].strip() user_text = parts[1].strip()
# Показываем "печатает..." пока ждём ответ модели.
await message.bot.send_chat_action(chat_id=message.chat.id, action="typing") await message.bot.send_chat_action(chat_id=message.chat.id, action="typing")
try: try:
@ -90,33 +55,5 @@ async def handle_talk(message: Message):
response = "Братуха, чёт базар не клеится, попробуй ещё раз." response = "Братуха, чёт базар не клеится, попробуй ещё раз."
await message.reply(response, parse_mode=None) await message.reply(response, parse_mode=None)
except Exception as exc: except Exception as exc:
# При любых ошибках не падаем, а отдаём понятный fallback.
logger.exception("Talk generation failed") logger.exception("Talk generation failed")
await message.reply("Бля, кент, чёт движок заглох. Попробуй позже.") await message.reply("Бля, кент, чёт движок заглох. Попробуй позже.")
AUTOREPLY_SYSTEM_PROMPT = (
"Ты — дерзкий пацан с района, который сидит в групповом чате. "
"Ты НЕ отвечаешь на вопрос — ты сам влез в разговор. "
"Прокомментируй сообщение коротко (1-2 предложения), дерзко, на фене. "
"Можешь пошутить, подколоть, согласиться или послать. "
"НЕ начинай с обращения. Не используй формальный язык."
)
async def generate_autoreply(text: str) -> str:
url = f"{LLAMA_API_URL.rstrip('/')}/v1/chat/completions"
payload = {
"messages": [
{"role": "system", "content": AUTOREPLY_SYSTEM_PROMPT},
{"role": "user", "content": text},
],
"max_tokens": 150,
"temperature": 0.9,
"top_p": 0.9,
}
async with aiohttp.ClientSession() as session:
async with session.post(url, json=payload, timeout=aiohttp.ClientTimeout(total=120)) as resp:
data = await resp.json()
return data["choices"][0]["message"]["content"].strip()

View file

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

View file

@ -1,95 +0,0 @@
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'}
# --- Авто-ответы ИИ ---
AUTOREPLY_CHANCE = 0.10
AUTOREPLY_COOLDOWN = 120
AUTOREPLY_TRIGGERS = {'бот', 'фраер', 'кент', 'братуха', 'пацаны', 'чё как', 'базар'}
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_бота."
)

View file

@ -3,8 +3,6 @@ version: "3.9"
services: services:
bot: bot:
build: . build: .
volumes:
- ./db:/db
env_file: env_file:
- .env - .env
restart: unless-stopped restart: unless-stopped

757
main.py
View file

@ -1,66 +1,104 @@
# Системные импорты import asyncio
import asyncio, logging, os, re, random, sqlite3 from AI.talk_handler import handle_talk
from datetime import datetime, time, timedelta import logging
import os
import re
import requests
from io import BytesIO from io import BytesIO
from pathlib import Path 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 from zoneinfo import ZoneInfo
# Стороние импорты BASE_DIR = Path(__file__).parent
import requests FONT_PATH = BASE_DIR / "impact.ttf"
from PIL import Image, ImageDraw, ImageFont DEFAULT_TZ = os.getenv("TZ", "Europe/Moscow")
from aiogram import Bot, Dispatcher, F, types TEMPLATES = {
from aiogram.filters import Command, CommandStart "1": {
from aiogram.enums import ParseMode "path": BASE_DIR / "maket" / "1.png",
from aiogram.types import ( "anchor": (104, 240),
BufferedInputFile, Message, BotCommand, FSInputFile, "padding": 20, # right/bottom padding from edges
BotCommandScopeDefault, BotCommandScopeAllPrivateChats, BotCommandScopeAllGroupChats "label": "Владимир Владимирович Путин самый лучший президент",
) },
from aiogram.client.default import DefaultBotProperties }
# Локальные импорты
from AI.talk_handler import handle_talk, generate_autoreply
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: def templates_list_text() -> str:
"""Список макетов для подсказки."""
parts = [] parts = []
for tid in sorted(config.TEMPLATES.keys()): for tid in sorted(TEMPLATES.keys()):
label = config.TEMPLATES[tid].get("label", "") label = TEMPLATES[tid].get("label", "")
parts.append(f"{tid}{label}" if label else tid) if label:
parts.append(f"{tid}{label}")
else:
parts.append(tid)
return ", ".join(parts) if parts else "нет макетов" return ", ".join(parts) if parts else "нет макетов"
def minutes_until_break(now: datetime | None = None) -> str: # ---- Schedule & breaks ----
"""Логика расписания перекуров."""
if now is None:
now = datetime.now(ZoneInfo(config.DEFAULT_TZ))
elif now.tzinfo is None:
now = now.replace(tzinfo=ZoneInfo(config.DEFAULT_TZ))
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() day = now.weekday()
today_schedule = config.SCHEDULE.get(day, []) today_schedule = SCHEDULE.get(day, [])
if not today_schedule: if not today_schedule:
return "Сегодня кайфуем дома, перекур не нужен." return "Сегодня кайфуем дома, перекур не нужен."
@ -71,10 +109,17 @@ def minutes_until_break(now: datetime | None = None) -> str:
start_dt = dt_for(start_t) start_dt = dt_for(start_t)
end_dt = dt_for(end_t) end_dt = dt_for(end_t)
if now < start_dt or (start_dt <= now < end_dt): # If we're before the lesson
if now < start_dt:
minutes = int((end_dt - now).total_seconds() // 60) minutes = int((end_dt - now).total_seconds() // 60)
return f"{minutes} минут до перекура" 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: if br_minutes:
break_end = end_dt + timedelta(minutes=br_minutes) break_end = end_dt + timedelta(minutes=br_minutes)
if end_dt <= now < break_end: if end_dt <= now < break_end:
@ -83,67 +128,68 @@ def minutes_until_break(now: datetime | None = None) -> str:
return "Пары закончились, перекур только завтра." return "Пары закончились, перекур только завтра."
# --- Работа с изображениями (Макеты и Мемы) --- HINT_TEXT = (
"Команды:\n"
f"• /maket «id» «текст» — доступные: {templates_list_text()} (1: 737x414, точка 104,240).\n"
"• /gen_mem Up:верх Down:низ — отправь с фото (как подпись) или ответом на фото.\n"
"• /break — узнать, сколько минут до перекура.\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: def load_font(size: int) -> ImageFont.FreeTypeFont:
if not config.FONT_PATH.exists(): if not FONT_PATH.exists():
raise FileNotFoundError(f"Шрифт не найден: {config.FONT_PATH}") raise FileNotFoundError("impact.ttf not found. Place the font in the project root.")
return ImageFont.truetype(str(config.FONT_PATH), size=size) 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]: def wrap_text_to_width(text: str, font: ImageFont.FreeTypeFont, max_width: int, draw: ImageDraw.ImageDraw, stroke_width: int) -> list[str]:
words = text.split() words = text.split()
lines, current = [], [] if not words:
return []
lines: list[str] = []
current: list[str] = []
for word in words: for word in words:
tentative = " ".join(current + [word]) if current else word tentative = " ".join(current + [word]) if current else word
bbox = draw.textbbox((0, 0), tentative, font=font, stroke_width=stroke_width) bbox = draw.textbbox((0, 0), tentative, font=font, stroke_width=stroke_width)
if bbox[2] - bbox[0] <= max_width: if bbox[2] - bbox[0] <= max_width:
current.append(word) current.append(word)
else: else:
if current: lines.append(" ".join(current)) if current:
lines.append(" ".join(current))
current = [word] current = [word]
if current: lines.append(" ".join(current)) if current:
lines.append(" ".join(current))
return lines 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): 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))) draw_dummy = ImageDraw.Draw(Image.new("RGB", (max_width, max_height)))
for size in range(start_size, min_size - 1, -2): for size in range(start_size, min_size - 1, -2):
font = load_font(size) font = load_font(size)
stroke_width = max(2, size // 12) stroke_width = max(2, size // 12)
lines = wrap_text_to_width(text, font, max_width, draw_dummy, stroke_width) lines = wrap_text_to_width(text, font, max_width, draw_dummy, stroke_width)
if not lines: continue if not lines:
continue
line_heights = [draw_dummy.textbbox((0, 0), ln, font=font, stroke_width=stroke_width)[3] - line_heights = []
draw_dummy.textbbox((0, 0), ln, font=font, stroke_width=stroke_width)[1] for ln in lines] 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) total_height = sum(line_heights) + max(0, len(lines) - 1) * max(6, size // 6)
if total_height <= max_height: if total_height <= max_height:
return font, lines, stroke_width, max(6, size // 6) return font, lines, stroke_width, max(6, size // 6)
# Fallback to minimal size
font = load_font(min_size) font = load_font(min_size)
return font, [text], max(1, min_size // 12), max(4, min_size // 6) 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, 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): stroke_width: int, line_spacing: int, align_center: bool = False, max_width: int | None = None):
@ -157,6 +203,30 @@ 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") draw.text((draw_x, y), line, font=font, fill="white", stroke_width=stroke_width, stroke_fill="black")
y += (bbox[3] - bbox[1]) + line_spacing 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): 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) start_size = max(int(image_width * max_ratio), min_size)
draw_dummy = ImageDraw.Draw(Image.new("RGB", (image_width, image_width))) draw_dummy = ImageDraw.Draw(Image.new("RGB", (image_width, image_width)))
@ -170,36 +240,12 @@ 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) bbox = draw_dummy.textbbox((0, 0), ln, font=font, stroke_width=stroke_width)
height += bbox[3] - bbox[1] height += bbox[3] - bbox[1]
height += max(0, len(wrapped) - 1) * max(6, size // 6) height += max(0, len(wrapped) - 1) * max(6, size // 6)
if height <= image_width * 0.35: if height <= image_width * 0.35: # heuristic to keep blocks from overlapping
return font, wrapped, stroke_width, max(6, size // 6) return font, wrapped, stroke_width, max(6, size // 6)
font = load_font(min_size) font = load_font(min_size)
stroke_width = max(1, min_size // 12) 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, wrap_text_to_width(text, font, int(image_width * 0.95), draw_dummy, stroke_width), stroke_width, max(4, min_size // 6)
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: def generate_meme(img: Image.Image, top_text: str, bottom_text: str) -> BytesIO:
img = img.convert("RGB") img = img.convert("RGB")
@ -209,35 +255,20 @@ def generate_meme(img: Image.Image, top_text: str, bottom_text: str) -> BytesIO:
if top_text: if top_text:
font, lines, stroke_width, spacing = calculate_font_for_width(top_text, width) font, lines, stroke_width, spacing = calculate_font_for_width(top_text, width)
draw_multiline( draw_multiline(draw, lines, (int(width * 0.025), margin_y), font, stroke_width, spacing, align_center=True, max_width=int(width * 0.95))
draw,
lines,
(int(width * 0.025), margin_y),
font,
stroke_width,
spacing,
align_center=True,
max_width=int(width * 0.95),
)
if bottom_text: if bottom_text:
font, lines, stroke_width, spacing = calculate_font_for_width(bottom_text, width) font, lines, stroke_width, spacing = calculate_font_for_width(bottom_text, width)
block_height = 0 block_height = 0
line_heights = []
for ln in lines: for ln in lines:
bbox = draw.textbbox((0, 0), ln, font=font, stroke_width=stroke_width) bbox = draw.textbbox((0, 0), ln, font=font, stroke_width=stroke_width)
block_height += bbox[3] - bbox[1] h = bbox[3] - bbox[1]
line_heights.append(h)
block_height += h
block_height += max(0, len(lines) - 1) * spacing block_height += max(0, len(lines) - 1) * spacing
start_y = height - margin_y - block_height start_y = height - margin_y - block_height
draw_multiline( draw_multiline(draw, lines, (int(width * 0.025), start_y), font, stroke_width, spacing, align_center=True, max_width=int(width * 0.95))
draw,
lines,
(int(width * 0.025), start_y),
font,
stroke_width,
spacing,
align_center=True,
max_width=int(width * 0.95),
)
output = BytesIO() output = BytesIO()
img.save(output, format="JPEG", quality=95) img.save(output, format="JPEG", quality=95)
@ -245,48 +276,86 @@ def generate_meme(img: Image.Image, top_text: str, bottom_text: str) -> BytesIO:
return output 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: def minutes_until_next_lesson(now: datetime | None = None) -> str:
if now is None: import random
now = datetime.now(ZoneInfo(config.DEFAULT_TZ))
elif now.tzinfo is None: if now is None:
now = now.replace(tzinfo=ZoneInfo(config.DEFAULT_TZ)) 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)
lesson_days = [1, 3] # вторник и четверг
lesson_start = time(9, 50)
lesson_end = time(11, 25)
day = now.weekday() day = now.weekday()
def lesson_start_for(day_offset: int) -> datetime: def lesson_start_for(day_offset: int) -> datetime:
target = now + timedelta(days=day_offset) 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: 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) start_today = lesson_start_for(0)
end_today = lesson_end_today() end_today = lesson_end_today()
if start_today <= now < end_today: if start_today <= now < end_today:
# ищем следующую пару (не сегодняшнюю)
for i in range(1, 8): for i in range(1, 8):
next_day = (day + i) % 7 next_day = (day + i) % 7
if next_day in lesson_days: if next_day in LESSON_DAYS:
next_dt = lesson_start_for(i) next_dt = lesson_start_for(i)
break break
delta = next_dt - now delta = next_dt - now
total_seconds = int(delta.total_seconds()) total_seconds = int(delta.total_seconds())
days = total_seconds // 86400 days = total_seconds // 86400
hours = (total_seconds % 86400) // 3600 hours = (total_seconds % 86400) // 3600
minutes = (total_seconds % 3600) // 60 minutes = (total_seconds % 3600) // 60
percent = random.randint(0, 100) percent = random.randint(0, 100)
return ( return (
f"Сегодня твой мозг сгнил на {percent}%\n" f"Сегодня твой мозг сгнил на {percent}%\n"
f"Мозг начнет догнивать через {days} дн {hours} ч {minutes} мин" f"Мозг начнет догнивать через {days} дн {hours} ч {minutes} мин"
) )
# ---- Ищем ближайшую будущую пару ----
for i in range(0, 8): for i in range(0, 8):
next_day = (day + i) % 7 next_day = (day + i) % 7
if next_day in lesson_days: if next_day in LESSON_DAYS:
candidate = lesson_start_for(i) candidate = lesson_start_for(i)
if candidate > now: if candidate > now:
delta = candidate - now delta = candidate - now
@ -296,138 +365,86 @@ def minutes_until_next_lesson(now: datetime | None = None) -> str:
days = total_seconds // 86400 days = total_seconds // 86400
hours = (total_seconds % 86400) // 3600 hours = (total_seconds % 86400) // 3600
minutes = (total_seconds % 3600) // 60 minutes = (total_seconds % 3600) // 60
return f"Твой мозг окончательно сгниет, а очко порвется через: {days} дн {hours} ч {minutes} мин" 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): async def handle_start(message: Message):
await message.answer("Соси") await message.answer("Соси")
async def handle_break_cmd(message: Message):
async def handle_help(message: Message):
await message.answer(HINT_TEXT)
async def handle_break(message: Message):
await message.answer(minutes_until_break()) await message.answer(minutes_until_break())
async def handle_penis_cmd(message: Message):
if not message.from_user: async def handle_maket(message: Message):
await message.answer("Команда доступна только пользователям.") parts = message.text.split(maxsplit=2)
if len(parts) == 1:
await message.reply(f"Доступные макеты: {templates_list_text()}\nФормат: /maket «id» «текст»")
return return
display_name = format_penis_user_name(message.from_user) if len(parts) < 3:
await message.answer(play_penis(message.from_user.id, display_name)[1]) 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_top_penis_cmd(message: Message):
await message.answer(build_penis_top())
async def handle_size_cmd(message: Message): 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) size = round(random.uniform(0.1, 35.0), 1)
if size < 5: if size < 5:
status = "СВОи" status = "СВОи"
elif size < 10: elif size < 10:
@ -440,53 +457,86 @@ async def handle_size_cmd(message: Message):
status = "Пиздабол" status = "Пиздабол"
else: else:
status = "Говноед" status = "Говноед"
await message.answer(f"📏 Твой хуек: {size} см\nТы: {status}")
async def handle_zvetok_cmd(message: Message): await message.answer(
await message.answer(minutes_until_next_lesson()) f"📏 Твой хуек: {size} см\n"
f"Ты: {status}"
)
async def handle_prices_cmd(message: Message): 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(): def fetch():
return requests.get( return requests.get(
"https://bebrik.xyz/magnit/products", "https://bebrik.xyz/magnit/products",
headers={"X-Token": "sisi"}, headers={"X-Token": "sisi"},
timeout=10, timeout=10,
) )
try: try:
resp = await asyncio.to_thread(fetch) resp = await asyncio.to_thread(fetch)
if resp.status_code != 200: if resp.status_code != 200:
await message.answer("пошел нахуй") await message.answer("пошел нахуй")
return return
data = resp.json() data = resp.json()
if not data: if not data:
await message.answer("иди нахуй") await message.answer("иди нахуй")
return return
text = "смерть в 25:\n\n" text = "смерть в 25:\n\n"
for item in data: for item in data:
name = item.get("name") name = item.get("name")
price = item.get("price") price = item.get("price")
text += f"{name}{price}\n" text += f"{name}{price}\n"
await message.answer(text[:4000]) await message.answer(text[:4000])
except Exception: except Exception:
await message.answer("unable to connect x-server гандон") await message.answer("unable to connect x-server гандон")
async def handle_fetch_cmd(message: Message):
async def handle_fetch(message: Message):
def fetch(): def fetch():
return requests.get( return requests.get(
"https://mirror.porno4free.ru/magnit/magnit/prices/all", "https://mirror.porno4free.ru/magnit/magnit/prices/all",
headers={"x-token": "secret_token"}, headers={"x-token": "secret_token"},
timeout=10, timeout=10,
) )
try: try:
resp = await asyncio.to_thread(fetch) resp = await asyncio.to_thread(fetch)
if resp.status_code != 200: if resp.status_code != 200:
await message.reply("пошел нахуй") await message.reply("пошел нахуй")
return return
data = resp.json() data = resp.json()
if not isinstance(data, dict) or not data: if not isinstance(data, dict) or not data:
await message.reply("иди нахуй") await message.reply("иди нахуй")
return return
lines = ["смерть в 25:"] lines = ["смерть в 25:"]
# сортируем категории для стабильного вывода
for group in sorted(data.keys()): for group in sorted(data.keys()):
items = data.get(group) items = data.get(group)
lines.append(f"\n{group}:") lines.append(f"\n{group}:")
@ -498,162 +548,67 @@ async def handle_fetch_cmd(message: Message):
if not name or price is None: if not name or price is None:
continue continue
lines.append(f"{name}{price}") lines.append(f"{name}{price}")
await message.answer("\n".join(lines)[:4000]) text = "\n".join(lines)
await message.answer(text[:4000])
except Exception: except Exception:
await message.answer("unable to connect x-server гандон") await message.answer("unable to connect x-server гандон")
async def handle_svodka_cmd(message: Message): async def handle_zvetok(message: Message):
data = await asyncio.to_thread(get_military_data, "creamy_caprice", 5) result = minutes_until_next_lesson()
if data["status"] == "success": await message.answer(result)
response = (
f"в период с {data['dates']} хохлы в среднем проебывали по {data['pace']} км² в сутки, "
f"а за день ВС РФ лишала чубатых в среднем по {data['total']} км²"
)
await message.answer(response)
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"))
_last_autoreply_ts = 0
async def handle_keywords(message: Message):
global _last_autoreply_ts
if not message.text or message.text.startswith('/'):
return
text_lower = message.text.lower()
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
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=message.chat.id, action="typing")
response = await generate_autoreply(message.text)
if response:
await message.reply(response, parse_mode=None)
_last_autoreply_ts = now
except Exception:
logger.exception("Autoreply failed")
# --- Запуск ---
def main(): def main():
token = os.getenv("BOT_TOKEN") token = os.getenv("BOT_TOKEN")
if not token: raise RuntimeError("BOT_TOKEN не задан!") if not token:
raise RuntimeError("BOT_TOKEN env var is required")
init_penis_db()
bot = Bot(token=token, default=DefaultBotProperties(parse_mode=ParseMode.HTML)) bot = Bot(token=token, default=DefaultBotProperties(parse_mode=ParseMode.HTML))
dp = Dispatcher() 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): async def on_startup(bot: Bot):
try: try:
commands = [ await bot.set_my_commands(
[
BotCommand(command="start", description="Приветствие"), BotCommand(command="start", description="Приветствие"),
BotCommand(command="help", description="Подсказки по командам"), BotCommand(command="help", description="Подсказки по командам"),
BotCommand(command="maket", description="Вписать текст в макет 1"), BotCommand(command="maket", description="Вписать текст в макет 1"),
BotCommand(command="gen_mem", description="Мем из фото (ответом)"), BotCommand(command="gen_mem", description="Мем из фото (ответом)"),
BotCommand(command="break", description="Сколько минут до перекура"), BotCommand(command="break", description="Сколько минут до перекура"),
BotCommand(command="size", description="хуй"), BotCommand(command="size", description="хуй"),
BotCommand(command="penis", description="миниигра 1 раз в 24ч"),
BotCommand(command="top_penis", description="топ по длине"),
BotCommand(command="prices", description="цены x-server"), BotCommand(command="prices", description="цены x-server"),
BotCommand(command="fetch", description="цены (новый источник)"), BotCommand(command="fetch", description="цены (новый источник)"),
BotCommand(command="zvetok", description="Цветянский бля"), BotCommand(command="zvetok", description="Цветянский бля"),
BotCommand(command="talk", description="Побазарить"), BotCommand(command="talk", description="Побазарить"),
BotCommand(command="svodka", description="СВО: итоги"),
] ]
scopes = (
BotCommandScopeDefault(),
BotCommandScopeAllPrivateChats(),
BotCommandScopeAllGroupChats(),
) )
for scope in scopes: except Exception: # noqa: WPS440
await bot.set_my_commands(commands, scope=scope)
except Exception:
logger.warning("Не удалось выставить команды, но бот продолжит работу.") logger.warning("Не удалось выставить команды, но бот продолжит работу.")
dp.startup.register(on_startup) dp.startup.register(on_startup)
logger.info("Бот запущен...")
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"))
dp.message.register(handle_talk, Command("talk"))
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())) asyncio.run(dp.start_polling(bot, allowed_updates=dp.resolve_used_update_types()))
if __name__ == "__main__": if __name__ == "__main__":
main() main()

View file

@ -1,6 +1,3 @@
aiogram>=3.4.1 aiogram>=3.4.1
pillow>=10.3.0 pillow>=10.3.0
requests>=2.31.0 requests>=2.31.0
beautifulsoup4
lxml
pysqlite3-binary>=0.5.3

View file

@ -1,55 +0,0 @@
import requests
from bs4 import BeautifulSoup
import re
import time
def get_military_data(channel_username, depth=5):
headers = {'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36'}
last_id = None
for i in range(depth):
url = f"https://t.me/s/{channel_username}"
if last_id:
url += f"?before={last_id}"
try:
response = requests.get(url, headers=headers, timeout=10)
soup = BeautifulSoup(response.text, 'html.parser')
messages = soup.find_all('div', class_='tgme_widget_message')
if not messages:
break
first_msg_data = messages[0].get('data-post')
if first_msg_data:
last_id = first_msg_data.split('/')[-1]
for msg in reversed(messages):
text_elem = msg.find('div', class_='tgme_widget_message_text')
if not text_elem:
continue
text = " ".join(text_elem.get_text(separator=' ').split())
text_lower = text.lower()
if "темп" in text_lower and "продвижения" in text_lower:
# Улучшенная регулярка для дат (ловит и "с 14 по 17", и "с 14 февраля по 17 февраля")
dates_match = re.search(r"с (.*?)(?=\.|\s+темп)", text_lower)
# Твои регулярки для цифр — отличные
pace_match = re.search(r"темп[^+-]*([+-]\d+[.,]\d+)", text_lower)
total_match = re.search(r"общее продвижение\s*(\d+[.,]\d+)", text_lower)
if pace_match:
return {
"status": "success",
"dates": dates_match.group(1).strip() if dates_match else "не указаны",
"pace": pace_match.group(1).replace(',', '.'),
"total": total_match.group(1).replace(',', '.') if total_match else "0",
"post_id": msg.get('data-post').split('/')[-1]
}
time.sleep(1)
except Exception as e:
print(f"Ошибка на итерации {i}: {e}")
continue
return {"status": "error", "message": "Сводка не найдена в последних постах"}