Merge githubstatus.py's functionality into github.py
This commit is contained in:
Erik Johnson
parent
394eaf52ee
commit
8afd1da5cf
@ -1,83 +1,544 @@
|
||||
import copy
|
||||
import json
|
||||
import re
|
||||
import threading
|
||||
import time
|
||||
from urllib.request import urlopen
|
||||
|
||||
import requests
|
||||
from i3pystatus import IntervalModule
|
||||
from i3pystatus import logger
|
||||
from i3pystatus import IntervalModule, formatp
|
||||
from i3pystatus.core import ConfigError
|
||||
from i3pystatus.core.desktop import DesktopNotification
|
||||
from i3pystatus.core.util import user_open, internet, require
|
||||
from requests import Timeout, ConnectionError
|
||||
|
||||
try:
|
||||
import requests
|
||||
HAS_REQUESTS = True
|
||||
except ImportError:
|
||||
HAS_REQUESTS = False
|
||||
|
||||
API_METHODS_URL = 'https://status.github.com/api.json'
|
||||
STATUS_URL = 'https://status.github.com'
|
||||
NOTIFICATIONS_URL = 'https://github.com/notifications'
|
||||
ACCESS_TOKEN_AUTH_URL = 'https://api.github.com/notifications?access_token=%s'
|
||||
BASIC_AUTH_URL = 'https://api.github.com/notifications'
|
||||
|
||||
|
||||
class Github(IntervalModule):
|
||||
"""
|
||||
Check GitHub for pending notifications.
|
||||
Requires `requests`
|
||||
'''
|
||||
This module checks the GitHub system status, and optionally the number of
|
||||
unread notifications.
|
||||
|
||||
Availables authentication methods:
|
||||
.. versionchanged:: 3.36
|
||||
Module now checks system status in addition to unread notifications.
|
||||
|
||||
* username + password
|
||||
* access_token (manually generate a new token at https://github.com/settings/tokens)
|
||||
.. note::
|
||||
For notification checking, the following is required:
|
||||
|
||||
See https://developer.github.com/v3/#authentication for more informations.
|
||||
- The requests_ module must be installed.
|
||||
- Either ``access_token`` (recommended) or ``username`` and
|
||||
``password`` must be used to authenticate to GitHub.
|
||||
|
||||
Formatters:
|
||||
Using an access token is the recommended authentication method. Click
|
||||
here__ to generate a new access token. Fill in the **Token
|
||||
description** box, and enable the **notifications** scope by checking
|
||||
the appropriate checkbox. Then, click the **Generate token** button.
|
||||
|
||||
* `{unread}` — contains the value of unread_marker when there are pending notifications
|
||||
* `{unread_count}` — number of unread notifications, empty if 0
|
||||
"""
|
||||
.. important::
|
||||
An access token is the only supported means of authentication for
|
||||
this module, if `2-factor authentication`_ is enabled.
|
||||
|
||||
max_error_len = 50
|
||||
unread_marker = "●"
|
||||
unread = ''
|
||||
color = '#78EAF2'
|
||||
.. _requests: https://pypi.python.org/pypi/requests
|
||||
.. __: https://github.com/settings/tokens/new
|
||||
.. _`2-factor authentication`: https://help.github.com/articles/about-two-factor-authentication/
|
||||
|
||||
See here__ for more information on GitHub's authentication API.
|
||||
|
||||
.. __: https://developer.github.com/v3/#authentication
|
||||
|
||||
If you would rather use a username and password pair, you can either
|
||||
pass them as arguments when registering the module, or use i3pystatus'
|
||||
:ref:`credential management <credentials>` support to store them in a
|
||||
keyring. Keep in mind that if you do not pass a ``username`` or
|
||||
``password`` parameter when registering the module, i3pystatus will
|
||||
still attempt to retrieve these values from a keyring if the keyring_
|
||||
Python module is installed. This could result in i3pystatus aborting
|
||||
during startup if it cannot find a usable keyring backend. If you do
|
||||
not plan to use credential management at all in i3pystatus, then you
|
||||
should either ensure that A) keyring_ is not installed, or B) both
|
||||
keyring_ and keyrings.alt_ are installed, to avoid this error.
|
||||
|
||||
.. _keyring: https://pypi.python.org/pypi/keyring
|
||||
.. _keyrings.alt: https://pypi.python.org/pypi/keyrings.alt
|
||||
|
||||
|
||||
.. rubric:: Available formatters
|
||||
|
||||
* `{status}` — Current GitHub status. This formatter can be different
|
||||
depending on the current status (``good``, ``minor``, or ``major``).
|
||||
The content displayed for each of these statuses is defined in the
|
||||
**status** config option.
|
||||
* `{unread}` — When there are unread notifications, this formatter will
|
||||
contain the value of the **unread_marker** marker config option.
|
||||
there are no unread notifications, it formatter will be an empty string.
|
||||
* `{unread_count}` — The number of unread notifications
|
||||
notifications, it will be an empty string.
|
||||
* `{update_error}` — When an error is encountered updating this module,
|
||||
this formatter will be set to the value of the **update_error**
|
||||
config option.
|
||||
|
||||
.. rubric:: Click events
|
||||
|
||||
This module responds to 4 different click events:
|
||||
|
||||
- **Left-click** — Forces an update of the module.
|
||||
- **Right-click** — Triggers a desktop notification showing the most recent
|
||||
update to the GitHub status. This is useful when the status changes when
|
||||
you are away from your computer, so that the updated status can be seen
|
||||
without visiting the `GitHub Status Dashboard`_. This click event
|
||||
requires **notify_status** to be set to ``True``.
|
||||
- **Double left-click** — Opens the GitHub `notifications page`_ in your web
|
||||
browser.
|
||||
- **Double right-click** — Opens the `GitHub Status Dashboard`_ in your web
|
||||
browser.
|
||||
|
||||
.. rubric:: Desktop notifications
|
||||
|
||||
.. versionadded:: 3.36
|
||||
|
||||
If **notify_status** is set to ``True``, a notification will be displayed
|
||||
when the status reported by the `GitHub Status API`_ changes.
|
||||
|
||||
If **notify_unread** is set to ``True``, a notification will be displayed
|
||||
when new unread notifications are found. Double-clicking the module will
|
||||
launch the GitHub notifications dashboard in your browser.
|
||||
|
||||
.. note::
|
||||
A notification will be displayed if there was a problem querying the
|
||||
`GitHub Status API`_, irrespective of whether or not **notify_status**
|
||||
or **notify_unread** is set to ``True``.
|
||||
|
||||
.. rubric:: Example configuration
|
||||
|
||||
The below example enables desktop notifications, enables Pango hinting for
|
||||
differently-colored **update_error** and **refresh_icon** text, and alters
|
||||
the both the status text and the colors used to visually denote the current
|
||||
status level.
|
||||
|
||||
.. code-block:: python
|
||||
|
||||
status.register(
|
||||
'github',
|
||||
notify_status=True,
|
||||
notify_unread=True,
|
||||
access_token='0123456789abcdef0123456789abcdef01234567',
|
||||
hints={'markup': 'pango'},
|
||||
update_error='<span color="#af0000">!</span>',
|
||||
refresh_icon='<span color="#ff5f00">⟳</span>',
|
||||
status={
|
||||
'good': '✓',
|
||||
'minor': '!',
|
||||
'major': '!!',
|
||||
},
|
||||
colors={
|
||||
'good': '#008700',
|
||||
'minor': '#d7ff00',
|
||||
'major': '#af0000',
|
||||
},
|
||||
)
|
||||
|
||||
.. _`GitHub Status API`: https://status.github.com/api
|
||||
.. _`GitHub Status Dashboard`: https://status.github.com
|
||||
.. _`notifications page`: https://github.com/notifications
|
||||
|
||||
.. rubric:: Extended string formatting
|
||||
|
||||
.. versionadded:: 3.36
|
||||
|
||||
This module supports the :ref:`formatp <formatp>` extended string format
|
||||
syntax. This allows for values to be hidden when they evaluate as False.
|
||||
The default ``format`` string value for this module makes use of this
|
||||
syntax to conditionally show the value of the ``update_error`` config value
|
||||
when the backend encounters an error during an update, but this can also
|
||||
be used to only show the number of unread notifications when that number is
|
||||
not **0**. The below example would show the unread count as **(3)** when
|
||||
there are 3 unread notifications, but would show nothing when there are no
|
||||
unread notifications.
|
||||
|
||||
.. code-block:: python
|
||||
|
||||
status.register(
|
||||
'github',
|
||||
notify_status=True,
|
||||
notify_unread=True,
|
||||
access_token='0123456789abcdef0123456789abcdef01234567',
|
||||
format='{status}[ ({unread_count})][ {update_error}]'
|
||||
)
|
||||
'''
|
||||
settings = (
|
||||
('format', 'format string'),
|
||||
('status', 'Dictionary mapping statuses to the text which represents '
|
||||
'that status type. This defaults to ``GitHub`` for all '
|
||||
'status types.'),
|
||||
('colors', 'Dictionary mapping statuses to the color used to display '
|
||||
'the status text'),
|
||||
('refresh_icon', 'Text to display (in addition to any text currently '
|
||||
'shown by the module) when refreshing the GitHub '
|
||||
'status. **NOTE:** Depending on how quickly the '
|
||||
'update is performed, the icon may not be displayed.'),
|
||||
('update_error', 'Value for the ``{update_error}`` formatter when an '
|
||||
'error is encountered while checking GitHub status'),
|
||||
('keyring_backend', 'alternative keyring backend for retrieving '
|
||||
'credentials'),
|
||||
('username', ''),
|
||||
('password', ''),
|
||||
('access_token', ''),
|
||||
('unread_marker', 'Defines the string that the ``{unread}`` formatter '
|
||||
'shows when there are pending notifications'),
|
||||
('notify_status', 'Set to ``True`` to display a desktop notification '
|
||||
'on status changes'),
|
||||
('notify_unread', 'Set to ``True`` to display a desktop notification '
|
||||
'when new notifications are detected'),
|
||||
('unread_notification_template',
|
||||
'String with no more than one ``%d``, which will be replaced by '
|
||||
'the number of new unread notifications. Useful for those with '
|
||||
'non-English locales who would like the notification to be in '
|
||||
'their native language. The ``%d`` can be omitted if desired.'),
|
||||
('api_methods_url', 'URL from which to retrieve the API endpoint URL '
|
||||
'which this module will use to check the GitHub '
|
||||
'Status'),
|
||||
('status_url', 'The URL to the status page (opened when the module is '
|
||||
'double-clicked with the right mouse button'),
|
||||
('notifications_url', 'The URL to the GitHub notifications page '
|
||||
'(opened when the module is double-clicked with '
|
||||
'the left mouse button'),
|
||||
)
|
||||
|
||||
# Defaults for module configurables
|
||||
_default_status = {
|
||||
'good': 'GitHub',
|
||||
'minor': 'GitHub',
|
||||
'major': 'GitHub',
|
||||
}
|
||||
_default_colors = {
|
||||
'good': '#2f895c',
|
||||
'minor': '#f29d50',
|
||||
'major': '#cc3300',
|
||||
}
|
||||
|
||||
# Module configurables
|
||||
format = '{status}[ {unread}][ {update_error}]'
|
||||
status = _default_status
|
||||
colors = _default_colors
|
||||
refresh_icon = '⟳'
|
||||
update_error = '!'
|
||||
username = ''
|
||||
password = ''
|
||||
access_token = ''
|
||||
format = '{unread}'
|
||||
unread_marker = '•'
|
||||
notify_status = False
|
||||
notify_unread = False
|
||||
unread_notification_template = 'You have %d new notification(s)'
|
||||
api_methods_url = API_METHODS_URL
|
||||
status_url = STATUS_URL
|
||||
notifications_url = NOTIFICATIONS_URL
|
||||
|
||||
# Global configurables
|
||||
interval = 600
|
||||
max_error_len = 50
|
||||
keyring_backend = None
|
||||
|
||||
on_leftclick = 'open_github'
|
||||
# Other
|
||||
unread = ''
|
||||
unknown_color = None
|
||||
unknown_status = '?'
|
||||
failed_update = False
|
||||
previous_status = None
|
||||
current_status = None
|
||||
new_unread = None
|
||||
previous_unread = None
|
||||
current_unread = None
|
||||
config_error = None
|
||||
data = {'status': '',
|
||||
'unread': 0,
|
||||
'unread_count': '',
|
||||
'update_error': ''}
|
||||
output = {'full_text': '', 'color': None}
|
||||
|
||||
settings = (
|
||||
('format', 'format string'),
|
||||
('keyring_backend', 'alternative keyring backend for retrieving credentials'),
|
||||
('unread_marker', 'sets the string that the "unread" formatter shows when there are pending notifications'),
|
||||
("username", ""),
|
||||
("password", ""),
|
||||
("access_token", "see https://developer.github.com/v3/#authentication"),
|
||||
("color", "")
|
||||
)
|
||||
|
||||
def open_github(self):
|
||||
user_open('https://github.com/' + self.username)
|
||||
# Click events
|
||||
on_leftclick = ['perform_update']
|
||||
on_rightclick = ['show_status_notification']
|
||||
on_doubleleftclick = ['launch_notifications_url']
|
||||
on_doublerightclick = ['launch_status_url']
|
||||
|
||||
@require(internet)
|
||||
def run(self):
|
||||
def launch_status_url(self):
|
||||
self.logger.debug('Launching %s in browser', self.status_url)
|
||||
user_open(self.status_url)
|
||||
|
||||
@require(internet)
|
||||
def launch_notifications_url(self):
|
||||
self.logger.debug('Launching %s in browser', self.notifications_url)
|
||||
user_open(self.notifications_url)
|
||||
|
||||
def init(self):
|
||||
if self.status != self._default_status:
|
||||
new_status = copy.copy(self._default_status)
|
||||
new_status.update(self.status)
|
||||
self.status = new_status
|
||||
|
||||
if self.colors != self._default_colors:
|
||||
new_colors = copy.copy(self._default_colors)
|
||||
new_colors.update(self.colors)
|
||||
self.colors = new_colors
|
||||
|
||||
self.logger.debug('status = %s', self.status)
|
||||
self.logger.debug('colors = %s', self.colors)
|
||||
|
||||
self.condition = threading.Condition()
|
||||
self.thread = threading.Thread(target=self.update_loop, daemon=True)
|
||||
self.thread.start()
|
||||
|
||||
def update_loop(self):
|
||||
try:
|
||||
self.perform_update()
|
||||
while True:
|
||||
with self.condition:
|
||||
self.condition.wait(self.interval)
|
||||
self.perform_update()
|
||||
except Exception:
|
||||
msg = 'Exception in {thread} at {time}, module {name}'.format(
|
||||
thread=threading.current_thread().name,
|
||||
time=time.strftime('%c'),
|
||||
name=self.__class__.__name__,
|
||||
)
|
||||
self.logger.error(msg, exc_info=True)
|
||||
|
||||
@require(internet)
|
||||
def api_request(self, url):
|
||||
self.logger.debug('Making API request to %s', url)
|
||||
try:
|
||||
with urlopen(url) as content:
|
||||
try:
|
||||
content_type = dict(content.getheaders())['Content-Type']
|
||||
charset = re.search(r'charset=(.*)', content_type).group(1)
|
||||
except AttributeError:
|
||||
charset = 'utf-8'
|
||||
response_json = content.read().decode(charset).strip()
|
||||
if not response_json:
|
||||
self.logger.debug('JSON response from %s was blank', url)
|
||||
return {}
|
||||
try:
|
||||
response = json.loads(response_json)
|
||||
except json.decoder.JSONDecodeError as exc:
|
||||
self.logger.error('Error loading JSON: %s', exc)
|
||||
self.logger.debug('JSON text that failed to load: %s',
|
||||
response_json)
|
||||
return {}
|
||||
self.logger.log(5, 'API response: %s', response)
|
||||
return response
|
||||
except Exception as exc:
|
||||
self.logger.error(
|
||||
'Failed to make API request to %s. Exception follows:', url,
|
||||
exc_info=True
|
||||
)
|
||||
return {}
|
||||
|
||||
def detect_status_change(self, response=None):
|
||||
if response is not None:
|
||||
# Compare last update to current and exit without displaying a
|
||||
# notification if one is not needed.
|
||||
if self.previous_status is None:
|
||||
# This is the first time status has been updated since
|
||||
# i3pystatus was started. Set self.previous_status and exit.
|
||||
self.previous_status = response
|
||||
return
|
||||
if response == self.previous_status:
|
||||
# No change, so no notification
|
||||
return
|
||||
self.previous_status = response
|
||||
|
||||
if self.previous_status is None:
|
||||
# The only way this would happen is if we invoked the right-click
|
||||
# event before we completed the initial status check.
|
||||
return
|
||||
|
||||
self.notify_status_change()
|
||||
|
||||
@staticmethod
|
||||
def notify(message):
|
||||
return DesktopNotification(title='GitHub', body=message).display()
|
||||
|
||||
def skip_notify(message):
|
||||
self.logger.debug(
|
||||
'Desktop notifications turned off. Skipped notification: %s',
|
||||
message
|
||||
)
|
||||
return False
|
||||
|
||||
def show_status_notification(self):
|
||||
message = self.current_status.get(
|
||||
'body',
|
||||
'Missing \'body\' param in API response'
|
||||
)
|
||||
return self.skip_notify(message) \
|
||||
if not self.notify_status \
|
||||
else self.notify(message)
|
||||
|
||||
def show_unread_notification(self):
|
||||
if '%d' not in self.unread_notification_template:
|
||||
formatted = self.unread_notification_template
|
||||
else:
|
||||
try:
|
||||
new_unread = len(self.new_unread)
|
||||
except TypeError:
|
||||
new_unread = 0
|
||||
try:
|
||||
formatted = self.unread_notification_template % new_unread
|
||||
except TypeError as exc:
|
||||
self.logger.error(
|
||||
'Failed to format {0!r}: {1}'.format(
|
||||
self.unread_notification_template,
|
||||
exc
|
||||
)
|
||||
)
|
||||
return False
|
||||
return self.skip_notify(formatted) \
|
||||
if not self.notify_unread \
|
||||
else self.notify(formatted)
|
||||
|
||||
@require(internet)
|
||||
def perform_update(self):
|
||||
self.output['full_text'] = \
|
||||
self.refresh_icon + self.output.get('full_text', '')
|
||||
self.failed_update = False
|
||||
|
||||
self.update_status()
|
||||
try:
|
||||
self.config_error = None
|
||||
self.update_unread()
|
||||
except ConfigError as exc:
|
||||
self.config_error = exc
|
||||
|
||||
self.data['update_error'] = self.update_error \
|
||||
if self.failed_update \
|
||||
else ''
|
||||
self.refresh_display()
|
||||
if self.failed_update:
|
||||
self.notify('Error occurred updating module. See log for details.')
|
||||
|
||||
@require(internet)
|
||||
def update_status(self):
|
||||
try:
|
||||
# Get most recent update
|
||||
if not hasattr(self, 'last_message_url'):
|
||||
self.last_message_url = \
|
||||
self.api_request(API_METHODS_URL)['last_message_url']
|
||||
self.current_status = self.api_request(self.last_message_url)
|
||||
if not self.current_status:
|
||||
self.failed_update = True
|
||||
return
|
||||
|
||||
self.data['status'] = self.status.get(
|
||||
self.current_status.get('status'),
|
||||
self.unknown_status)
|
||||
|
||||
if self.previous_status is not None:
|
||||
if self.current_status != self.previous_status:
|
||||
self.show_status_notification()
|
||||
self.previous_status = self.current_status
|
||||
|
||||
except Exception:
|
||||
# Don't let an uncaught exception kill the update thread
|
||||
self.logger.error(
|
||||
'Uncaught error occurred while checking GitHub status. '
|
||||
'Exception follows:', exc_info=True
|
||||
)
|
||||
self.failed_update = True
|
||||
|
||||
@require(internet)
|
||||
def update_unread(self):
|
||||
# Reset the new_unread attribute to prevent spurious notifications
|
||||
self.new_unread = None
|
||||
|
||||
try:
|
||||
if self.access_token:
|
||||
response = requests.get('https://api.github.com/notifications?access_token=' + self.access_token)
|
||||
else:
|
||||
response = requests.get('https://api.github.com/notifications', auth=(self.username, self.password))
|
||||
data = json.loads(response.text)
|
||||
except (ConnectionError, Timeout) as e:
|
||||
logger.warn(e)
|
||||
data = []
|
||||
if not self.username and not self.password and not self.access_token:
|
||||
# Auth not configured
|
||||
self.logger.debug(
|
||||
'No auth configured, notifications will not be checked')
|
||||
return False
|
||||
|
||||
# Bad credentials
|
||||
if isinstance(data, dict):
|
||||
err_msg = data['message']
|
||||
raise ConfigError(err_msg)
|
||||
if not HAS_REQUESTS:
|
||||
self.log.error(
|
||||
'The requests module is required to check GitHub notifications')
|
||||
self.failed_update = True
|
||||
return False
|
||||
|
||||
format_values = dict(unread_count='', unread='')
|
||||
unread = len(data)
|
||||
if unread > 0:
|
||||
format_values['unread_count'] = unread
|
||||
format_values['unread'] = self.unread_marker
|
||||
try:
|
||||
self.logger.debug(
|
||||
'Making API request to retrieve unread notifications')
|
||||
if self.access_token:
|
||||
self.logger.debug('Authenticating using access_token')
|
||||
response = requests.get(
|
||||
ACCESS_TOKEN_AUTH_URL % self.access_token)
|
||||
else:
|
||||
self.logger.debug('Authenticating using username/password')
|
||||
response = requests.get(BASIC_AUTH_URL,
|
||||
auth=(self.username, self.password))
|
||||
self.logger.log(5,
|
||||
'Raw return from GitHub notification check: %s',
|
||||
response.text)
|
||||
unread_data = json.loads(response.text)
|
||||
except (requests.ConnectionError, requests.Timeout) as exc:
|
||||
self.logger.error('Failed to check unread notifications: %s', exc)
|
||||
self.failed_update = True
|
||||
return False
|
||||
except json.decoder.JSONDecodeError as exc:
|
||||
self.logger.error('Error loading JSON: %s', exc)
|
||||
self.logger.debug('JSON text that failed to load: %s', response.text)
|
||||
self.failed_update = True
|
||||
return False
|
||||
|
||||
self.data = format_values
|
||||
self.output = {
|
||||
'full_text': self.format.format(**format_values),
|
||||
'color': self.color
|
||||
}
|
||||
# Bad credentials or some other error
|
||||
if isinstance(unread_data, dict):
|
||||
raise ConfigError(
|
||||
unread_data.get(
|
||||
'message',
|
||||
'Unknown error encountered retrieving unread notifications'
|
||||
)
|
||||
)
|
||||
|
||||
self.current_unread = set([x['id'] for x in unread_data if 'id' in x])
|
||||
self.data['unread_count'] = len(self.current_unread)
|
||||
self.data['unread'] = self.unread_marker \
|
||||
if self.data['unread_count'] > 0 \
|
||||
else ''
|
||||
|
||||
if self.previous_unread is not None:
|
||||
if not self.current_unread.issubset(self.previous_unread):
|
||||
self.new_unread = self.current_unread - self.previous_unread
|
||||
if self.new_unread:
|
||||
self.show_unread_notification()
|
||||
self.previous_unread = self.current_unread
|
||||
return True
|
||||
except ConfigError as exc:
|
||||
# This will be caught by the calling function
|
||||
raise exc
|
||||
except Exception as exc:
|
||||
# Don't let an uncaught exception kill the update thread
|
||||
self.logger.error(
|
||||
'Uncaught error occurred while checking GitHub notifications. '
|
||||
'Exception follows:', exc_info=True
|
||||
)
|
||||
self.failed_update = True
|
||||
return False
|
||||
|
||||
def refresh_display(self):
|
||||
color = self.colors.get(
|
||||
self.current_status.get('status'),
|
||||
self.unknown_color)
|
||||
self.output = {'full_text': formatp(self.format, **self.data).strip(),
|
||||
'color': color}
|
||||
|
||||
def run(self):
|
||||
if self.config_error is not None:
|
||||
raise self.config_error
|
||||
|
||||
@ -1,271 +0,0 @@
|
||||
import copy
|
||||
import json
|
||||
import re
|
||||
import threading
|
||||
import time
|
||||
from urllib.request import urlopen
|
||||
|
||||
from i3pystatus import IntervalModule, formatp
|
||||
from i3pystatus.core.desktop import DesktopNotification
|
||||
from i3pystatus.core.util import user_open, internet, require
|
||||
|
||||
API_METHODS_URL = 'https://status.github.com/api.json'
|
||||
|
||||
|
||||
class GitHubStatus(IntervalModule):
|
||||
'''
|
||||
This module uses the `GitHub Status API`_ to show whether or not there are
|
||||
currently any issues with github.com. Clicking the module will force it to
|
||||
update, and double-clicking it will launch the `GitHub Status Dashboard`_
|
||||
(the URL for the status page can be overriden using the **status_page**
|
||||
option).
|
||||
|
||||
.. _`GitHub Status API`: https://status.github.com/api
|
||||
.. _`GitHub Status Dashboard`: https://status.github.com
|
||||
|
||||
.. rubric:: Available formatters
|
||||
|
||||
* `{status}` — Current GitHub status. This formatter can be different
|
||||
depending on the current status (``good``, ``minor``, or ``major``).
|
||||
The content displayed for each of these statuses is defined in the
|
||||
**status** config option.
|
||||
* `{update_error}` — When an error is encountered updating the GitHub
|
||||
status, this formatter will be set to the value of the **update_error**
|
||||
config option. Otherwise, this formatter will be an empty string.
|
||||
|
||||
.. rubric:: Desktop notifications
|
||||
|
||||
If **notify** is set to ``True``, then desktop notifications will be
|
||||
enabled for this module. A notification will be displayed if there was a
|
||||
problem querying the `GitHub Status API`_, and also when the status changes.
|
||||
Additionally, right-clicking the module will replay the notification for
|
||||
the most recent status change.
|
||||
|
||||
.. rubric:: Example configuration
|
||||
|
||||
The below example enables desktop notifications, enables Pango hinting for
|
||||
differently-colored **update_error** and **refresh_icon** text, and alters
|
||||
the colors used to visually denote the current status level.
|
||||
|
||||
.. code-block:: python
|
||||
|
||||
status.register(
|
||||
'githubstatus',
|
||||
notify=True,
|
||||
hints={'markup': 'pango'},
|
||||
update_error='<span color="#af0000">!</span>',
|
||||
refresh_icon='<span color="#ff5f00">⟳</span>',
|
||||
colors={
|
||||
'good': '#008700',
|
||||
'minor': '#d7ff00',
|
||||
'major': '#af0000',
|
||||
},
|
||||
)
|
||||
|
||||
'''
|
||||
|
||||
settings = (
|
||||
('status', 'Dictionary mapping statuses to the text which represents '
|
||||
'that status type'),
|
||||
('colors', 'Dictionary mapping statuses to the color used to display '
|
||||
'the status text'),
|
||||
('refresh_icon', 'Text to display (in addition to any text currently '
|
||||
'shown by the module) when refreshing the GitHub '
|
||||
'status. **NOTE:** Depending on how quickly the '
|
||||
'update is performed, the icon may not be displayed.'),
|
||||
('status_page', 'Page to launch when module is double-clicked'),
|
||||
('notify', 'Set to ``True`` to enable desktop notifications'),
|
||||
('update_error', 'Value for the ``{update_error}`` formatter when an '
|
||||
'error is encountered while checking GitHub status'),
|
||||
('format', 'Format to use for displaying status info'),
|
||||
)
|
||||
|
||||
_default_status = {
|
||||
'good': 'GitHub',
|
||||
'minor': 'GitHub',
|
||||
'major': 'GitHub',
|
||||
}
|
||||
_default_colors = {
|
||||
'good': '#2f895c',
|
||||
'minor': '#f29d50',
|
||||
'major': '#cc3300',
|
||||
}
|
||||
|
||||
status = _default_status
|
||||
colors = _default_colors
|
||||
refresh_icon = '⟳'
|
||||
status_page = 'https://status.github.com'
|
||||
notify = False
|
||||
update_error = '!'
|
||||
format = '{status}[ {update_error}]'
|
||||
|
||||
# A color of None == a fallback to the default status bar color
|
||||
unknown_color = None
|
||||
unknown_status = '?'
|
||||
previous_change = None
|
||||
current_status = {}
|
||||
|
||||
data = {'status': '', 'update_error': ''}
|
||||
output = {'full_text': '', 'color': None}
|
||||
interval = 300
|
||||
|
||||
on_leftclick = ['check_status']
|
||||
on_rightclick = ['notify_change']
|
||||
on_doubleleftclick = ['launch_status_page']
|
||||
|
||||
def init(self):
|
||||
if self.status != self._default_status:
|
||||
new_status = copy.copy(self._default_status)
|
||||
new_status.update(self.status)
|
||||
self.status = new_status
|
||||
|
||||
if self.colors != self._default_colors:
|
||||
new_colors = copy.copy(self._default_colors)
|
||||
new_colors.update(self.colors)
|
||||
self.colors = new_colors
|
||||
|
||||
self.logger.debug('status = %s', self.status)
|
||||
self.logger.debug('colors = %s', self.colors)
|
||||
|
||||
self.condition = threading.Condition()
|
||||
self.thread = threading.Thread(target=self.update_thread, daemon=True)
|
||||
self.thread.start()
|
||||
|
||||
def update_thread(self):
|
||||
try:
|
||||
self.check_status()
|
||||
while True:
|
||||
with self.condition:
|
||||
self.condition.wait(self.interval)
|
||||
self.check_status()
|
||||
except Exception:
|
||||
msg = 'Exception in {thread} at {time}, module {name}'.format(
|
||||
thread=threading.current_thread().name,
|
||||
time=time.strftime('%c'),
|
||||
name=self.__class__.__name__,
|
||||
)
|
||||
self.logger.error(msg, exc_info=True)
|
||||
|
||||
@require(internet)
|
||||
def launch_status_page(self):
|
||||
self.logger.debug('Launching %s in browser', self.status_page)
|
||||
user_open(self.status_page)
|
||||
|
||||
@require(internet)
|
||||
def api_request(self, url):
|
||||
self.logger.debug('Making API request to %s', url)
|
||||
try:
|
||||
with urlopen(url) as content:
|
||||
try:
|
||||
content_type = dict(content.getheaders())['Content-Type']
|
||||
charset = re.search(r'charset=(.*)', content_type).group(1)
|
||||
except AttributeError:
|
||||
charset = 'utf-8'
|
||||
response_json = content.read().decode(charset).strip()
|
||||
if not response_json:
|
||||
self.logger.debug('JSON response from %s was blank', url)
|
||||
return {}
|
||||
try:
|
||||
response = json.loads(response_json)
|
||||
except json.decoder.JSONDecodeError as exc:
|
||||
self.logger.error('Error loading JSON: %s', exc)
|
||||
self.logger.debug('JSON text that failed to load: %s',
|
||||
response_json)
|
||||
return {}
|
||||
self.logger.log(5, 'API response: %s', response)
|
||||
return response
|
||||
except Exception as exc:
|
||||
self.logger.error(
|
||||
'Failed to make API request to %s. Exception follows:', url,
|
||||
exc_info=True
|
||||
)
|
||||
return {}
|
||||
|
||||
def detect_status_change(self, response=None):
|
||||
if response is not None:
|
||||
# Compare last update to current and exit without displaying a
|
||||
# notification if one is not needed.
|
||||
if self.previous_change is None:
|
||||
# This is the first time status has been updated since
|
||||
# i3pystatus was started. Set self.previous_change and exit.
|
||||
self.previous_change = response
|
||||
return
|
||||
if response == self.previous_change:
|
||||
# No change, so no notification
|
||||
return
|
||||
self.previous_change = response
|
||||
|
||||
if self.previous_change is None:
|
||||
# The only way this would happen is if we invoked the right-click
|
||||
# event before we completed the initial status check.
|
||||
return
|
||||
|
||||
self.notify_change()
|
||||
|
||||
def notify_change(self):
|
||||
message = self.previous_change.get(
|
||||
'body',
|
||||
'Missing \'body\' param in API response'
|
||||
)
|
||||
self.display_notification(message)
|
||||
|
||||
def display_notification(self, message):
|
||||
if not self.notify:
|
||||
self.logger.debug(
|
||||
'Skipping notification, desktop notifications turned off'
|
||||
)
|
||||
return
|
||||
DesktopNotification(
|
||||
title='GitHub Status',
|
||||
body=message).display()
|
||||
|
||||
@require(internet)
|
||||
def check_status(self):
|
||||
'''
|
||||
Check the weather using the configured backend
|
||||
'''
|
||||
self.output['full_text'] = \
|
||||
self.refresh_icon + self.output.get('full_text', '')
|
||||
self.query_github()
|
||||
if self.current_status:
|
||||
# Show desktop notification if status changed (and alerts enabled)
|
||||
self.detect_status_change(self.current_status)
|
||||
self.refresh_display()
|
||||
|
||||
def query_github(self):
|
||||
self.data['update_error'] = ''
|
||||
color = None
|
||||
try:
|
||||
# Get most recent update
|
||||
if not hasattr(self, 'last_message_url'):
|
||||
self.last_message_url = \
|
||||
self.api_request(API_METHODS_URL)['last_message_url']
|
||||
self.current_status = self.api_request(self.last_message_url)
|
||||
if not self.current_status:
|
||||
self.data['update_error'] = self.update_error
|
||||
return
|
||||
|
||||
self.data['status'] = self.status.get(
|
||||
self.current_status.get('status'),
|
||||
self.unknown_status)
|
||||
except Exception:
|
||||
# Don't let an uncaught exception kill the update thread
|
||||
self.logger.error(
|
||||
'Uncaught error occurred while checking GitHub status. '
|
||||
'Exception follows:', exc_info=True
|
||||
)
|
||||
self.data['update_error'] = self.update_error
|
||||
|
||||
def refresh_display(self):
|
||||
color = self.colors.get(
|
||||
self.current_status.get('status'),
|
||||
self.unknown_color)
|
||||
self.output = {'full_text': formatp(self.format, **self.data).strip(),
|
||||
'color': color}
|
||||
if self.data['update_error']:
|
||||
self.display_notification(
|
||||
'Error occurred checking GitHub status. See log for details.'
|
||||
)
|
||||
|
||||
def run(self):
|
||||
pass
|
||||
Loading…
Reference in New Issue
Block a user