Compare commits

..

1 commit

6 changed files with 187 additions and 333 deletions

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 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"]

105
config.py Normal file
View file

@ -0,0 +1,105 @@
# ВСЕ КОНСТАНТНЫЕ ЗНАЧЕНИЯ
import os
from pathlib import Path
from datetime import timedelta, time
def make_time(h: int, m: int) -> time:
return time(hour=h, minute=m)
# Базовые пути
BASE_DIR = Path(__file__).resolve().parent
FONT_PATH = BASE_DIR / "impact.ttf"
DEFAULT_TZ = os.getenv("TZ", "Europe/Moscow")
# Состояние мини-игры
PENIS_STATE: dict[int, dict[str, object]] = {}
PENIS_START_LENGTH = 10.0
PENIS_MIN_DELTA = 0.1
PENIS_MAX_DELTA = 7.0
PENIS_COOLDOWN = timedelta(hours=24)
PENIS_TOP_LIMIT = 10
# Твои шаблоны и тот самый огромный кортеж
TEMPLATES = {
"1": {
"path": BASE_DIR / "maket" / "1.png",
"anchor": (104, 240),
"padding": 20,
"label": "Владимир Владимирович Путин самый лучший президент",
},
# Сюда вставь весь остальной огромный массив
}
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 — выходной
}
# Ключевые слова, на которые бот кидает трек Creep.
KEYWORDS = {'крип', 'радиохэд', 'creep', 'radiohead'}
def templates_list_text() -> str:
parts = []
for tid in sorted(TEMPLATES.keys()):
label = TEMPLATES[tid].get("label", "")
if label:
parts.append(f"{tid}{label}")
else:
parts.append(tid)
return ", ".join(parts) if parts else "нет макетов"
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)."
)

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

353
main.py
View file

