From ffa9604027820982f700701359f68f151efbc317 Mon Sep 17 00:00:00 2001 From: Tristan Cacqueray Date: Wed, 8 Sep 2021 13:13:09 +0000 Subject: [PATCH] Introduce a BackendInterface This change replaces object inheritance with object composition to enable adding backend other than WikiPage. Change-Id: Id1a5453a1635f17c67594696d58e1d5c17b9566f --- statusbot/bot.py | 117 +++++++++++++++++++++++++++-------------------- 1 file changed, 67 insertions(+), 50 deletions(-) mode change 100644 => 100755 statusbot/bot.py diff --git a/statusbot/bot.py b/statusbot/bot.py old mode 100644 new mode 100755 index 1923eb9..e3de362 --- a/statusbot/bot.py +++ b/statusbot/bot.py @@ -78,7 +78,21 @@ irc.client.ServerConnection.buffer_class.errors = 'replace' ANTI_FLOOD_SLEEP = 2 -class WikiPage(object): +class BackendInterface(object): + def __init__(self, config): + pass + + def login(self): + pass + + def load(self) -> str: + pass + + def save(self, data: str): + pass + + +class WikiPage(BackendInterface): def __init__(self, config): self.url = config.get('wiki', 'url') self.pageid = config.get('wiki', 'pageid') @@ -97,11 +111,6 @@ class WikiPage(object): format='json')) return data['query']['pages'][str(self.pageid)]['revisions'][0]['*'] - def timestamp(self, ts=None): - if not ts: - ts = datetime.datetime.now() - return ts.strftime("%Y-%m-%d %H:%M:%S UTC") - def save(self, text): data = self.wiki.call(dict(action='query', prop='info', @@ -115,26 +124,36 @@ class WikiPage(object): token=token)) -class SuccessPage(WikiPage): - def __init__(self, config): - super(SuccessPage, self).__init__(config) - if config.has_option('wiki', 'successpageid'): - self.pageid = config.get('wiki', 'successpageid') - else: - self.pageid = None - if config.has_option('wiki', 'successpageurl'): - self.pageurl = config.get('wiki', 'successpageurl') - else: - self.pageurl = None +def timestamp(ts=None): + if not ts: + ts = datetime.datetime.now() + return ts.strftime("%Y-%m-%d %H:%M:%S UTC") + + +def get_opt(config, section, name): + """Return an option value when present or None""" + if config.has_option(section, name): + return config.get(section, name) + return None + + +class SuccessPage(object): + def __init__(self, config, backend): + self.backend = backend(config) + self.ready = False + if isinstance(self.backend, WikiPage): + self.backend.pageid = get_opt(config, 'wiki', 'successpageid') + self.backend.pageurl = get_opt(config, 'wiki', 'successpageurl') + self.ready = self.backend.pageid is not None if config.has_option('irclogs', 'url'): self.irclogs_url = config.get('irclogs', 'url') else: self.irclogs_url = None def log(self, channel, nick, msg): - if self.pageid: - self.login() - ts = self.timestamp() + if self.ready: + self.backend.login() + ts = timestamp() if self.irclogs_url: url = self.irclogs_url % { 'chan': urllib.parse.quote(channel), @@ -142,33 +161,29 @@ class SuccessPage(WikiPage): onchan = "[%s %s]" % (url, channel) else: onchan = channel - data = self.load() + data = self.backend.load() current = data.split("\n") newtext = "%s\n|-\n| %s || %s (on %s) || %s\n%s" % ( current[0], ts, nick, onchan, msg, '\n'.join(current[1:])) - self.save(newtext) + self.backend.save(newtext) -class ThanksPage(WikiPage): - def __init__(self, config): - super(ThanksPage, self).__init__(config) - if config.has_option('wiki', 'thankspageid'): - self.pageid = config.get('wiki', 'thankspageid') - else: - self.pageid = None - if config.has_option('wiki', 'thankspageurl'): - self.pageurl = config.get('wiki', 'thankspageurl') - else: - self.pageurl = None +class ThanksPage(object): + def __init__(self, config, backend): + self.backend = backend(config) + if isinstance(self.backend, WikiPage): + self.backend.pageid = get_opt(config, 'wiki', 'thankspageid') + self.backend.pageurl = get_opt(config, 'wiki', 'thankspageurl') + self.ready = self.backend.pageid is not None if config.has_option('irclogs', 'url'): self.irclogs_url = config.get('irclogs', 'url') else: self.irclogs_url = None def log(self, channel, nick, msg): - if self.pageid: - self.login() - ts = self.timestamp() + if self.ready: + self.backend.login() + ts = timestamp() if self.irclogs_url: url = self.irclogs_url % { 'chan': urllib.parse.quote(channel), @@ -176,11 +191,11 @@ class ThanksPage(WikiPage): onchan = "[%s %s]" % (url, channel) else: onchan = channel - data = self.load() + data = self.backend.load() current = data.split("\n") newtext = "%s\n|-\n| %s || %s (on %s) || %s\n%s" % ( current[0], ts, nick, onchan, msg, '\n'.join(current[1:])) - self.save(newtext) + self.backend.save(newtext) class UpdateInterface(object): @@ -229,12 +244,12 @@ class Tweet(UpdateInterface): self.update("Everything back to normal") -class StatusPage(WikiPage, UpdateInterface): +class StatusPage(UpdateInterface): alert_re = re.compile(r'{{CI Alert\|(.*?)}}') item_re = re.compile(r'^\* (.*)$') - def __init__(self, config): - super(StatusPage, self).__init__(config) + def __init__(self, config, backend): + self.backend = backend(config) self.current_alert = None self.items = [] @@ -251,7 +266,7 @@ class StatusPage(WikiPage, UpdateInterface): self.update(clear_alert=True, msg=msg) def update(self, set_alert=None, clear_alert=None, msg=None): - self.login() + self.backend.login() self.loadItems() if set_alert: self.setAlert(msg) @@ -264,7 +279,7 @@ class StatusPage(WikiPage, UpdateInterface): def loadItems(self): self.current_alert = None self.items = [] - text = self.load() + text = self.backend.load() for line in text.split('\n'): m = self.alert_re.match(line) if m: @@ -279,10 +294,10 @@ class StatusPage(WikiPage, UpdateInterface): text += '{{CI Alert|%s}}\n\n' % self.current_alert for item in self.items: text += '* %s\n' % item - self.save(text) + self.backend.save(text) def addItem(self, item, ts=None): - text = '%s %s' % (self.timestamp(ts=ts), item) + text = '%s %s' % (timestamp(ts=ts), item) self.items.insert(0, text) def setAlert(self, current_alert): @@ -345,7 +360,7 @@ class BaseStatusBot(SSL, irc.bot.SingleServerIRCBot): self.successlog.log(channel, nick, text) self.send(channel, "%s: Added success to Success page " "(%s)" - % (nick, self.successlog.pageurl)) + % (nick, self.successlog.backend.pageurl)) def handle_thanks_command(self, channel, nick, msg): parts = msg.split() @@ -354,7 +369,7 @@ class BaseStatusBot(SSL, irc.bot.SingleServerIRCBot): self.thankslog.log(channel, nick, text) self.send(channel, "%s: Added your thanks to Thanks page " "(%s)" - % (nick, self.thankslog.pageurl)) + % (nick, self.thankslog.backend.pageurl)) def handle_status_command(self, channel, nick, msg): parts = msg.split() @@ -495,14 +510,16 @@ def _main(configpath): config.read(configpath) setup_logging(config) + backend = WikiPage + channels = ['#' + name.strip() for name in config.get('ircbot', 'channels').split(',')] nicks = [name.strip() for name in config.get('ircbot', 'nicks').split(',')] - publishers = [StatusPage(config), + publishers = [StatusPage(config, backend), AlertFile(config)] - successlog = SuccessPage(config) - thankslog = ThanksPage(config) + successlog = SuccessPage(config, backend) + thankslog = ThanksPage(config, backend) if config.has_section('twitter'): publishers.append(Tweet(config))