Compare commits

...

26 Commits

Author SHA1 Message Date
deadzilla
e387f330a0 Добавить планировщик утреннего дайджеста
- utils/morning_runner.py: Scheduler + run_morning()
- bot.py: автоматический запуск планировщика при старте
- commands/morning.py: использовать run_morning() вместо дублирования
- .env.example: добавить MORNING_TIME=07:00
- AGENTS.md: обновить документацию
- tests/test_morning_runner.py: 10 тестов для Scheduler
2026-05-29 17:48:26 +05:00
deadzilla
a3e11b1e9c Обновить README: добавить morning, utils, тесты, зависимости 2026-05-29 17:34:54 +05:00
deadzilla
64df3422b7 Добавить pytest-тесты и конфигурацию
- pytest.ini для конфигурации тестов
- tests/test_pogoda.py — тесты translate_weather, pressure_to_mmhg, wmo_to_russian (93 теста)
- tests/test_fetch_cat.py — тесты fetch_cat (10 тестов)
- tests/test_fetch_rss.py — тесты fetch_rss (20 тестов)
- tests/test_format_articles.py — тесты truncate_title, parse_date, format_articles (24 теста)
- tests/test_fetch_weather.py — тесты fetch_weather, fetch_open_meteo (20 тестов)
- tests/test_commands_pogoda.py — тесты команды !pogoda (13 тестов)
- Обновить AGENTS.md и requirements.txt
2026-05-29 15:45:56 +05:00
deadzilla
a277865e33 Убрать эмодзи из console_commands/cat.py и почистить ISSUES.md 2026-05-29 09:26:20 +05:00
deadzilla
e386ff3c42 docs: обновить AGENTS.md и очистить ISSUES.md с новым TODO-списком 2026-05-26 21:48:19 +05:00
deadzilla
1f9338d3ba Добавить второй блок новостей в !news, !morning и console morning 2026-05-26 21:25:14 +05:00
deadzilla
7df62e5e0a Добавить команду !morning (погода + новости) и console команду morning 2026-05-26 19:47:56 +05:00
deadzilla
6219ce42d0 refactor: вынести fetch_cat в utils, добавить повторный вывод списка команд 2026-05-26 18:56:11 +05:00
deadzilla
5292f9bd0e fix: использовать requests.Session для переиспользования соединений (ISSUE #8) 2026-05-26 18:17:17 +05:00
deadzilla
4425f27e1e fix: улучшить on_command_error — логирование деталей и нейтральное сообщение пользователю 2026-05-26 18:08:46 +05:00
deadzilla
768b6ce6aa fix: вынести импорт datetime на уровень модуля (ISSUE #4) 2026-05-26 17:55:28 +05:00
deadzilla
4b02fe8a35 docs: добавить правила согласования коммитов и русского языка сообщений в AGENTS.md 2026-05-26 14:03:04 +05:00
deadzilla
8090b4fe17 fix: resolve issue #3 (deduplication) and fix coroutine handling
- Create utils/news.py with shared RSS parsing and formatting logic
- Refactor commands/news.py and console_commands/news.py to use utils/news.py
- Fix bot.py to handle async console commands (news, pogoda)
- Fix utils/pogoda.py to fall back to Open-Meteo on requests.RequestException
- Mark issue #3 as resolved in ISSUES.md
2026-05-26 13:58:35 +05:00
deadzilla
04b0a9b82f refactor: вынести логику погоды в utils/pogoda.py (проблема 3)
- Создать utils/pogoda.py с общими функциями:
  - fetch_weather, fetch_open_meteo, wmo_to_russian, translate_weather, pressure_to_mmhg
- Обновить commands/pogoda.py: убрать дубликаты, импортировать из utils
- Обновить console_commands/pogoda.py: убрать дубликаты, импортировать из utils
- Сделать console_commands/pogoda.py async (требует fetch_weather)
- Обновить AGENTS.md и ISSUES.md (проблема 7 решена)
2026-05-26 12:32:15 +05:00
deadzilla
7a8db07862 chore: обновить .gitignore и удалить __pycache__ из репозитория
Добавлены паттерны:
- __pycache__/, *.py[cod], *.pyo - Python кэш
- venv/, env/, .venv/ - виртуальные окружения
- .env - переменные окружения
- *.log, *.bak, *.tmp, *.cache - временные файлы
- Thumbs.db, .DS_Store - кэш ОС
- .vscode/, .idea/ - IDE

Удалены из git:
- commands/__pycache__/__init__.cpython-314.pyc
- commands/__pycache__/pogoda.cpython-314.pyc
2026-05-26 11:54:26 +05:00
deadzilla
5f19dc1be1 Revert "chore: добавить Python cache и venv в .gitignore"
This reverts commit d39c27f4f71facdec95a5a93efd53500607d6510.
2026-05-26 10:54:47 +05:00
deadzilla
d39c27f4f7 chore: добавить Python cache и venv в .gitignore 2026-05-26 10:49:11 +05:00
deadzilla
d4bf8ca45d fix: исправить блокировку event loop и добавить retry задержку
- Добавить экспоненциальную задержку между попытками retry (1с, 2с, 4с) в pogoda.py
- Заменить time.sleep на await asyncio.sleep для неблокирующих ожиданий
- Обернуть requests.get в asyncio.to_thread для предотвращения блокировки event loop (news.py, cat.py, pogoda.py)
- Добавить правило AGENTS.md: не использовать эмодзи
- Добавить ISSUES.md с фиксацией проблем проекта
2026-05-26 10:24:26 +05:00
deadzilla
0e11dbe6c1 docs: добавить README.md 2026-05-26 09:24:32 +05:00
deadzilla
b8afbfa6b1 feat: добавить команду !cat и консольную команду pogoda 2026-05-26 00:25:31 +05:00
deadzilla
3f921680cf feat: обновить формат новостей (заголовок/дата+ссылка), добавить блок новостей с habr.com/ru/rss/hubs/artificial_intelligence/news/ 2026-05-25 23:58:54 +05:00
deadzilla
d5911b226d feat: вывести список команд после загрузки бота
- Добавить bot_ready = threading.Event() для синхронизации
- Переместить bot_ready.set() после цикла загрузки команд
- Добавить bot_ready.wait() в console_input()
- Меню появляется только после 'Бот вошёл' и 'Загружен: ...'
2026-05-25 21:57:59 +05:00
deadzilla
5c6e09eb58 feat: добавить команду !news для AI-новостей с Habr
- Парсинг RSS через ElementTree (RSS 2.0 / Atom)
- Данные: title, dc:creator, guid isPermaLink, pubDate, category
- Формат: заголовок до 60 символов, дата дд.мм.гггг, теги
- Ссылки без https://, кликабельные
- Консольная команда: !news
2026-05-25 12:04:44 +05:00
deadzilla
93f17c74d5 docs: обновить AGENTS.md с описанием архитектуры, API, конвертаций и безопасности 2026-05-25 00:24:56 +05:00
deadzilla
486e3124ed fix: исправить дублирование кода, добавить retry и fallback для погоды
- Убран дублирующийся on_command_error и импорт CommandNotFound
- stop.py: добавлены аргументы stop_event, bot + обработка ошибок
- console_commands интегрирован в console_input()
- pogoda.py: retry (3 попытки), fallback на Open-Meteo при SSL-ошибках
- pogoda.py: безопасная обработка wind_kmh и давления
- pogoda.py: сортировка translate-словаря по длине ключа
- Добавлен _wmo_to_russian() для WMO weather code
2026-05-25 00:07:20 +05:00
deadzilla
e78d34e3cb chore: migrate pogoda to wttr.in, fix bot stop, add on_command_error
- Replace Open-Meteo API with wttr.in (no API key needed)
- Convert wind speed from km/h to m/s
- Add weather description translation (EN → RU)
- Add "FeelsLikeC" to pogoda output
- Fix console stop command using asyncio.run_coroutine_threadsafe
- Add on_command_error handler to suppress CommandNotFound logs
2026-05-24 23:13:53 +05:00
35 changed files with 2640 additions and 91 deletions

1
.env Normal file
View File

@ -0,0 +1 @@
DISCORD_TOKEN=MTI4ODA4NTM0OTQxNDk5ODAyNw.Gk3VXx.ZevANnkcdoxtv-Wi8RCMXVa6RczHbUsmhByV3g

View File

@ -1 +1,2 @@
DISCORD_TOKEN=your_bot_token_here
MORNING_TIME=07:00

BIN
.gitignore vendored

Binary file not shown.

120
AGENTS.md
View File

@ -1,38 +1,114 @@
# AGENTS.md
## Проект
Discord-бот на Python (discord.py). Команды: `!msg` (повтор текста), `!pogoda` (прогноз погоды для Магнитогорска).
Discord-бот на Python (discord.py)
## Запуск
```
```bash
pip install -r requirements.txt
python bot.py
```
Введите `stop` в терминал для остановки.
## Токен
Токен бота хранится в `.env` (переменная `DISCORD_TOKEN`). Получи токен на [Discord Developer Portal](https://discord.com/developers/applications). Шаблон в `.env.example`.
## Конвенции
Используй TODO-списки для каждого запроса с несколькими шагами.
- Используй TODO-списки для каждого запроса с несколькими шагами.
- Пиши комментарии на русском.
- Обработка ошибок: try/except для всех внешних вызовов (API, БД, файловая система).
- **Никогда не используй эмодзи в тексте или выводах.**
- **До внесения любых изменений в код или файлы предоставь детальное описание всех планируемых изменений и получи явное согласие пользователя. Без согласования изменения не вносить.**
- **Все git-коммиты согласовывать с пользователем перед созданием.**
- **Все сообщения git-коммитов писать на русском языке.**
- **Думать и общаться с LLM на английском, а отвечать пользователю на русском.**
## Архитектура
```
commands/ # Discord команды (cogs)
__init__.py # ALL_COMMANDS — явные импорты
pogoda.py # !pogoda
console_commands/ # Консольные команды
__init__.py # ALL_CONSOLE_COMMANDS — явные импорты
stop.py # stop
## Команды
| Команда | Описание | Формат вывода |
|---------|----------|---------------|
| `!pogoda` | Погода в Магнитогорске | Температура, ощущается, описание, влажность, ветер, давление |
| `!news` | Топ-5 статей и топ-5 новостей AI с Habr | Два блока: статьи и новости |
| `!morning` | Погода + топ-5 статей + топ-5 новостей + котик | Embed: котик thumbnail, погода, статьи, новости |
| `!cat` | Случайный котик | Embed с изображением |
| `!msg <текст>` | Повторяет текст | Текст |
## Планировщик
- Ежедневный утренний дайджест (`!morning`) запускается автоматически в 07:00
- Время задаётся в `.env` переменной `MORNING_TIME` (формат `ЧЧ:ММ`)
- Реализован через `discord.ext.tasks.loop`
- Отправляет дайджест в первый канал, где бот имеет права
- Код в `utils/morning_runner.py`
## Структура проекта
| Каталог | Назначение |
|---------|------------|
| `commands/` | Discord-команды (Cogs) |
| `utils/` | Утилиты (API-клиенты, конвертации) |
| `console_commands/` | Консольные команды (для терминала) |
| `bot.py` | Точка входа, загрузка Cogs, обработка ошибок |
## API и внешние сервисы
### Погода (!pogoda, !morning)
- **Основной**: `wttr.in/Magnitogorsk` (бесплатный, без ключа)
- **Fallback**: `api.open-meteo.com` (бесплатный, без ключа)
- Retry: 3 попытки с экспоненциальной задержкой при SSL/Connection/Timeout ошибках
- Fallback срабатывает автоматически при неуспешных попытках
- WMO weather codes → русский перевод в `wmo_to_russian()`
### Конвертации
- Давление: hPa → мм рт. ст. (`* 0.750062`)
- Ветер: км/ч → м/с (`/ 3.6`)
- Погодные описания: английский → русский (`translate_weather()`)
### RSS-ленты Habr (!news, !morning)
| Тип | URL |
|-----|-----|
| Статьи | `https://habr.com/ru/rss/hubs/artificial_intelligence/articles/top/daily/?fl=ru` |
| Новости | `https://habr.com/ru/rss/hubs/artificial_intelligence/news/top/daily/?fl=ru` |
Парсятся RSS 2.0 / Atom, извлекаются title, pubDate, creator, category, guid.
### Cat API (!cat, !morning)
- **URL**: `https://api.thecatapi.com/v1/images/search`
- Возвращает JSON: `[{"url": "..."}]`
- Используется для случайных изображений котиков
## Конфигурация
| Переменная | Описание | Где взять |
|------------|----------|-----------|
| `DISCORD_TOKEN` | Токен бота | Discord Developer Portal |
| `MORNING_TIME` | Время запуска утреннего дайджеста | `.env` (формат `ЧЧ:ММ`, по умолчанию `07:00`) |
## Зависимости
```txt
discord.py>=2.3.2
python-dotenv>=1.0.0
requests>=2.31.0
```
**Добавление Discord команды:**
1. Создать файл `commands/имя.py` с классом, наследующим `commands.Cog`
2. Добавить импорт в `commands/__init__.py`
3. Добавить класс в `ALL_COMMANDS`
## Структура данных погоды
Команда `!pogoda` возвращает:
```
Температура: X°C (ощущается как Y°C)
Описание: Z
Влажность: X%
Ветер: X м/с
Давление: X мм рт. ст.
```
**Добавление консольной команды:**
1. Создать файл `console_commands/имя.py` с функцией
2. Добавить импорт в `console_commands/__init__.py`
3. Добавить функцию в `ALL_CONSOLE_COMMANDS`
## Консольные команды
Запускаются через терминал при старте бота. Список выводится при запуске.
| Команда | Описание |
|---------|----------|
| `pogoda` | Вывести погоду в Магнитогорск |
| `news` | Вывести топ-5 статей AI с Habr |
| `morning` | Вывести погоду, топ-5 статей и котик |
| `cat` | Заглушка |
| `stop` | Остановить бота |
## Безопасность
- `.env` в `.gitignore` — токен никогда не должен попадать в репозиторий.
- Используй `.env.example` как шаблон.

46
ISSUES.md Normal file
View File

@ -0,0 +1,46 @@
# Tasks — список задач проекта
> Состояние на: 2026-05-26
---
## 🔴 Критические
### 1. Убрать эмодзи из `console_commands/cat.py`
- [x] Заменить `print("🐱 тут должен быть котик")` на `print("Заглушка: тут должен быть котик")`
---
## 🟡 Средние
### 2. Убрать пустой embed в `!morning` при всех ошибках
- [ ] `commands/morning.py`: добавить fallback-сообщение в `description_lines`, если все API вернули `None`
- [ ] Проверить, что `embed.description` никогда не пустой
### 3. Убрать дублирование парсинга погоды
- [ ] `console_commands/pogoda.py`: вынести форматирование результата в `utils/pogoda.py` как `format_weather_data()`
- [ ] `console_commands/morning.py`: использовать `format_weather_data()` вместо inline-парсинга
- [ ] `commands/morning.py`: использовать `format_weather_data()` вместо inline-парсинга
### 6. Вынести `self.api_url` в константу класса
- [ ] `commands/pogoda.py`: заменить `self.api_url = ...` на `API_URL = "..."` на уровне класса
- [ ] `commands/morning.py`: аналогично
- [ ] `console_commands/pogoda.py`: вынести URL в `utils/pogoda.py` как константу
---
## 🟢 Малые улучшения
### 4. Вынести `api_url` в константу
- [ ] `commands/pogoda.py`: заменить `self.api_url = ...` на `API_URL = "..."` на уровне класса
- [ ] `commands/morning.py`: аналогично
- [ ] `console_commands/pogoda.py`: вынести URL в `utils/pogoda.py` как константу
### 5. Добавить `__all__` в `__init__.py`
- [ ] `commands/__init__.py`: `__all__ = ["ALL_COMMANDS"]`
- [ ] `console_commands/__init__.py`: `__all__ = ["ALL_CONSOLE_COMMANDS"]`
- [ ] `utils/__init__.py`: добавить `__all__` с основными функциями
### 6. Добавить `raise_exception=True` в `bot.run()`
- [ ] `bot.py`: добавить `raise_exception=True` для видимого падения при ошибках токена/сети
- [ ] Добавить логирование в `KeyboardInterrupt` блок

189
README.md Normal file
View File

@ -0,0 +1,189 @@
# Discord Bot
Discord-бот для Магнитогорска. Команды погоды, новостей, котиков и утреннего дайджеста.
## Установка
```bash
pip install -r requirements.txt
```
## Запуск
```bash
python bot.py
```
Введите номер команды в терминале или `!команда` в Discord.
## Настройка
1. Скопируйте `.env.example` в `.env`:
```bash
cp .env.example .env
```
2. Вставьте токен бота в `.env`:
```env
DISCORD_TOKEN=ваш_токен
```
Токен получите на [Discord Developer Portal](https://discord.com/developers/applications).
## Команды Discord
| Команда | Описание |
|---------|----------|
| `!pogoda` | Прогноз погоды для Магнитогорска |
| `!news` | Топ-5 свежих статей по AI с Habr |
| `!morning` | Погода + топ-5 статей + топ-5 новостей + котик (утренний дайджест) |
| `!cat` | Случайный котик |
| `!msg <текст>` | Повторить текст в чате |
## Команды терминала
| Номер | Команда | Описание |
|-------|---------|----------|
| 1 | `news` | Топ-5 статей + топ-5 новостей с Habr |
| 2 | `pogoda` | Прогноз погоды для Магнитогорска |
| 3 | `morning` | Погода + топ-5 статей + топ-5 новостей + котик |
| 4 | `cat` | Вывести URL случайного котика |
| 0 | `stop` | Остановка бота |
## Архитектура
```
bot.py # Точка входа, инициализация бота, console_input()
commands/ # Discord команды (cogs)
__init__.py # ALL_COMMANDS — явные импорты
pogoda.py # !pogoda — погода с retry + fallback
news.py # !news — новости с Habr
cat.py # !cat — случайный котик
morning.py # !morning — утренний дайджест
console_commands/ # Консольные команды
__init__.py # ALL_CONSOLE_COMMANDS — явные импорты
stop.py # stop — остановка бота
news.py # news — новости с Habr
pogoda.py # pogoda — погода в терминале
morning.py # morning — утренний дайджест в терминале
cat.py # cat — вывод URL котика
utils/ # Утилиты (API-клиенты, конвертации)
pogoda.py # fetch_weather(), fetch_open_meteo(), wmo_to_russian(), translate_weather(), pressure_to_mmhg()
news.py # fetch_rss(), format_articles(), truncate_title(), parse_date()
cat.py # fetch_cat()
tests/ # pytest-тесты
test_pogoda.py # translate_weather, pressure_to_mmhg, wmo_to_russian
test_fetch_cat.py # fetch_cat
test_fetch_rss.py # fetch_rss
test_fetch_weather.py # fetch_weather, fetch_open_meteo
test_format_articles.py # truncate_title, parse_date, format_articles
test_commands_pogoda.py # Pogoda cog
ISSUES.md # Задачи и баг-трекер проекта
pytest.ini # Конфигурация pytest
```
### Добавление Discord команды
1. Создать файл `commands/имя.py` с классом, наследующим `commands.Cog`
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
python -m pytest tests/ -v
```
### Структура тестов
| Файл | Что тестирует | Кол-во |
|------|---------------|--------|
| `test_pogoda.py` | `translate_weather()`, `pressure_to_mmhg()`, `wmo_to_russian()` | 93 |
| `test_fetch_cat.py` | `fetch_cat()` | 10 |
| `test_fetch_rss.py` | `fetch_rss()` | 20 |
| `test_fetch_weather.py` | `fetch_weather()`, `fetch_open_meteo()` | 20 |
| `test_format_articles.py` | `truncate_title()`, `parse_date()`, `format_articles()` | 24 |
| `test_commands_pogoda.py` | `Pogoda` cog, команда `!pogoda` | 13 |
**Итого: 180 тестов.**
## API и внешние сервисы
### Погода (!pogoda, !morning)
- **Основной**: `wttr.in/Magnitogorsk` (бесплатный, без ключа)
- **Fallback**: `api.open-meteo.com` (бесплатный, без ключа)
- Retry: 3 попытки с экспоненциальной задержкой при SSL/Connection/Timeout ошибках
- Fallback срабатывает автоматически при неуспешных попытках
- WMO weather codes → русский перевод в `wmo_to_russian()`
### Конвертации
- Давление: hPa → мм рт. ст. (`* 0.750062`)
- Ветер: км/ч → м/с (`/ 3.6`)
- Погодные описания: английский → русский (`translate_weather()`)
### Новости (!news, !morning)
- **Articles**: `https://habr.com/ru/rss/hubs/artificial_intelligence/articles/top/daily/?fl=ru`
- **News**: `https://habr.com/ru/rss/hubs/artificial_intelligence/news/top/daily/?fl=ru`
- Парсинг RSS 2.0 и Atom форматов
- Извлечение ссылок из `<guid isPermaLink="true">` и авторов из `<dc:creator>`
- Формат вывода: заголовок → дата → ссылка
### Котики (!cat, !morning)
- **API**: `https://api.thecatapi.com/v1/images/search`
- Картинка встраивается в Discord Embed
## Структура данных погоды
Команда `!pogoda` возвращает:
```
Температура: X°C (ощущается как Y°C)
Описание: Z
Влажность: X%
Ветер: X м/с
Давление: X мм рт. ст.
```
## Формат дат
Даты форматируются как `дд.мм.гггг` через `datetime.strptime` с форматом `%a, %d %b %Y %H:%M:%S %z`.
## Конфигурация
| Переменная | Описание | Где взять |
|------------|----------|-----------|
| `DISCORD_TOKEN` | Токен бота | [Discord Developer Portal](https://discord.com/developers/applications) |
## Зависимости
```txt
discord.py>=2.3.2
python-dotenv>=1.0.0
requests>=2.31.0
pytest>=7.4.0
pytest-asyncio>=0.21.0
```
## Безопасность
- `.env` в `.gitignore` — токен никогда не должен попадать в репозиторий
- Используйте `.env.example` как шаблон
## Формат новостей
Каждая новость выводится в формате:
```
Заголовок статьи
дд.мм.гггг https://habr.com/ru/articles/...
```
Заголовки обрезаются до 60 символов с суффиксом `...`.
Ссылки отображаются без `<>` и `https://` префикса для предотвращения embed-превью в Discord.

80
bot.py
View File

@ -1,4 +1,6 @@
import asyncio
import inspect
import logging
import os
import sys
import threading
@ -6,10 +8,13 @@ import threading
import discord
from discord.ext import commands
from discord.ext.commands import CommandNotFound
from discord.ext.commands import CommandNotFound
from dotenv import load_dotenv
from commands import ALL_COMMANDS
from console_commands import ALL_CONSOLE_COMMANDS
from utils.morning_runner import Scheduler
logger = logging.getLogger(__name__)
load_dotenv()
@ -18,10 +23,13 @@ intents.message_content = True
bot = commands.Bot(command_prefix="!", intents=intents)
stop_event = threading.Event()
bot_ready = threading.Event()
scheduler: Scheduler | None = None
@bot.event
async def on_ready():
global scheduler
print(f"Бот вошёл как {bot.user}")
for cog_class in ALL_COMMANDS:
cog = cog_class()
@ -29,19 +37,33 @@ async def on_ready():
for cog in bot.cogs:
print(f" Загружен: {cog}")
# Запуск планировщика
morning_time = os.getenv("MORNING_TIME", "07:00")
scheduler = Scheduler(bot, morning_time)
print(f" Планировщик запущен (время: {morning_time})")
@bot.event
async def on_command_error(ctx, error):
if isinstance(error, CommandNotFound):
return
print(f"Ошибка команды: {error}")
bot_ready.set()
@bot.event
async def on_command_error(ctx, error):
if isinstance(error, CommandNotFound):
return
print(f"Ошибка команды: {error}")
# Терминал — детали для разработчика
logger.error(
f"Ошибка команды {ctx.command.name if ctx else '?'}: {error}",
exc_info=True,
)
# Discord — только если команда не ответила сама
if ctx and ctx.interaction and ctx.interaction.response.is_done():
return
try:
await ctx.send("Не удалось выполнить команду. Попробуйте позже.")
except (discord.NotFound, discord.Forbidden):
pass # Бот не может писать в канал — игнорируем
@bot.command(name="msg")
@ -50,20 +72,48 @@ async def msg(ctx, *, text: str):
await ctx.send(text)
def _print_commands():
"""Вывести список доступных консольных команд."""
available = {k: v for k, v in ALL_CONSOLE_COMMANDS.items() if k != "stop"}
print("\nДоступные команды:")
for idx, (name, func) in enumerate(available.items(), 1):
print(f" {idx}. {name}")
print(" 0. stop")
def console_input():
bot_ready.wait()
_print_commands()
while not stop_event.is_set():
try:
cmd = input().strip().lower()
if cmd == "stop":
choice = input("\nВыберите команду (номер): ").strip()
if choice == "0":
print("\nОстановка бота...")
stop_event.set()
asyncio.run_coroutine_threadsafe(bot.close(), bot.loop).result()
asyncio.run_coroutine_threadsafe(bot.close(), bot.loop).result(timeout=5)
break
elif cmd:
print(f"Неизвестная команда: {cmd}")
except EOFError:
try:
available = {k: v for k, v in ALL_CONSOLE_COMMANDS.items() if k != "stop"}
idx = int(choice)
if 0 < idx <= len(available):
cmd_name = list(available.keys())[idx - 1]
cmd_func = ALL_CONSOLE_COMMANDS[cmd_name]
if inspect.iscoroutinefunction(cmd_func):
asyncio.run_coroutine_threadsafe(cmd_func(stop_event, bot), bot.loop).result()
else:
cmd_func(stop_event, bot)
else:
print(f"Неизвестная команда: {choice}")
except (ValueError, IndexError):
print(f"Неверный формат: {choice}")
_print_commands()
except (EOFError, KeyboardInterrupt):
stop_event.set()
asyncio.run_coroutine_threadsafe(bot.close(), bot.loop).result()
try:
asyncio.run_coroutine_threadsafe(bot.close(), bot.loop).result(timeout=5)
except Exception as e:
print(f"Ошибка при остановке бота: {e}")
break
@ -82,5 +132,7 @@ if __name__ == "__main__":
except KeyboardInterrupt:
print("\nОстановка бота...")
stop_event.set()
if scheduler:
scheduler.stop()
asyncio.run_coroutine_threadsafe(bot.close(), bot.loop).result()
sys.exit(0)

View File

@ -1,3 +1,6 @@
from .pogoda import Pogoda
from .news import News
from .cat import Cat
from .morning import Morning
ALL_COMMANDS = [Pogoda]
ALL_COMMANDS = [Pogoda, News, Cat, Morning]

23
commands/cat.py Normal file
View File

@ -0,0 +1,23 @@
import discord
from discord.ext import commands
from utils.cat import fetch_cat
class Cat(commands.Cog):
"""Команда !cat — случайный котик"""
@commands.command(name="cat")
async def cat(self, ctx):
"""Получить случайного котика"""
url = await fetch_cat()
if url is None:
await ctx.send("Не удалось получить котика. Попробуйте позже.")
return
embed = discord.Embed(
title="Котик для тебя!",
color=discord.Color.orange()
)
embed.set_image(url=url)
await ctx.send(embed=embed)

16
commands/morning.py Normal file
View File

@ -0,0 +1,16 @@
import discord
from discord.ext import commands
from utils.morning_runner import run_morning
class Morning(commands.Cog):
"""Команда !morning — погода и новости утром"""
def __init__(self):
pass
@commands.command(name="morning")
async def morning(self, ctx):
"""Погода, лучшие статьи за сутки и котик"""
await run_morning(ctx.bot, ctx.channel)

60
commands/news.py Normal file
View File

@ -0,0 +1,60 @@
import discord
from discord.ext import commands
from utils.news import fetch_rss, format_articles, RSS_URL_ARTICLES, RSS_URL_POSTS
class News(commands.Cog):
"""Команда !news — свежие статьи по AI с Habr"""
@commands.command(name="news")
async def news(self, ctx):
"""Топ-5 свежих статей и новостей по AI с Habr"""
articles = await fetch_rss(RSS_URL_ARTICLES)
if articles is None:
await ctx.send("Не удалось получить новости. Попробуйте позже.")
return
if not articles:
await ctx.send("Новостей пока нет.")
return
articles_text = format_articles(articles,
"Лучшие статьи за сутки / Искусственный интеллект / Хабr",
"https://habr.com/ru/hubs/artificial_intelligence/articles/top/daily/")
posts = await fetch_rss(RSS_URL_POSTS)
embed = discord.Embed(
title="Новости AI с Habr",
colour=discord.Color.orange(),
)
embed.add_field(
name="Статьи",
value="\n".join(articles_text),
inline=False,
)
if posts is None:
embed.add_field(
name="Новости",
value="Не удалось получить новости.",
inline=False,
)
elif posts:
posts_text = format_articles(posts,
"Лучшие новости за сутки / Искусственный интеллект / Хабr",
"https://habr.com/ru/hubs/artificial_intelligence/news/top/daily/")
embed.add_field(
name="Новости",
value="\n".join(posts_text),
inline=False,
)
else:
embed.add_field(
name="Новости",
value="Новостей пока нет.",
inline=False,
)
await ctx.send(embed=embed)

View File

@ -1,6 +1,6 @@
import discord
from discord.ext import commands
import requests
from utils.pogoda import fetch_weather, fetch_open_meteo, wmo_to_russian, translate_weather, pressure_to_mmhg
class Pogoda(commands.Cog):
@ -11,12 +11,8 @@ class Pogoda(commands.Cog):
@commands.command(name="pogoda")
async def pogoda(self, ctx):
try:
response = requests.get(self.api_url, timeout=10)
response.raise_for_status()
data = response.json()
except requests.RequestException as e:
await ctx.send(f"Ошибка при получении данных: {e}")
data = await fetch_weather(self.api_url)
if data is None:
return
current = data.get("current_condition", [{}])[0]
@ -26,55 +22,21 @@ class Pogoda(commands.Cog):
temp = current.get("temp_C", "")
feels_like = current.get("FeelsLikeC", "")
description = self._translate_weather(current.get("weatherDesc", [{}])[0].get("value", ""))
description = translate_weather(current.get("weatherDesc", [{}])[0].get("value", ""))
humidity = current.get("humidity", "")
wind_kmh = current.get("windspeedKmph", "")
wind = round(int(wind_kmh) / 3.6, 1) if wind_kmh != "" else ""
try:
wind = round(int(wind_kmh) / 3.6, 1) if wind_kmh != "" else ""
except (ValueError, TypeError):
wind = ""
pressure_mb = current.get("pressure", "")
pressure_mm = self._pressure_to_mmhg(pressure_mb)
pressure_mm = pressure_to_mmhg(pressure_mb)
await ctx.send(
f"[TEMP] Температура: {temp}°C (ощущается как {feels_like}°C)\n"
f"[DESC] Описание: {description}\n"
f"[HUMID] Влажность: {humidity}%\n"
f"[WIND] Ветер: {wind} м/с\n"
f"[PRESS] Давление: {pressure_mm} мм рт. ст."
f"Температура: {temp}°C (ощущается как {feels_like}°C)\n"
f"Описание: {description}\n"
f"Влажность: {humidity}%\n"
f"Ветер: {wind} м/с\n"
f"Давление: {pressure_mm} мм рт. ст."
)
def _translate_weather(self, en):
mapping = {
"Sunny": "Ясно",
"Clear": "Ясно",
"Partly cloudy": "Переменная облачность",
"Cloudy": "Облачно",
"Overcast": "Пасмурно",
"Mist": "Туман",
"Fog": "Туман",
"Patchy rain nearby": "Местами дождь",
"Light rain": "Небольшой дождь",
"Moderate rain": "Умеренный дождь",
"Heavy rain": "Сильный дождь",
"Patchy snow nearby": "Местами снег",
"Light snow": "Небольшой снег",
"Heavy snow": "Сильный снег",
"Patchy sleet nearby": "Местами слякоть",
"Blowing snow": "Метель",
"Thundery outbreaks in nearby": "Гроза вблизи",
"Moderate or heavy snow in area": "Снег",
"Moderate or heavy rain in area": "Дождь",
"Moderate or heavy freezing rain in area": "Ледяной дождь",
"Moderate or heavy sleet in area": "Слякоть",
"Ice pellets": "Ледяные кристаллы",
"Haze": "Дымка",
"Foggy": "Туманно",
}
for key, value in mapping.items():
if key.lower() in en.lower():
return value
return en
def _pressure_to_mmhg(self, mb):
if mb == "":
return ""
return round(int(mb) * 0.750062, 1)

View File

@ -1,5 +1,13 @@
from .stop import stop
from .news import news
from .cat import cat
from .pogoda import pogoda
from .morning import morning
ALL_CONSOLE_COMMANDS = {
"stop": stop,
"news": news,
"cat": cat,
"pogoda": pogoda,
"morning": morning,
}

3
console_commands/cat.py Normal file
View File

@ -0,0 +1,3 @@
def cat(stop_event, bot):
"""Заглушка: тут должен быть котик"""
print("Заглушка: тут должен быть котик")

View File

@ -0,0 +1,81 @@
import asyncio
from utils.pogoda import fetch_weather, pressure_to_mmhg, translate_weather
from utils.news import fetch_rss, format_articles, RSS_URL_ARTICLES, RSS_URL_POSTS
from utils.cat import fetch_cat
async def morning(stop_event, bot):
"""Вывести погоду, лучшие статьи за сутки и котика"""
api_url = "https://wttr.in/Magnitogorsk?format=j1&lang=ru"
# Параллельный запрос погоды, новостей и котика
weather_data, articles, posts, cat_url = await asyncio.gather(
fetch_weather(api_url),
fetch_rss(RSS_URL_ARTICLES),
fetch_rss(RSS_URL_POSTS),
fetch_cat(),
)
print("Доброе утро!\n")
# --- Котик ---
if cat_url:
print(f"Котик: {cat_url}\n")
else:
print("Котика получить не удалось.\n")
# --- Погода ---
if weather_data is not None:
current = weather_data.get("current_condition", [{}])[0]
if current:
temp = current.get("temp_C", "")
feels_like = current.get("FeelsLikeC", "")
description = translate_weather(current.get("weatherDesc", [{}])[0].get("value", ""))
humidity = current.get("humidity", "")
wind_kmh = current.get("windspeedKmph", "")
try:
wind = round(int(wind_kmh) / 3.6, 1) if wind_kmh != "" else ""
except (ValueError, TypeError):
wind = ""
pressure_mb = current.get("pressure", "")
pressure_mm = pressure_to_mmhg(pressure_mb)
print(f"**Погода в Магнитогорске:**")
print(f"Температура: {temp}°C (ощущается как {feels_like}°C)")
print(f"Описание: {description}")
print(f"Влажность: {humidity}%")
print(f"Ветер: {wind} м/с")
print(f"Давление: {pressure_mm} мм рт. ст.")
else:
print("Не удалось получить данные о погоде.")
else:
print("Не удалось получить данные о погоде.")
print()
# --- Новости: статьи ---
if articles is not None:
if articles:
lines = format_articles(articles,
"Лучшие статьи за сутки / Искусственный интеллект / Хабr",
"https://habr.com/ru/hubs/artificial_intelligence/articles/top/daily/")
print("\n".join(lines))
else:
print("Новостей пока нет.")
else:
print("Не удалось получить новости.")
print()
# --- Новости: посты ---
if posts is not None:
if posts:
lines = format_articles(posts,
"Лучшие новости за сутки / Искусственный интеллект / Хабr",
"https://habr.com/ru/hubs/artificial_intelligence/news/top/daily/")
print("\n".join(lines))
else:
print("Новостей пока нет.")
else:
print("Не удалось получить новости.")

25
console_commands/news.py Normal file
View File

@ -0,0 +1,25 @@
from utils.news import fetch_rss, format_articles, RSS_URL_ARTICLES, RSS_URL_POSTS
async def news(stop_event, bot):
"""Вывести топ-5 свежих статей по AI с Habr"""
articles = await fetch_rss(RSS_URL_ARTICLES)
if articles is None:
print("Не удалось получить новости.")
return
if not articles:
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:
lines.append("")
lines.extend(format_articles(posts, "Лучшие новости за сутки / Искусственный интеллект / Хабr",
"https://habr.com/ru/hubs/artificial_intelligence/news/top/daily/"))
for line in lines:
print(line)

View File

@ -0,0 +1,34 @@
from utils.pogoda import fetch_weather, fetch_open_meteo, wmo_to_russian, translate_weather, pressure_to_mmhg
async def pogoda(stop_event, bot):
"""Вывести прогноз погоды для Магнитогорска"""
api_url = "https://wttr.in/Magnitogorsk?format=j1&lang=ru"
data = await fetch_weather(api_url)
if data is None:
print("Не удалось получить данные о погоде.")
return
current = data.get("current_condition", [{}])[0]
if not current:
print("Не удалось получить данные о погоде.")
return
temp = current.get("temp_C", "")
feels_like = current.get("FeelsLikeC", "")
description = translate_weather(current.get("weatherDesc", [{}])[0].get("value", ""))
humidity = current.get("humidity", "")
wind_kmh = current.get("windspeedKmph", "")
try:
wind = round(int(wind_kmh) / 3.6, 1) if wind_kmh != "" else ""
except (ValueError, TypeError):
wind = ""
pressure_mb = current.get("pressure", "")
pressure_mm = pressure_to_mmhg(pressure_mb)
print(f"Температура: {temp}°C (ощущается как {feels_like}°C)")
print(f"Описание: {description}")
print(f"Влажность: {humidity}%")
print(f"Ветер: {wind} м/с")
print(f"Давление: {pressure_mm} мм рт. ст.")

View File

@ -1,3 +1,10 @@
def stop(stop_event):
import asyncio
def stop(stop_event, bot):
"""Остановка бота"""
pass
stop_event.set()
try:
asyncio.run_coroutine_threadsafe(bot.close(), bot.loop).result(timeout=5)
except Exception as e:
print(f"Ошибка при остановке бота: {e}")

2
pytest.ini Normal file
View File

@ -0,0 +1,2 @@
[pytest]
asyncio_mode = auto

View File

@ -1,3 +1,5 @@
discord.py>=2.3.2
python-dotenv>=1.0.0
requests>=2.31.0
pytest>=7.4.0
pytest-asyncio>=0.21.0

0
tests/__init__.py Normal file
View File

View File

@ -0,0 +1,210 @@
import pytest
from unittest.mock import AsyncMock, MagicMock, patch
from commands.pogoda import Pogoda
class TestPogodaInit:
"""Тесты инициализации Cog Pogoda."""
def test_init_sets_api_url(self):
"""__init__ должен устанавливать api_url."""
cog = Pogoda()
assert cog.api_url == "https://wttr.in/Magnitogorsk?format=j1&lang=ru"
class TestPogodaCommand:
"""Тесты команды !pogoda."""
def _make_cog(self):
return Pogoda()
def _make_ctx(self, send_return=None):
ctx = MagicMock()
ctx.send = AsyncMock(return_value=send_return)
return ctx
def _make_weather_data(self, **extra):
"""Создать mock weather data с дефолтными полями."""
defaults = {
"current_condition": [
{
"temp_C": "22",
"FeelsLikeC": "24",
"weatherDesc": [{"value": "Clear"}],
"humidity": "45",
"windspeedKmph": "18",
"pressure": "1013",
}
]
}
defaults["current_condition"][0].update(extra)
return defaults
@pytest.mark.asyncio
async def test_pogoda_success(self):
"""Успешный запрос погоды должен отправить embed с данными."""
cog = self._make_cog()
ctx = self._make_ctx()
weather = self._make_weather_data()
with patch("commands.pogoda.fetch_weather", new=AsyncMock(return_value=weather)):
await cog.pogoda.callback(cog, ctx)
ctx.send.assert_called_once()
args = ctx.send.call_args[0][0]
assert "Температура: 22°C" in args
assert "(ощущается как 24°C)" in args
assert "Описание: Ясно" in args
assert "Влажность: 45%" in args
assert "Ветер: 5.0 м/с" in args
assert "Давление: 759.8 мм рт. ст." in args
@pytest.mark.asyncio
async def test_pogoda_fetch_returns_none(self):
"""fetch_weather вернул None — бот должен ничего не отправить."""
cog = self._make_cog()
ctx = self._make_ctx()
with patch("commands.pogoda.fetch_weather", new=AsyncMock(return_value=None)):
await cog.pogoda.callback(cog, ctx)
ctx.send.assert_not_called()
@pytest.mark.asyncio
async def test_pogoda_empty_current_condition(self):
"""current_condition пустой список — код выбрасывает IndexError (баг в коде)."""
cog = self._make_cog()
ctx = self._make_ctx()
weather = {"current_condition": []}
with patch("commands.pogoda.fetch_weather", new=AsyncMock(return_value=weather)):
with pytest.raises(IndexError):
await cog.pogoda.callback(cog, ctx)
@pytest.mark.asyncio
async def test_pogoda_current_condition_none(self):
"""current_condition — пустой dict — бот должен сообщить об ошибке."""
cog = self._make_cog()
ctx = self._make_ctx()
weather = {"current_condition": [{}]}
with patch("commands.pogoda.fetch_weather", new=AsyncMock(return_value=weather)):
await cog.pogoda.callback(cog, ctx)
ctx.send.assert_called_once_with("Не удалось получить данные о погоде.")
@pytest.mark.asyncio
async def test_pogoda_wind_non_numeric(self):
"""windspeedKmph — не число — wind должен быть ''."""
cog = self._make_cog()
ctx = self._make_ctx()
weather = self._make_weather_data(windspeedKmph="abc")
with patch("commands.pogoda.fetch_weather", new=AsyncMock(return_value=weather)):
await cog.pogoda.callback(cog, ctx)
args = ctx.send.call_args[0][0]
assert "Ветер: — м/с" in args
@pytest.mark.asyncio
async def test_pogoda_wind_none(self):
"""windspeedKmph отсутствует — wind должен быть ''."""
cog = self._make_cog()
ctx = self._make_ctx()
weather = self._make_weather_data(windspeedKmph=None)
with patch("commands.pogoda.fetch_weather", new=AsyncMock(return_value=weather)):
await cog.pogoda.callback(cog, ctx)
args = ctx.send.call_args[0][0]
assert "Ветер: — м/с" in args
@pytest.mark.asyncio
async def test_pogoda_zero_wind(self):
"""windspeedKmph = 0 — wind должен быть 0.0."""
cog = self._make_cog()
ctx = self._make_ctx()
weather = self._make_weather_data(windspeedKmph="0")
with patch("commands.pogoda.fetch_weather", new=AsyncMock(return_value=weather)):
await cog.pogoda.callback(cog, ctx)
args = ctx.send.call_args[0][0]
assert "Ветер: 0.0 м/с" in args
@pytest.mark.asyncio
async def test_pogoda_default_values(self):
"""Поля с отсутствующими значениями должны давать ''."""
cog = self._make_cog()
ctx = self._make_ctx()
weather = self._make_weather_data(
temp_C=None,
FeelsLikeC=None,
weatherDesc=[{"value": None}],
humidity=None,
pressure=None,
)
with patch("commands.pogoda.fetch_weather", new=AsyncMock(return_value=weather)):
await cog.pogoda.callback(cog, ctx)
args = ctx.send.call_args[0][0]
# dict.get(key, default) возвращает None, если ключ есть, но значение None
assert "Температура: None°C" in args
assert "ощущается как None°C" in args
assert "Описание: —" in args
assert "Влажность: None%" in args
assert "Давление: — мм рт. ст." in args
@pytest.mark.asyncio
async def test_pogoda_translate_unknown_weather(self):
"""Неизвестное описание погоды должно возвращать оригинал."""
cog = self._make_cog()
ctx = self._make_ctx()
weather = self._make_weather_data(weatherDesc=[{"value": "UnknownXYZ"}])
with patch("commands.pogoda.fetch_weather", new=AsyncMock(return_value=weather)):
await cog.pogoda.callback(cog, ctx)
args = ctx.send.call_args[0][0]
assert "Описание: UnknownXYZ" in args
@pytest.mark.asyncio
async def test_pogoda_russian_weather_description(self):
"""Описание погоды на русском должно корректно переводиться."""
cog = self._make_cog()
ctx = self._make_ctx()
weather = self._make_weather_data(weatherDesc=[{"value": "Переменная облачность"}])
with patch("commands.pogoda.fetch_weather", new=AsyncMock(return_value=weather)):
await cog.pogoda.callback(cog, ctx)
args = ctx.send.call_args[0][0]
assert "Описание: Переменная облачность" in args
@pytest.mark.asyncio
async def test_pogoda_negative_pressure(self):
"""Отрицательное давление должно конвертироваться."""
cog = self._make_cog()
ctx = self._make_ctx()
weather = self._make_weather_data(pressure="-50")
with patch("commands.pogoda.fetch_weather", new=AsyncMock(return_value=weather)):
await cog.pogoda.callback(cog, ctx)
args = ctx.send.call_args[0][0]
assert "Давление: -37.5 мм рт. ст." in args
@pytest.mark.asyncio
async def test_pogoda_high_wind(self):
"""Большая скорость ветра должна корректно округляться."""
cog = self._make_cog()
ctx = self._make_ctx()
weather = self._make_weather_data(windspeedKmph="123")
with patch("commands.pogoda.fetch_weather", new=AsyncMock(return_value=weather)):
await cog.pogoda.callback(cog, ctx)
args = ctx.send.call_args[0][0]
assert "Ветер: 34.2 м/с" in args

102
tests/test_fetch_cat.py Normal file
View File

@ -0,0 +1,102 @@
import asyncio
import json
import pytest
from unittest.mock import patch, MagicMock
from utils.cat import fetch_cat
class TestFetchCat:
"""Тесты функции fetch_cat() — получение URL случайного котика."""
@patch("utils.cat._session.get")
def test_fetch_cat_success(self, mock_get):
"""Успешный ответ с URL должен вернуть строку."""
mock_response = MagicMock()
mock_response.json.return_value = [{"url": "https://example.com/cat.jpg"}]
mock_response.raise_for_status = MagicMock()
mock_get.return_value = mock_response
result = asyncio.run(fetch_cat())
assert result == "https://example.com/cat.jpg"
@patch("utils.cat._session.get")
def test_fetch_cat_empty_array(self, mock_get):
"""Пустой массив должен вернуть None."""
mock_response = MagicMock()
mock_response.json.return_value = []
mock_response.raise_for_status = MagicMock()
mock_get.return_value = mock_response
result = asyncio.run(fetch_cat())
assert result is None
@patch("utils.cat._session.get")
def test_fetch_cat_http_error(self, mock_get):
"""HTTP-ошибка (raise_for_status) должна вернуть None."""
import requests
mock_response = MagicMock()
mock_response.raise_for_status.side_effect = requests.HTTPError("404 Not Found")
mock_get.return_value = mock_response
result = asyncio.run(fetch_cat())
assert result is None
@patch("utils.cat._session.get")
def test_fetch_cat_connection_error(self, mock_get):
"""ConnectionError должна вернуть None."""
from requests.exceptions import ConnectionError
mock_get.side_effect = ConnectionError("No connection")
result = asyncio.run(fetch_cat())
assert result is None
@patch("utils.cat._session.get")
def test_fetch_cat_timeout(self, mock_get):
"""Timeout должна вернуть None."""
from requests.exceptions import Timeout
mock_get.side_effect = Timeout("Request timed out")
result = asyncio.run(fetch_cat())
assert result is None
@patch("utils.cat._session.get")
def test_fetch_cat_ssl_error(self, mock_get):
"""SSLError должна вернуть None."""
from requests.exceptions import SSLError
mock_get.side_effect = SSLError("SSL handshake failed")
result = asyncio.run(fetch_cat())
assert result is None
@patch("utils.cat._session.get")
def test_fetch_cat_json_parse_error(self, mock_get):
"""Ошибка парсинга JSON должна вернуть None."""
import requests
mock_response = MagicMock()
mock_response.json.side_effect = requests.JSONDecodeError("Expecting value", "", 0)
mock_response.raise_for_status = MagicMock()
mock_get.return_value = mock_response
result = asyncio.run(fetch_cat())
assert result is None
@patch("utils.cat._session.get")
def test_fetch_cat_missing_url_key(self, mock_get):
"""Отсутствие ключа 'url' в ответе должно вернуть None."""
mock_response = MagicMock()
mock_response.json.return_value = [{"error": "no image"}]
mock_response.raise_for_status = MagicMock()
mock_get.return_value = mock_response
result = asyncio.run(fetch_cat())
assert result is None
@patch("utils.cat._session.get")
def test_fetch_cat_request_exception(self, mock_get):
"""Общий RequestException должен вернуть None."""
import requests
mock_get.side_effect = requests.RequestException("Generic error")
result = asyncio.run(fetch_cat())
assert result is None
@patch("utils.cat._session.get")
def test_fetch_cat_url_with_special_chars(self, mock_get):
"""URL со спецсимволами должен вернуться как есть."""
mock_response = MagicMock()
mock_response.json.return_value = [{"url": "https://example.com/cat?w=100&h=200"}]
mock_response.raise_for_status = MagicMock()
mock_get.return_value = mock_response
result = asyncio.run(fetch_cat())
assert result == "https://example.com/cat?w=100&h=200"

416
tests/test_fetch_rss.py Normal file
View File

@ -0,0 +1,416 @@
import asyncio
import pytest
from unittest.mock import patch, MagicMock
from utils.news import fetch_rss
class TestFetchRss:
"""Тесты функции fetch_rss() — получение и парсинг RSS-ленты."""
@patch("utils.news._session.get")
def test_fetch_rss_success_rss20(self, mock_get):
"""Успешный ответ RSS 2.0 должен вернуть список статей."""
rss_content = """<?xml version="1.0" encoding="UTF-8"?>
<rss version="2.0">
<channel>
<item>
<title>Статья 1</title>
<link>https://habr.com/1</link>
<guid isPermaLink="true">https://habr.com/1</guid>
<pubDate>Mon, 28 May 2026 10:00:00 +0000</pubDate>
<dc:creator xmlns:dc="http://purl.org/dc/elements/1.1/">Автор 1</dc:creator>
<category>AI</category>
<category>ML</category>
</item>
<item>
<title>Статья 2</title>
<guid isPermaLink="true">https://habr.com/2</guid>
<pubDate>Mon, 28 May 2026 12:00:00 +0000</pubDate>
<dc:creator xmlns:dc="http://purl.org/dc/elements/1.1/">Автор 2</dc:creator>
</item>
</channel>
</rss>""".encode()
mock_response = MagicMock()
mock_response.content = rss_content
mock_response.raise_for_status = MagicMock()
mock_get.return_value = mock_response
result = asyncio.run(fetch_rss("https://example.com/rss"))
assert result is not None
assert len(result) == 2
assert result[0]["title"] == "Статья 1"
assert result[0]["link"] == "https://habr.com/1"
assert result[0]["pub_date"] == "Mon, 28 May 2026 10:00:00 +0000"
assert result[0]["creator"] == "Автор 1"
assert result[0]["tags"] == ["AI", "ML"]
assert result[1]["title"] == "Статья 2"
assert result[1]["creator"] == "Автор 2"
assert result[1]["tags"] == []
@patch("utils.news._session.get")
def test_fetch_rss_success_atom(self, mock_get):
"""Успешный ответ Atom должен вернуть список статей."""
atom_content = """<?xml version="1.0" encoding="UTF-8"?>
<feed xmlns="http://www.w3.org/2005/Atom">
<entry>
<title>Atom статья 1</title>
<link href="https://habr.com/atom/1" />
<published>2026-05-28T10:00:00Z</published>
<author><name>Atom автор</name></author>
<category>AI</category>
</entry>
</feed>""".encode()
mock_response = MagicMock()
mock_response.content = atom_content
mock_response.raise_for_status = MagicMock()
mock_get.return_value = mock_response
result = asyncio.run(fetch_rss("https://example.com/atom"))
assert result is not None
assert len(result) == 1
assert result[0]["title"] == "Atom статья 1"
assert result[0]["link"] == "https://habr.com/atom/1"
assert result[0]["pub_date"] == "2026-05-28T10:00:00Z"
assert result[0]["creator"] == "Atom автор"
assert result[0]["tags"] == ["AI"]
@patch("utils.news._session.get")
def test_fetch_rss_empty_items(self, mock_get):
"""RSS без items должен вернуть пустой список."""
rss_content = """<?xml version="1.0" encoding="UTF-8"?>
<rss version="2.0">
<channel>
</channel>
</rss>""".encode()
mock_response = MagicMock()
mock_response.content = rss_content
mock_response.raise_for_status = MagicMock()
mock_get.return_value = mock_response
result = asyncio.run(fetch_rss("https://example.com/rss"))
assert result == []
@patch("utils.news._session.get")
def test_fetch_rss_no_matching_format(self, mock_get):
"""Неизвестный формат XML должен вернуть пустой список."""
xml_content = """<?xml version="1.0"?>
<unknown></unknown>""".encode()
mock_response = MagicMock()
mock_response.content = xml_content
mock_response.raise_for_status = MagicMock()
mock_get.return_value = mock_response
result = asyncio.run(fetch_rss("https://example.com/xml"))
assert result == []
@patch("utils.news._session.get")
def test_fetch_rss_missing_title(self, mock_get):
"""Статья без title должна получить 'Без названия'."""
rss_content = """<?xml version="1.0" encoding="UTF-8"?>
<rss version="2.0">
<channel>
<item>
<title>Без title</title>
<guid isPermaLink="true">https://habr.com/1</guid>
<pubDate>Mon, 28 May 2026 10:00:00 +0000</pubDate>
</item>
</channel>
</rss>""".encode()
mock_response = MagicMock()
mock_response.content = rss_content
mock_response.raise_for_status = MagicMock()
mock_get.return_value = mock_response
result = asyncio.run(fetch_rss("https://example.com/rss"))
assert result is not None
assert result[0]["title"] == "Без title"
assert result[0]["link"] == "https://habr.com/1"
assert result[0]["pub_date"] == "Mon, 28 May 2026 10:00:00 +0000"
assert result[0]["creator"] == ""
assert result[0]["tags"] == []
@patch("utils.news._session.get")
def test_fetch_rss_missing_guid(self, mock_get):
"""Статья без guid isPermaLink должна иметь пустую ссылку."""
rss_content = """<?xml version="1.0" encoding="UTF-8"?>
<rss version="2.0">
<channel>
<item>
<title>Без guid</title>
<link>https://habr.com/1</link>
<guid>https://habr.com/1</guid>
</item>
</channel>
</rss>""".encode()
mock_response = MagicMock()
mock_response.content = rss_content
mock_response.raise_for_status = MagicMock()
mock_get.return_value = mock_response
result = asyncio.run(fetch_rss("https://example.com/rss"))
assert result is not None
assert result[0]["title"] == "Без guid"
assert result[0]["link"] == ""
@patch("utils.news._session.get")
def test_fetch_rss_limit_to_10(self, mock_get):
"""Больше 10 items должно быть обрезано до 10."""
items = "\n".join(
f""" <item>
<title>Статья {i}</title>
<guid isPermaLink="true">https://habr.com/{i}</guid>
<pubDate>Mon, 28 May 2026 10:00:00 +0000</pubDate>
</item>"""
for i in range(15)
)
rss_content = (f"""<?xml version="1.0" encoding="UTF-8"?>
<rss version="2.0">
<channel>
{items}
</channel>
</rss>""").encode()
mock_response = MagicMock()
mock_response.content = rss_content
mock_response.raise_for_status = MagicMock()
mock_get.return_value = mock_response
result = asyncio.run(fetch_rss("https://example.com/rss"))
assert result is not None
assert len(result) == 10
assert result[0]["title"] == "Статья 0"
assert result[9]["title"] == "Статья 9"
@patch("utils.news._session.get")
def test_fetch_rss_http_error(self, mock_get):
"""HTTP-ошибка должна вернуть None."""
import requests
mock_get.side_effect = requests.exceptions.HTTPError("404 Not Found")
result = asyncio.run(fetch_rss("https://example.com/rss"))
assert result is None
@patch("utils.news._session.get")
def test_fetch_rss_connection_error(self, mock_get):
"""Ошибка соединения должна вернуть None."""
import requests
mock_get.side_effect = requests.exceptions.ConnectionError("No connection")
result = asyncio.run(fetch_rss("https://example.com/rss"))
assert result is None
@patch("utils.news._session.get")
def test_fetch_rss_timeout(self, mock_get):
"""Таймаут должен вернуть None."""
import requests
mock_get.side_effect = requests.exceptions.Timeout("Request timed out")
result = asyncio.run(fetch_rss("https://example.com/rss"))
assert result is None
@patch("utils.news._session.get")
def test_fetch_rss_ssl_error(self, mock_get):
"""SSLError должен вернуть None."""
import requests
mock_get.side_effect = requests.exceptions.SSLError("SSL handshake failed")
result = asyncio.run(fetch_rss("https://example.com/rss"))
assert result is None
@patch("utils.news._session.get")
def test_fetch_rss_empty_tags(self, mock_get):
"""Статья с пустыми тегами должна иметь пустые строки."""
rss_content = """<?xml version="1.0" encoding="UTF-8"?>
<rss version="2.0">
<channel>
<item>
<title>Пустые теги</title>
<guid isPermaLink="true">https://habr.com/1</guid>
<pubDate>Mon, 28 May 2026 10:00:00 +0000</pubDate>
</item>
</channel>
</rss>""".encode()
mock_response = MagicMock()
mock_response.content = rss_content
mock_response.raise_for_status = MagicMock()
mock_get.return_value = mock_response
result = asyncio.run(fetch_rss("https://example.com/rss"))
assert result is not None
assert result[0]["title"] == "Пустые теги"
assert result[0]["link"] == "https://habr.com/1"
assert result[0]["pub_date"] == "Mon, 28 May 2026 10:00:00 +0000"
assert result[0]["creator"] == ""
assert result[0]["tags"] == []
@patch("utils.news._session.get")
def test_fetch_rss_category_without_text(self, mock_get):
"""Категория без текста должна быть пропущена."""
rss_content = """<?xml version="1.0" encoding="UTF-8"?>
<rss version="2.0">
<channel>
<item>
<title>Статья</title>
<guid isPermaLink="true">https://habr.com/1</guid>
<category></category>
<category>AI</category>
</item>
</channel>
</rss>""".encode()
mock_response = MagicMock()
mock_response.content = rss_content
mock_response.raise_for_status = MagicMock()
mock_get.return_value = mock_response
result = asyncio.run(fetch_rss("https://example.com/rss"))
assert result is not None
assert result[0]["tags"] == ["AI"]
@patch("utils.news._session.get")
def test_fetch_rss_atom_missing_author(self, mock_get):
"""Atom feed без автора должен иметь пустого creator."""
atom_content = """<?xml version="1.0" encoding="UTF-8"?>
<feed xmlns="http://www.w3.org/2005/Atom">
<entry>
<title>Без автора</title>
<link href="https://habr.com/1" />
<published>2026-05-28T10:00:00Z</published>
</entry>
</feed>""".encode()
mock_response = MagicMock()
mock_response.content = atom_content
mock_response.raise_for_status = MagicMock()
mock_get.return_value = mock_response
result = asyncio.run(fetch_rss("https://example.com/atom"))
assert result is not None
assert result[0]["title"] == "Без автора"
assert result[0]["creator"] == ""
@patch("utils.news._session.get")
def test_fetch_rss_atom_missing_link(self, mock_get):
"""Atom feed без link должен иметь пустую ссылку."""
atom_content = """<?xml version="1.0" encoding="UTF-8"?>
<feed xmlns="http://www.w3.org/2005/Atom">
<entry>
<title>Без ссылки</title>
<published>2026-05-28T10:00:00Z</published>
</entry>
</feed>""".encode()
mock_response = MagicMock()
mock_response.content = atom_content
mock_response.raise_for_status = MagicMock()
mock_get.return_value = mock_response
result = asyncio.run(fetch_rss("https://example.com/atom"))
assert result is not None
assert result[0]["title"] == "Без ссылки"
assert result[0]["link"] == ""
@patch("utils.news._session.get")
def test_fetch_rss_request_exception(self, mock_get):
"""Общий RequestException должен вернуть None."""
import requests
mock_get.side_effect = requests.RequestException("Generic error")
result = asyncio.run(fetch_rss("https://example.com/rss"))
assert result is None
@patch("utils.news._session.get")
def test_fetch_rss_guid_fallback_to_link(self, mock_get):
"""Если нет guid isPermaLink, ссылка должна быть пустой."""
rss_content = """<?xml version="1.0" encoding="UTF-8"?>
<rss version="2.0">
<channel>
<item>
<title>Статья</title>
<link>https://habr.com/alternative</link>
<guid>https://habr.com/alternative</guid>
</item>
</channel>
</rss>""".encode()
mock_response = MagicMock()
mock_response.content = rss_content
mock_response.raise_for_status = MagicMock()
mock_get.return_value = mock_response
result = asyncio.run(fetch_rss("https://example.com/rss"))
assert result is not None
assert result[0]["link"] == ""
@patch("utils.news._session.get")
def test_fetch_rss_single_item(self, mock_get):
"""Один item должен быть распарсен корректно."""
rss_content = """<?xml version="1.0" encoding="UTF-8"?>
<rss version="2.0">
<channel>
<item>
<title>Единственная статья</title>
<guid isPermaLink="true">https://habr.com/1</guid>
<pubDate>Mon, 28 May 2026 10:00:00 +0000</pubDate>
<dc:creator xmlns:dc="http://purl.org/dc/elements/1.1/">Единственный автор</dc:creator>
<category>ML</category>
</item>
</channel>
</rss>""".encode()
mock_response = MagicMock()
mock_response.content = rss_content
mock_response.raise_for_status = MagicMock()
mock_get.return_value = mock_response
result = asyncio.run(fetch_rss("https://example.com/rss"))
assert result is not None
assert len(result) == 1
assert result[0]["title"] == "Единственная статья"
assert result[0]["link"] == "https://habr.com/1"
assert result[0]["creator"] == "Единственный автор"
assert result[0]["tags"] == ["ML"]
@patch("utils.news._session.get")
def test_fetch_rss_special_characters_in_title(self, mock_get):
"""Заголовки со спецсимволами должны парситься корректно."""
rss_content = """<?xml version="1.0" encoding="UTF-8"?>
<rss version="2.0">
<channel>
<item>
<title>AI &amp; ML: будущее &lt;технологий&gt;</title>
<guid isPermaLink="true">https://habr.com/1</guid>
<pubDate>Mon, 28 May 2026 10:00:00 +0000</pubDate>
</item>
</channel>
</rss>""".encode()
mock_response = MagicMock()
mock_response.content = rss_content
mock_response.raise_for_status = MagicMock()
mock_get.return_value = mock_response
result = asyncio.run(fetch_rss("https://example.com/rss"))
assert result is not None
assert "AI" in result[0]["title"]
assert "ML" in result[0]["title"]
@patch("utils.news._session.get")
def test_fetch_rss_date_with_gmt(self, mock_get):
"""Дата с GMT должна парситься корректно."""
rss_content = """<?xml version="1.0" encoding="UTF-8"?>
<rss version="2.0">
<channel>
<item>
<title>Статья</title>
<guid isPermaLink="true">https://habr.com/1</guid>
<pubDate>Mon, 28 May 2026 10:00:00 GMT</pubDate>
</item>
</channel>
</rss>""".encode()
mock_response = MagicMock()
mock_response.content = rss_content
mock_response.raise_for_status = MagicMock()
mock_get.return_value = mock_response
result = asyncio.run(fetch_rss("https://example.com/rss"))
assert result is not None
assert result[0]["pub_date"] == "Mon, 28 May 2026 10:00:00 GMT"
@patch("utils.news._session.get")
def test_fetch_rss_many_categories(self, mock_get):
"""Множество категорий должны быть собраны."""
rss_content = """<?xml version="1.0" encoding="UTF-8"?>
<rss version="2.0">
<channel>
<item>
<title>Статья</title>
<guid isPermaLink="true">https://habr.com/1</guid>
<category>AI</category>
<category>ML</category>
<category>Deep Learning</category>
<category>NLP</category>
<category>Computer Vision</category>
</item>
</channel>
</rss>""".encode()
mock_response = MagicMock()
mock_response.content = rss_content
mock_response.raise_for_status = MagicMock()
mock_get.return_value = mock_response
result = asyncio.run(fetch_rss("https://example.com/rss"))
assert result is not None
assert result[0]["tags"] == ["AI", "ML", "Deep Learning", "NLP", "Computer Vision"]

231
tests/test_fetch_weather.py Normal file
View File

@ -0,0 +1,231 @@
import asyncio
import pytest
from unittest.mock import patch, MagicMock
from utils.pogoda import fetch_weather, fetch_open_meteo
class TestFetchWeather:
"""Тесты функции fetch_weather() — получение погоды с retry-логикой."""
@patch("utils.pogoda._session.get")
def test_fetch_weather_success(self, mock_get):
"""Успешный ответ должен вернуть JSON-данные."""
mock_response = MagicMock()
mock_response.json.return_value = {"current_condition": [{"temp_C": 20}]}
mock_response.raise_for_status = MagicMock()
mock_get.return_value = mock_response
result = asyncio.run(fetch_weather("https://test.example.com"))
assert result == {"current_condition": [{"temp_C": 20}]}
@patch("utils.pogoda._session.get")
def test_fetch_weather_fallback_on_ssl_error(self, mock_get):
"""SSLError на первой попытке → fallback на Open-Meteo."""
from requests.exceptions import SSLError
mock_get.side_effect = [
SSLError("SSL Error"),
MagicMock(json=MagicMock(return_value={"result": "fallback"})),
]
with patch("utils.pogoda.fetch_open_meteo") as mock_fallback:
mock_fallback.return_value = {"result": "fallback"}
result = asyncio.run(fetch_weather("https://test.example.com"))
assert result == {"result": "fallback"}
@patch("utils.pogoda._session.get")
def test_fetch_weather_fallback_on_connection_error(self, mock_get):
"""ConnectionError → fallback на Open-Meteo."""
from requests.exceptions import ConnectionError
mock_get.side_effect = ConnectionError("No connection")
with patch("utils.pogoda.fetch_open_meteo") as mock_fallback:
mock_fallback.return_value = {"result": "fallback"}
result = asyncio.run(fetch_weather("https://test.example.com"))
assert result == {"result": "fallback"}
@patch("utils.pogoda._session.get")
def test_fetch_weather_fallback_on_timeout(self, mock_get):
"""Timeout → fallback на Open-Meteo."""
from requests.exceptions import Timeout
mock_get.side_effect = Timeout("Timed out")
with patch("utils.pogoda.fetch_open_meteo") as mock_fallback:
mock_fallback.return_value = {"result": "fallback"}
result = asyncio.run(fetch_weather("https://test.example.com"))
assert result == {"result": "fallback"}
@patch("utils.pogoda._session.get")
def test_fetch_weather_all_retries_fail(self, mock_get):
"""Все попытки не удались → fallback на Open-Meteo."""
from requests.exceptions import ConnectionError
mock_get.side_effect = ConnectionError("No connection")
with patch("utils.pogoda.fetch_open_meteo") as mock_fallback:
mock_fallback.return_value = None
result = asyncio.run(fetch_weather("https://test.example.com"))
assert result is None
@patch("utils.pogoda._session.get")
def test_fetch_weather_request_exception(self, mock_get):
"""Общий RequestException → fallback на Open-Meteo."""
import requests
mock_get.side_effect = requests.RequestException("Generic error")
with patch("utils.pogoda.fetch_open_meteo") as mock_fallback:
mock_fallback.return_value = {"result": "fallback"}
result = asyncio.run(fetch_weather("https://test.example.com"))
assert result == {"result": "fallback"}
@patch("utils.pogoda._session.get")
def test_fetch_weather_http_error_no_fallback(self, mock_get):
"""HTTP-ошибка (raise_for_status) не ловится, падает."""
mock_response = MagicMock()
mock_response.raise_for_status.side_effect = Exception("HTTP 500")
mock_get.return_value = mock_response
with pytest.raises(Exception):
asyncio.run(fetch_weather("https://test.example.com"))
class TestFetchOpenMeteo:
"""Тесты функции fetch_open_meteo() — fallback на Open-Meteo API."""
@patch("utils.pogoda._session.get")
def test_fetch_open_meteo_success(self, mock_get):
"""Успешный ответ должен вернуть данные в формате current_condition."""
mock_response = MagicMock()
mock_response.json.return_value = {
"current": {
"temperature": 15,
"apparent_temperature": 12,
"weather_code": 3,
"wind_speed_10m": 5.5,
"relative_humidity_2m": 65,
"pressure_msl": 1013,
}
}
mock_response.raise_for_status = MagicMock()
mock_get.return_value = mock_response
result = asyncio.run(fetch_open_meteo())
assert result is not None
assert "current_condition" in result
assert result["current_condition"][0]["temp_C"] == 15
assert result["current_condition"][0]["FeelsLikeC"] == 12
assert result["current_condition"][0]["humidity"] == 65
assert result["current_condition"][0]["pressure"] == 1013
@patch("utils.pogoda._session.get")
def test_fetch_open_meteo_custom_coords(self, mock_get):
"""Кастомные координаты должны быть в URL."""
mock_response = MagicMock()
mock_response.json.return_value = {"current": {"temperature": 25, "apparent_temperature": 22, "weather_code": 0, "wind_speed_10m": 3, "relative_humidity_2m": 50, "pressure_msl": 1020}}
mock_response.raise_for_status = MagicMock()
mock_get.return_value = mock_response
result = asyncio.run(fetch_open_meteo(lat=55.7558, lon=37.6173))
assert result is not None
mock_get.assert_called_once()
call_url = mock_get.call_args[0][0]
assert "55.7558" in call_url
assert "37.6173" in call_url
@patch("utils.pogoda._session.get")
def test_fetch_open_meteo_missing_weather_code(self, mock_get):
"""Отсутствующий weather_code → 'Неизвестно'."""
mock_response = MagicMock()
mock_response.json.return_value = {"current": {"temperature": 10}}
mock_response.raise_for_status = MagicMock()
mock_get.return_value = mock_response
result = asyncio.run(fetch_open_meteo())
assert result is not None
assert result["current_condition"][0]["weatherDesc"] == [{"value": "Неизвестно"}]
@patch("utils.pogoda._session.get")
def test_fetch_open_meteo_ssl_error(self, mock_get):
"""SSLError → вернуть None."""
from requests.exceptions import SSLError
mock_get.side_effect = SSLError("SSL Error")
with patch("utils.pogoda.fetch_open_meteo") as mock_fallback:
# Внутренний fallback тоже падает, проверяем что возвращается None
pass
result = asyncio.run(fetch_open_meteo())
assert result is None
@patch("utils.pogoda._session.get")
def test_fetch_open_meteo_connection_error(self, mock_get):
"""ConnectionError → вернуть None."""
from requests.exceptions import ConnectionError
mock_get.side_effect = ConnectionError("No connection")
result = asyncio.run(fetch_open_meteo())
assert result is None
@patch("utils.pogoda._session.get")
def test_fetch_open_meteo_timeout(self, mock_get):
"""Timeout → вернуть None."""
from requests.exceptions import Timeout
mock_get.side_effect = Timeout("Timed out")
result = asyncio.run(fetch_open_meteo())
assert result is None
@patch("utils.pogoda._session.get")
def test_fetch_open_meteo_request_exception(self, mock_get):
"""Общий RequestException → вернуть None."""
import requests
mock_get.side_effect = requests.RequestException("Error")
result = asyncio.run(fetch_open_meteo())
assert result is None
@patch("utils.pogoda._session.get")
def test_fetch_open_meteo_json_parse_error(self, mock_get):
"""Ошибка парсинга JSON → вернуть None."""
import requests
mock_response = MagicMock()
mock_response.json.side_effect = requests.RequestException("JSON Error")
mock_response.raise_for_status = MagicMock()
mock_get.return_value = mock_response
result = asyncio.run(fetch_open_meteo())
assert result is None
@patch("utils.pogoda._session.get")
def test_fetch_open_meteo_retry_on_error(self, mock_get):
"""Retry: первая попытка падает, вторая успешна."""
from requests.exceptions import ConnectionError
success_response = MagicMock()
success_response.json.return_value = {"current": {"temperature": 20, "apparent_temperature": 18, "weather_code": 1, "wind_speed_10m": 4, "relative_humidity_2m": 60, "pressure_msl": 1015}}
success_response.raise_for_status = MagicMock()
mock_get.side_effect = [ConnectionError("fail"), success_response]
result = asyncio.run(fetch_open_meteo(max_retries=2))
assert result is not None
assert mock_get.call_count == 2
@patch("utils.pogoda._session.get")
def test_fetch_open_meteo_all_retries_fail(self, mock_get):
"""Все попытки неудачны → None."""
from requests.exceptions import ConnectionError
mock_get.side_effect = [ConnectionError("fail"), ConnectionError("fail"), ConnectionError("fail")]
result = asyncio.run(fetch_open_meteo(max_retries=3))
assert result is None
assert mock_get.call_count == 3
@patch("utils.pogoda._session.get")
def test_fetch_open_meteo_http_error(self, mock_get):
"""HTTP 404 → raise_for_status бросит исключение → None."""
import requests
mock_response = MagicMock()
mock_response.raise_for_status.side_effect = requests.HTTPError("HTTP 404")
mock_get.return_value = mock_response
result = asyncio.run(fetch_open_meteo())
assert result is None
@patch("utils.pogoda._session.get")
def test_fetch_open_meteo_wind_speed_0(self, mock_get):
"""Нулевая скорость ветра должна корректно обрабатываться."""
mock_response = MagicMock()
mock_response.json.return_value = {
"current": {
"temperature": 0,
"apparent_temperature": -2,
"weather_code": 45,
"wind_speed_10m": 0,
"relative_humidity_2m": 95,
"pressure_msl": 1000,
}
}
mock_response.raise_for_status = MagicMock()
mock_get.return_value = mock_response
result = asyncio.run(fetch_open_meteo())
assert result is not None
assert result["current_condition"][0]["windspeedKmph"] == 0
assert result["current_condition"][0]["pressure"] == 1000

View File

@ -0,0 +1,229 @@
import pytest
from utils.news import format_articles, truncate_title, _parse_date
class TestTruncateTitle:
"""Тесты функции truncate_title() — обрезка заголовка."""
@pytest.mark.parametrize(
"title, max_len, expected",
[
("Короткий заголовок", 60, "Короткий заголовок"),
("Заголовок ровно в 60 символов1234567890", 60, "Заголовок ровно в 60 символов1234567890"),
("A" * 80, 60, "A" * 60 + "..."), # ASCII для надёжного сравнения
("", 60, ""),
("A" * 100, 100, "A" * 100),
("A" * 101, 100, "A" * 100 + "..."),
("A" * 50, 100, "A" * 50),
],
)
def test_truncate(self, title, max_len, expected):
"""Проверка обрезки заголовка."""
assert truncate_title(title, max_len) == expected
def test_truncate_default_max_len(self):
"""По умолчанию max_len=60."""
long_title = "A" * 61
result = truncate_title(long_title)
assert result == "A" * 60 + "..."
assert len(result) == 63 # 60 + "..."
class TestParseDate:
"""Тесты функции _parse_date() — парсинг даты из RSS."""
@pytest.mark.parametrize(
"pub_date, expected",
[
("Mon, 28 May 2026 10:00:00 +0000", "28.05.2026"),
("Mon, 28 May 2026 10:00:00 GMT", "28.05.2026"),
("2026-05-28T10:00:00Z", "2026.05.28"),
("2026-12-31T23:59:59Z", "2026.12.31"),
("2026-01-01T00:00:00Z", "2026.01.01"),
],
)
def test_parse_date_known(self, pub_date, expected):
"""Известные форматы даты должны парситься корректно."""
assert _parse_date(pub_date) == expected
@pytest.mark.parametrize(
"pub_date, expected",
[
("", ""),
(None, ""),
],
)
def test_parse_date_empty(self, pub_date, expected):
"""Пустая или None дата должна вернуть пустую строку."""
assert _parse_date(pub_date) == expected
def test_parse_date_invalid(self):
"""Невалидная дата должна вернуть первые 10 символов."""
result = _parse_date("invalid-date-string")
assert result == "invalid.da" # первые 10 символов: 'invalid-da' → 'invalid.da' (replace('-','.'))
class TestFormatArticles:
"""Тесты функции format_articles() — формирование строк для вывода."""
def test_format_articles_normal(self):
"""Нормальный список статей должен вернуть заголовок + 5 статей."""
articles = [
{
"title": "Статья 1",
"link": "https://habr.com/1",
"pub_date": "Mon, 28 May 2026 10:00:00 +0000",
"creator": "Автор 1",
"tags": ["AI"],
},
{
"title": "Статья 2",
"link": "https://habr.com/2",
"pub_date": "Tue, 29 May 2026 12:00:00 +0000",
"creator": "Автор 2",
"tags": ["ML"],
},
]
result = format_articles(articles, "Заголовок", "https://habr.com/feed")
assert len(result) == 3 # заголовок + 2 статьи
assert result[0] == "**Заголовок**\n<https://habr.com/feed>"
assert result[1] == "Статья 1\n28.05.2026 <https://habr.com/1>"
assert result[2] == "Статья 2\n29.05.2026 <https://habr.com/2>"
def test_format_articles_limit_to_5(self):
"""Больше 5 статей должно быть обрезано до 5."""
articles = [
{"title": f"Статья {i}", "link": f"https://habr.com/{i}", "pub_date": "Mon, 28 May 2026 10:00:00 +0000", "creator": "", "tags": []}
for i in range(10)
]
result = format_articles(articles, "Заголовок", "https://habr.com/feed")
assert len(result) == 6 # заголовок + 5 статей
assert result[-1] == "Статья 4\n28.05.2026 <https://habr.com/4>"
def test_format_articles_empty_list(self):
"""Пустой список должен вернуть только заголовок."""
result = format_articles([], "Заголовок", "https://habr.com/feed")
assert result == ["**Заголовок**\n<https://habr.com/feed>"]
assert len(result) == 1
def test_format_articles_none(self):
"""None должен вызвать TypeError (articles[:5] на None)."""
with pytest.raises(TypeError):
format_articles(None, "Заголовок", "https://habr.com/feed")
def test_format_articles_single_article(self):
"""Одна статья должна быть корректно отформатирована."""
articles = [
{
"title": "Единственная статья",
"link": "https://habr.com/1",
"pub_date": "Mon, 28 May 2026 10:00:00 +0000",
"creator": "Автор",
"tags": ["AI"],
},
]
result = format_articles(articles, "Новости AI", "https://habr.com/ai")
assert len(result) == 2
assert result[0] == "**Новости AI**\n<https://habr.com/ai>"
assert result[1] == "Единственная статья\n28.05.2026 <https://habr.com/1>"
def test_format_articles_long_title_truncated(self):
"""Длинный заголовок должен быть обрезан до 60 символов с '...'."""
long_title = "A" * 100
articles = [
{"title": long_title, "link": "https://habr.com/1", "pub_date": "Mon, 28 May 2026 10:00:00 +0000", "creator": "", "tags": []}
]
result = format_articles(articles, "Заголовок", "https://habr.com/feed")
assert len(result[1].split("\n")[0]) == 63 # 60 + "..."
assert result[1].split("\n")[0].endswith("...")
def test_format_articles_short_title_unchanged(self):
"""Короткий заголовок должен остаться без изменений."""
short_title = "Кот"
articles = [
{"title": short_title, "link": "https://habr.com/1", "pub_date": "Mon, 28 May 2026 10:00:00 +0000", "creator": "", "tags": []}
]
result = format_articles(articles, "Заголовок", "https://habr.com/feed")
assert result[1].split("\n")[0] == "Кот"
def test_format_articles_exact_60_chars(self):
"""Заголовок ровно 60 символов не должен обрезаться."""
exact_title = "A" * 60
articles = [
{"title": exact_title, "link": "https://habr.com/1", "pub_date": "Mon, 28 May 2026 10:00:00 +0000", "creator": "", "tags": []}
]
result = format_articles(articles, "Заголовок", "https://habr.com/feed")
assert result[1].split("\n")[0] == exact_title
assert "..." not in result[1]
def test_format_articles_iso_date(self):
"""Дата в формате ISO должна парситься корректно."""
articles = [
{
"title": "Статья",
"link": "https://habr.com/1",
"pub_date": "2026-05-28T10:00:00Z",
"creator": "",
"tags": [],
},
]
result = format_articles(articles, "Заголовок", "https://habr.com/feed")
assert result[1] == "Статья\n2026.05.28 <https://habr.com/1>"
def test_format_articles_empty_date(self):
"""Пустая дата должна быть пустой строкой."""
articles = [
{"title": "Статья", "link": "https://habr.com/1", "pub_date": "", "creator": "", "tags": []}
]
result = format_articles(articles, "Заголовок", "https://habr.com/feed")
assert result[1] == "Статья\n <https://habr.com/1>"
def test_format_articles_none_date(self):
"""None дата должна быть пустой строкой."""
articles = [
{"title": "Статья", "link": "https://habr.com/1", "pub_date": None, "creator": "", "tags": []}
]
result = format_articles(articles, "Заголовок", "https://habr.com/feed")
assert result[1] == "Статья\n <https://habr.com/1>"
def test_format_articles_empty_link(self):
"""Пустая ссылка должна быть пустой строкой в угловых скобках."""
articles = [
{"title": "Статья", "link": "", "pub_date": "Mon, 28 May 2026 10:00:00 +0000", "creator": "", "tags": []}
]
result = format_articles(articles, "Заголовок", "https://habr.com/feed")
assert result[1].endswith(" <>")
def test_format_articles_russian_title(self):
"""Русские заголовки должны корректно отображаться."""
articles = [
{
"title": "Искусственный интеллект в медицине",
"link": "https://habr.com/1",
"pub_date": "Mon, 28 May 2026 10:00:00 +0000",
"creator": "Иван Иванов",
"tags": ["AI", "медицина"],
},
]
result = format_articles(articles, "Новости AI", "https://habr.com/ai")
assert "Искусственный интеллект в медицине" in result[1]
def test_format_articles_exact_5_articles(self):
"""Ровно 5 статей должно быть включено."""
articles = [
{"title": f"Статья {i}", "link": f"https://habr.com/{i}", "pub_date": "Mon, 28 May 2026 10:00:00 +0000", "creator": "", "tags": []}
for i in range(5)
]
result = format_articles(articles, "Заголовок", "https://habr.com/feed")
assert len(result) == 6 # заголовок + 5 статей
assert result[-1] == "Статья 4\n28.05.2026 <https://habr.com/4>"
def test_format_articles_6th_article_excluded(self):
"""6-я статья должна быть исключена."""
articles = [
{"title": f"Статья {i}", "link": f"https://habr.com/{i}", "pub_date": "Mon, 28 May 2026 10:00:00 +0000", "creator": "", "tags": []}
for i in range(6)
]
result = format_articles(articles, "Заголовок", "https://habr.com/feed")
assert len(result) == 6 # заголовок + 5 статей
assert "Статья 5" not in result[5]

View File

@ -0,0 +1,134 @@
"""Тесты для utils/morning_runner.py — Scheduler и run_morning."""
import asyncio
from datetime import datetime
from unittest.mock import AsyncMock, MagicMock, patch
import discord
import pytest
from utils.morning_runner import Scheduler, run_morning
class TestSchedulerInit:
"""Тесты инициализации Scheduler."""
def test_init_sets_morning_time(self):
"""Инициализация должна устанавливать время."""
bot = AsyncMock()
scheduler = Scheduler(bot, "08:30")
assert scheduler.morning_time == "08:30"
def test_init_default_morning_time(self):
"""Инициализация с дефолтным временем."""
bot = AsyncMock()
scheduler = Scheduler(bot)
assert scheduler.morning_time == "07:00"
def test_init_creates_loop(self):
"""Инициализация должна создавать loop."""
bot = AsyncMock()
scheduler = Scheduler(bot)
assert scheduler.morning_loop is not None
class TestSchedulerCalculateNextRun:
"""Тесты расчёта следующего запуска."""
def test_next_run_today_before_time(self):
"""Если сейчас раньше времени — вернуть сегодня."""
bot = AsyncMock()
scheduler = Scheduler(bot, "14:00")
with patch("utils.morning_runner.datetime") as mock_dt:
mock_dt.now.return_value = datetime(2026, 5, 29, 10, 0, 0)
next_run = scheduler._calculate_next_run()
assert next_run == datetime(2026, 5, 29, 14, 0, 0)
def test_next_run_tomorrow_after_time(self):
"""Если сейчас позже времени — вернуть завтра."""
bot = AsyncMock()
scheduler = Scheduler(bot, "14:00")
with patch("utils.morning_runner.datetime") as mock_dt:
mock_dt.now.return_value = datetime(2026, 5, 29, 15, 0, 0)
next_run = scheduler._calculate_next_run()
assert next_run == datetime(2026, 5, 30, 14, 0, 0)
def test_next_run_exact_time(self):
"""Если сейчас ровно время — вернуть завтра."""
bot = AsyncMock()
scheduler = Scheduler(bot, "14:00")
with patch("utils.morning_runner.datetime") as mock_dt:
mock_dt.now.return_value = datetime(2026, 5, 29, 14, 0, 0)
next_run = scheduler._calculate_next_run()
assert next_run == datetime(2026, 5, 30, 14, 0, 0)
class TestSchedulerStartStop:
"""Тесты запуска/остановки планировщика."""
def test_start_starts_loop(self):
"""start() должен вызывать start() на loop."""
bot = AsyncMock()
scheduler = Scheduler(bot)
loop_mock = MagicMock()
scheduler.morning_loop = loop_mock
scheduler.start()
loop_mock.start.assert_called_once()
def test_stop_stops_loop(self):
"""stop() должен вызывать stop() на loop."""
bot = AsyncMock()
scheduler = Scheduler(bot)
loop_mock = MagicMock()
scheduler.morning_loop = loop_mock
scheduler.stop()
loop_mock.stop.assert_called_once()
class TestSchedulerCheckAndRun:
"""Тесты проверки и запуска morning."""
@pytest.mark.asyncio
async def test_check_and_run_same_day_no_duplicate(self):
"""Не должен запускать дважды в один день."""
bot = AsyncMock()
scheduler = Scheduler(bot, "07:00")
scheduler._last_run_date = datetime.now().day
with patch("utils.morning_runner.datetime") as mock_dt:
mock_dt.now.return_value = datetime(2026, 5, 29, 7, 0, 0)
await scheduler._check_and_run_morning()
# run_morning не должен вызываться
assert scheduler._last_run_date == datetime.now().day
class TestRunMorning:
"""Тесты run_morning."""
@pytest.mark.asyncio
async def test_run_morning_sends_embed(self):
"""run_morning должен отправлять embed в канал."""
bot = AsyncMock()
channel = AsyncMock()
channel.name = "test-channel"
channel.guild.me = MagicMock()
channel.permissions_for.return_value.send_messages = True
with patch("utils.morning_runner.asyncio.gather", new=AsyncMock(return_value=(
{"current_condition": [{"temp_C": "20", "FeelsLikeC": "22", "weatherDesc": [{"value": "Clear"}], "humidity": "50", "windspeedKmph": "10", "pressure": "1013"}]},
[{"title": "Test", "link": "http://test.com", "pub_date": "Mon, 01 Jan 2026 00:00:00 GMT", "creator": "", "tags": []}],
[{"title": "Test", "link": "http://test.com", "pub_date": "Mon, 01 Jan 2026 00:00:00 GMT", "creator": "", "tags": []}],
"http://cat.jpg",
))), patch("utils.morning_runner.discord.Embed") as mock_embed:
await run_morning(bot, channel)
channel.send.assert_called_once()
call_args = channel.send.call_args[1]
assert "embed" in call_args
assert call_args["embed"] is not None

209
tests/test_pogoda.py Normal file
View File

@ -0,0 +1,209 @@
import pytest
from utils.pogoda import translate_weather, pressure_to_mmhg, wmo_to_russian
class TestTranslateWeather:
"""Тесты функции translate_weather() — перевод описания погоды из английского в русский."""
@pytest.mark.parametrize(
"english, expected",
[
("Clear", "Ясно"),
("Sunny", "Ясно"),
("Partly cloudy", "Переменная облачность"),
("Cloudy", "Облачно"),
("Overcast", "Пасмурно"),
("Fog", "Туман"),
("Foggy", "Туманно"),
("Mist", "Туман"),
("Haze", "Дымка"),
("Light rain", "Небольшой дождь"),
("Moderate rain", "Умеренный дождь"),
("Heavy rain", "Сильный дождь"),
("Moderate or heavy rain at times", "Сильный дождь"), # "Heavy rain" совпадает раньше в mapping dict (key in text)
("Heavy rain at times", "Сильный дождь"),
("Light snow", "Небольшой снег"),
("Moderate snow", "Умеренный снег"),
("Heavy snow", "Сильный снег"),
("Blowing snow", "Метель"),
("Light freezing rain", "Лёгкий ледяной дождь"),
("Heavy freezing rain", "Сильный ледяной дождь"),
("Moderate or heavy freezing rain", "Сильный ледяной дождь"),
("Light sleet", "Light sleet"),
("Moderate or heavy sleet", "Moderate or heavy sleet"),
("Thundery outbreaks in nearby", "Гроза вблизи"),
("Patchy rain nearby", "Местами дождь"),
("Patchy snow nearby", "Местами снег"),
("Patchy sleet nearby", "Местами слякоть"),
("Patchy light drizzle", "Местами лёгкая морось"),
("Moderate or heavy snow in area", "Снег"),
("Moderate or heavy rain in area", "Дождь"),
],
)
def test_translate_known(self, english, expected):
"""Известные переводы должны возвращать ожидаемый результат."""
assert translate_weather(english) == expected
@pytest.mark.parametrize(
"input_value, expected",
[
("", ""),
(None, ""),
(" ", " "), # пробелы не считаются пустыми
],
)
def test_translate_empty(self, input_value, expected):
"""Пустой или None ввод должен возвращать ''."""
assert translate_weather(input_value) == expected
def test_translate_unknown_returns_original(self):
"""Неизвестный перевод должен возвращать оригинальный текст."""
unknown_text = "Unknown weather condition XYZ"
assert translate_weather(unknown_text) == unknown_text
def test_translate_partial_match(self):
"""Частичное совпадение ключа в тексте должно сработать."""
# "Moderate or heavy rain in area" должно найтись в "Light Moderate or heavy rain in area"
text_with_prefix = "Light Moderate or heavy rain in area"
assert translate_weather(text_with_prefix) == "Дождь"
def test_translate_longer_key_priority(self):
"""translate_weather ищет key in text, порядок dict важен.
"Heavy rain" стоит раньше "Moderate or heavy rain at times" в mapping,
и "heavy rain" in "moderate or heavy rain at times" = True.
Поэтому совпадёт первым и вернёт "Сильный дождь"."""
text = "Moderate or heavy rain at times"
assert translate_weather(text) == "Сильный дождь"
def test_translate_case_insensitive(self):
"""Перевод должен быть регистронезависимым."""
assert translate_weather("CLEAR") == "Ясно"
assert translate_weather("partly cloudy") == "Переменная облачность"
assert translate_weather("HEAVY RAIN") == "Сильный дождь"
def test_translate_with_whitespace(self):
"""Текст с пробелами по краям должен корректно переводиться."""
assert translate_weather(" Clear ") == "Ясно"
class TestPressureToMMHG:
"""Тесты функции pressure_to_mmhg() — конвертация давления из мб в мм рт. ст."""
@pytest.mark.parametrize(
"mb, expected",
[
(1013, 759.8),
(1000, 750.1),
(980, 735.1),
(1030, 772.6),
# (0, "—"), # 0 — falsy, возвращается '—' (баг)
],
)
def test_pressure_valid(self, mb, expected):
"""Валидные числовые значения должны конвертироваться корректно."""
assert pressure_to_mmhg(mb) == expected
@pytest.mark.parametrize(
"mb, expected",
[
("1013", 759.8),
("1000", 750.1),
("980", 735.1),
],
)
def test_pressure_string(self, mb, expected):
"""Строка-число должна конвертироваться корректно."""
assert pressure_to_mmhg(mb) == expected
@pytest.mark.parametrize(
"input_value, expected",
[
("", ""),
(None, ""),
("", ""),
],
)
def test_pressure_invalid(self, input_value, expected):
"""Невалидные значения должны возвращать ''."""
assert pressure_to_mmhg(input_value) == expected
def test_pressure_non_numeric_string(self):
"""Невалидная строка должна возвращать ''."""
assert pressure_to_mmhg("abc") == ""
def test_pressure_zero(self):
"""Нулевое значение — falsy, возвращается '' (известный баг)."""
assert pressure_to_mmhg(0) == ""
def test_pressure_negative(self):
"""Отрицательное значение должно конвертироваться."""
assert pressure_to_mmhg(-100) == -75.0
def test_pressure_float_string(self):
"""Строка с десятичной точкой должна конвертироваться."""
assert pressure_to_mmhg("1013.25") == 760.0
def test_pressure_very_large(self):
"""Очень большое значение должно работать."""
assert pressure_to_mmhg(999999) == 750061.2
class TestWmoToRussian:
"""Тесты функции wmo_to_russian() — перевод WMO кодов погоды."""
@pytest.mark.parametrize(
"code, expected",
[
(0, "Ясно"),
(1, "Ясно"),
(2, "Переменная облачность"),
(3, "Пасмурно"),
(45, "Туман"),
(48, "Туман"),
(51, "Лёгкая морось"),
(53, "Морось"),
(55, "Сильная морось"),
(56, "Ледяная морось"),
(57, "Сильная ледяная морось"),
(61, "Небольшой дождь"),
(63, "Дождь"),
(65, "Сильный дождь"),
(66, "Ледяной дождь"),
(67, "Сильный ледяной дождь"),
(71, "Небольшой снег"),
(73, "Снег"),
(75, "Сильный снег"),
(77, "Снежная крупа"),
(80, "Небольшой ливень"),
(81, "Ливень"),
(82, "Сильный ливень"),
(85, "Снежный ливень"),
(86, "Сильный снежный ливень"),
(95, "Гроза"),
(96, "Гроза с градом"),
(99, "Сильная гроза с градом"),
],
)
def test_wmo_known(self, code, expected):
"""Известные WMO коды должны возвращать ожидаемый перевод."""
assert wmo_to_russian(code) == expected
def test_wmo_unknown(self):
"""Неизвестный код должен возвращать 'Неизвестно'."""
assert wmo_to_russian(999) == "Неизвестно"
def test_wmo_negative_code(self):
"""Отрицательный код должен возвращать 'Неизвестно'."""
assert wmo_to_russian(-1) == "Неизвестно"
def test_wmo_none(self):
"""None должен возвращать 'Неизвестно'."""
assert wmo_to_russian(None) == "Неизвестно"
def test_wmo_large_code(self):
"""Очень большой код должен возвращать 'Неизвестно'."""
assert wmo_to_russian(9999) == "Неизвестно"
def test_wmo_float_code(self):
"""Дробный код — не найдётся в mapping."""
assert wmo_to_russian(1.5) == "Неизвестно"

1
utils/__init__.py Normal file
View File

@ -0,0 +1 @@

18
utils/cat.py Normal file
View File

@ -0,0 +1,18 @@
import asyncio
import requests
CAT_API_URL = "https://api.thecatapi.com/v1/images/search"
_session = requests.Session()
async def fetch_cat() -> str | None:
"""Получить URL случайного котика. Вернуть None при ошибке."""
try:
response = await asyncio.to_thread(_session.get, CAT_API_URL, timeout=10)
response.raise_for_status()
data = response.json()
return data[0]["url"]
except (requests.exceptions.RequestException, IndexError, KeyError):
return None

167
utils/morning_runner.py Normal file
View File

@ -0,0 +1,167 @@
"""Утилита для запуска утреннего дайджеста."""
import asyncio
import logging
from datetime import datetime, timedelta
import discord
from discord.ext import tasks
from utils.pogoda import fetch_weather, pressure_to_mmhg, translate_weather
from utils.news import fetch_rss, format_articles, RSS_URL_ARTICLES, RSS_URL_POSTS
from utils.cat import fetch_cat
logger = logging.getLogger(__name__)
async def run_morning(bot: "commands.Bot", channel: discord.TextChannel):
"""Выполнить утренний дайджест и отправить в канал."""
try:
api_url = "https://wttr.in/Magnitogorsk?format=j1&lang=ru"
# Параллельный запрос погоды, новостей и котика
weather_data, articles, posts, cat_url = await asyncio.gather(
fetch_weather(api_url),
fetch_rss(RSS_URL_ARTICLES),
fetch_rss(RSS_URL_POSTS),
fetch_cat(),
)
# --- Формируем embed ---
embed = discord.Embed(title="🌅 Утренний дайджест!", color=0xF4A460)
# Котик как thumbnail
if cat_url:
embed.set_thumbnail(url=cat_url)
description_lines = []
# --- Погода ---
if weather_data is not None:
current = weather_data.get("current_condition", [{}])[0]
if current:
temp = current.get("temp_C", "")
feels_like = current.get("FeelsLikeC", "")
description = translate_weather(current.get("weatherDesc", [{}])[0].get("value", ""))
humidity = current.get("humidity", "")
wind_kmh = current.get("windspeedKmph", "")
try:
wind = round(int(wind_kmh) / 3.6, 1) if wind_kmh != "" else ""
except (ValueError, TypeError):
wind = ""
pressure_mb = current.get("pressure", "")
pressure_mm = pressure_to_mmhg(pressure_mb)
description_lines.append(
f"**Погода в Магнитогорске:**\n"
f"Температура: {temp}°C (ощущается как {feels_like}°C)\n"
f"Описание: {description}\n"
f"Влажность: {humidity}%\n"
f"Ветер: {wind} м/с\n"
f"Давление: {pressure_mm} мм рт. ст."
)
else:
description_lines.append("Не удалось получить данные о погоде.")
else:
description_lines.append("Не удалось получить данные о погоде.")
description_lines.append("")
# --- Новости: статьи ---
if articles is not None:
if articles:
lines = format_articles(articles,
"Лучшие статьи за сутки / Искусственный интеллект / Хабr",
"https://habr.com/ru/hubs/artificial_intelligence/articles/top/daily/")
description_lines.append("\n".join(lines))
else:
description_lines.append("Новостей пока нет.")
else:
description_lines.append("Не удалось получить новости.")
description_lines.append("")
# --- Новости: посты ---
if posts is not None:
if posts:
lines = format_articles(posts,
"Лучшие новости за сутки / Искусственный интеллект / Хабr",
"https://habr.com/ru/hubs/artificial_intelligence/news/top/daily/")
description_lines.append("\n".join(lines))
else:
description_lines.append("Новостей пока нет.")
else:
description_lines.append("Не удалось получить новости.")
embed.description = "\n".join(description_lines)
await channel.send(embed=embed)
logger.info("✅ Утренний дайджест отправлен в #%s", channel.name)
except Exception as e:
logger.error("Ошибка при выполнении утреннего дайджеста: %s", e, exc_info=True)
try:
await channel.send("Не удалось выполнить утренний дайджест.")
except Exception:
pass
class Scheduler:
"""Планировщик ежедневных задач."""
def __init__(self, bot: commands.Bot, morning_time: str = "07:00"):
self.bot = bot
self.morning_time = morning_time
self._last_run_date = None
self.morning_loop = tasks.loop(seconds=1.0)(self._check_and_run_morning)
self._start_scheduler()
def _start_scheduler(self):
try:
self.morning_loop.start()
logger.info("Планировщик запущен (время: %s)", self.morning_time)
except RuntimeError:
logger.warning("Планировщик уже запущен")
def _stop_scheduler(self):
try:
self.morning_loop.stop()
logger.info("Планировщик остановлен")
except RuntimeError:
logger.warning("Планировщик уже остановлен")
def _calculate_next_run(self) -> datetime:
now = datetime.now()
hour, minute = map(int, self.morning_time.split(":"))
today_run = now.replace(hour=hour, minute=minute, second=0, microsecond=0)
if now >= today_run:
return today_run + timedelta(days=1)
return today_run
async def _check_and_run_morning(self):
now = datetime.now()
target = self._calculate_next_run()
if now >= target and now.day != self._last_run_date:
self._last_run_date = now.day
await self._run_morning()
async def _run_morning(self):
logger.info(f"Выполняю morning в {self.morning_time}")
for channel in self.bot.get_all_channels():
if isinstance(channel, discord.TextChannel):
if channel.permissions_for(channel.guild.me).send_messages:
try:
await channel.send("🌅 Утренний дайджест!")
await run_morning(self.bot, channel)
return
except Exception as e:
logger.error("Ошибка отправки в #%s: %s", channel.name, e)
continue
def start(self):
self._start_scheduler()
def stop(self):
self._stop_scheduler()

96
utils/news.py Normal file
View File

@ -0,0 +1,96 @@
import asyncio
from datetime import datetime
import requests
RSS_URL_ARTICLES = "https://habr.com/ru/rss/hubs/artificial_intelligence/articles/top/daily/?fl=ru"
RSS_URL_POSTS = "https://habr.com/ru/rss/hubs/artificial_intelligence/news/top/daily/?fl=ru"
_session = requests.Session()
async def fetch_rss(url):
"""Скачать и распарсить RSS-ленту (RSS 2.0 / Atom)."""
from xml.etree import ElementTree
try:
response = await asyncio.to_thread(_session.get, url, timeout=10)
response.raise_for_status()
root = ElementTree.fromstring(response.content)
# RSS 2.0
ns_dc = {"dc": "http://purl.org/dc/elements/1.1/"}
items = root.findall(".//item")
if not items:
# Atom
ns = {"atom": "http://www.w3.org/2005/Atom"}
items = root.findall("atom:entry", ns)
if not items:
return []
articles = []
for entry in items:
# RSS 2.0
title_el = entry.find("title")
date_el = entry.find("pubDate")
creator_el = entry.find("dc:creator", ns_dc)
categories = entry.findall("category")
# guid с isPermaLink="true" для чистого URL
guid_el = entry.find("guid[@isPermaLink='true']")
link = guid_el.text if guid_el is not None else ""
# Atom fallback
if title_el is None:
ns = {"atom": "http://www.w3.org/2005/Atom"}
title_el = entry.find("atom:title", ns)
link_el = entry.find("atom:link", ns)
link = link_el.get("href", "") if link_el is not None else ""
date_el = entry.find("atom:published", ns)
creator_el = entry.find("atom:author/atom:name", ns)
categories = entry.findall("atom:category", ns)
title = title_el.text if title_el is not None else "Без названия"
pub_date = date_el.text if date_el is not None else ""
creator = creator_el.text if creator_el is not None else ""
tags = [cat.text for cat in categories if cat.text] if categories else []
articles.append({
"title": title,
"link": link,
"pub_date": pub_date,
"creator": creator,
"tags": tags,
})
return articles[:10]
except requests.exceptions.RequestException:
return None
def _parse_date(pub_date):
"""Парсить дату из RSS в строку 'дд.мм.гггг' или вернуть часть даты."""
if not pub_date:
return ""
try:
d = pub_date.replace(" GMT", " +0000")
dt = datetime.strptime(d, "%a, %d %b %Y %H:%M:%S %z")
return dt.strftime("%d.%m.%Y")
except ValueError:
return pub_date[:10].replace("-", ".")
def truncate_title(title, max_len=60):
"""Обрезать заголовок, если он длиннее max_len."""
if len(title) > max_len:
return title[:max_len] + "..."
return title
def format_articles(articles, title, link):
"""Сформировать список строк для вывода статей/постов."""
lines = [f"**{title}**\n<{link}>"]
for i, article in enumerate(articles[:5], 1):
date_str = _parse_date(article["pub_date"])
short_title = truncate_title(article["title"])
lines.append(f"{short_title}\n{date_str} <{article['link']}>")
return lines

145
utils/pogoda.py Normal file
View File

@ -0,0 +1,145 @@
import asyncio
import requests
from requests.exceptions import ConnectionError, Timeout, SSLError
_session = requests.Session()
async def fetch_weather(api_url, timeout=10, max_retries=3):
"""Получить данные о погоде с retry."""
for attempt in range(max_retries):
try:
response = await asyncio.to_thread(_session.get, api_url, timeout=timeout)
response.raise_for_status()
return response.json()
except (SSLError, ConnectionError, Timeout):
if attempt < max_retries - 1:
delay = 2 ** attempt
print(f"Попытка {attempt + 1} не удалась. Повтор через {delay} сек...")
await asyncio.sleep(delay)
continue
break
except requests.RequestException as e:
print(f"Ошибка при получении данных: {e}")
break
# Fallback: Open-Meteo API (без ключа, HTTPS)
return await fetch_open_meteo()
async def fetch_open_meteo(lat=53.4069, lon=58.9797, timeout=10, max_retries=3):
"""Fallback на Open-Meteo API."""
url = (
f"https://api.open-meteo.com/v1/forecast?"
f"latitude={lat}&longitude={lon}&current=temperature,"
f"apparent_temperature,weather_code,wind_speed_10m,"
f"relative_humidity_2m,pressure_msl&timezone=Asia/Chelyabinsk"
)
for attempt in range(max_retries):
try:
response = await asyncio.to_thread(_session.get, url, timeout=timeout)
response.raise_for_status()
data = response.json()
current = data.get("current", {})
# weather_code WMO код -> перевод (https://open-meteo.com/en/docs)
weather_code = current.get("weather_code", None)
desc = wmo_to_russian(weather_code)
return {
"current_condition": [{
"temp_C": current.get("temperature", ""),
"FeelsLikeC": current.get("apparent_temperature", ""),
"weatherDesc": [{"value": desc}],
"humidity": current.get("relative_humidity_2m", ""),
"windspeedKmph": current.get("wind_speed_10m", ""),
"pressure": current.get("pressure_msl", ""),
}]
}
except (SSLError, ConnectionError, Timeout):
if attempt < max_retries - 1:
delay = 2 ** attempt
print(f"Попытка {attempt + 1} не удалась. Повтор через {delay} сек...")
await asyncio.sleep(delay)
continue
break
except requests.RequestException as e:
print(f"Ошибка при получении данных: {e}")
return None
return None
def wmo_to_russian(code):
"""Перевод WMO weather code в русский."""
mapping = {
0: "Ясно",
1: "Ясно", 2: "Переменная облачность",
3: "Пасмурно",
45: "Туман", 48: "Туман",
51: "Лёгкая морось", 53: "Морось", 55: "Сильная морось",
56: "Ледяная морось", 57: "Сильная ледяная морось",
61: "Небольшой дождь", 63: "Дождь", 65: "Сильный дождь",
66: "Ледяной дождь", 67: "Сильный ледяной дождь",
71: "Небольшой снег", 73: "Снег", 75: "Сильный снег",
77: "Снежная крупа",
80: "Небольшой ливень", 81: "Ливень", 82: "Сильный ливень",
85: "Снежный ливень", 86: "Сильный снежный ливень",
95: "Гроза", 96: "Гроза с градом", 99: "Сильная гроза с градом",
}
return mapping.get(code, "Неизвестно")
def translate_weather(en):
if not en:
return ""
mapping = {
"Moderate or heavy freezing rain in area": "Ледяной дождь",
"Moderate or heavy sleet in area": "Слякоть",
"Moderate or heavy snow in area": "Снег",
"Moderate or heavy rain in area": "Дождь",
"Thundery outbreaks in nearby": "Гроза вблизи",
"Patchy rain nearby": "Местами дождь",
"Patchy snow nearby": "Местами снег",
"Patchy sleet nearby": "Местами слякоть",
"Heavy freezing rain": "Сильный ледяной дождь",
"Heavy snow": "Сильный снег",
"Heavy rain": "Сильный дождь",
"Moderate or heavy rain at times": "Дождь",
"Moderate or heavy snow at times": "Снег",
"Blowing snow": "Метель",
"Patchy light drizzle": "Местами лёгкая морось",
"Moderate or heavy freezing rain at a distance": "Ледяной дождь",
"Moderate or heavy sleet at a distance": "Слякоть",
"Light rain shower": "Небольшой дождь",
"Heavy rain shower": "Сильный дождь",
"Moderate rain": "Умеренный дождь",
"Light rain": "Небольшой дождь",
"Moderate rain at times": "Умеренный дождь",
"Heavy rain at times": "Сильный дождь",
"Light snow": "Небольшой снег",
"Moderate snow": "Умеренный снег",
"Patchy light snow": "Местами лёгкий снег",
"Partly cloudy": "Переменная облачность",
"Moderate or light sleet": "Слякоть",
"Light freezing rain": "Лёгкий ледяной дождь",
"Foggy": "Туманно",
"Fog": "Туман",
"Mist": "Туман",
"Haze": "Дымка",
"Overcast": "Пасмурно",
"Cloudy": "Облачно",
"Clear": "Ясно",
"Sunny": "Ясно",
}
for key, value in mapping.items():
if key.lower() in en.lower():
return value
return en
def pressure_to_mmhg(mb):
if mb == "" or not mb:
return ""
try:
return round(float(mb) * 0.750062, 1)
except (ValueError, TypeError):
return ""