@ -1,155 +1,40 @@
import asyncio # Сделал чуть компактнее и читабельнее
from AI.talk_handler import handle_talk import asyncio, logging, os, re, requests, random
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 from textwrap import wrap
import random from datetime import datetime, time, timedelta
import sqlite3 from zoneinfo import ZoneInfo
from aiogram import Bot, Dispatcher, F, types from aiogram import Bot, Dispatcher, F, types
from aiogram.enums import ParseMode from aiogram.enums import ParseMode
from aiogram.filters import Command, CommandStart from aiogram.filters import Command, CommandStart
from aiogram.types import ( from aiogram.types import BufferedInputFile, Message, BotCommand, FSInputFile
BufferedInputFile,
Message,
BotCommand,
FSInputFile,
BotCommandScopeDefault,
BotCommandScopeAllPrivateChats,
BotCommandScopeAllGroupChats,
)
from aiogram.client.default import DefaultBotProperties from aiogram.client.default import DefaultBotProperties
from PIL import Image, ImageDraw, ImageFont from PIL import Image, ImageDraw, ImageFont
from datetime import datetime, time, timedelta from AI.talk_handler import handle_talk
from zoneinfo import ZoneInfo
from zparser import get_military_data
# Базовые пути/настройки проекта. # Конфигурационный файл, куда я вынес все константы
BASE_DIR = Path(__file__).parent import config
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": "Владимир Владимирович Путин самый лучший президент",
},
}
def init_penis_db() -> None:
# Инициализация таблицы для миниигры /penis.
with sqlite3.connect(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(TEMPLATES.keys()):
label = TEMPLATES[tid].get("label", "")
if label:
parts.append(f"{tid}{label}")
else:
parts.append(tid)
return ", ".join(parts) if parts else "нет макетов"
# ---- Schedule & breaks ---- # ---- 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: def minutes_until_break(now: datetime | None = None) -> str:
# Возвращает статус: до перекура / идёт перекур / пары окончены. # Возвращает статус: до перекура / идёт перекур / пары окончены.
if now is None: if now is None:
now = datetime.now(ZoneInfo(DEFAULT_TZ)) now = datetime.now(ZoneInfo(config.DEFAULT_TZ))
elif now.tzinfo is None: elif now.tzinfo is None:
now = now.replace(tzinfo=ZoneInfo(DEFAULT_TZ)) now = now.replace(tzinfo=ZoneInfo(config.DEFAULT_TZ))
day = now.weekday() day = now.weekday()
today_schedule = SCHEDULE.get(day, []) today_SCHEDULE = config.SCHEDULE.get(day, [])
if not today_schedule: if not today_SCHEDULE:
return "Сегодня кайфуем дома, перекур не нужен." return "Сегодня кайфуем дома, перекур не нужен."
def dt_for(t: time) -> datetime: def dt_for(t: time) -> datetime:
return now.replace(hour=t.hour, minute=t.minute, second=0, microsecond=0) return now.replace(hour=t.hour, minute=t.minute, second=0, microsecond=0)
for start_t, end_t, br_minutes in today_schedule: for start_t, end_t, br_minutes in today_SCHEDULE:
start_dt = dt_for(start_t) start_dt = dt_for(start_t)
end_dt = dt_for(end_t) end_dt = dt_for(end_t)
@ -172,19 +57,6 @@ 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"
"• /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) logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
@ -192,9 +64,9 @@ logger = logging.getLogger(__name__)
def load_font(size: int) -> ImageFont.FreeTypeFont: def load_font(size: int) -> ImageFont.FreeTypeFont:
# Грузим Impact нужного размера. # Грузим Impact нужного размера.
if not FONT_PATH.exists(): if not config.FONT_PATH.exists():
raise FileNotFoundError("impact.ttf not found. Place the font in the project root.") raise FileNotFoundError("impact.ttf not found. Place the font in the project root.")
return ImageFont.truetype(str(FONT_PATH), size=size) 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]: def wrap_text_to_width(text: str, font: ImageFont.FreeTypeFont, max_width: int, draw: ImageDraw.ImageDraw, stroke_width: int) -> list[str]:
@ -257,7 +129,7 @@ def draw_multiline(draw: ImageDraw.ImageDraw, lines: list[str], position: tuple[
def generate_maket(text: str, template_id: str = "1") -> BytesIO: def generate_maket(text: str, template_id: str = "1") -> BytesIO:
# Генерируем картинку макета с текстом в заранее заданной области. # Генерируем картинку макета с текстом в заранее заданной области.
tpl = TEMPLATES.get(template_id) tpl = config.TEMPLATES.get(template_id)
if not tpl: if not tpl:
raise FileNotFoundError(f"Template {template_id} is missing") raise FileNotFoundError(f"Template {template_id} is missing")
template_path = tpl["path"] template_path = tpl["path"]
@ -356,9 +228,9 @@ def minutes_until_next_lesson(now: datetime | None = None) -> str:
import random import random
if now is None: if now is None:
now = datetime.now(ZoneInfo(DEFAULT_TZ)) now = datetime.now(ZoneInfo(config.DEFAULT_TZ))
elif now.tzinfo is None: elif now.tzinfo is None:
now = now.replace(tzinfo=ZoneInfo(DEFAULT_TZ)) now = now.replace(tzinfo=ZoneInfo(config.DEFAULT_TZ))
LESSON_DAYS = [1, 3] # 1 = вторник, 3 = четверг LESSON_DAYS = [1, 3] # 1 = вторник, 3 = четверг
LESSON_START = time(9, 50) LESSON_START = time(9, 50)
@ -429,66 +301,38 @@ def minutes_until_next_lesson(now: datetime | None = None) -> str:
def play_penis(user_id: int, user_name: str | None = None, now: datetime | None = None) -> tuple[bool, str]: def play_penis(user_id: int, user_name: str | None = None, now: datetime | None = None) -> tuple[bool, str]:
# Миниигра /penis: изменение длины и кулдаун 24 часа от последнего броска. # Миниигра /penis: изменение длины и кулдаун 24 часа от последнего броска.
if now is None: if now is None:
now = datetime.now(ZoneInfo(DEFAULT_TZ)) now = datetime.now(ZoneInfo(config.DEFAULT_TZ))
elif now.tzinfo is None: 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: state = config.PENIS_STATE.get(user_id)
with sqlite3.connect(PENIS_DB_PATH) as conn: if state is None:
conn.row_factory = sqlite3.Row state = {"length": config.PENIS_START_LENGTH, "last_used_at": None}
row = conn.execute( config.PENIS_STATE[user_id] = state
"SELECT length, last_used_ts, display_name FROM penis_stats WHERE user_id = ?",
(user_id,),
).fetchone()
if row is None: # Обновляем отображаемое имя для топа.
current_length = PENIS_START_LENGTH if user_name:
last_used_ts = None state["display_name"] = user_name
stored_name = 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: last_used_at = state.get("last_used_at")
next_ts = int(last_used_ts) + PENIS_COOLDOWN_SECONDS if isinstance(last_used_at, datetime):
if now_ts < next_ts: next_time = last_used_at + config.PENIS_COOLDOWN
# Даже при кулдауне обновляем имя, чтобы топ показывал актуальный ник. if now < next_time:
if row is not None and row["display_name"] != display_name: # Если рано, считаем оставшееся время до следующей попытки.
conn.execute( left = next_time - now
"UPDATE penis_stats SET display_name = ? WHERE user_id = ?", total_seconds = int(left.total_seconds())
(display_name, user_id), hours = total_seconds // 3600
) minutes = (total_seconds % 3600) // 60
left_seconds = next_ts - now_ts return False, f"Сегодня уже кидал. Попробуй через {hours} ч {minutes} мин."
hours = left_seconds // 3600
minutes = (left_seconds % 3600) // 60
conn.commit()
return False, f"Сегодня уже кидал. Попробуй через {hours} ч {minutes} мин."
delta = round(random.uniform(PENIS_MIN_DELTA, PENIS_MAX_DELTA), 1) current_length = float(state.get("length", config.PENIS_START_LENGTH))
sign = random.choice([-1, 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) # Не даём длине уйти в отрицательные значения.
new_length = round(max(0.1, current_length + sign * delta), 1)
# Для нового пользователя создаём строку, для существующего — обновляем. state["length"] = new_length
if row is None: state["last_used_at"] = now
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 "-" sign_text = "+" if sign > 0 else "-"
return ( return (
@ -508,34 +352,25 @@ def format_penis_user_name(user: types.User) -> str:
return f"user_{user.id}" return f"user_{user.id}"
def build_penis_top(limit: int = PENIS_TOP_LIMIT) -> str: def build_penis_top(limit: int = config.PENIS_TOP_LIMIT) -> str:
# Формируем динамический топ по текущим значениям /penis. # Формируем динамический топ по текущим значениям /penis.
try: rows: list[tuple[float, str]] = []
with sqlite3.connect(PENIS_DB_PATH) as conn: for user_id, state in config.PENIS_STATE.items():
conn.row_factory = sqlite3.Row length = state.get("length")
rows = conn.execute( if not isinstance(length, (int, float)):
""" continue
SELECT user_id, display_name, length display_name = state.get("display_name")
FROM penis_stats if not isinstance(display_name, str) or not display_name.strip():
ORDER BY length DESC, user_id ASC display_name = f"user_{user_id}"
LIMIT ? rows.append((float(length), display_name.strip()))
""",
(limit,),
).fetchall()
except sqlite3.Error:
logger.exception("Failed to build /top_penis")
return "Не удалось загрузить топ. Ошибка БД."
if not rows: if not rows:
return "🏆 Топ пуст. Сначала пусть кто-то крутнет /penis." return "🏆 Топ пуст. Сначала пусть кто-то крутнет /penis."
rows.sort(key=lambda item: item[0], reverse=True)
medals = {1: "🥇", 2: "🥈", 3: "🥉"} medals = {1: "🥇", 2: "🥈", 3: "🥉"}
lines = ["🏆 Топ по длине:"] lines = ["🏆 Топ по длине:"]
for index, row in enumerate(rows, start=1): for index, (length, name) in enumerate(rows[:limit], 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}.") prefix = medals.get(index, f"{index}.")
lines.append(f"{prefix} {name}{length:.1f} см") lines.append(f"{prefix} {name}{length:.1f} см")
return "\n".join(lines) return "\n".join(lines)
@ -547,7 +382,7 @@ async def handle_start(message: Message):
async def handle_help(message: Message): async def handle_help(message: Message):
# Вывод подсказки по возможностям бота. # Вывод подсказки по возможностям бота.
await message.answer(HINT_TEXT) await message.answer(config.HINT_TEXT)
async def handle_break(message: Message): async def handle_break(message: Message):
# Команда расписания перекура. # Команда расписания перекура.
@ -558,14 +393,14 @@ async def handle_maket(message: Message):
# Генерация текста в подготовленном шаблоне /maket. # Генерация текста в подготовленном шаблоне /maket.
parts = message.text.split(maxsplit=2) parts = message.text.split(maxsplit=2)
if len(parts) == 1: if len(parts) == 1:
await message.reply(f"Доступные макеты: {templates_list_text()}\nФормат: /maket «id» «текст»") await message.reply(f"Доступные макеты: {config.templates_list_text()}\nФормат: /maket «id» «текст»")
return return
if len(parts) < 3: if len(parts) < 3:
await message.reply(f"Формат: /maket «id» «текст»\nДоступные макеты: {templates_list_text()}\n\n" + HINT_TEXT) await message.reply(f"Формат: /maket «id» «текст»\nДоступные макеты: {config.templates_list_text()}\n\n" + config.HINT_TEXT)
return return
template_id = parts[1] template_id = parts[1]
if template_id not in TEMPLATES: if template_id not in config.TEMPLATES:
await message.reply(f"Такого макета нет. Доступные: {templates_list_text()}") await message.reply(f"Такого макета нет. Доступные: {config.templates_list_text()}")
return return
text = parts[2].strip() text = parts[2].strip()
if not text: if not text:
@ -658,9 +493,7 @@ async def handle_top_penis(message: Message):
await message.answer(build_penis_top()) await message.answer(build_penis_top())
# Ключевые слова, на которые бот кидает трек Creep. # Смотреть config.py
KEYWORDS = {'крип', 'радиохэд', 'creep', 'radiohead'}
# I'M CREEP
async def handle_keywords(message: types.Message): async def handle_keywords(message: types.Message):
# Не реагируем на команды, только на обычный текст. # Не реагируем на команды, только на обычный текст.
@ -669,8 +502,8 @@ async def handle_keywords(message: types.Message):
text = message.text.lower() text = message.text.lower()
if any(re.search(rf'\b{word}\b', text) for word in KEYWORDS): if any(re.search(rf'\b{word}\b', text) for word in config.KEYWORDS):
audio_path = BASE_DIR / "maket" / "Radiohead - Creep.mp3" audio_path = config.BASE_DIR / "maket" / "Radiohead - Creep.mp3"
if audio_path.exists(): if audio_path.exists():
await message.answer_audio( await message.answer_audio(
@ -758,58 +591,34 @@ async def handle_zvetok(message: Message):
await message.answer(result) await message.answer(result)
async def handle_svodka(message: Message):
channel = "creamy_caprice"
data = get_military_data(channel, depth=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']}")
def main(): def main():
# Точка входа: создаём бота, регистрируем команды и запускаем polling. # Точка входа: создаём бота, регистрируем команды и запускаем polling.
token = os.getenv("BOT_TOKEN") token = os.getenv("BOT_TOKEN")
if not token: if not token:
raise RuntimeError("BOT_TOKEN env var is required") raise RuntimeError("BOT_TOKEN env var is required")
# Готовим SQLite-таблицу для /penis и /top_penis.
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()
async def on_startup(bot: Bot): async def on_startup(bot: Bot):
# Выставляем меню команд в интерфейсе Telegram. # Выставляем меню команд в интерфейсе Telegram.
try: try:
commands = [ await bot.set_my_commands(
BotCommand(command="start", description="Приветствие"), [
BotCommand(command="help", description="Подсказки по командам"), BotCommand(command="start", description="Приветствие"),
BotCommand(command="maket", description="Вписать текст в макет 1"), BotCommand(command="help", description="Подсказки по командам"),
BotCommand(command="gen_mem", description="Мем из фото (ответом)"), BotCommand(command="maket", description="Вписать текст в макет 1"),
BotCommand(command="break", description="Сколько минут до перекура"), BotCommand(command="gen_mem", description="Мем из фото (ответом)"),
BotCommand(command="size", description="хуй"), BotCommand(command="break", description="Сколько минут до перекура"),
BotCommand(command="penis", description="миниигра 1 раз в 24ч"), BotCommand(command="size", description="хуй"),
BotCommand(command="top_penis", description="топ по длине"), BotCommand(command="penis", description="миниигра 1 раз в 24ч"),
BotCommand(command="prices", description="цены x-server"), BotCommand(command="top_penis", description="топ по длине"),
BotCommand(command="fetch", description="цены (новый источник)"), BotCommand(command="prices", description="цены x-server"),
BotCommand(command="zvetok", description="Цветянский бля"), BotCommand(command="fetch", description="цены (новый источник)"),
BotCommand(command="talk", description="Побазарить"), BotCommand(command="zvetok", description="Цветянский бля"),
BotCommand(command="svodka", description="СВО: итоги"), BotCommand(command="talk", description="Побазарить"),
] ]
# Явно задаём команды для default/private/group, чтобы меню через "/" появлялось стабильно.
scopes = (
BotCommandScopeDefault(),
BotCommandScopeAllPrivateChats(),
BotCommandScopeAllGroupChats(),
) )
for scope in scopes:
await bot.set_my_commands(commands, scope=scope)
except Exception: # noqa: WPS440 except Exception: # noqa: WPS440
logger.warning("Не удалось выставить команды, но бот продолжит работу.") logger.warning("Не удалось выставить команды, но бот продолжит работу.")
@ -827,7 +636,7 @@ def main():
dp.message.register(handle_fetch, Command("fetch")) dp.message.register(handle_fetch, Command("fetch"))
dp.message.register(handle_zvetok, Command("zvetok")) dp.message.register(handle_zvetok, Command("zvetok"))
dp.message.register(handle_talk, Command("talk")) dp.message.register(handle_talk, Command("talk"))
dp.message.register(handle_svodka, Command("svodka"))
async def handle_gen_mem_cmd(message: Message, bot: Bot): async def handle_gen_mem_cmd(message: Message, bot: Bot):

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": "Сводка не найдена в последних постах"}