mirror of
https://github.com/tropicoo/yt-dlp-bot.git
synced 2025-03-04 02:33:06 +08:00
33 lines
1.1 KiB
Python
33 lines
1.1 KiB
Python
import logging
|
|
|
|
from yt_shared.enums import TaskSource
|
|
from yt_shared.rabbit.publisher import Publisher
|
|
from yt_shared.schemas.url import URL
|
|
from yt_shared.schemas.video import VideoPayload
|
|
|
|
|
|
class URLService:
|
|
def __init__(self) -> None:
|
|
self._log = logging.getLogger(self.__class__.__name__)
|
|
self._publisher = Publisher()
|
|
|
|
async def process_url(self, url: URL) -> bool:
|
|
return await self._send_to_worker(url)
|
|
|
|
async def process_urls(self, urls: list[URL]) -> None:
|
|
for url in urls:
|
|
await self._send_to_worker(url)
|
|
|
|
async def _send_to_worker(self, url: URL) -> bool:
|
|
payload = VideoPayload(
|
|
url=url.url,
|
|
message_id=url.message_id,
|
|
from_user_id=url.from_user_id,
|
|
from_chat_id=url.from_chat_id,
|
|
from_chat_type=url.from_chat_type,
|
|
source=TaskSource.BOT,
|
|
)
|
|
is_sent = await self._publisher.send_for_download(payload)
|
|
if not is_sent:
|
|
self._log.error('Failed to publish URL %s to message broker', url.url)
|
|
return is_sent
|