__version__ = (1, 1, 0)
# █ █ ▀ █▄▀ ▄▀█ █▀█ ▀
# █▀█ █ █ █ █▀█ █▀▄ █
# © Copyright 2022
#
# https://t.me/hikariatama
#
# 🔒 Licensed under the GNU AGPLv3
# 🌐 https://www.gnu.org/licenses/agpl-3.0.html
# meta pic: https://static.dan.tatar/artai_icon.png
# meta banner: https://mods.hikariatama.ru/badges/artai.jpg
# meta developer: @hikarimods
# scope: hikka_only
# scope: hikka_min 1.2.10
import asyncio
import base64
import io
import random
from typing import Union
import requests
from telethon.tl.types import Message
from .. import loader, utils
from ..inline.types import InlineCall
def base(bytes_: bytes) -> str:
return f"data:image/jpeg;base64,{base64.b64encode(bytes_).decode()}"
async def poll(queue_hash: str) -> str:
for _ in range(50):
answ = await utils.run_sync(
requests.post,
"https://akhaliq-jojogan.hf.space/api/queue/status/",
params={"hash": queue_hash},
)
if answ.json()["status"] == "COMPLETE":
return answ.json()["data"]["data"][0]
elif answ.json()["status"] != "PENDING":
return False
await asyncio.sleep(3)
async def animefy(image: bytes, engine: str) -> Union[bytes, bool]:
file = io.BytesIO(
base64.decodebytes(
(
await poll(
(
await utils.run_sync(
requests.post,
"https://akhaliq-jojogan.hf.space/api/queue/push/",
headers={
"accept": "*/*",
"accept-encoding": "gzip, deflate, br",
"accept-language": "en-US,en;q=0.9,ru;q=0.8",
"cache-control": "no-cache",
"content-type": "application/json",
"origin": "https://akhaliq-jojogan.hf.space",
"pragma": "no-cache",
"referer": (
"https://akhaliq-jojogan.hf.space/?__theme=light"
),
"sec-fetch-dest": "empty",
"sec-fetch-mode": "cors",
"sec-fetch-site": "same-origin",
"sec-gpc": "1",
"user-agent": (
"Mozilla/5.0 (Windows NT 10.0; Win64; x64)"
" AppleWebKit/537.36 (KHTML, like Gecko)"
" Chrome/92.0.4515.131 Safari/537.36"
),
},
json={
"action": "predict",
"data": [base(image), engine],
"fn_index": 0,
"session_hash": utils.rand(11).lower(),
},
)
).json()["hash"]
)
)
.split("base64,")[1]
.encode()
)
)
file.name = "photo.jpg"
return file
@loader.tds
class ArtAIMod(loader.Module):
"""Ultimate module, which uses AI to draw ppl"""
paint = "🎨"
strings = {
"name": "ArtAI",
"no_reply": (
"🚫 Reply to a photo"
" required"
),
"pick_engine": "👩🎤 Please, choose engine to process this photo",
"uploading": "☁️ Uploading...",
"success": (
f"{paint} This is nice",
f"{paint} Shee-e-esh",
f"{paint} I'm the artist, this is my POV!",
f"{paint} Do not blame me, I'm the artist",
),
}
strings_ru = {
"no_reply": (
"🚫 Ответь на фото"
),
"pick_engine": "👩🎤 Выбери движок для обработки этой фотографии",
"uploading": "☁️ Загружаю...",
"success": (
f"{paint} Это классно",
f"{paint} Shee-e-esh",
f"{paint} Я художник, я так вижу!",
f"{paint} Не обвиняй меня, я художник",
),
}
strings_de = {
"no_reply": (
"🚫 Antworte auf ein"
" Foto"
),
"pick_engine": "👩🎤 Wähle einen Motor, um dieses Foto zu verarbeiten",
"uploading": "☁️ Hochladen...",
"success": (
f"{paint} Das ist schön",
f"{paint} Shee-e-esh",
f"{paint} Ich bin der Künstler, das ist meine Sicht!",
f"{paint} Verurteile mich nicht, ich bin der Künstler",
),
}
strings_uz = {
"no_reply": (
"🚫 Fotoya javob"
" bering"
),
"pick_engine": "👩🎤 Ushbu rasmni ishlash uchun injinani tanlang",
"uploading": "☁️ Yuklanmoqda...",
"success": (
f"{paint} Bu yaxshi",
f"{paint} Shee-e-esh",
),
}
strings_es = {
"no_reply": (
"🚫 Responde a una"
" foto"
),
"pick_engine": "👩🎤 Elige un motor para procesar esta foto",
"uploading": "☁️ Subiendo...",
"success": (
f"{paint} Esto es bueno",
f"{paint} Shee-e-esh",
f"{paint} Soy el artista, esta es mi visión",
f"{paint} No me culpes, soy el artista",
),
}
strings_tr = {
"no_reply": (
"🚫 Bir fotoğrafa yanıt"
" verin"
),
"pick_engine": "👩🎤 Bu fotoğrafı işlemek için bir motor seçin",
"uploading": "☁️ Yükleniyor...",
"success": (
f"{paint} Bu güzel",
f"{paint} Shee-e-esh",
f"{paint} Ben sanatçıyım, bu benim bakış açım!",
f"{paint} Sana suçlamayın, ben sanatçıyım",
),
}
async def artaicmd(self, message: Message):
""" - Create anime art from photo"""
reply = await message.get_reply_message()
if not reply or not reply.photo:
await utils.answer(message, self.strings("no_reply"))
return
await self.inline.form(
message=message,
text=self.strings("pick_engine"),
reply_markup=self._gen_markup(reply),
)
async def _process_engine(
self,
call: InlineCall,
engine: str,
chat_id: int,
message_id: int,
):
await call.edit(self.strings("uploading"))
media = await self._client.download_media(
(
await self._client.get_messages(
entity=chat_id,
ids=[message_id],
limit=1,
)
)[0],
bytes,
)
if engine != "All":
await self._client.send_file(
chat_id,
file=await animefy(media, engine),
reply_to=message_id,
caption=random.choice(self.strings("success")),
)
await call.delete()
return
else:
res = []
statuses = {
"JoJo": "⬜️",
"Disney": "⬜️",
"Jinx": "⬜️",
"Caitlyn": "⬜️",
"Yasuho": "⬜️",
"Arcane Multi": "⬜️",
"Art": "⬜️",
"Spider-Verse": "⬜️",
"Sketch": "⬜️",
}
for engine in statuses:
suffix = (
lambda: (
f"Processing image...\n\n{''.join(statuses.values())}"
)
)
res += [await animefy(media, engine)]
statuses[engine] = "🟩"
await call.edit(suffix())
await call.delete()
try:
await self._client.send_file(
chat_id,
file=res,
reply_to=message_id,
caption=random.choice(self.strings("success")),
)
except TypeError:
pass
def _gen_markup(self, reply: Message) -> list:
engines = [
"👊 JoJo",
"👸 Disney",
"🥷 Jinx",
"😥 Caitlyn",
"👩🎤 Yasuho",
"👨🎤 Arcane Multi",
"🎨 Art",
"🕸 Spider-Verse",
"✒️ Sketch",
"🎁 All",
]
return utils.chunks(
[
{
"text": engine,
"callback": self._process_engine,
"args": (
engine.split(maxsplit=1)[1],
utils.get_chat_id(reply),
reply.id,
),
}
for engine in engines
],
2,
)