devzilla/discordBot
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

удаление консольных команд

Browse Source
This commit is contained in:
deadzilla 2026-07-07 20:52:11 +05:00
parent cc2808fb40
commit 900128fbd4
21 changed files with 3 additions and 1011 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
Dockerfile
View File

@ -18,8 +18,6 @@ RUN pip install --no-cache-dir -r requirements.txt && \
COPY bot.py .
COPY commands/ commands/
COPY utils/ utils/
COPY console_commands/ console_commands
# .env передаётся через docker-compose environment:
# DISCORD_TOKEN=${DISCORD_TOKEN}
# MORNING_TIME=${MORNING_TIME:-07:00}

9
ISSUES.md
View File

@ -2,20 +2,11 @@
## Средний приоритет
- [ ] **Добавить тесты для `console_commands/`** — есть `logs`, `reload`, `trigger_morning`, `help`. Не хватает: `pogoda`, `news`, `morning`, `cat`, `status`, `stats`
- [ ] **Проверка наличия `.env`** — есть проверка DISCORD_TOKEN в `_validate_config()`, но нет проверки существования самого файла `.env` (без файла бот падает с непонятной ошибкой)
- [ ] **Graceful shutdown (SIGTERM/SIGINT)** — нет обработчиков сигналов, бот завершается резко без закрытия handler-ов и cleanup
- [ ] **`asyncio.iscoroutinefunction` deprecated** — 35 warnings, функция удалена в Python 3.16; заменить на `inspect.iscoroutinefunction` (решение: обновление discord.py или monkey-patch в `bot.py`)
- [ ] **Утечка корутины в тестах** — `RuntimeWarning: coroutine 'trigger_morning.<locals>._run' was never awaited` в `test_trigger_morning.py`
## Консольные команды
### Низкий приоритет
- [ ] **`memory`** — текущее потребление памяти процесса
- [ ] **`health`** — проверка доступности внешних API (wttr.in, TheCatAPI, Habr)
- [ ] **`debug <on|off>`** — переключить verbose-режим бота
## Низкий приоритет
- [ ] **Добавить pre-commit хуки** — lint (flake8/ruff), форматирование (black)

73
README.md
View File

