LongArticleAlgServer/routes/AccountArticleRank.py

"""
@author: luojunhui
"""
from applications.functions import ArticleRank
from applications.functions import title_sim_v2_by_list
from applications.functions import get_article_title_url_list


def has_same(title, account_nickname):
    """
    判断是否存储
    :param title:
    :param account_nickname:
    :return:
    """
    account_title_list = get_article_title_url_list(
        account_nickname,
        # max_time='20240603'
    )
    sim_res = title_sim_v2_by_list(title, account_title_list)
    if sim_res:
        return True
    return False

class AccountArticleRank(object):
    """
    文章排序
    """

    def __init__(self, params, mysql_client):
        """
        :param params: 请求参数
        :param mysql_client: 数据库链接池
        """
        self.publishArticleList = None
        self.publishNum = None
        self.strategy = None
        self.ghId = None
        self.accountName = None
        self.accountId = None
        self.params = params
        self.mysql_client = mysql_client

    async def check_params(self):
        """
        校验参数
        :return:
        """
        try:
            self.accountId = self.params['accountId']
            self.accountName = self.params['accountName']
            self.ghId = self.params['ghId']
            self.strategy = self.params['strategy']
            self.publishNum = self.params['publishNum']
            self.publishArticleList = self.params['publishArticleList']
            self.title_list = [i['title'] for i in self.publishArticleList]
            # try:
            self.title_list = [i for i in self.title_list if not has_same(i, self.accountName)]
            # except:
            #     self.title_list = self.title_list
            self.content_list = [i['content'] for i in self.publishArticleList]
            return None
        except Exception as e:
            response = {
                "msg": "params error",
                "info": "params check failed, params : {} is not correct".format(e),
                "code": 0
            }
            return response

    async def rank_v1(self):
        """
        Rank Version 1
        :return:
        """
        try:
            rank_info = ArticleRank().rank(account_list=[self.accountName], text_list=self.title_list)
            score_list = rank_info[self.accountName]['score_list']
            title_score_dict = {}
            for index, item in enumerate(self.title_list):
                title_score_dict[item] = score_list[index]
            result_list = []
            for obj in self.publishArticleList:
                if title_score_dict.get(obj['title']):
                    obj['score'] = title_score_dict[obj['title']]
                    result_list.append(obj)
            sorted_list = sorted(result_list, key=lambda x: x['score'], reverse=True)
            result = {
                "accountId": self.accountId,
                "accountName": self.accountName,
                "ghId": self.ghId,
                "strategy": self.strategy,
                "publishNum": self.publishNum,
                "rank_list": sorted_list[:self.publishNum]
            }
        except Exception as e:
            print("error", e)
            result = {
                "accountId": self.accountId,
                "accountName": self.accountName,
                "ghId": self.ghId,
                "strategy": self.strategy,
                "publishNum": self.publishNum,
                "rank_list": self.publishArticleList[:self.publishNum]
            }
        response = {
            "status": "Rank Success",
            "data": result
        }
        return response

    async def rank_v2(self):
        """
        Rank Version 2
        :return:
        """
        return await self.rank_v1()

    async def rank_v3(self):
        """
        Rank Version 3
        :return:
        """
        return await self.rank_v1()

    async def rank_v4(self):
        """
        Rank Version 4
        :return:
        """
        return await self.rank_v1()

    async def rank_v5(self):
        """
        Rank Version 5
        :return:
        """
        return await self.rank_v1()

    async def choose_strategy(self):
        """
        选择排序策略
        :return:
        """
        match self.strategy:
            case "ArticleRankV1":
                return await self.rank_v1()
            case "ArticleRankV2":
                return await self.rank_v2()
            case "ArticleRankV3":
                return await self.rank_v3()
            case "ArticleRankV4":
                return await self.rank_v4()
            case "ArticleRankV5":
                return await self.rank_v5()

    async def deal(self):
        """
        Deal Function
        :return:
        """
        error_params = await self.check_params()
        if error_params:
            return error_params
        else:
            return await self.choose_strategy()