vt_decor.py

   1 import base64
   2 import inspect
   3 import ipaddress
   4 import re
   5 import textwrap
   6 import time
   7 import sys
   8 import weakref
   9
  10 from .. import chunks, messages, roost, util, zcode
  11
  12 ## Config
  13 # I'm not sure why I can't just use the standard form of
  14 # _enabled = util.Configurable(...) here, but this works.
  15
  16 SETTINGS = {
  17     'enabled': True,
  18     'min_width': 100,
  19     'time': 'receiveTime',
  20     'show_seconds': False,
  21     'show_addr': False,
  22     'show_signatures': False,
  23 }
  24
  25 def _gen_setter(key):
  26     def setter(_context, value):
  27         SETTINGS[key] = value
  28     return setter
  29
  30 util.Configurable(
  31     'user.vt_decor.enabled',
  32     True,
  33     action=_gen_setter('enabled'),
  34     oneof=('True', 'False'),
  35     validate=lambda val: isinstance(val, bool),
  36     coerce=util.coerce_bool,
  37     )
  38
  39 util.Configurable(
  40     'user.vt_decor.time',
  41     'receiveTime',
  42     action=_gen_setter('time'),
  43     oneof=('time', 'receiveTime'),
  44     )
  45
  46 util.Configurable(
  47     'user.vt_decor.show_seconds',
  48     False,
  49     action=_gen_setter('show_seconds'),
  50     oneof=('True', 'False'),
  51     validate=lambda val: isinstance(val, bool),
  52     coerce=util.coerce_bool,
  53     )
  54
  55 util.Configurable(
  56     'user.vt_decor.show_addr',
  57     False,
  58     action=_gen_setter('show_addr'),
  59     oneof=('True', 'False'),
  60     validate=lambda val: isinstance(val, bool),
  61     coerce=util.coerce_bool,
  62     )
  63
  64 util.Configurable(
  65     'user.vt_decor.show_signatures',
  66     False,
  67     action=_gen_setter('show_signatures'),
  68     oneof=('True', 'False'),
  69     validate=lambda val: isinstance(val, bool),
  70     coerce=util.coerce_bool,
  71     )
  72
  73
  74 ## Utils
  75
  76 UI = lambda: None
  77
  78 def _get_width():
  79     global UI
  80     try:
  81         return UI().maxx
  82     except AttributeError:
  83         pass
  84
  85     frame = inspect.currentframe()
  86     try:
  87         while True:
  88             frame = frame.f_back
  89             try:
  90                 ui = frame.f_locals['self'].ui
  91                 UI = weakref.ref(ui)
  92                 return ui.maxx
  93             except (KeyError, AttributeError):
  94                 pass
  95     finally:
  96         del frame
  97
  98
  99 ## The fun part
 100
 101 MSG_NO_WRAP_PATTERNS = [re.compile(x) for x in (
 102     r'\n(?: |>)',
 103     r'\t',
 104     r'[^.]  ',
 105 )]
 106
 107 REALM_MAP = {
 108     'ANDREW.CMU.EDU': 'AN',
 109     'CS.CMU.EDU': 'CS',
 110     'IASTATE.EDU': 'IA',
 111 }
 112
 113 class RoostVTDecor(roost.RoostMessage.Decor):
 114
 115
 116     @classmethod
 117     def decorate(cls, msg, decoration):
 118         try:
 119             if SETTINGS['enabled'] and _get_width() >= SETTINGS['min_width']:
 120                 return cls.vt_decorate(msg, decoration)
 121         except Exception:
 122             pass
 123         return super().decorate(msg, decoration)
 124
 125     @staticmethod
 126     def rewrap_p(msg):
 127         force_wrap_tuples = {
 128             ('moira', 'incremental'),
 129             ('scripts', 'nagios.multivalue-key.mysql-s'),
 130             ('scripts', 'nagios.unique-key.mysql-s'),
 131         }
 132
 133         msg_tuples = {
 134             (msg.data['classKey'], '*'),
 135             (msg.data['classKey'], msg.data['instanceKey']),
 136         }
 137
 138         return (bool(force_wrap_tuples & msg_tuples) or
 139                 not any(x.search(msg.body) for x in MSG_NO_WRAP_PATTERNS))
 140
 141
 142     @classmethod
 143     def vt_decorate(cls, msg, decoration):
 144         width = _get_width()
 145
 146         tags = cls.decotags(decoration)
 147
 148         realm = msg.backend.realm
 149         sender = msg.data['sender']
 150         recipient = msg.data['recipient']
 151         if msg.data['isPersonal'] and msg.data['isOutgoing']:
 152             sender = f'➤{recipient}'
 153         if sender.endswith(realm):
 154             sender = sender[:sender.index('@')]
 155         klass = msg.data['class'] if msg.data['classKey'] != 'message' else ''
 156         inst = msg.data['instance'] or "''"
 157         if klass and inst.lower() == 'personal':
 158             dest = klass
 159         else:
 160             dest = f'{klass}[{inst}]'
 161         auth = '+' if msg.data['auth'] else '-'
 162         if not msg.data['auth'] and  msg.data['opcode'] == 'mattermost':
 163             auth = '¤'
 164         t = time.strftime(
 165             '%H:%M:%S' if SETTINGS['show_seconds'] else '%H:%M',
 166             time.localtime(msg.data[SETTINGS['time']] / 1000))
 167
 168         if recipient.startswith('@'):
 169             mrealm = REALM_MAP.get(recipient[1:], '??')
 170             dest = f'{mrealm} {dest}'
 171
 172         dest_width = 18 - (util.glyphwidth(dest) - len(dest))
 173
 174         prefix = f'{sender:10.10} {t} {dest:{dest_width}.{dest_width}} {auth} '
 175         indent = ' ' * util.glyphwidth(prefix)
 176
 177         body = msg.body.rstrip()
 178         if cls.rewrap_p(msg):
 179             body = textwrap.fill(body, width,
 180                                  initial_indent=indent,
 181                                  subsequent_indent=indent,
 182                                  break_long_words=False,
 183                                  break_on_hyphens=False).lstrip()
 184         else:
 185             body = textwrap.indent(body, indent).lstrip()
 186
 187         # body = f'{list(decoration.keys())}\n'
 188         if msg.backend.format_body == 'format':
 189             cbody = zcode.tag(body, frozenset(tags))
 190         elif msg.backend.format_body == 'clear':
 191             cbody = chunks.Chunk([(tags, '')])
 192         else:
 193             if msg.backend.format_body == 'strip':
 194                 body = zcode.strip(body)
 195             cbody = chunks.Chunk([(tags, body)])
 196
 197
 198         chunk = chunks.Chunk([(tags, f'{prefix}')]) + cbody
 199
 200         if SETTINGS['show_signatures'] and msg.backend.format_zsig != 'clear':
 201             zsig = '\n' + textwrap.fill(
 202                 msg.data['signature'], width,
 203                 initial_indent='                  -- ',
 204                 subsequent_indent='                     ',
 205                 break_long_words=False,
 206                 break_on_hyphens=False)
 207
 208             if msg.backend.format_zsig == 'format':
 209                 chunk += zcode.tag(zsig, frozenset(tags))
 210             elif msg.backend.format_zsig == 'strip':
 211                 chunk.append((tags, zcode.strip(zsig)))
 212             else:
 213                 chunk.append((tags, zsig))
 214
 215         chunk.append(
 216             (tags | {'right'},
 217              f'## {ipaddress.ip_address(base64.b64decode(msg.data["uid"])[:4])}'
 218              if SETTINGS['show_addr'] else ''
 219             ))
 220
 221         return chunk