137 lines
4.7 KiB
Python
137 lines
4.7 KiB
Python
"""Bot's callbacks implementation"""
|
||
|
||
import asyncio
|
||
import glob
|
||
import html
|
||
import os
|
||
import shutil
|
||
import traceback
|
||
|
||
from mab import MatrixBot
|
||
from mab.types import RoomEventData
|
||
from mab.filters.text import TextFilter, TextStartsWithFilter, TextEndsWithFilter
|
||
|
||
from nio import MatrixRoom, RoomMessageText
|
||
|
||
from downloader import Downloader
|
||
|
||
#
|
||
# PRIVATE DATA
|
||
#
|
||
_bot: MatrixBot
|
||
_tasks: set[asyncio.Task] = set()
|
||
|
||
#
|
||
# PRIVATE
|
||
#
|
||
async def worker_download(room: MatrixRoom, text: RoomMessageText) -> None:
|
||
# respond
|
||
response = (
|
||
f"<strong>Скачивание видео по "
|
||
f"<a href=\"{html.escape(text.body)}\">вашей ссылке</a></strong>"
|
||
)
|
||
await _bot.send_text(room, response)
|
||
# get formats
|
||
d = Downloader(text.body)
|
||
try:
|
||
result = await d.download_video("temp")
|
||
except:
|
||
await _bot.send_text(room, traceback.format_exc())
|
||
return
|
||
await _bot.send_text(room, "<strong>Скачано, загрузка в Matrix...</strong>")
|
||
# upload
|
||
try:
|
||
await _bot.send_video(room, result, text=f"Скачано из {text.body}")
|
||
except:
|
||
await _bot.send_text(room.room_id, traceback.format_exc())
|
||
finally:
|
||
os.unlink(result)
|
||
|
||
#
|
||
# CALLBACKS
|
||
#
|
||
async def command_help(e: RoomEventData) -> None:
|
||
"""!help"""
|
||
if e.event.sender == _bot.get_client().user_id:
|
||
return
|
||
resp = "<strong>2026-matrix-down</strong><br>"
|
||
resp += "Этот бот используется для скачивания видео/аудио с хостингов, "
|
||
resp += "поддерживаемых <code>yt-dlp</code>. Просто отправьте ссылку в эту "
|
||
resp += "комнату, чтобы бот предложил варианты загрузки."
|
||
await e.bot.send_text(e.room, resp)
|
||
|
||
async def command_http_https(e: RoomEventData) -> None:
|
||
"""The message starts with `http://` or `https://`"""
|
||
if e.event.sender == _bot.get_client().user_id:
|
||
return
|
||
if not isinstance(e.event, RoomMessageText):
|
||
return
|
||
t = asyncio.create_task(worker_download(e.room, e.event))
|
||
_tasks.add(t)
|
||
t.add_done_callback(_tasks.discard)
|
||
|
||
async def command_test(e: RoomEventData) -> None:
|
||
"""The message ends with ` TesT` (case matters)"""
|
||
if e.event.sender == _bot.get_client().user_id:
|
||
return
|
||
# glob patterns to use
|
||
glob_img = ["*.png", "*.jpg", ".jpeg", "*.gif", "*.webp"]
|
||
glob_vid = ["*.mp4", "*.avi", "*.mkv"]
|
||
# TODO
|
||
glob_aud = ["*.mp3", "*.m4a", ".wav", "*.ogg", "*.flac"]
|
||
# help message
|
||
resp = "<strong>Сообщение заканчивается на <code> TesT</code>.</strong><br>"
|
||
resp += "Сейчас будут отправлены все файлы из директории <code>test_media"
|
||
resp += "</code>, у которых следующие расширения:<br>"
|
||
resp += f"<ul><li>Изображения: {", ".join([f"<code>{g}</code>" for g in glob_img])}</li></ul>"
|
||
resp += f"<ul><li>Видео: {", ".join([f"<code>{g}</code>" for g in glob_vid])}</li></ul>"
|
||
await _bot.send_text(e.room, resp)
|
||
await asyncio.sleep(1)
|
||
# send images
|
||
for g in [f"test_media/{g}" for g in glob_img]:
|
||
imgs = glob.glob(g)
|
||
for file in imgs:
|
||
try:
|
||
await _bot.send_image(
|
||
e.room,
|
||
file,
|
||
text=f"Картинка <code>{html.escape(file)}</code>",
|
||
is_html=True
|
||
)
|
||
except:
|
||
await _bot.send_text(e.room, traceback.format_exc())
|
||
await asyncio.sleep(0.5)
|
||
# send videos
|
||
for g in [f"test_media/{g}" for g in glob_vid]:
|
||
imgs = glob.glob(g)
|
||
for file in imgs:
|
||
try:
|
||
await _bot.send_video(
|
||
e.room,
|
||
file,
|
||
text=f"Видео <code>{html.escape(file)}</code>",
|
||
is_html=True
|
||
)
|
||
except:
|
||
await _bot.send_text(e.room, traceback.format_exc())
|
||
await asyncio.sleep(0.5)
|
||
|
||
#
|
||
# PUBLIC
|
||
#
|
||
async def initialize(bot: MatrixBot) -> None:
|
||
"""This function initializes the bot. It does not start it."""
|
||
global _bot
|
||
shutil.rmtree("temp", ignore_errors=True)
|
||
os.makedirs("temp", exist_ok=True)
|
||
_bot = bot
|
||
bot.add_callback(
|
||
TextStartsWithFilter(["https://", "http://"]), command_http_https)
|
||
bot.add_callback(
|
||
TextEndsWithFilter(" TesT", any_case=False), command_test)
|
||
bot.add_callback(
|
||
TextFilter(), command_help)
|
||
|
||
async def deinitizalize(bot: MatrixBot) -> None:
|
||
"""This function deinitializes the bot. It does not stop it."""
|
||
shutil.rmtree("temp", ignore_errors=True) |