LongArticlesJob/applications/functions/functions.py

"""
@author: luojunhui
"""
import threading
import hashlib

from datetime import datetime, timezone


class Functions(object):
    """
    functions class
    """

    @classmethod
    def show_desc_to_sta(cls, show_desc):
        """

        :return:
        """

        def decode_show_v(show_v):
            """

            :param show_v:
            :return:
            """
            foo = show_v.replace('千', 'e3').replace('万', 'e4').replace('亿', 'e8')
            foo = eval(foo)
            return int(foo)

        def decode_show_k(show_k):
            """

            :param show_k:
            :return:
            """
            this_dict = {
                '阅读': 'show_view_count',  # 文章
                '看过': 'show_view_count',  # 图文
                '观看': 'show_view_count',  # 视频
                '赞': 'show_like_count',
                '付费': 'show_pay_count',
                '赞赏': 'show_zs_count',
            }
            if show_k not in this_dict:
                print(f'error from decode_show_k, show_k not found: {show_k}')
            return this_dict.get(show_k, 'show_unknown')

        show_desc = show_desc.replace('+', '')
        sta = {}
        for show_kv in show_desc.split('\u2004\u2005'):
            if not show_kv:
                continue
            show_k, show_v = show_kv.split('\u2006')
            k = decode_show_k(show_k)
            v = decode_show_v(show_v)
            sta[k] = v
        res = {
            'show_view_count': sta.get('show_view_count', 0),
            'show_like_count': sta.get('show_like_count', 0),
            'show_pay_count': sta.get('show_pay_count', 0),
            'show_zs_count': sta.get('show_zs_count', 0),
        }
        return res

    @classmethod
    def generateGzhId(cls, url):
        """
        generate url
        :param url:
        :return:
        """
        biz = url.split("biz=")[1].split("&")[0]
        idx = url.split("&idx=")[1].split("&")[0]
        sn = url.split("&sn=")[1].split("&")[0]
        url_bit = "{}-{}-{}".format(biz, idx, sn).encode()
        md5_hash = hashlib.md5()
        md5_hash.update(url_bit)
        md5_value = md5_hash.hexdigest()
        return md5_value

    @classmethod
    def job_with_thread(cls, job_func):
        """
        每个任务放到单个线程中
        :param job_func:
        :return:
        """
        job_thread = threading.Thread(target=job_func)
        job_thread.start()

    @classmethod
    def str_to_md5(cls, strings):
        """
        字符串转化为 md5 值
        :param strings:
        :return:
        """
        # 将字符串转换为字节
        original_bytes = strings.encode('utf-8')
        # 创建一个md5 hash对象
        md5_hash = hashlib.md5()
        # 更新hash对象，传入原始字节
        md5_hash.update(original_bytes)
        # 获取16进制形式的MD5哈希值
        md5_value = md5_hash.hexdigest()
        return md5_value

    @classmethod
    def float_to_percentage(cls, value, decimals=3) -> str:
        """
        把小数转化为百分数
        :param value:
        :param decimals:
        :return:
        """
        percentage_value = round(value * 100, decimals)
        return "{}%".format(percentage_value)

    @classmethod
    def str_to_timestamp(cls, date_string, string_format='%Y-%m-%d') -> int:
        """
        :param string_format:
        :param date_string:
        :return:
        """
        date_obj = datetime.strptime(date_string, string_format)
        timestamp = date_obj.timestamp()
        return int(timestamp)

    @classmethod
    def timestamp_to_str(cls, timestamp, string_format='%Y-%m-%d %H:%M:%S') -> str:
        """
        :param string_format:
        :param timestamp:
        """
        dt_object = datetime.utcfromtimestamp(timestamp).replace(tzinfo=timezone.utc).astimezone()
        date_string = dt_object.strftime(string_format)
        return date_string