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".appsMapviewlogs")
        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()