@ -14,7 +14,7 @@ pip install -r requirements.txt
python bot.py
```
Введите номер команды в терминале или `!команда` в Discord.
Используйте `!команда` в Discord.
## Настройка
@ -43,28 +43,10 @@ python bot.py
| `!status` | Статус бота: пинг к Discord gateway, uptime |
| `!stats` | Количество серверов, каналов, пользователей |
## Команды терминала
| Номер | Команда | Описание |
|-------|---------|----------|
| 1 | `news` | Топ-5 статей + топ-5 новостей с Habr |
| 2 | `pogoda` | Прогноз погоды для Магнитогорска |
| 3 | `morning` | Погода + топ-5 статей + топ-5 новостей + котик |
| 4 | `cat` | Вывести URL случайного котика |
| 5 | `help` | Показать список всех команд |
| 6 | `status` | Статус бота: пинг и uptime |
| 7 | `stats` | Количество серверов, каналов, пользователей |
| 8 | `logs` | Последние строки лога (tail -20) |
| 9 | `reload` | Горячая перезагрузка всех cogs |
| 10 | `trigger morning` | Ручной запуск morning-дайджеста |
| 0 | `stop` | Остановка бота |
> Номера команд генерируются автоматически из `ALL_CONSOLE_COMMANDS` в порядке определения в `console_commands/__init__.py`.
## Архитектура
```
bot.py # Точка входа, инициализация бота, console_input()
bot.py # Точка входа, инициализация бота
commands/ # Discord команды (cogs)
__init__.py # ALL_COMMANDS — явные импорты
pg.py # !pg — погода (обёртка над utils.pogoda)
@ -74,20 +56,6 @@ commands/ # Discord команды (cogs)
help.py # !hp — список команд (автогенерация из bot.commands)
status.py # !status — статус бота: пинг, uptime
stats.py # !stats — серверы, каналы, пользователи
console_commands/ # Консольные команды
__init__.py # ALL_CONSOLE_COMMANDS — явные импорты
admin.py # admin — CLI для docker exec (pogoda, news, cat, morning, help)
stop.py # stop — остановка бота
news.py # news — новости с Habr
pogoda.py # pogoda — погода в терминале
morning.py # morning — утренний дайджест в терминале
cat.py # cat — вывод URL котика
help.py # help — список всех команд
status.py # status — статус бота в терминале
stats.py # stats — статистика серверов в терминале
logs.py # logs — последние строки лога (tail)
reload.py # reload — горячая перезагрузка cogs
trigger_morning.py # trigger morning — ручной запуск morning-дайджеста
utils/ # Утилиты (API-клиенты, конвертации)
__init__.py # __all__ — публичный API утилит
pogoda.py # fetch_weather(), fetch_open_meteo(), wmo_to_russian(), translate_weather(), pressure_to_mmhg(), format_weather_data_for_console(), format_weather_for_embed()
@ -106,9 +74,7 @@ tests/ # pytest-тесты
test_bot.py # инициализация бота
test_morning_runner.py# тесты morning runner-а
test_help_discord.py # команда !hp — проверка формата вывода и контента
test_help_console.py # консольная help — проверка списка команд
test_logger.py # setup_logging — уровни, обработчики, формат
test_admin.py # admin.py — CLI-скрипт для docker exec
test_commands_status.py # команда !status — embed и uptime
test_commands_stats.py # команда !stats — подсчёт серверов и каналов
ISSUES.md # Задачи и баг-трекер проекта
@ -124,12 +90,6 @@ docker-compose.yml # Запуск бота в Docker
2. Добавить импорт в `commands/__init__.py`
3. Добавить класс в `ALL_COMMANDS`
### Добавление консольной команды
1. Создать файл `console_commands/имя.py` с функцией `func(stop_event, bot)`
2. Добавить импорт в `console_commands/__init__.py`
3. Добавить функцию в `ALL_CONSOLE_COMMANDS`
## Запуск тестов
```bash
@ -149,17 +109,11 @@ python -m pytest tests/ -v
| `test_bot.py` | инициализация бота | 7 |
| `test_morning_runner.py` | morning runner-а | 68 |
| `test_help_discord.py` | команда `!hp` | 2 |
| `test_help_console.py` | консольная `help` | 2 |
| `test_logger.py` | `setup_logging` (уровни, обработчики, формат) | 9 |
| `test_rate_limiter.py` | `RateLimiter` (токен-бакет) | 5 |
| `test_admin.py` | `admin.py` — CLI для docker exec | 5 |
| `test_commands_status.py` | команда `!status` (embed, uptime) | 6 |
| `test_commands_stats.py` | команда `!stats` (серверы, каналы) | 4 |
| `test_console_logs.py` | команда `logs` (чтение лога) | 4 |
| `test_console_reload.py` | команда `reload` (перезагрузка cogs) | 2 |
| `test_console_trigger_morning.py` | команда `trigger morning` (запуск дайджеста) | 3 |
**Итого: 243 теста.**
**Итого: 227 теста.**
## Запуск в Docker
@ -179,27 +133,8 @@ DISCORD_TOKEN=ваш_токен docker-compose up
- База: `python:3.14-slim`
- Healthcheck: проверка каждые 30 сек (старт-период 60 сек)
- Консольный ввод отключён в Docker (stdin недоступен)
- Версия Python настраивается через `ARG PYTHON_VERSION`
### Администрирование через docker exec
Для управления ботом из терминала (без Discord-чата) используйте `admin.py`:
```bash
docker exec discord-bot python console_commands/admin.py pogoda
docker exec discord-bot python console_commands/admin.py news
docker exec discord-bot python console_commands/admin.py cat
docker exec discord-bot python console_commands/admin.py morning
docker exec discord-bot python console_commands/admin.py logs # последние строки лога
docker exec discord-bot python console_commands/admin.py help
docker stop discord-bot # остановка бота
```
Команды `reload` и `trigger morning` доступны только через интерактивный терминал бота (требуют запущенного экземпляра бота).
Результат выводится в stdout терминала. Команды используют те же `utils`, что и Discord-команды.
## API и внешние сервисы
### Погода (!pg, !morning)
@ -271,8 +206,6 @@ docker stop discord-bot # остановка бота
- Максимальный объём: ~25 МБ
- **Уровень**: настраивается через `LOG_LEVEL` в `.env` (по умолчанию `INFO`)
Консольная команда `logs` (номер 8) выводит последние 20 строк `logs/bot.log`.
## Зависимости
```txt

25
console_commands/__init__.py
View File

