- console_commands/logs.py — чтение последних строк лога (tail -20) - console_commands/reload.py — горячая перезагрузка всех cogs - console_commands/trigger_morning.py — ручной запуск morning-дайджеста - logs доступна через admin.py (docker exec) и интерактивный терминал - reload и trigger morning доступны через интерактивный терминал - сохранён bot._scheduler для доступа к планировщику - обновлены __init__.py, admin.py, README.md, ISSUES.md - добавлены тесты: test_console_logs.py (4), test_console_reload.py (2), test_console_trigger_morning.py (3) - итого 243 теста
235 lines
7.0 KiB
Python
235 lines
7.0 KiB
Python
"""Админ-скрипт для управления ботом через 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()
|