MatchRate/functions.py

"""
@author: luojunhui
"""

import pymysql
from datetime import datetime, timedelta


class MatchRate(object):
    """
    匹配率
    """

    @classmethod
    def generate_stamp_list(cls, start_date, end_date):
        """
        Generate daily date_str
        :param start_date:
        :param end_date:
        :return:
        """
        start = datetime.strptime(start_date, "%Y%m%d")
        end = datetime.strptime(end_date, "%Y%m%d")
        current = start
        timestamp_list = []
        while current <= end:
            timestamp_list.append(current.timestamp() * 1000)
            current += timedelta(days=1)
        return timestamp_list

    @classmethod
    def generate_today_stamp(cls, date_string):
        """

        :param date_string:
        :return:
        """
        return datetime.strptime(date_string, "%Y%m%d").timestamp() * 1000

    @classmethod
    def generate_yesterday_stamp(cls, now_dt):
        """
        Generate date in 3 days
        :param now_dt:
        :return:
        """
        now_date = datetime.strptime(now_dt, "%Y%m%d")
        yesterday = now_date - timedelta(days=1)
        return yesterday.timestamp() * 1000

    @classmethod
    def match_rate(cls, start_time_stamp, end_time_stamp):
        """
        sensitive words
        :return:
        """
        connection = pymysql.connect(
            host="rm-t4na9qj85v7790tf84o.mysql.singapore.rds.aliyuncs.com",  # 数据库IP地址，内网地址
            port=3306,  # 端口号
            user="crawler_readonly",  # mysql用户名
            passwd="cyber#crawler_2023",  # mysql用户登录密码
            db="aigc-admin-prod",  # 数据库名
            charset="utf8mb4"  # 如果数据库里面的文本是utf8编码的，charset指定是utf8
        )
        sql = f"""
            select status, trace_id, error_msg
            from publish_content_miniprogram 
            where create_timestamp >= {start_time_stamp} and create_timestamp < {end_time_stamp};
            """
        cursor = connection.cursor()
        cursor.execute(sql)
        data = cursor.fetchall()
        result = [list(line) for line in data]
        return result


class RateDetail(object):
    """
    Rate Detail
    """
    @classmethod
    def rate_and_error_list(cls, result_list):
        """
        成功，失败的数据
        :param result_list:
        :return:
        """
        success_count = 0
        fail_count = 0
        processing_count = 0
        total_requests = len(result_list)
        error_list = []
        if result_list:
            for temp in result_list:
                status = temp[0]
                error = temp[2]
                if status == 1:
                    processing_count += 1
                elif status == 2:
                    success_count += 1
                elif status == 3:
                    fail_count += 1
                else:
                    continue
                if error:
                    error_list.append(temp)
            obj = {
                "success_count": success_count,
                "fail_count": fail_count,
                "processing_count": processing_count,
                "total_count": total_requests,
                "error_list": error_list
                }
        else:
            obj = {
                "success_count": None,
                "fail_count": None,
                "processing_count": None,
                "total_count": None,
                "error_list": []
                }
        return obj