minor fixes, added result processing
This commit is contained in:
nikili0n
@ -1,16 +1,16 @@
|
||||
import asyncio
|
||||
import concurrent.futures as pool
|
||||
import subprocess
|
||||
import traceback
|
||||
|
||||
from functools import partial
|
||||
|
||||
from fastapi import HTTPException
|
||||
from urllib.parse import urlparse
|
||||
|
||||
from src.core.async_queue import AsyncQueue
|
||||
from src.core.rabbitmq import get_messages
|
||||
from src.core.rabbitmq import get_messages, publish_message_with_task_done
|
||||
from src.core.redis_client import RedisClient
|
||||
from src.core.result import Result, ResultTypeEnum
|
||||
from src.exceptions.download_exceptions import SiteNotImplementedException
|
||||
from src.exceptions.download_exceptions import FileAlreadyExistException, SiteNotImplementedException
|
||||
from src.parsers.MyMail.my_mail_parser import MyMailParser
|
||||
from src.parsers.Yappy.yappy_parser import YappyParser
|
||||
from src.parsers.base_parser import BaseParser
|
||||
@ -39,55 +39,82 @@ class MasterService:
|
||||
async def create_workers(self):
|
||||
while True:
|
||||
video_params = await self.queue.get()
|
||||
#TODO: позднее написать функцию для определения парсера автоматически
|
||||
# TODO: позднее написать функцию для определения парсера автоматически
|
||||
redis = RedisClient()
|
||||
# TODO: проверить что в редисе задача либо уже выполнена, т.е. сразу отдать ссылку, либо что она ранее была закончена с ошибкой
|
||||
# и проверять словарь self.currently_underway, для надёжности
|
||||
await redis.del_task_from_queue_and_add_to_tasks(task=video_params)
|
||||
self.currently_underway[video_params['link']] = video_params
|
||||
|
||||
download_task = self.loop.run_in_executor(self.executor, partial(
|
||||
MasterService.video_processing_executor, video_params=video_params
|
||||
)
|
||||
)
|
||||
))
|
||||
|
||||
result = await download_task
|
||||
await redis.del_task_from_tasks_and_add_to_task_done(task={"link": video_params["link"], "result": result})
|
||||
result: Result = await download_task
|
||||
if result.result_type in [ResultTypeEnum.DONE, ResultTypeEnum.EXIST]:
|
||||
await redis.del_task_from_tasks_and_add_to_task_done(task=result.value)
|
||||
await publish_message_with_task_done(task=result.value)
|
||||
self.queue.task_done()
|
||||
else:
|
||||
error_message = {
|
||||
"link": video_params["link"],
|
||||
"result": result.value,
|
||||
"status": "error"
|
||||
}
|
||||
await redis.del_task_from_tasks_and_add_to_task_done(task=error_message)
|
||||
await publish_message_with_task_done(task=error_message)
|
||||
|
||||
if video_params['link'] in self.currently_underway:
|
||||
del self.currently_underway[video_params['link']]
|
||||
# TODO process result
|
||||
'''
|
||||
Result.Done \ Result.Exist - уведомить что задача выполнена, и отослать во вторую очередь сообщений RabbitMQ сообщение об этом
|
||||
Result.Error - в таблице Редиса для выполненых задач, пометить, что это ошибка и уведомить об этом по второй очереди сообщений
|
||||
и потом почистить self.currently_underway
|
||||
'''
|
||||
self.queue.task_done()
|
||||
# Result.Done \ Result.Exist - уведомить что задача выполнена, и отослать во вторую очередь сообщений
|
||||
# RabbitMQ сообщение об этом
|
||||
# Result.Error - в таблице Редиса для выполненых задач, пометить, что это ошибка и уведомить об этом
|
||||
# по второй очереди сообщений и потом почистить self.currently_underway
|
||||
|
||||
@staticmethod
|
||||
def video_download(video_params: dict):
|
||||
downloader: BaseParser | YappyParser | MyMailParser = MasterService.get_parser(video_params)
|
||||
try:
|
||||
result = downloader.video_download()
|
||||
return result
|
||||
except SiteNotImplementedException as ex:
|
||||
raise HTTPException(
|
||||
status_code=400,
|
||||
detail=ex.message
|
||||
)
|
||||
result = downloader.video_download()
|
||||
return result
|
||||
|
||||
@staticmethod
|
||||
def get_parser(params: dict):
|
||||
parser_mapping = {
|
||||
"MyMailRu": MyMailParser(params),
|
||||
"base": BaseParser(params),
|
||||
"Yappy": YappyParser(params),
|
||||
}
|
||||
return parser_mapping[params["parser"]]
|
||||
try:
|
||||
domain = urlparse(params["link"]).netloc
|
||||
parser_mapping = {
|
||||
"my.mail.ru": MyMailParser(params),
|
||||
"www.youtube.com": BaseParser(params),
|
||||
"vk.com": BaseParser(params),
|
||||
"ok.ru": BaseParser(params),
|
||||
"likee.video": BaseParser(params),
|
||||
"dzen.ru": BaseParser(params),
|
||||
"yappy.media": YappyParser(params),
|
||||
}
|
||||
return parser_mapping[domain]
|
||||
except KeyError:
|
||||
raise SiteNotImplementedException
|
||||
|
||||
@staticmethod
|
||||
def video_processing_executor(video_params: dict):
|
||||
try:
|
||||
result = MasterService.video_download(video_params=video_params)
|
||||
return result
|
||||
return Result(result_type=ResultTypeEnum.DONE, value={
|
||||
"link": video_params["link"],
|
||||
"result": result,
|
||||
"status": "done"
|
||||
})
|
||||
except FileAlreadyExistException as ex:
|
||||
return Result(result_type=ResultTypeEnum.EXIST, value={
|
||||
"link": video_params["link"],
|
||||
"result": ex.message,
|
||||
"status": "exist"
|
||||
})
|
||||
except SiteNotImplementedException as ex:
|
||||
return Result(result_type=ResultTypeEnum.EXCEPTION, value=ex.default_message)
|
||||
|
||||
except Exception as ex:
|
||||
return Result(result_type=ResultTypeEnum.EXCEPTION, value=ex)
|
||||
return Result(result_type=ResultTypeEnum.EXCEPTION, value=traceback.format_exc())
|
||||
# TODO upload to server
|
||||
|
||||
|
||||
@ -95,5 +122,3 @@ def executor_initializer():
|
||||
import setproctitle
|
||||
setproctitle.setproctitle(f'video_downloader_executor_process')
|
||||
return True
|
||||
|
||||
|
||||
|
||||
@ -2,7 +2,7 @@ import asyncio
|
||||
import json
|
||||
from functools import partial
|
||||
|
||||
from aio_pika import connect
|
||||
from aio_pika import connect, Message, DeliveryMode
|
||||
from aio_pika.abc import AbstractIncomingMessage
|
||||
|
||||
|
||||
@ -26,3 +26,24 @@ async def get_messages(inner_queue) -> None:
|
||||
|
||||
print(" [*] Waiting for messages. To exit press CTRL+C")
|
||||
await asyncio.Future()
|
||||
|
||||
|
||||
async def publish_message_with_task_done(task: dict) -> None:
|
||||
queue_name = "tasks_done"
|
||||
async with await connect("amqp://guest:guest@localhost/") as connection:
|
||||
# Creating channel
|
||||
channel = await connection.channel()
|
||||
|
||||
# Will take no more than 10 messages in advance
|
||||
await channel.set_qos(prefetch_count=1)
|
||||
|
||||
# Declaring queue
|
||||
queue = await channel.declare_queue(queue_name)
|
||||
message = Message(
|
||||
json.dumps(task, indent=4).encode('utf-8'), delivery_mode=DeliveryMode.PERSISTENT,
|
||||
)
|
||||
await channel.default_exchange.publish(
|
||||
message,
|
||||
routing_key=queue_name,
|
||||
)
|
||||
|
||||
|
||||
@ -1,4 +1,6 @@
|
||||
import types
|
||||
from enum import Enum
|
||||
from inspect import Traceback
|
||||
|
||||
|
||||
class ResultTypeEnum(Enum):
|
||||
@ -8,9 +10,9 @@ class ResultTypeEnum(Enum):
|
||||
|
||||
|
||||
class Result:
|
||||
def __init__(self, result_type: ResultTypeEnum, value: Exception | bool = None):
|
||||
def __init__(self, result_type: ResultTypeEnum, value: str | dict = None):
|
||||
self.result_type = result_type
|
||||
self.value = value
|
||||
|
||||
def __repr__(self):
|
||||
return f'Result: {self.result_type.value}. Traceback: {self.value if self.value else None}'
|
||||
return f'Result: {self.result_type.value}. Value: {self.value if self.value else None}'
|
||||
|
||||
Reference in New Issue
Block a user