@ -1,25 +0,0 @@
from .stop import stop
from .news import news
from .cat import cat
from .pogoda import pogoda
from .morning import morning
from .help import help as console_help
from .status import status
from .stats import stats
from .logs import logs
from .reload import reload as reload_cogs
from .trigger_morning import trigger_morning
ALL_CONSOLE_COMMANDS = {
"stop": stop,
"news": news,
"cat": cat,
"pogoda": pogoda,
"morning": morning,
"help": console_help,
"status": status,
"stats": stats,
"logs": logs,
"reload": reload_cogs,
"trigger morning": trigger_morning,
}

234
console_commands/admin.py
View File

@ -1,234 +0,0 @@
"""Админ-скрипт для управления ботом через docker exec.
Использование:
docker exec discord-bot python admin.py pogoda
docker exec discord-bot python admin.py news
docker exec discord-bot python admin.py cat
docker exec discord-bot python admin.py morning
docker exec discord-bot python admin.py help
"""
import sys
import asyncio
async def run_logs():
"""Вывести последние строки лога."""
from console_commands.logs import LOG_FILE, DEFAULT_LINES
lines = DEFAULT_LINES
if len(sys.argv) > 2:
try:
lines = int(sys.argv[2])
except ValueError:
pass
from pathlib import Path
log_file = LOG_FILE
if not log_file.exists():
print(f"Файл лога не найден: {log_file}")
return
try:
with open(log_file, "r", encoding="utf-8") as f:
all_lines = f.readlines()
tail = all_lines[-lines:] if len(all_lines) > lines else all_lines
print(f"\nПоследние {len(tail)} строк {log_file}:")
print("-" * 40)
print("".join(tail), end="")
print("-" * 40)
except OSError as e:
print(f"Ошибка чтения лога: {e}")
def print_help():
"""Показать список команд."""
print("\nАдмин-команды:")
print("-" * 40)
commands = [
("pogoda", "Прогноз погоды в Магнитогорске"),
("news", "Топ-5 статей и новостей AI с Habr"),
("cat", "URL случайного котика"),
("morning", "Утренний дайджест: погода + новости + котик"),
("logs", "Последние строки лога (tail -20)"),
("help", "Показать этот список"),
]
for name, desc in commands:
print(f" {name:<12} — {desc}")
print("-" * 40)
print("\n status — через Discord: !status")
print(" stats — через Discord: !stats")
print(" reload — через терминал бота (интерактив)")
print(" trigger morning — через терминал бота (интерактив)")
print(" stop — docker stop discord-bot\n")
async def run_pogoda():
"""Вывести прогноз погоды для Магнитогорска."""
from utils.pogoda import (
API_URL_WEATHER,
fetch_weather,
format_weather_data_for_console,
)
data = await fetch_weather(API_URL_WEATHER)
if data is None:
print("Не удалось получить данные о погоде.")
return
formatted = format_weather_data_for_console(data)
if not formatted:
print("Не удалось получить данные о погоде.")
return
for line in formatted:
print(line)
async def run_news():
"""Вывести топ-5 свежих статей по AI с Habr."""
from utils.news import (
RSS_URL_ARTICLES,
RSS_URL_POSTS,
fetch_rss,
format_articles,
)
articles = await fetch_rss(RSS_URL_ARTICLES)
if articles is None:
print("Не удалось получить новости.")
return
if not articles:
print("Новостей пока нет.")
return
lines = format_articles(
articles,
"Статьи AI / Хабр",
"https://habr.com/ru/hubs/artificial_intelligence/articles/top/daily/",
)
posts = await fetch_rss(RSS_URL_POSTS)
if posts is not None and posts:
lines.append("")
lines.extend(
format_articles(
posts,
"Новости AI / Хабр",
"https://habr.com/ru/hubs/artificial_intelligence/news/top/daily/",
)
)
for line in lines:
print(line)
async def run_cat():
"""Вывести URL случайного котика."""
from utils.cat import fetch_cat
url = await fetch_cat()
if url is None:
print("Не удалось получить котика.")
return
print(f"Котик: {url}")
async def run_morning():
"""Вывести утренний дайджест: погода + новости + котик."""
from utils.cat import fetch_cat
from utils.morning_runner import gather_morning
from utils.news import format_articles
from utils.pogoda import format_weather_data_for_console
data = await gather_morning()
print("Доброе утро!\n")
# --- Котик ---
if data.cat_url:
print(f"Котик: {data.cat_url}\n")
else:
print("Котика получить не удалось.\n")
# --- Погода ---
formatted = format_weather_data_for_console(data.weather)
if formatted:
print("**Погода в Магнитогорске:**")
for line in formatted:
print(line)
else:
print("Не удалось получить данные о погоде.")
print()
# --- Новости: статьи ---
if data.articles is not None:
if data.articles:
print(
"\n".join(
format_articles(
data.articles,
"Статьи AI / Хабр",
"https://habr.com/ru/hubs/"
"artificial_intelligence/articles/top/daily/",
)
)
)
else:
print("Новостей пока нет.")
else:
print("Не удалось получить новости.")
print()
# --- Новости: посты ---
if data.posts is not None:
if data.posts:
print(
"\n".join(
format_articles(
data.posts,
"Новости AI / Хабр",
"https://habr.com/ru/hubs/"
"artificial_intelligence/news/top/daily/",
)
)
)
else:
print("Новостей пока нет.")
else:
print("Не удалось получить новости.")
COMMANDS = {
"pogoda": run_pogoda,
"news": run_news,
"cat": run_cat,
"morning": run_morning,
"logs": run_logs,
"help": lambda: print_help(),
}
def main():
"""Точка входа: парсит аргумент и вызывает команду."""
if len(sys.argv) < 2:
print("Укажите команду.")
print_help()
sys.exit(1)
command = sys.argv[1].lower()
handler = COMMANDS.get(command)
if handler is None:
print(f"Неизвестная команда: {command}")
print_help()
sys.exit(1)
if command == "help":
handler()
else:
asyncio.run(handler())
if __name__ == "__main__":
main()

