Python实现日志实时监测的示例详解

#抽象订阅者
from abc import ABCMeta,abstractmethod
class Subscriber(metaclass=ABCMeta):
    #向具体订阅者发送消息的方法
    @abstractmethod
    def update(self):
        pass

#具体订阅者
#1、短信订阅者
class SMSSubscriber(Subscriber):
    def __init__(self,publisher):
        self.publisher = publisher
        self.publisher.attach(self)

    def update(self):
        print(type(self).__name__,self.publisher.getNews())


#2、邮件订阅者
class EmailSubscriber(Subscriber):
    def __init__(self, publisher):
        self.publisher = publisher
        self.publisher.attach(self)

    def update(self):
        print(type(self).__name__,self.publisher.getNews())
        info = self.publisher.getNews()
        # 发送邮件
        Sender_mail(info).sender_mail()

#3、日志订阅（文件存储）
class LoggerSubscriber(Subscriber):

    def __init__(self, publisher):
        log_dir = os.path.expanduser(r".\apps\Mapview\logs")
        log_file = os.path.join(log_dir, "file_{time}.log")
        logger.add(log_file, rotation="100KB", retention=2)
        self.publisher = publisher
        self.publisher.attach(self)

    def update(self):
        print(type(self).__name__,self.publisher.getNews())
        info=self.publisher.getNews()
        logger.info(f"{info}")

#创建发布者
class NewsPublisher:
    def __init__(self):
        self.__subscribers = []
        self.__latestNews = None
    
    #将订阅者添加到队列中
    def attach(self,subscriber):
        self.__subscribers.append(subscriber)
    
    #从订阅的主题里面移除
    def detach(self):
        return self.__subscribers.pop()

    #生成观察者列表
    def subscribers(self):
        return [type(x).__name__ for x in self.__subscribers]

    #发送通知给相关的主题订阅者
    def notifySubscribers(self):
        for sub in self.__subscribers:
            #update（）方法由具体的观察者或订阅者实现的
            sub.update()  #推送更新
    
    #创建新消息
    def addNews(self,news):
        self.__latestNews = news
    
    #返回最新消息，并通知观察者
    def getNews(self):
        return "Got News:",self.__latestNews

from apps.tools.subscription_model import NewsPublisher,LoggerSubscriber,EmailSubscriber
import functools


#如果加载失败，调用订阅者
def publisher(info):
    news_publisher = NewsPublisher()
    # for Subscribers in [EmailSubscriber, LoggerSubscriber]:
    for Subscribers in [LoggerSubscriber]:
        # eval(LoggerSubscriber)(news_publisher)
        Subscribers(news_publisher)
        print("\nSubscribers", news_publisher.subscribers())
        news_publisher.addNews(f"{info}")
        news_publisher.notifySubscribers()


#处理异常的装饰器
def fail_data(msg='地图加载失败'):
    def catch_exception(origin_func):
        @functools.wraps(origin_func)
        def wrapper(*args, **kwargs):
            try:
                u = origin_func(*args, **kwargs)
                print("这个函数正常执行：%s" % origin_func.__name__)
                return u

            except Exception as e:
                info = f"{msg}:{e.__doc__}"
                """
                接口调用失败处理机制，追加日志
                """
                print(info)
                publisher(info)
                # news_publisher = NewsPublisher()
                # LoggerSubscriber(news_publisher)
                # print("\nSubscribers", news_publisher.subscribers())
                # news_publisher.addNews(f"{info}")
                # news_publisher.notifySubscribers()
        return wrapper
    return catch_exception

if __name__ == '__main__':

    from loguru import logger
    from apps.tools.Sender_Email import Sender_mail

    news_publisher =NewsPublisher()
    for Subscribers in [LoggerSubscriber]:
        print(Subscribers)
        Subscribers(news_publisher)

    print("\nSubscribers",news_publisher.subscribers())

    news_publisher.addNews("地图加载失败！")
    news_publisher.notifySubscribers()