scripts/bot-run.py

   1 #!/usr/bin/env python3
   2 """
   3 Script for running a simple bot.
   4 """
   5 import json
   6 import subprocess
   7 from urllib.parse import urljoin
   8 from urllib.request import Request, urlopen
   9
  10
  11 class API:
  12     def __init__(self, base_url, token):
  13         self.base_url = base_url
  14         self.token = token
  15
  16     def request(self, url, method='GET', data=None):
  17         url = urljoin(self.base_url, url)
  18         if data:
  19             data = json.dumps(data).encode('utf-8')
  20         else:
  21             data = None
  22
  23         headers = {
  24                 "Content-type": "application/json",
  25         }
  26         if self.token:
  27             headers['Authorization'] = 'Token ' + self.token
  28
  29         req = Request(url, data=data, method=method, headers=headers)
  30         try:
  31             resp = urlopen(req)
  32         except Exception as e:
  33             print(e.reason)
  34             print(e.read().decode('utf-8'))
  35             raise
  36         else:
  37             return json.load(resp)
  38
  39     def my_chunks(self):
  40         me = self.request('me/')['id']
  41         return self.request('documents/chunks/?user={}'.format(me))
  42
  43
  44 def process_chunk(chunk, api, executable):
  45     print(chunk['id'])
  46     head = chunk['head']
  47     text = api.request(head)['text']
  48     text = text.encode('utf-8')
  49
  50     try:
  51         p = subprocess.run(
  52             [executable],
  53             input=text,
  54             capture_output=True,
  55             check=True
  56         )
  57     except subprocess.CalledProcessError as e:
  58         print('Ditching the update. Bot exited with error code {} and output:'.format(e.returncode))
  59         print(e.stderr.decode('utf-8'))
  60         return
  61     result_text = p.stdout.decode('utf-8')
  62     stderr_text = p.stderr.decode('utf-8')
  63     api.request(chunk['revisions'], 'POST', {
  64         "parent": head,
  65         "description": stderr_text or 'Automatic update.',
  66         "text": result_text
  67     })
  68     # Remove the user assignment.
  69     api.request(chunk['id'], 'PUT', {
  70         "user": None
  71     })
  72
  73
  74 if __name__ == '__main__':
  75     import argparse
  76     parser = argparse.ArgumentParser(
  77         description='Runs a bot for Redakcja. '
  78         'You need to provide an executable which will take current revision '
  79         'of text as stdin, and output the new version on stdout. '
  80         'Any output given on stderr will be used as revision description. '
  81         'If bot exits with non-zero return code, the update will be ditched.'
  82     )
  83     parser.add_argument(
  84         'token', metavar='TOKEN', help='A Redakcja API token.'
  85     )
  86     parser.add_argument(
  87         'executable', metavar='EXECUTABLE', help='An executable to run as bot.'
  88     )
  89     parser.add_argument(
  90         '--api', metavar='API', help='A base URL for the API.',
  91         default='https://redakcja.wolnelektury.pl/api/',
  92     )
  93     args = parser.parse_args()
  94
  95
  96     api = API(args.api, args.token)
  97
  98     chunks = api.my_chunks()
  99     if chunks:
 100         for chunk in api.my_chunks():
 101             process_chunk(chunk, api, args.executable)
 102     else:
 103         print('No assigned chunks found.')