17
console_commands/cat.py
View File

@ -1,17 +0,0 @@
import logging
from utils.cat import fetch_cat
logger = logging.getLogger(__name__)
async def cat(stop_event, bot):
"""Вывести URL случайного котика"""
logger.info("Консольная команда: cat")
url = await fetch_cat()
if url is None:
logger.warning("Консольная команда cat: не удалось получить котика")
print("Не удалось получить котика.")
return
print(f"Котик: {url}")
logger.info("Консольная команда cat: выполнена")

46
console_commands/help.py
View File

@ -1,46 +0,0 @@
from typing import TYPE_CHECKING
def help(stop_event, bot):
"""Показать список доступных команд"""
# Проверка на завершение бота
if stop_event.is_set():
return None
commands = [
("!pg", "Прогноз погоды в Магнитогорске"),
("!nw", "Топ-5 статей и топ-5 новостей AI с Habr"),
("!morning", "Утренний дайджест: погода + новости + котик"),
("!cat", "Случайный котик"),
("!msg <текст>", "Повторить текст в чате"),
]
console_commands = [
("help", "Показать список всех команд"),
("pogoda", "Прогноз погоды в Магнитогорске"),
("news", "Топ-5 статей и новостей AI с Habr"),
("morning", "Утренний дайджест: погода + новости + котик"),
("cat", "Случайный котик"),
("stop", "Остановить бота"),
]
print("\n" + "=" * 60)
print("Discord Bot — Список команд")
print("=" * 60)
print()
print("Discord команды:")
print("-" * 40)
for cmd, desc in commands:
print(f" • {cmd:<20} — {desc}")
print()
print("-" * 40)
print("Консольные команды:")
print("-" * 40)
for cmd, desc in console_commands:
print(f" • {cmd:<20} — {desc}")
print()
print("=" * 60)

28
console_commands/logs.py
View File

@ -1,28 +0,0 @@
"""Просмотр последних строк лога."""
from pathlib import Path
LOG_FILE = Path("logs/bot.log")
DEFAULT_LINES = 20
def logs(stop_event, bot, lines: int = DEFAULT_LINES):
"""Показать последние строки лога (tail -N)."""
if stop_event.is_set():
return None
if not LOG_FILE.exists():
print(f"Файл лога не найден: {LOG_FILE}")
return None
try:
with open(LOG_FILE, "r", encoding="utf-8") as f:
all_lines = f.readlines()
tail = all_lines[-lines:] if len(all_lines) > lines else all_lines
print(f"\nПоследние {len(tail)} строк {LOG_FILE}:")
print("-" * 40)
print("".join(tail), end="")
print("-" * 40)
except OSError as e:
print(f"Ошибка чтения лога: {e}")

