装饰器模块

文件名： func_decorators.py

# -*- coding: UTF-8 -*-
'''
装饰函数集合
'''

import functools
import time
import logging

# 装饰函数 - 时间装饰函数
def timer(func):
    def wrapper(*args, **kwargs):
        # start the timer
        start_time = time.time()
        # call the decorated function
        result = func(*args, **kwargs)
        # remeasure the time
        end_time = time.time()
        # compute the elapsed time and print it
        execution_time = end_time - start_time
        print(f"Execution time: {execution_time} seconds")
        print(f"本模块消耗时间: {execution_time} 秒")
        m, s = divmod(execution_time, 60)
        h, m = divmod(m, 60)
        time_string = f'{h}小时,{m}分{s}秒'
        print(f'项目累计耗时： {time_string}')
        # return the result of the decorated function execution
        return result
    # return reference to the wrapper function
    return wrapper

# 使用装饰器 @retry(exceptions=(Exception,), tries=5, initial_delay=3)
# 设置tries=15，需要 15 次尝试才会达到或超过1天的等待时间。
def retry(exceptions, tries, initial_delay):
    """
    创建一个装饰器，用于在函数执行失败时进行重试。
    参数:
        exceptions (tuple): 一个异常类型元组，指定在哪些异常发生时进行重试。
        tries (int): 最大重试次数，包括最初的一次尝试。
        initial_delay (float): 重试前的初始等待时间（秒）。
    返回:
        decorator: 一个装饰器，用于包装需要重试逻辑的函数。
    """

    # 配置日志记录
    logging.basicConfig(level=logging.INFO)
    logger = logging.getLogger(__name__)

    def decorator(func):
        """
        装饰器函数，用于包装需要重试逻辑的函数。
        参数:
            func (function): 需要被包装的函数。
        返回:
            function: 包装后的函数，包含重试逻辑。
        """
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            """
            包装函数，用于实现重试逻辑。
            参数:
                *args: 被装饰函数的位置参数。
                **kwargs: 被装饰函数的关键字参数。
            返回:
                result: 被装饰函数的返回值。
            """
            delay = initial_delay  # 设置初始延迟时间
            for i in range(tries):
                try:
                    # 尝试执行被装饰的函数
                    return func(*args, **kwargs)
                except exceptions as e:
                    # 如果捕获到指定的异常，记录错误并等待后重试
                    timestamp = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
                    logger.error(f"{timestamp} - {func.__name__} failed: {e}, retrying in {delay} seconds.")
                    time.sleep(delay)
                    delay *= initial_delay  # 指数级增长等待时间

            # 如果达到最大重试次数仍然失败，抛出异常
            raise Exception(f"{func.__name__} has reached the maximum retry limit.")

        return wrapper

    return decorator

重试模块

普通重试

import time
import csv
import os

# 装饰函数 - 重试次数
def retry(max_attempts=1000, delay=1):
    def decorator(func):
        def wrapper(*args, **kwargs):
            attempts = 0
            while attempts < max_attempts:
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    attempts += 1
                    print(f'{time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())} 正在重试第 {attempts} 次；错误提示信息为: {e}')
                    # 确保日志目录存在
                    log_dir = './log'
                    if not os.path.exists(log_dir):
                        os.makedirs(log_dir)
                    with open(os.path.join(log_dir, 'errorlog.csv'), mode='a', encoding='utf-8', newline="") as f:
                        csv_write = csv.writer(f)
                        csv_write.writerow([time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()), func.__name__, str(e), args])
                    time.sleep(delay)
            # 重试失败后抛出异常
            raise Exception(f"Function '{func.__name__}' failed after {max_attempts} attempts")
        return wrapper
    return decorator

'''
为了调用该函数，我们可以指定最大的尝试次数和每次调用函数之间的时间长度（秒）。

@retry(max_attempts=3, delay=2)
def fetch_data(url):
    print("Fetching the data..")
    # raise timeout error to simulate a server not responding..
    raise TimeoutError("Server is not responding.")
fetch_data("https://example.com/data")  
# Retries 3 times with a 2-second delay between attempts
'''

