VideoToArticles/deal/publish_deal.py

"""
@author: luojunhui
"""
from applications.upload import download_and_upload, auto_upload_aigc
from applications.functions import hash_title


class PublishDeal(object):
    """
    自动发布接口处理
    """

    def __init__(self, params):
        self.process_list = []
        self.taskName = None
        self.params = params

    def check_params(self):
        """
        check params
        """
        try:
            self.taskName = self.params['task_list']
            obj_list = self.params['article_list']
            self.process_list = [
                {
                    "title": obj['title'],
                    "text": obj['text'],
                    "img_list": obj['img_list'],
                    "video_id": obj['video_id'],
                    "cover": obj['cover']
                } for obj in obj_list
            ]
            return None
        except Exception as e:
            res = {
                "error": "params error",
                "info": "check your param: {}".format(e)
            }
            return res

    def deal(self):
        """
        处理请求
        """
        params_error = self.check_params()
        if params_error:
            return params_error
        else:
            L = []
            for article_obj in self.process_list:
                # hash title
                h_title = hash_title(article_obj['title'])
                # process imgList
                img_list = []
                for index, url in enumerate(article_obj['img_list'], 1):
                    save_path = "temp/{}-{}.png".format(h_title, index)
                    img_obj = download_and_upload(save_path, url)
                    img_list.append(img_obj)

                # process Cover
                cover_path = "temp/cover-{}.png".format(h_title)
                cover_obj = download_and_upload(cover_path, article_obj['cover'])
                result_obj = {
                    "title": article_obj['title'] + "video_id={}".format(article_obj['video_id']),
                    "text": article_obj['text'],
                    "img_list": img_list,
                    "cover": cover_obj
                }
                L.append(result_obj)
            res = auto_upload_aigc(task_name=self.taskName, obj_list=L)
            return res