71
console_commands/morning.py
View File

@ -1,71 +0,0 @@
import logging
from utils.cat import fetch_cat
from utils.morning_runner import gather_morning
from utils.news import RSS_URL_ARTICLES, RSS_URL_POSTS, format_articles
from utils.pogoda import format_weather_data_for_console
logger = logging.getLogger(__name__)
async def morning(stop_event, bot):
"""Вывести погоду, лучшие статьи за сутки и котик"""
logger.info("Консольная команда: morning")
data = await gather_morning()
print("Доброе утро!\n")
# --- Котик ---
if data.cat_url:
print(f"Котик: {data.cat_url}\n")
else:
logger.warning("Консольная команда morning: не удалось получить котика")
print("Котика получить не удалось.\n")
# --- Погода ---
formatted = format_weather_data_for_console(data.weather)
if formatted:
print("**Погода в Магнитогорске:**")
for line in formatted:
print(line)
else:
logger.warning("Консольная команда morning: не удалось получить погоду")
print("Не удалось получить данные о погоде.")
print()
# --- Новости: статьи ---
if data.articles is not None:
if data.articles:
lines = format_articles(
data.articles,
"Лучшие статьи за сутки / Искусственный интеллект / Хабr",
"https://habr.com/ru/hubs/artificial_intelligence/articles/top/daily/",
)
print("\n".join(lines))
else:
logger.info("Консольная команда morning: статей нет в RSS")
print("Новостей пока нет.")
else:
logger.warning("Консольная команда morning: не удалось получить статьи")
print("Не удалось получить новости.")
print()
# --- Новости: посты ---
if data.posts is not None:
if data.posts:
lines = format_articles(
data.posts,
"Лучшие новости за сутки / Искусственный интеллект / Хабr",
"https://habr.com/ru/hubs/artificial_intelligence/news/top/daily/",
)
print("\n".join(lines))
else:
logger.info("Консольная команда morning: постов нет в RSS")
print("Новостей пока нет.")
else:
logger.warning("Консольная команда morning: не удалось получить посты")
print("Не удалось получить новости.")
logger.info("Консольная команда morning: завершен")

37
console_commands/news.py
View File

@ -1,37 +0,0 @@
import logging
from utils.news import fetch_rss, format_articles, RSS_URL_ARTICLES, RSS_URL_POSTS
logger = logging.getLogger(__name__)
async def news(stop_event, bot):
"""Вывести топ-5 свежих статей по AI с Habr"""
logger.info("Консольная команда: news")
articles = await fetch_rss(RSS_URL_ARTICLES)
if articles is None:
logger.warning("Консольная команда news: не удалось получить статьи")
print("Не удалось получить новости.")
return
if not articles:
logger.info("Консольная команда news: статей нет в RSS")
print("Новостей пока нет.")
return
lines = format_articles(articles, "Лучшие статьи за сутки / Искусственный интеллект / Хабr",
"https://habr.com/ru/hubs/artificial_intelligence/articles/top/daily/")
posts = await fetch_rss(RSS_URL_POSTS)
if posts is None:
logger.warning("Консольная команда news: не удалось получить посты")
elif posts:
lines.append("")
lines.extend(format_articles(posts, "Лучшие новости за сутки / Искусственный интеллект / Хабr",
"https://habr.com/ru/hubs/artificial_intelligence/news/top/daily/"))
else:
logger.info("Консольная команда news: постов нет в RSS")
for line in lines:
print(line)
logger.info("Консольная команда news: выполнена (статей: %d, постов: %d)", len(articles), len(posts) if posts else 0)

26
console_commands/pogoda.py
View File

@ -1,26 +0,0 @@
import logging
from utils.pogoda import API_URL_WEATHER, fetch_weather, format_weather_data_for_console
logger = logging.getLogger(__name__)
async def pogoda(stop_event, bot):
"""Вывести прогноз погоды для Магнитогорска"""
logger.info("Консольная команда: pogoda")
data = await fetch_weather(API_URL_WEATHER)
if data is None:
logger.warning("Консольная команда pogoda: не удалось получить данные о погоде")
print("Не удалось получить данные о погоде.")
return
formatted = format_weather_data_for_console(data)
if not formatted:
logger.warning("Консольная команда pogoda: данные погоды пустые")
print("Не удалось получить данные о погоде.")
return
for line in formatted:
print(line)
logger.info("Консольная команда pogoda: выполнена")

