648 lines
25 KiB
Python
648 lines
25 KiB
Python
import copy
|
|
import json
|
|
import re
|
|
import threading
|
|
import time
|
|
from urllib.request import urlopen
|
|
|
|
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
|
|
|
|
try:
|
|
import requests
|
|
HAS_REQUESTS = True
|
|
except ImportError:
|
|
HAS_REQUESTS = False
|
|
|
|
API_METHODS_URL = 'https://www.githubstatus.com/api/v2/summary.json'
|
|
STATUS_URL = 'https://www.githubstatus.com'
|
|
NOTIFICATIONS_URL = 'https://github.com/notifications'
|
|
AUTH_URL = 'https://api.github.com/notifications'
|
|
|
|
|
|
class Github(IntervalModule):
|
|
'''
|
|
This module checks the GitHub system status, and optionally the number of
|
|
unread notifications.
|
|
|
|
.. versionchanged:: 3.36
|
|
Module now checks system status in addition to unread notifications.
|
|
|
|
.. note::
|
|
For notification checking, the following is required:
|
|
|
|
- The requests_ module must be installed.
|
|
- Either ``access_token`` (recommended) or ``username`` and
|
|
``password`` must be used to authenticate to GitHub.
|
|
|
|
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.
|
|
|
|
.. important::
|
|
An access token is the only supported means of authentication for
|
|
this module, if `2-factor authentication`_ is enabled.
|
|
|
|
.. _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 outage status (``none``, ``minor``, ``major``,
|
|
or ``critical``). 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. It also sets the log level to debug, for troubleshooting
|
|
purposes.
|
|
|
|
.. code-block:: python
|
|
|
|
status.register(
|
|
'github',
|
|
log_level=logging.DEBUG,
|
|
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={
|
|
'none': '✓',
|
|
'minor': '!',
|
|
'major': '!!',
|
|
'critical': '!!!',
|
|
},
|
|
colors={
|
|
'none': '#008700',
|
|
'minor': '#d7ff00',
|
|
'major': '#af0000',
|
|
'critical': '#640000',
|
|
},
|
|
)
|
|
|
|
.. note::
|
|
Setting debug logging and authenticating with an access token will
|
|
include the access token in the log file, as the notification URL is
|
|
logged at this level.
|
|
|
|
.. _`GitHub Status API`: https://www.githubstatus.com/api
|
|
.. _`GitHub Status Dashboard`: https://www.githubstatus.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_colors = {
|
|
'none': '#28a745',
|
|
'maintenance': '#4f8cc9',
|
|
'minor': '#dbab09',
|
|
'major': '#e36209',
|
|
'critical': '#dc3545',
|
|
}
|
|
|
|
# Module configurables
|
|
format = '{status}[ {unread}][ {update_error}]'
|
|
status = {}
|
|
colors = _default_colors
|
|
refresh_icon = '⟳'
|
|
update_error = '!'
|
|
username = ''
|
|
password = ''
|
|
access_token = ''
|
|
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
|
|
|
|
# Other
|
|
unread = ''
|
|
unknown_color = None
|
|
unknown_status = '?'
|
|
failed_update = False
|
|
__previous_json = None
|
|
__current_json = 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}
|
|
|
|
# Click events
|
|
on_leftclick = ['perform_update']
|
|
on_rightclick = ['show_status_notification']
|
|
on_doubleleftclick = ['launch_notifications_url']
|
|
on_doublerightclick = ['launch_status_url']
|
|
|
|
@require(internet)
|
|
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.colors != self._default_colors:
|
|
new_colors = copy.copy(self._default_colors)
|
|
new_colors.update(self.colors)
|
|
self.colors = new_colors
|
|
|
|
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 status_api_request(self, url):
|
|
self.logger.debug('Making GitHub Status 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.error('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_json is None:
|
|
# This is the first time status has been updated since
|
|
# i3pystatus was started. Set self.__previous_json and exit.
|
|
self.__previous_json = response
|
|
return
|
|
if response.get('status', {}).get('description') == self.__previous_json.get('status', {}).get('description'):
|
|
# No change, so no notification
|
|
return
|
|
self.__previous_json = response
|
|
|
|
if self.__previous_json 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.show_status_notification()
|
|
|
|
@staticmethod
|
|
def notify(message):
|
|
return DesktopNotification(title='GitHub', body=message).display()
|
|
|
|
def skip_notify(self, message):
|
|
self.logger.debug(
|
|
'Desktop notifications turned off. Skipped notification: %s',
|
|
message
|
|
)
|
|
return False
|
|
|
|
def show_status_notification(self):
|
|
message = self.current_status_description
|
|
self.skip_notify(message) \
|
|
if not self.notify_status or (self.previous_status is None and self.current_status == 'none') \
|
|
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()
|
|
|
|
@property
|
|
def current_incidents(self):
|
|
try:
|
|
return self.__current_json['incidents']
|
|
except (KeyError, TypeError):
|
|
return []
|
|
|
|
@property
|
|
def previous_incidents(self):
|
|
try:
|
|
return self.__previous_json['incidents']
|
|
except (KeyError, TypeError):
|
|
return []
|
|
|
|
@property
|
|
def current_status(self):
|
|
try:
|
|
return self.__current_json['status']['indicator']
|
|
except (KeyError, TypeError):
|
|
return None
|
|
|
|
@property
|
|
def previous_status(self):
|
|
try:
|
|
return self.__previous_json['status']['indicator']
|
|
except (KeyError, TypeError):
|
|
return None
|
|
|
|
@property
|
|
def current_status_description(self):
|
|
try:
|
|
return self.__current_json['status']['description']
|
|
except (KeyError, TypeError):
|
|
return None
|
|
|
|
@require(internet)
|
|
def update_status(self):
|
|
try:
|
|
# Get most recent update
|
|
self.__current_json = self.status_api_request(self.api_methods_url)
|
|
if not self.__current_json:
|
|
self.failed_update = True
|
|
return
|
|
|
|
self.logger.debug('Current GitHub Status: %s', self.current_status)
|
|
self.data['status'] = self.status.get(self.current_status, 'GitHub')
|
|
if self.current_incidents != self.previous_incidents:
|
|
self.show_status_notification()
|
|
self.__previous_json = self.__current_json
|
|
|
|
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 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 True
|
|
|
|
if not HAS_REQUESTS:
|
|
self.logger.error(
|
|
'The requests module is required to check GitHub notifications')
|
|
self.failed_update = True
|
|
return False
|
|
|
|
self.logger.debug(
|
|
'Checking unread notifications using %s',
|
|
'access token' if self.access_token else 'username/password'
|
|
)
|
|
|
|
if self.access_token:
|
|
request_kwargs = {
|
|
'headers': {
|
|
'Authorization': 'token {}'.format(self.access_token),
|
|
},
|
|
}
|
|
else:
|
|
request_kwargs = {
|
|
'auth': (self.username, self.password),
|
|
}
|
|
|
|
self.current_unread = set()
|
|
page_num = 0
|
|
old_unread_url = None
|
|
unread_url = AUTH_URL
|
|
while old_unread_url != unread_url:
|
|
old_unread_url = unread_url
|
|
page_num += 1
|
|
self.logger.debug(
|
|
'Reading page %d of notifications (%s)',
|
|
page_num, unread_url
|
|
)
|
|
try:
|
|
response = requests.get(unread_url, **request_kwargs)
|
|
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
|
|
|
|
# Bad credentials or some other error
|
|
if isinstance(unread_data, dict):
|
|
raise ConfigError(
|
|
unread_data.get(
|
|
'message',
|
|
'Unknown error encountered retrieving unread notifications'
|
|
)
|
|
)
|
|
|
|
# Update the current count of unread notifications
|
|
self.current_unread.update(
|
|
[x['id'] for x in unread_data if 'id' in x]
|
|
)
|
|
|
|
# Check 'Link' header for next page of notifications
|
|
# (https://tools.ietf.org/html/rfc5988#section-5)
|
|
self.logger.debug('Checking for next page of notifications')
|
|
try:
|
|
link_header = response.headers['Link']
|
|
except AttributeError:
|
|
self.logger.error(
|
|
'No headers present in response. This might be due to '
|
|
'an API change in the requests module.'
|
|
)
|
|
self.failed_update = True
|
|
continue
|
|
except KeyError:
|
|
self.logger.debug('Only one page of notifications present')
|
|
continue
|
|
else:
|
|
# Process 'Link' header
|
|
try:
|
|
links = requests.utils.parse_header_links(link_header)
|
|
except Exception as exc:
|
|
self.logger.error(
|
|
'Failed to parse \'Link\' header: %s', exc
|
|
)
|
|
self.failed_update = True
|
|
continue
|
|
|
|
for link in links:
|
|
try:
|
|
link_rel = link['rel']
|
|
if link_rel != 'next':
|
|
# Link does not refer to the next page, skip it
|
|
continue
|
|
# Set the unread_url so that when we reach the top
|
|
# of the outer loop, we have a new URL to check.
|
|
unread_url = link['url']
|
|
break
|
|
except TypeError:
|
|
# Malformed hypermedia link
|
|
self.logger.warning(
|
|
'Malformed hypermedia link (%s) in \'Link\' '
|
|
'header (%s)', link, links
|
|
)
|
|
continue
|
|
else:
|
|
self.logger.debug('No more pages of notifications remain')
|
|
|
|
if self.failed_update:
|
|
return False
|
|
|
|
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):
|
|
previous_color = self.output.get('color')
|
|
try:
|
|
color = self.colors.get(
|
|
self.current_status,
|
|
self.unknown_color)
|
|
except TypeError:
|
|
# Shouldn't get here, but this would happen if this function is
|
|
# called before we check the current status for the first time.
|
|
color = previous_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
|