Source code for errbot.core_plugins.vcheck

import sys
import threading
from json import JSONDecodeError
from urllib.error import HTTPError, URLError

import requests
from requests.exceptions import ConnectionError

from errbot import BotPlugin
from errbot.utils import version2tuple
from errbot.version import VERSION

HOME = ""

installed_version = version2tuple(VERSION)

PY_VERSION = ".".join(str(e) for e in sys.version_info[:3])

[docs]class VersionChecker(BotPlugin): connected = False activated = False
[docs] def activate(self): if self.mode not in ( "null", "test", "Dummy", "text", ): # skip in all test confs. self.activated = True self.version_check() # once at startup anyway self.start_poller(3600 * 24, self.version_check) # once every 24H super().activate() else:"Skip version checking under %s mode.", self.mode)
[docs] def deactivate(self): self.activated = False super().deactivate()
def _get_version(self): """Get errbot version based on python version.""" version = VERSION major_py_version = PY_VERSION.partition(".")[0] # noinspection PyBroadException try: possible_versions = requests.get(HOME).json() version = possible_versions.get( "python{}".format(major_py_version), VERSION ) self.log.debug("Latest Errbot version is: %s", version) except (HTTPError, URLError, ConnectionError, JSONDecodeError):"Could not establish connection to retrieve latest version.") return version def _async_vcheck(self): current_version_txt = self._get_version() self.log.debug("Installed Errbot version is: %s", current_version_txt) current_version = version2tuple(current_version_txt) if installed_version < current_version: self.log.debug( "A new version %s has been found, notify the admins!", current_version_txt, ) self.warn_admins( f"Version {current_version_txt} of Errbot is available. " f"{current_version_txt}. " f"To disable this check do: {self._bot.prefix}plugin blacklist VersionChecker" )
[docs] def version_check(self): if not self.activated: self.log.debug("Version check disabled") return self.log.debug("Checking version in background.") threading.Thread(target=self._async_vcheck).start()
[docs] def callback_connect(self): if not self.connected: self.connected = True