mirror of
https://github.com/otter18/tg_logger.git
synced 2025-08-04 21:32:56 +02:00
v3.1
This commit is contained in:
@@ -6,6 +6,8 @@
|
||||
|
||||
Bridging python logging and files to tg bot
|
||||
|
||||
Documentation is available at [Read the Docs](https://tg-logger.readthedocs.io/)
|
||||
|
||||
## Table of Contents
|
||||
- [Installation & Usage](#installation--usage)
|
||||
- [Screenshot](#screenshot)
|
||||
|
2
setup.py
2
setup.py
@@ -2,7 +2,7 @@ from setuptools import setup, find_packages
|
||||
|
||||
setup(
|
||||
name="tg_logger",
|
||||
version="3.0",
|
||||
version="3.1",
|
||||
description="A tool to bridge python logging and user files to telegram bot",
|
||||
long_description=open("README.md").read(),
|
||||
long_description_content_type="text/markdown",
|
||||
|
@@ -1,17 +1,13 @@
|
||||
# Copyright (c) ChernV (@otter18), 2021.
|
||||
|
||||
from .files import *
|
||||
from .handler import *
|
||||
|
||||
import logging
|
||||
from logging import StreamHandler
|
||||
from time import time, sleep
|
||||
from typing import List
|
||||
|
||||
import telebot
|
||||
|
||||
# logging setup
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def setup(base_logger: logging.Logger,
|
||||
def setup(base_logger: logging.Logger = logging.getLogger(),
|
||||
token: str = '',
|
||||
users: List[int] = [],
|
||||
timeout: int = 10,
|
||||
@@ -38,76 +34,3 @@ def setup(base_logger: logging.Logger,
|
||||
)
|
||||
tg_handler.setFormatter(formatter)
|
||||
base_logger.addHandler(tg_handler)
|
||||
|
||||
|
||||
class TgLoggerHandler(StreamHandler):
|
||||
"""Logger handler for tg_logger"""
|
||||
|
||||
def __init__(self, token: str, users: List[int], timeout: int = 10):
|
||||
"""
|
||||
Setup TgLoggerHandler class
|
||||
|
||||
:param token: tg bot token to log form
|
||||
:param users: list of used_id to log to
|
||||
:param timeout: seconds for retrying to send log if error occupied
|
||||
"""
|
||||
|
||||
super().__init__()
|
||||
self.token = token
|
||||
self.users = users
|
||||
self.timeout = timeout
|
||||
|
||||
self.bot = telebot.TeleBot(token=self.token)
|
||||
|
||||
def emit(self, record):
|
||||
msg = self.format(record)
|
||||
for user_id in self.users:
|
||||
t0 = time()
|
||||
while time() - t0 < self.timeout:
|
||||
try:
|
||||
self.bot.send_message(user_id, msg, parse_mode="HTML")
|
||||
break
|
||||
except Exception as ex:
|
||||
logger.exception("Exception while sending %s to %s:", msg, user_id)
|
||||
sleep(1)
|
||||
|
||||
|
||||
class TgFileLogger:
|
||||
"""tg_logger tool to send files"""
|
||||
|
||||
def __init__(self, token: str, users: List[int], timeout: int = 10):
|
||||
"""
|
||||
Setup TgFileLogger tool
|
||||
|
||||
:param token: tg bot token to log form
|
||||
:param users: list of used_id to log to
|
||||
:param timeout: seconds for retrying to send log if error occupied
|
||||
"""
|
||||
|
||||
self.token = token
|
||||
self.users = users
|
||||
self.timeout = timeout
|
||||
|
||||
self.bot = telebot.TeleBot(token=self.token)
|
||||
|
||||
def send(self, file_path: str, caption: str = ''):
|
||||
"""
|
||||
Function to send file
|
||||
|
||||
:param file_path: file path to log
|
||||
:param caption: text to file with file
|
||||
|
||||
:return: None
|
||||
"""
|
||||
|
||||
with open(file_path, 'rb') as data:
|
||||
for user_id in self.users:
|
||||
t0 = time()
|
||||
while time() - t0 < self.timeout:
|
||||
try:
|
||||
self.bot.send_document(user_id, data=data, caption=caption, parse_mode="HTML")
|
||||
logger.info("File %s successfully send to %s", file_path, user_id)
|
||||
break
|
||||
except Exception as ex:
|
||||
logger.exception("Exception while sending %s to %s:", file_path, user_id)
|
||||
sleep(1)
|
||||
|
51
tg_logger/files.py
Normal file
51
tg_logger/files.py
Normal file
@@ -0,0 +1,51 @@
|
||||
# Copyright (c) ChernV (@otter18), 2021.
|
||||
|
||||
import logging
|
||||
from time import time, sleep
|
||||
from typing import List
|
||||
|
||||
import telebot
|
||||
|
||||
# logging setup
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class TgFileLogger:
|
||||
"""tg_logger tool to send files"""
|
||||
|
||||
def __init__(self, token: str, users: List[int], timeout: int = 10):
|
||||
"""
|
||||
Setup TgFileLogger tool
|
||||
|
||||
:param token: tg bot token to log form
|
||||
:param users: list of used_id to log to
|
||||
:param timeout: seconds for retrying to send log if error occupied
|
||||
"""
|
||||
|
||||
self.token = token
|
||||
self.users = users
|
||||
self.timeout = timeout
|
||||
|
||||
self.bot = telebot.TeleBot(token=self.token)
|
||||
|
||||
def send(self, file_path: str, caption: str = ''):
|
||||
"""
|
||||
Function to send file
|
||||
|
||||
:param file_path: file path to log
|
||||
:param caption: text to file with file
|
||||
|
||||
:return: None
|
||||
"""
|
||||
|
||||
with open(file_path, 'rb') as data:
|
||||
for user_id in self.users:
|
||||
t0 = time()
|
||||
while time() - t0 < self.timeout:
|
||||
try:
|
||||
self.bot.send_document(user_id, data=data, caption=caption, parse_mode="HTML")
|
||||
logger.info("File %s successfully send to %s", file_path, user_id)
|
||||
break
|
||||
except Exception:
|
||||
logger.exception("Exception while sending %s to %s:", file_path, user_id)
|
||||
sleep(1)
|
43
tg_logger/handler.py
Normal file
43
tg_logger/handler.py
Normal file
@@ -0,0 +1,43 @@
|
||||
# Copyright (c) ChernV (@otter18), 2021.
|
||||
|
||||
import logging
|
||||
from logging import StreamHandler
|
||||
from time import time, sleep
|
||||
from typing import List
|
||||
|
||||
import telebot
|
||||
|
||||
# logging setup
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class TgLoggerHandler(StreamHandler):
|
||||
"""Logger handler for tg_logger"""
|
||||
|
||||
def __init__(self, token: str, users: List[int], timeout: int = 10):
|
||||
"""
|
||||
Setup TgLoggerHandler class
|
||||
|
||||
:param token: tg bot token to log form
|
||||
:param users: list of used_id to log to
|
||||
:param timeout: seconds for retrying to send log if error occupied
|
||||
"""
|
||||
|
||||
super().__init__()
|
||||
self.token = token
|
||||
self.users = users
|
||||
self.timeout = timeout
|
||||
|
||||
self.bot = telebot.TeleBot(token=self.token)
|
||||
|
||||
def emit(self, record):
|
||||
msg = self.format(record)
|
||||
for user_id in self.users:
|
||||
t0 = time()
|
||||
while time() - t0 < self.timeout:
|
||||
try:
|
||||
self.bot.send_message(user_id, msg, parse_mode="HTML")
|
||||
break
|
||||
except Exception as ex:
|
||||
logger.exception("Exception while sending %s to %s:", msg, user_id)
|
||||
sleep(1)
|
Reference in New Issue
Block a user