darkshy/derpibooru-telegram-bot
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Add tests

Browse Source
This commit is contained in:
DarkShyMW
2025-08-23 09:54:40 +03:00
parent 176dd968d5
commit 16ee57b158
6 changed files with 170 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files

29
.github/workflows/tests.yml vendored Normal file
View File

@@ -0,0 +1,29 @@
name: Tests
on:
push:
branches: [ "main" ]
pull_request:
branches: [ "main" ]
jobs:
test:
runs-on: ubuntu-latest
steps:
- name: Checkout repository
uses: actions/checkout@v4
- name: Set up Python
uses: actions/setup-python@v5
with:
python-version: "3.11"
- name: Install dependencies
run: |
python -m pip install --upgrade pip
pip install -r requirements.txt
pip install pytest pytest-asyncio
- name: Run tests
run: pytest

1
README.md
View File

@@ -1,4 +1,5 @@
# 🐴 Derpibooru Telegram Bot
![Tests](https://github.com/DarkShyMW/derpibooru-telegram-bot/actions/workflows/tests.yml/badge.svg)
Telegram-бот, который автоматически публикует изображения с [Derpibooru](https://derpibooru.org) по заданным тегам.

93
app.py
View File

@@ -3,26 +3,28 @@ import aiohttp
import logging
import random
import json
from datetime import datetime, timedelta
from telegram import Bot
from pathlib import Path
import sys
from datetime import datetime, timedelta
from pathlib import Path
from telegram import Bot
# === Настройки ===
TELEGRAM_TOKEN = "************************"
CHANNEL_ID = -100*********** # числовой ID канала
DERPIBOORU_TOKEN = "токен от дерпибуру"
TELEGRAM_TOKEN = "токен_вашего_тг_аккаунта" # Токен от Telegram
CHANNEL_ID = -100000000000000 # числовой ID канала
DERPIBOORU_TOKEN = "xxxxxxxxxxxxxxxx" # Токен от DerpiBooru
TAGS_LIST = [
["тег1"],
["тег2"],
["тег3"],
["тег4"]
["gay"],
["female"],
["vulva"],
["creampie"],
]
DERPYBOORU_API_SEARCH = "https://derpibooru.org/api/v1/json/search/images"
DERPIBOORU_API = "https://derpibooru.org/api/v1/json/search/images"
SENT_IMAGES_FILE = Path("sent_images.json")
FILTER_ID = 56027 # ID фильтра, разрешающего NSFW изображения
# Логи
# === Логи ===
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s [%(levelname)s] %(message)s",
@@ -34,7 +36,7 @@ logging.basicConfig(
bot = Bot(token=TELEGRAM_TOKEN)
# Загрузка уже отправленных изображений
# === Загружаем уже отправленные изображения ===
if SENT_IMAGES_FILE.exists():
try:
with SENT_IMAGES_FILE.open("r", encoding="utf-8") as f:
@@ -47,11 +49,12 @@ else:
async def fetch_image(session, tags):
"""Запрос изображения с Derpibooru по тегам."""
params = {
"q": " ".join(tags),
"per_page": 50,
"page": 1,
"key": DERPIBOORU_TOKEN
"key": DERPIBOORU_TOKEN,
}
if FILTER_ID:
params["filter_id"] = FILTER_ID
@@ -59,10 +62,11 @@ async def fetch_image(session, tags):
backoff = 1
while True:
try:
async with session.get(DERPYBOORU_API_SEARCH, params=params) as response:
async with session.get(DERPIBOORU_API, params=params) as response:
if response.status == 200:
data = await response.json()
images = data.get("images", [])
if not images:
logging.warning("Ничего не найдено по тегам: %s", tags)
return None
@@ -71,6 +75,7 @@ async def fetch_image(session, tags):
for img in images:
reps = img.get("representations", {})
url = reps.get("full") or reps.get("large") or reps.get("medium")
if not url:
continue
if url not in sent_images:
@@ -80,16 +85,21 @@ async def fetch_image(session, tags):
logging.info("Выбрано изображение: %s", url)
return url, author, source, img_tags
logging.info("Все найденные изображения уже отправлены для тегов: %s", tags)
logging.info("Все найденные изображения уже отправлены: %s", tags)
return None
elif response.status in [500, 501]:
logging.warning("Ошибка %s от Derpibooru. Ждем %s секунд перед повтором...", response.status, backoff)
elif response.status in (500, 501):
logging.warning(
"Ошибка %s от Derpibooru. Ждем %s секунд...",
response.status, backoff
)
await asyncio.sleep(backoff)
backoff = min(backoff * 2, 900)
else:
logging.error("Ошибка API Derpibooru: %s", response.status)
return None
except aiohttp.ClientError as e:
logging.error("Ошибка запроса к Derpibooru: %s", e)
await asyncio.sleep(backoff)
@@ -97,19 +107,31 @@ async def fetch_image(session, tags):
async def post_image(tags=None):
"""Публикация изображения в канал."""
async with aiohttp.ClientSession() as session:
if tags is None:
tags = random.choice(TAGS_LIST)
result = await fetch_image(session, tags)
if result:
url, author, source, img_tags = result
caption = f"Автор: {author}\nИсточник: {source}\nТеги: {', '.join(img_tags)}"
caption = (
f"Автор: {author}\n"
f"Источник: {source}\n"
f"Теги: {', '.join(img_tags)}"
)
try:
await bot.send_photo(chat_id=CHANNEL_ID, photo=url, caption=caption)
await bot.send_photo(
chat_id=CHANNEL_ID,
photo=url,
caption=caption
)
logging.info("Отправлено изображение: %s", url)
sent_images.add(url)
with SENT_IMAGES_FILE.open("w", encoding="utf-8") as f:
json.dump(list(sent_images), f)
except Exception as e:
logging.error("Ошибка при отправке изображения: %s", e)
else:
@@ -117,15 +139,19 @@ async def post_image(tags=None):
async def scheduler():
"""Публикует изображение каждый час (в начале часа)."""
while True:
now = datetime.now()
# Публикуем догоняющее изображение, если запуск не в начале часа
# Если бот запущен не в начале часа — публикуем сразу
if now.minute != 0 or now.second != 0:
logging.info("Пропущенный час — публикуем сразу")
await post_image()
# Ждем до следующей 00 минуты
next_run = (now + timedelta(hours=1)).replace(minute=0, second=0, microsecond=0)
# Ждем до следующего часа
next_run = (now + timedelta(hours=1)).replace(
minute=0, second=0, microsecond=0
)
wait_seconds = (next_run - datetime.now()).total_seconds()
logging.info("Ждем %.0f секунд до следующей публикации", wait_seconds)
await asyncio.sleep(wait_seconds)
@@ -135,26 +161,26 @@ async def scheduler():
async def command_listener():
"""Слушает команды из stdin (postnow, posttags, timetopost)."""
loop = asyncio.get_event_loop()
while True:
cmd = await loop.run_in_executor(None, sys.stdin.readline)
cmd = cmd.strip().lower()
if cmd == "postnow" or cmd == ">> postnow":
logging.info("Команда postnow получена, публикуем случайное изображение...")
if cmd in {"postnow", ">> postnow"}:
logging.info("Команда postnow получена — публикуем...")
await post_image()
elif cmd.startswith("posttags "):
tags = cmd[len("posttags "):].split()
logging.info("Команда posttags получена, публикуем изображение с тегами: %s", tags)
logging.info("Команда posttags получена: %s", tags)
await post_image(tags)
elif cmd.startswith("timetopost"):
now = datetime.now()
# Публикуем догоняющее изображение, если запуск не в начале часа
if now.minute != 0 or now.second != 0:
logging.info("Пропущенный час — публикуем сразу")
await post_image()
# Ждем до следующей 00 минуты
next_run = (now + timedelta(hours=1)).replace(minute=0, second=0, microsecond=0)
next_run = (now + timedelta(hours=1)).replace(
minute=0, second=0, microsecond=0
)
wait_seconds = (next_run - datetime.now()).total_seconds()
logging.info("Времени до публикации: %s", wait_seconds)
@@ -163,6 +189,7 @@ async def command_listener():
async def main():
"""Главный запуск: планировщик + команды."""
await asyncio.gather(
scheduler(),
command_listener()

6
requirements.txt
View File

@@ -1,2 +1,4 @@
python-telegram-bot==20.5
aiohttp==3.9.5
pytest
pytest-asyncio
aiohttp
aioresponses

51
tests/test_fetch.py Normal file
View File

@@ -0,0 +1,51 @@
import pytest
from aioresponses import aioresponses
from app import fetch_image, DERPYBOORU_API_SEARCH
@pytest.mark.asyncio
async def test_fetch_image_success():
"""Тестируем успешный возврат картинки из API"""
tags = ["gay"]
mock_data = {
"images": [
{
"id": 123,
"uploader": "tester",
"view_url": "https://derpibooru.org/images/123",
"tags": ["gay", "cute"],
"representations": {
"full": "https://example.com/full.png",
"large": "https://example.com/large.png"
}
}
]
}
with aioresponses() as mocked:
mocked.get(DERPYBOORU_API_SEARCH, status=200, payload=mock_data)
async with __import__("aiohttp").ClientSession() as session:
result = await fetch_image(session, tags)
assert result is not None
url, author, source, img_tags = result
assert url == "https://example.com/full.png"
assert author == "tester"
assert source == "https://derpibooru.org/images/123"
assert "gay" in img_tags
@pytest.mark.asyncio
async def test_fetch_image_empty():
"""Тестируем ситуацию, когда API не вернул картинок"""
tags = ["nonexistent"]
with aioresponses() as mocked:
mocked.get(DERPYBOORU_API_SEARCH, status=200, payload={"images": []})
async with __import__("aiohttp").ClientSession() as session:
result = await fetch_image(session, tags)
assert result is None

24
tests/test_post.py Normal file
View File

@@ -0,0 +1,24 @@
import pytest
from unittest.mock import AsyncMock, patch
from app import post_image
@pytest.mark.asyncio
@patch("app.bot.send_photo", new_callable=AsyncMock)
@patch("app.fetch_image", return_value=(
"https://example.com/full.png",
"tester",
"https://derpibooru.org/images/123",
["gay", "cute"]
))
async def test_post_image_success(mock_fetch, mock_send_photo, tmp_path, monkeypatch):
"""Тестируем успешную публикацию картинки"""
# временный файл для sent_images
monkeypatch.setattr("app.SENT_IMAGES_FILE", tmp_path / "sent.json")
monkeypatch.setattr("app.sent_images", set())
await post_image(tags=["gay"])
mock_fetch.assert_called_once()
mock_send_photo.assert_awaited_once()