23
console_commands/reload.py
View File

@ -1,23 +0,0 @@
"""Горячая перезагрузка cogs без остановки бота."""
from commands import ALL_COMMANDS
def reload(stop_event, bot):
"""Перезагрузить все cogs бота."""
if stop_event.is_set():
return None
# Удалить все текущие cogs
cog_names = list(bot.cogs.keys())
for cog_name in cog_names:
bot.remove_cog(cog_name)
# Пересоздать и добавить все cogs
reloaded = []
for cog_class in ALL_COMMANDS:
cog = cog_class()
bot.add_cog(cog)
reloaded.append(cog.__class__.__name__)
print(f"\nПерезагружено cogs: {', '.join(reloaded)}")

31
console_commands/stats.py
View File

@ -1,31 +0,0 @@
import logging
import discord
logger = logging.getLogger(__name__)
def stats(stop_event, bot):
"""Показать статистику серверов: серверы, каналы, пользователи"""
if stop_event.is_set():
return None
logger.info("Консольная команда: stats")
guilds = bot.guilds
total_guilds = len(guilds)
total_channels = sum(
len([ch for ch in guild.channels if not isinstance(ch, discord.CategoryChannel)])
for guild in guilds
)
total_members = sum(guild.member_count or 0 for guild in guilds)
latency_ms = round(bot.latency * 1000, 1)
print("\n" + "=" * 40)
print("Статистика серверов")
print("=" * 40)
print(f" Серверов: {total_guilds}")
print(f" Каналов: {total_channels}")
print(f" Пользователей: {total_members}")
print(f" Пинг: {latency_ms} мс")
print("=" * 40)
logger.info("Консольная команда stats: выполнена")

44
console_commands/status.py
View File

