上传文件至 Log4p/plugins

This commit is contained in:
芙宁娜 2024-03-17 14:08:54 +00:00
parent 34bdabc4db
commit 4f88ab21ad
4 changed files with 74 additions and 0 deletions

View File

@ -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

View File

@ -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)

BIN
Log4p/plugins/__init__.py Normal file

Binary file not shown.

View File

@ -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)