上传文件至 Log4p

Log4p/DecoratorsTools.py

@@ -0,0 +1,11 @@
+import warnings
+
+def deprecated(use_func, d_version):
+   def decorator(func):
+        async def wrapper(*args, **kwargs):
+            warnings.warn(f"The '{func.__name__}' function is deprecated, "
+                          f"use '{use_func}' instead. Deprecated in version {d_version}.",
+                          DeprecationWarning, stacklevel=2)
+            return await func(*args, **kwargs)
+        return wrapper
+   return decorator

Log4p/HttpHander.py

@@ -0,0 +1,35 @@
+import logging
+import requests
+import httpx
+
+class HTTPhandler(logging.Handler):
+    def __init__(self, url):
+        super().__init__()
+        self.url = url
+
+    def emit(self, record):
+        log_entry = self.format(record)
+        payload = {'log': log_entry}
+        try:
+            response = requests.post(self.url, json=payload)
+            if not response.ok:
+                raise ValueError(response.text)
+        except Exception as e:
+            logging.error("Failed to send log to %s: %s", self.url, e)
+
+class AsyncHTTPhandler(logging.Handler):
+    def __init__(self, url):
+        super().__init__()
+        self.url = url
+
+    async def emit(self, record):
+        log_entry = self.format(record)
+        payload = {'log': log_entry}
+        try:
+            async with httpx.AsyncClient(timeout=120,max_redirects=5) as client:
+                response = await client.post(self.url, json=payload)
+                if not response.is_success:
+                    raise ValueError(await response.text())
+        except Exception as e:
+            logging.error("Failed to send log to %s: %s", self.url, e)
+

Log4p/websocketHander.py

@@ -0,0 +1,28 @@
+import logging
+import asyncio
+import websockets
+
+class WebsocketHandler(logging.Handler):
+    def __init__(self, server_address):
+        super().__init__()
+        self.server_address = server_address
+    
+    async def send_log_async(self, message):
+        async with websockets.connect(self.server_address) as websocket:
+            await websocket.send(message)
+    
+    def send_log_sync(self, message):
+        loop = asyncio.new_event_loop()
+        asyncio.set_event_loop(loop)
+        loop.run_until_complete(self.send_log_async(message))
+
+    def emit(self, record):
+        log_entry = self.format(record)
+        
+        if asyncio.get_event_loop().is_running():
+            asyncio.create_task(self.send_log_async(log_entry))
+        else:
+            try:
+                self.send_log_sync(log_entry)
+            except Exception as e:
+                logging.warning("Failed to send log synchronously: %s", e)