v1.3 版本大升级增加py3 新特性工具类

Signed-off-by: Eric <[email protected]>

README.md

@@ -1,6 +1,6 @@
 # Tools
 [![License](https://img.shields.io/:license-apache-blue.svg)](https://opensource.org/licenses/Apache-2.0)
-![latest 1.2.6](https://img.shields.io/badge/latest-1.2.6-green.svg?style=flat)
+![latest 1.3](https://img.shields.io/badge/latest-1.3-green.svg?style=flat)
 ![GitHub commits since latest release](https://img.shields.io/github/commits-since/eric-jxl/Tools/latest)
 
 
@@ -12,32 +12,40 @@
 pip install eric_tools
 ```
 
-```
+```markdown
 encryption_classmethod.py  Python HMAC+MD5加密签名算法
 
 exception_class.py  异常类
 
-resize_image.py 图片压缩
+resize_image.py     图片压缩
+
+ip.py               ip地址定位API
+
+logger.py           日志模块类和高级日志装饰器
+
+remove.py           删除文件
 
-ip.py           ip地址定位API
+send_email.py       发送邮件
 
-logger.py       日志模块类和高级日志装饰器
+sftp.py             ssh远程下载文件
 
-remove.py       删除文件
+pgsql.py            对postgresql 增删改查操作
 
-send_email.py   发送邮件
+readconfig           针对读取配置文件
 
-sftp.py         ssh远程下载文件
+jwt_encrypt         生成jwt Access Token 加密及解密
 
-pgsql.py        对postgresql 增删改查操作
+convert_json        支持json和object之间转换
 
-readconfig      针对读取配置文件
+Abstract.py         抽象类模型
 
-jwt_encrypt     生成jwt Access Token 加密及解密
+decorator.py        惰性属性装饰器
 
-convert_json    支持json和object之间转换
+**新增内容**
+async_queue.py        异步队列操作
+downloader.py         下载器
+logMixIn.py           日志元类和高级日志装饰器
+nginx_log.py          nginx日志解析(默认access.log)
 
-Abstract.py     抽象类模型
 
-decorator.py    惰性属性装饰器
 ```

__init__.py

@@ -1,4 +1,4 @@
-#-*- coding=utf-8 -*-
-name='Eric-Tools'
+# -*- coding=utf-8 -*-
+name = 'Eric-Tools'
 License = 'Apache'
-
+Version = '1.3'

eric_tools/__init__.py

@@ -21,7 +21,7 @@
 name = 'Eric-Tools'
 __title__ = 'tools'
 __description__ = 'Python HTTP for Humans.'
-__version__ = "1.2.6"
+__version__ = "1.3"
 __author__ = 'Eric'
 __doc__ = ["Python Daily Development Tools"]
 __url__ = "https://github.com/Eric-jxl/Tools"

eric_tools/async_queue.py

@@ -0,0 +1,45 @@
+import asyncio
+
+
+class AsyncMessageQueue:
+    def __init__(self, num_workers=3):
+        self.queue = asyncio.Queue()
+        self.num_workers = num_workers
+
+    async def process_message(self, message):
+        raise NotImplementedError
+
+    async def worker(self):
+        while True:
+            message = await self.queue.get()
+            await self.process_message(message)
+            self.queue.task_done()
+
+    async def start(self):
+        tasks = [asyncio.create_task(self.worker())
+                 for _ in range(self.num_workers)]
+
+        # 等待所有任务完成
+        await self.queue.join()
+        for task in tasks:
+            task.cancel()
+
+
+class EmailMessageQueue(AsyncMessageQueue):
+    async def process_message(self, message):
+        # 模拟发送邮件的耗时操作
+        await asyncio.sleep(2)
+        print(f"发送邮件: {message}")
+
+
+async def main():
+    email_queue = EmailMessageQueue()
+
+    # 向队列中添加消息
+    for i in range(5):
+        await email_queue.queue.put(f"邮件{i}")
+
+    await email_queue.start()
+
+if __name__ == "__main__":
+    asyncio.run(main())

eric_tools/downloader.py

@@ -0,0 +1,57 @@
+import asyncio
+import aiohttp
+import os
+import time
+from collections import deque
+
+class Downloader:
+    def __init__(self, urls, max_concurrent=5, delay=0):
+        self.urls = deque(urls)
+        self.max_concurrent = max_concurrent
+        self.delay = delay
+        self.queue = asyncio.Queue()
+        self.session = aiohttp.ClientSession()
+
+    async def download(self, url):
+        async with self.session.get(url) as response:
+            filename = os.path.basename(url)
+            with open(filename, 'wb') as f:
+                while True:
+                    chunk = await response.content.read(1024)
+                    if not chunk:
+                        break
+                    f.write(chunk)
+        print(f"{url} downloaded")
+
+    async def worker(self):
+        while True:
+            url = await self.queue.get()
+            try:
+                if self.delay:
+                    time.sleep(self.delay)
+                await self.download(url)
+            except Exception as e:
+                print(f"Error downloading {url}: {e}")
+            self.queue.task_done()
+
+    async def run(self):
+        for url in self.urls:
+            self.queue.put_nowait(url)
+
+        tasks = []
+        for _ in range(self.max_concurrent):
+            task = asyncio.create_task(self.worker())
+            tasks.append(task)
+
+        await self.queue.join()
+
+        for task in tasks:
+            task.cancel()
+
+        await asyncio.gather(*tasks, return_exceptions=True)
+        await self.session.close()
+
+if __name__ == "__main__":
+    urls = ["https://example.com/file1.txt", "https://example.com/file2.txt", "https://example.com/file3.txt"]
+    downloader = Downloader(urls, max_concurrent=2, delay=1)
+    asyncio.run(downloader.run())

eric_tools/excel_generator.py

@@ -0,0 +1,49 @@
+import xlsxwriter
+
+
+class ExcelGenerator:
+    def __init__(self, file_name='output.xlsx'):
+        self.workbook = xlsxwriter.Workbook(file_name)
+        self.current_sheet = None
+
+    def create_sheet(self, sheet_name='Sheet1'):
+        self.current_sheet = self.workbook.add_worksheet(sheet_name)
+
+    def write_data(self, row, col, data):
+        if self.current_sheet:
+            self.current_sheet.write(row, col, data)
+        else:
+            print("No sheet selected. Create a sheet first.")
+
+    def generate_from_data_structure(self, data_structure):
+        if not self.current_sheet:
+            self.create_sheet()
+
+        for row_index, row_data in enumerate(data_structure):
+            for col_index, cell_data in enumerate(row_data):
+                self.write_data(row_index, col_index, cell_data)
+
+    def save_file(self):
+        if self.workbook:
+            self.workbook.close()
+        else:
+            print("No workbook to save.")
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, exc_type, exc_value, exc_traceback):
+        self.save_file()
+
+
+if __name__ == '__main__':
+    data_structure = [
+        ['Name', 'Age', 'City'],
+        ['John', 25, 'New York'],
+        ['Alice', 30, 'San Francisco'],
+    ]
+
+    with ExcelGenerator() as f:
+        f.create_sheet("Mysheet")
+        # TODO:Generate Excel from data structure
+        f.generate_from_data_structure(data_structure)

eric_tools/logMixin.py

@@ -0,0 +1,87 @@
+class Meta(type):
+    """自定义元类，用于动态添加Mixin"""
+
+    def __new__(meta, name, bases, dct):
+        if "use_logging" in dct and dct["use_logging"]:
+            dct.update(LogMixin.__dict__)  # 动态添加LogMixin的属性和方法
+        return super().__new__(meta, name, bases, dct)
+
+
+class LogMixin:
+    def log(self, message):
+        print(f"{self.__class__.__name__}: {message}")
+
+
+class MyClass(metaclass=Meta):
+    use_logging = True  # 设置标志位，指示是否使用日志Mixin
+
+    def do_something(self):
+        self.log("Doing something...")
+
+
+class StrategyMeta(type):
+    """更智能的元类，根据类属性动态选择Mixin"""
+
+    def __new__(meta, name, bases, dct):
+        mixins_to_apply = []
+        if "security_required" in dct and dct["security_required"]:
+            mixins_to_apply.append(SecurityMixin)
+        if "cache_enabled" in dct and dct["cache_enabled"]:
+            mixins_to_apply.extend([CacheMixin, CacheControlMixin])
+
+        for mixin in mixins_to_apply:
+            dct.update(mixin.__dict__)
+
+        return super().__new__(meta, name, bases, dct)
+
+# 假设SecurityMixin, CacheMixin, CacheControlMixin已定义
+
+
+class MySecureClass(metaclass=StrategyMeta):
+    security_required = True
+    cache_enabled = True
+
+    def perform_operation(self):
+        self.secure_access()
+        self.cache_data()
+        self.control_cache()
+
+
+_registry = []
+
+
+class RegisterMixinMeta(type):
+    def __new__(cls, name, bases, dct):
+        new_class = super().__new__(cls, name, bases, dct)
+        if 'register_me' in dct and dct['register_me']:
+            _registry.append(new_class)
+        return new_class
+
+
+class IPlugin(metaclass=RegisterMixinMeta):
+    """接口类 ，定义了插件应实现的方法"""
+
+    def plugin_action(self):
+        raise NotImplementedError("Subclasses must implement plugin_action.")
+
+
+class PluginA(IPlugin):
+    register_me = True
+
+    def plugin_action(self):
+        print("Plugin A is active.")
+
+
+class PluginB(IPlugin):
+    def plugin_action(self):
+        print("Plugin B is active but not registered explicitly.")
+
+
+# 使用示例
+for plugin_class in _registry:
+    plugin_class().plugin_action()
+
+
+if __name__ == "__main__":
+    my_instance = MyClass()
+    my_instance.do_something()

eric_tools/nginx_log.py

@@ -0,0 +1,44 @@
+import sys
+import os
+import datetime
+
+# 定义一个函数来获取用户的输入
+
+
+def get_user_input(prompt):
+    return input(prompt)
+
+
+# 获取当前日期
+today = datetime.date.today()
+
+# 获取过去三天的日期
+three_days_ago = today - datetime.timedelta(days=3)
+
+# 获取要切割的日志文件路径
+log_file_path = '/var/log/nginx/access.log'
+
+# 获取要匹配的IP地址
+ip_address = get_user_input("Enter the IP address to match: ")
+
+# 创建一个新文件来存储切割后的日志
+output_file_path = 'nginx_access_log_cut.txt'
+with open(output_file_path, 'w') as output_file:
+    # 逐行读取日志文件
+    with open(log_file_path, 'r') as log_file:
+        for line in log_file:
+            # 分割日志行
+            parts = line.split()
+
+            # 获取日志日期
+            log_date = parts[3][1:]
+
+            # 检查日志日期是否在过去三天内
+            if log_date >= three_days_ago.strftime('%d/%b/%Y'):
+                # 检查日志行是否包含匹配的IP地址
+                if ip_address in line:
+                    # 将日志行写入输出文件
+                    output_file.write(line)
+
+# 打印切割后的日志文件路径
+print(f"Cut nginx access log saved to: {output_file_path}")

setup.py

@@ -7,7 +7,7 @@
 
 setuptools.setup(
     name="eric_tools",
-    version="1.2.6",
+    version="1.3",
     author="Eric",
     author_email="[email protected]",
     description="Python Daily Development Tools",
@@ -27,5 +27,6 @@
         'requests',
         'paramiko',
         'Pillow',
+        'xlsxwriter'
     ]
 )