导入的模块日志记录调用未显示



我一直在阅读正确的日志记录,到目前为止,我喜欢它的进展。一切都很好,直到我尝试在我编写的主文件和模块中进行日志记录。主文件能够写入文件和控制台,但导入的模块在两者中都不显示任何内容。如果我不得不猜测,我假设我必须像在代码配置中使用的那样单独配置模块输出。问题是我不确定如何或是否是原因。我已经尽力用谷歌搜索而不是问,但我现在在这里。这是源代码的链接。如果您尝试运行它,您可能需要更改导入,因为当我直接导入文件时,pycharm 不喜欢它。所以从"从测试导入速度测试"到"导入速度测试"的文件是 main.py 和 speedtest.py

主要

import logging
from tests import speedtest
import time
# Logging configuration
logFormatter = logging.Formatter("%(asctime)s [%(threadName)-12.12s] [%(levelname)-5.5s]  %(message)s")
logger = logging.getLogger(__name__)
logger.setLevel(logging.DEBUG)
# creates a handler to deal with writing to the file
file_handler = logging.FileHandler("log.txt", mode="w")
file_handler.setFormatter(logFormatter)
# handler for writing to the console
console_handler = logging.StreamHandler()
console_handler.setFormatter(logFormatter)
# adds the handlers to the root logger
logger.addHandler(file_handler)
logger.addHandler(console_handler)
# max speed provided
NOMINAL_SPEED = 50
# threshold in percentage 60% seems to be a decent amount to expect.
THRESHOLD = 60
# padding in percentage for severe warning
PAD = 10
# amount of time in between runs
INTERVAL = 300

class Main:
    """
    Main running class
    """
    def __init__(self):
        self.speedtest = speedtest.SpeedTest(share=True)
        self.threshold = THRESHOLD
        self.pad = PAD
        self.nominal = NOMINAL_SPEED
        self.done = False
        logger.debug("Starting main loop.")
        while not self.done:
            self.loop()
            time.sleep(INTERVAL)
    def loop(self):
        try:
            results = self.speedtest.run()
        except Exception as e:
            logger.error("Skipped running speed test this run. Will try again next time")
            return
        download = float(results["download"][:-7])
        upload = float(results["upload"][:-7])
        url = results["url"]
        host = results["host"]
        diff_download = (download / self.nominal) * 100
        logger.debug("Current download is {} Mbps upload is {} Mbps. Share url: {} host: {}".format(download, upload, url, host))
        if (((self.threshold - self.pad)/100) * self.nominal) <= diff_download <= ((self.threshold/100) * self.nominal):
            logger.info("Speed is currently at {}% nominal.".format(diff_download))
            self.warning()
        elif diff_download <= ((self.threshold - self.pad)/100) * self.nominal:
            logger.info("Speed is currently at {}% nominal. This is a problem.".format(diff_download))
            self.critical()
    def warning(self):
        pass
    def critical(self):
        pass

if __name__ == "__main__":
    Main()

速度测试

import subprocess
import logging
import os

class SpeedTest:
    """
    Class to run speed test and return the results in an easy to use manner
    """
    def __init__(self, share=False):
        """
        Init method
        :param share: When set to true it will also return a url to the speed test image
        :return:
        """
        self.logger = logging.getLogger(__name__)
        self.logger.addHandler(logging.NullHandler())
        self._share = share
        if share is True:
            self.logger.debug("Share flag set to True")
            self.cmd = ["speedtest-cli", "--share"]
        else:
            self.logger.debug("Share not set to true. Ignoring share url")
            self.cmd = ["speedtest-cli"]
    def run(self):
        """
        Runs the speed test returning a dict containing upload, download, ping, and share url if wanted.
        :return:
        """
        self.logger.debug("Starting speedtest!")
        # check_output returns the output in bytes so we use decode() to turn it into a simple string. Then we split
        # the lines giving us a list.
        try:
            stdout = subprocess.check_output(self.cmd).decode().splitlines()
        except subprocess.CalledProcessError as e:
            self.logger.error(e)
            raise e
        res = {}
        for i in stdout:
            if "Download:" in i:
                res["download"] = i[10:]
            if "Upload:" in i:
                res["upload"] = i[8:]
            if "Hosted" in i:
                res["host"] = i[2:]
            if self._share is True and "Share results:" in i:
                res["url"] = i[15:]
            else:
                res["url"] = None
        return res
    def ping(self, addr):
        """
        Pings an address and returns a 1 if the connection can not be made or a 0 if it succeeds
        :param addr: IPv4 address
        :return:
        """
        try:
            if os.name is "nt":
                self.logger.debug("Windows OS detected")
                self.logger.info("Pinging {}".format(addr))
                subprocess.check_output(["ping", "-n", "1", addr])
            elif os.name is "posix":
                self.logger.debug("Nix OS detected")
                subprocess.check_output(["ping", "-c", "1", addr])
        except subprocess.CalledProcessError:
            self.logger.warning("Returned non zero value. Is the internet working?")
            return 1
        return 0
if __name__ == "__main__":
    logging.basicConfig(level=logging.DEBUG)
    for i in SpeedTest(share=True).run().items():
        print(i)
    print(SpeedTest().ping("8.8.8.0"))

在 speedtest.py 中调用时:

logging.getLogger(__name__)

它将为 speedtest.py 创建一个记录器对象,因此您必须单独配置它。如果您希望它与主记录器相同,只需添加:

self.speedtest.logger = logger

在 Main 的构造函数中创建 SpeedTest 对象之后

您的另一种选择是将__name__作为参数传递给 SpeedTest() 并使用该参数创建记录器(我认为这对您来说是一个更好的选择,因为您在构造函数中写入记录器)。

最新更新