指数级重试

import functools
import time
import logging

# 使用装饰器 @retry(exceptions=(Exception,), tries=5, initial_delay=3)
# 设置tries=15，需要 15 次尝试才会达到或超过1天的等待时间。
def retry(exceptions, tries, initial_delay):
    """
    创建一个装饰器，用于在函数执行失败时进行重试。
    参数:
        exceptions (tuple): 一个异常类型元组，指定在哪些异常发生时进行重试。
        tries (int): 最大重试次数，包括最初的一次尝试。
        initial_delay (float): 重试前的初始等待时间（秒）。
    返回:
        decorator: 一个装饰器，用于包装需要重试逻辑的函数。
    """

    # 配置日志记录
    logging.basicConfig(level=logging.INFO)
    logger = logging.getLogger(__name__)

    def decorator(func):
        """
        装饰器函数，用于包装需要重试逻辑的函数。
        参数:
            func (function): 需要被包装的函数。
        返回:
            function: 包装后的函数，包含重试逻辑。
        """
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            """
            包装函数，用于实现重试逻辑。
            参数:
                *args: 被装饰函数的位置参数。
                **kwargs: 被装饰函数的关键字参数。
            返回:
                result: 被装饰函数的返回值。
            """
            delay = initial_delay  # 设置初始延迟时间
            for i in range(tries):
                try:
                    # 尝试执行被装饰的函数
                    return func(*args, **kwargs)
                except exceptions as e:
                    # 如果捕获到指定的异常，记录错误并等待后重试
                    timestamp = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
                    logger.error(f"{timestamp} - {func.__name__} failed: {e}, retrying in {delay} seconds.")
                    time.sleep(delay)
                    delay *= initial_delay  # 指数级增长等待时间

            # 如果达到最大重试次数仍然失败，抛出异常
            raise Exception(f"{func.__name__} has reached the maximum retry limit.")

        return wrapper

    return decorator

打印输出

文件名：print_coloredtext.py

import time
class ColoredText:
    def __init__(self, text):
        self.text = text

    def red(self):
        return f"\033[91m{self.text}\033[0m"

    def yellow(self):
        return f"\033[93m{self.text}\033[0m"

    def blue(self):
        return f"\033[94m{self.text}\033[0m"

    def cyan(self):
        return f"\033[96m{self.text}\033[0m"

    def pink(self):
        return f"\033[95m{self.text}\033[0m"

    def purple(self):
        return f"\033[94m{self.text}\033[0m"

    def green(self):
        return f"\033[92m{self.text}\033[0m"

# 示例用法

# my_text = "This is colored text!"
# colored_text = ColoredText(my_text)
# print(ColoredText(my_text).blue(), 'haibucuo')
# print(ColoredText(my_text).cyan(), 'haibucuo')
# # print(colored_text.red())
# print(colored_text.yellow())
# str = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
# print(type(time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())))
# print(ColoredText(str).yellow())
# print(ColoredText(time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())).yellow())
# goods_title = 'Apple 1ea'
# print(ColoredText(time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())).yellow(), f'>> PL-商品名称: {goods_title}')

字符串处理

文件名：string_cleaner.py

import re


def string_to_float(string_raw):
    """
    尝试将字符串转换为整数，如果失败则返回0。
    参数:
        string_raw (str): 待转换的原始字符串。
    返回:
        int: 转换成功的整数，或在转换失败时返回0。
    """
    # 定义需要移除的字符模式，包括逗号和日元符号
    pattern = r"[,円]"

    # 检查string_raw是否为真值（非None且非空字符串）
    if string_raw:
        # 移除非数字字符并去除首尾空格
        s = re.sub(pattern, "", string_raw).strip()

        try:
            # 尝试转换为整数
            return float(s)
        except (ValueError, TypeError):
            # 如果转换失败，返回0
            return 0
    else:
        # 如果string_raw是None或空字符串，返回0
        return 0


if __name__ == "__main__":
    string_raw = "0.1ss"
    s = string_to_int(string_raw)
    print("float", s)
    print("int", int(s))