@ -1,44 +0,0 @@
import logging
import time
logger = logging.getLogger(__name__)
def status(stop_event, bot):
"""Показать статус бота: пинг и время работы"""
if stop_event.is_set():
return None
logger.info("Консольная команда: status")
latency_ms = round(bot.latency * 1000, 1)
start_time = getattr(bot, "_start_time", time.time())
uptime_seconds = time.time() - start_time
uptime_str = _format_uptime(uptime_seconds)
print("\n" + "=" * 40)
print("Статус бота")
print("=" * 40)
print(f" Пинг: {latency_ms} мс")
print(f" Uptime: {uptime_str}")
print(f" Статус: Online")
print("=" * 40)
logger.info("Консольная команда status: выполнена")
def _format_uptime(total_seconds: float) -> str:
"""Форматировать секунды в человекочитаемый вид."""
days = int(total_seconds // 86400)
hours = int((total_seconds % 86400) // 3600)
minutes = int((total_seconds % 3600) // 60)
seconds = int(total_seconds % 60)
parts = []
if days:
parts.append(f"{days}д")
if hours:
parts.append(f"{hours}ч")
if minutes:
parts.append(f"{minutes}м")
parts.append(f"{seconds}с")
return " ".join(parts)

13
console_commands/stop.py
View File

@ -1,13 +0,0 @@
import asyncio
import logging
logger = logging.getLogger(__name__)
def stop(stop_event, bot):
"""Остановка бота"""
stop_event.set()
try:
asyncio.run_coroutine_threadsafe(bot.close(), bot.loop).result(timeout=5)
except Exception as e:
logger.error("Ошибка при остановке бота: %s", e)

22
console_commands/trigger_morning.py
View File

@ -1,22 +0,0 @@
"""Ручной запуск morning-дайджеста в канал."""
import asyncio
def trigger_morning(stop_event, bot):
"""Ручной запуск morning-дайджеста через scheduler."""
if stop_event.is_set():
return None
scheduler = getattr(bot, "_scheduler", None)
if scheduler is None:
print("Планировщик не запущен. Утренний дайджест недоступен.")
return None
async def _run():
await scheduler._run_morning()
# Запустить async-метод в event loop бота
future = asyncio.run_coroutine_threadsafe(_run(), bot.loop)
future.result()
print("\nУтренний дайджест запущен вручную.")

65
tests/test_admin.py
View File

@ -1,65 +0,0 @@
"""Тесты для admin.py — управление через docker exec."""
import subprocess
import sys
import pytest
ADMIN_SCRIPT = "console_commands/admin.py"
class TestAdminHelp:
"""Тесты команды help."""
def test_help_shows_commands(self, capfd):
"""Команда help выводит список команд."""
result = subprocess.run(
[sys.executable, ADMIN_SCRIPT, "help"],
capture_output=True,
text=True,
)
assert result.returncode == 0
output = result.stdout
assert "pogoda" in output
assert "news" in output
assert "cat" in output
assert "morning" in output
assert "help" in output
def test_no_args_shows_help(self, capfd):
"""Без аргументов выводит help и завершается с кодом 1."""
result = subprocess.run(
[sys.executable, ADMIN_SCRIPT],
capture_output=True,
text=True,
)
assert result.returncode == 1
assert "pogoda" in result.stdout
def test_unknown_command(self, capfd):
"""Неизвестная команда завершается с кодом 1."""
result = subprocess.run(
[sys.executable, ADMIN_SCRIPT, "unknown"],
capture_output=True,
text=True,
)
assert result.returncode == 1
assert "unknown" in result.stdout.lower()
class TestAdminImport:
"""Тесты импорта модуля."""
def test_admin_module_imports(self):
"""Модуль admin импортируется без ошибок."""
from console_commands import admin
assert hasattr(admin, "main")
assert hasattr(admin, "COMMANDS")
assert len(admin.COMMANDS) == 6
def test_commands_mapping(self):
"""Все команды зарегистрированы в маппинге."""
from console_commands.admin import COMMANDS
expected = {"pogoda", "news", "cat", "morning", "logs", "help"}
assert set(COMMANDS.keys()) == expected

87
tests/test_console_logs.py
View File

@ -1,87 +0,0 @@
"""Тесты для console_commands/logs.py."""
import importlib
from pathlib import Path
from unittest.mock import MagicMock
class TestLogsCommand:
"""Тесты консольной команды logs."""
def _get_module(self):
"""Импортировать модуль logs."""
return importlib.import_module("console_commands.logs")
def _get_logs_func(self):
"""Импортировать функцию logs."""
from console_commands.logs import logs
return logs
def test_logs_no_file(self, capfd):
"""Выводит сообщение если файл лога не существует."""
stop_event = MagicMock()
stop_event.is_set.return_value = False
bot = MagicMock()
mod = self._get_module()
original = mod.LOG_FILE
try:
mod.LOG_FILE = Path("/nonexistent/bot.log")
self._get_logs_func()(stop_event, bot)
finally:
mod.LOG_FILE = original
captured = capfd.readouterr()
assert "не найден" in captured.out
def test_logs_reads_last_lines(self, capfd, tmp_path):
"""Выводит последние N строк лога."""
log_file = tmp_path / "bot.log"
lines = [f"Line {i}\n" for i in range(30)]
log_file.write_text("".join(lines))
stop_event = MagicMock()
stop_event.is_set.return_value = False
bot = MagicMock()
mod = self._get_module()
original = mod.LOG_FILE
try:
mod.LOG_FILE = log_file
self._get_logs_func()(stop_event, bot, lines=5)
finally:
mod.LOG_FILE = original
captured = capfd.readouterr()
assert "Line 25" in captured.out
assert "Line 29" in captured.out
assert "Line 1" not in captured.out
def test_logs_fewer_lines_than_requested(self, capfd, tmp_path):
"""Если строк меньше N, выводит все."""
log_file = tmp_path / "bot.log"
log_file.write_text("Line 1\nLine 2\n")
stop_event = MagicMock()
stop_event.is_set.return_value = False
bot = MagicMock()
mod = self._get_module()
original = mod.LOG_FILE
try:
mod.LOG_FILE = log_file
self._get_logs_func()(stop_event, bot, lines=10)
finally:
mod.LOG_FILE = original
captured = capfd.readouterr()
assert "Line 1" in captured.out
assert "Line 2" in captured.out
def test_logs_stop_event(self):
"""Не выполняется если stop_event установлен."""
stop_event = MagicMock()
stop_event.is_set.return_value = True
bot = MagicMock()
result = self._get_logs_func()(stop_event, bot)
assert result is None

54
tests/test_console_reload.py
View File

@ -1,54 +0,0 @@
"""Тесты для console_commands/reload.py."""
import importlib
from unittest.mock import MagicMock
class TestReloadCommand:
"""Тесты консольной команды reload."""
def _get_module(self):
"""Импортировать модуль reload."""
return importlib.reload(importlib.import_module("console_commands.reload"))
def test_reload_removes_and_adds_cogs(self, capfd):
"""Удаляет старые cogs и добавляет новые."""
stop_event = MagicMock()
stop_event.is_set.return_value = False
bot = MagicMock()
bot.cogs = {"TestCog1": MagicMock(), "TestCog2": MagicMock()}
cog_class1 = MagicMock()
cog_class1.__name__ = "TestCog1"
cog_instance1 = MagicMock()
cog_instance1.__class__.__name__ = "TestCog1"
cog_class1.return_value = cog_instance1
cog_class2 = MagicMock()
cog_class2.__name__ = "TestCog2"
cog_instance2 = MagicMock()
cog_instance2.__class__.__name__ = "TestCog2"
cog_class2.return_value = cog_instance2
mod = self._get_module()
original = mod.ALL_COMMANDS
try:
mod.ALL_COMMANDS = [cog_class1, cog_class2]
mod.reload(stop_event, bot)
finally:
mod.ALL_COMMANDS = original
assert bot.remove_cog.call_count == 2
assert bot.add_cog.call_count == 2
captured = capfd.readouterr()
assert "Перезагружено" in captured.out
def test_reload_stop_event(self):
"""Не выполняется если stop_event установлен."""
stop_event = MagicMock()
stop_event.is_set.return_value = True
bot = MagicMock()
mod = self._get_module()
result = mod.reload(stop_event, bot)
assert result is None

53
tests/test_console_trigger_morning.py
View File

@ -1,53 +0,0 @@
"""Тесты для console_commands/trigger_morning.py."""
import importlib
from unittest.mock import MagicMock, patch
class TestTriggerMorningCommand:
"""Тесты консольной команды trigger morning."""
def test_trigger_morning_no_scheduler(self, capfd):
"""Выводит сообщение если scheduler не найден."""
stop_event = MagicMock()
stop_event.is_set.return_value = False
bot = MagicMock()
# Убедимся что _scheduler не установлен через getattr
type(bot)._scheduler = None
mod = importlib.import_module("console_commands.trigger_morning")
mod.trigger_morning(stop_event, bot)
captured = capfd.readouterr()
assert "не запущен" in captured.out
def test_trigger_morning_stop_event(self):
"""Не выполняется если stop_event установлен."""
stop_event = MagicMock()
stop_event.is_set.return_value = True
bot = MagicMock()
mod = importlib.import_module("console_commands.trigger_morning")
result = mod.trigger_morning(stop_event, bot)
assert result is None
def test_trigger_morning_calls_scheduler(self, capfd):
"""Вызывает scheduler._run_morning()."""
stop_event = MagicMock()
stop_event.is_set.return_value = False
scheduler = MagicMock()
bot = MagicMock()
bot._scheduler = scheduler
mock_future = MagicMock()
mock_future.result.return_value = None
mod = importlib.import_module("console_commands.trigger_morning")
with patch.object(
mod, "asyncio",
MagicMock(run_coroutine_threadsafe=MagicMock(return_value=mock_future))
):
mod.trigger_morning(stop_event, bot)
captured = capfd.readouterr()
assert "запущен вручную" in captured.out

54
tests/test_help_console.py
View File

@ -1,54 +0,0 @@
"""Тесты для консольной команды help."""
import io
from contextlib import redirect_stdout
from unittest.mock import MagicMock
class TestHelpCommandConsole:
"""Тесты для консольной команды help."""
def test_stop_event_check(self):
"""Проверка работы с остановленным ботом."""
from console_commands.help import help
stop_event = MagicMock()
stop_event.is_set.return_value = True
f = io.StringIO()
with redirect_stdout(f):
result = help(stop_event, MagicMock())
assert result is None
def test_output_contains_all_commands(self):
"""Проверка наличия всех команд в выводе."""
from console_commands.help import help
stop_event = MagicMock()
stop_event.is_set.return_value = False
f = io.StringIO()
with redirect_stdout(f):
help(stop_event, MagicMock())
output = f.getvalue()
# Discord команды
assert "!pg" in output
assert "!nw" in output
assert "!morning" in output
assert "!cat" in output
assert "!msg" in output
# Консольные команды
assert "help" in output
assert "pogoda" in output
assert "news" in output
assert "morning" in output
assert "cat" in output
assert "stop" in output
if __name__ == "__main__":
pytest.main([__file__, "-v"])