# =======================================
# _ __ __ __ _
# | |/ /___ | \/ | ___ __| |___
# | ' // _ \ | |\/| |/ _ \ / _` / __|
# | . \ __/ | | | | (_) | (_| \__ \
# |_|\_\___| |_| |_|\___/ \__,_|___/
# @ke_mods
# =======================================
# meta developer: @ke_mods
# requires: pillow
import asyncio
import logging
import traceback
from logging import basicConfig
from io import BytesIO
import requests
from PIL import Image
from .. import loader, utils
basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
@loader.tds
class RandomAnimePicMod(loader.Module):
strings = {
"name": "RandomAnimePic",
"img": "✅ Your anime pic\n🔗 URL: {}",
"loading": "✨ Loading image...",
"categories_loading": "✨ Loading categories...",
"categories": "✅ Available categories\n{}
",
"no_categories": "🚫 Categories not found",
"error": "🚫 An unexpected error occurred...",
}
strings_ru = {
"img": "✅ Ваша аниме-картинка\n🔗 Ссылка: {}",
"loading": "✨ Загрузка изображения...",
"categories_loading": "✨ Загрузка категорий...",
"categories": "✅ Доступные категории\n{}
",
"no_categories": "🚫 Категории не найдены",
"error": "🚫 Произошла непредвиденная ошибка...",
}
RANDOM_API_URL = "https://api.nekosapi.com/v4/images/random"
IMAGES_API_URL = "https://api.nekosapi.com/v4/images"
CATEGORIES_SCAN_LIMIT = 500
@loader.command(ru_doc="- получить рандомную аниме-картинку 👀")
async def rapiccmd(self, message):
"""- fetch random anime-pic 👀"""
await utils.answer(message, self.strings["loading"])
try:
category = await utils.get_args_raw().strip()
def fetch_image():
params = {"limit": 1, "rating": ["safe"]}
if category:
params["tags"] = [category]
response = requests.get(self.RANDOM_API_URL, params=params, timeout=15)
response.raise_for_status()
data = response.json()
if not isinstance(data, list) or not data:
raise ValueError("API returned empty response")
url = data[0].get("url")
if not url:
raise ValueError("API response does not contain image url")
image_response = requests.get(url, timeout=20)
image_response.raise_for_status()
image_stream = BytesIO(image_response.content)
image = Image.open(image_stream)
image.load()
output = BytesIO()
if "A" in image.getbands() or image.mode == "P":
image.convert("RGBA").save(output, format="PNG")
output.name = "anime.png"
else:
image.convert("RGB").save(output, format="JPEG", quality=95)
output.name = "anime.jpg"
output.seek(0)
return url, output
url, file = await asyncio.to_thread(fetch_image)
await utils.answer(
message,
self.strings["img"].format(url),
file=file
)
except Exception:
logger.error(
"Error fetching random anime pic: %s",
traceback.format_exc(),
)
await utils.answer(message, self.strings["error"])
@loader.command(ru_doc="- получить список категорий из API 👀")
async def racategoriescmd(self, message):
"""- fetch categories from api 👀"""
await utils.answer(message, self.strings["categories_loading"])
try:
def fetch_categories() -> list[str]:
tags = set()
offset = 0
while offset < self.CATEGORIES_SCAN_LIMIT:
response = requests.get(
self.IMAGES_API_URL,
params={
"limit": 100,
"offset": offset,
"rating": ["safe"],
},
timeout=20,
)
response.raise_for_status()
data = response.json()
items = data.get("items") or data.get("results") or []
if not items:
break
for item in items:
tags.update(item.get("tags", []))
if len(items) < 100:
break
offset += 100
return sorted(tags)
categories = await asyncio.to_thread(fetch_categories)
if not categories:
await utils.answer(message, self.strings["no_categories"])
return
formatted_categories = ", ".join(
f"{category}" for category in categories
)
await utils.answer(
message,
self.strings["categories"].format(formatted_categories),
)
except Exception:
logger.error(
"Error fetching categories: %s",
traceback.format_exc(),
)
await utils.answer(message, self.strings["error"])