Reconnect every 15 minutes

Author: Xi-Plus

rc.py

@@ -62,13 +62,10 @@
 
 
 @celery.task(name='wmrcm.process', queue='wmrcm')
-def process(rawdata):
-    logging.info(rawdata)
-    try:
-        change = json.loads(rawdata)
-    except json.decoder.JSONDecodeError as e:
-        msg = 'UnicodeDecodeError: {}. {}'.format(e, rawdata)
-        logging.error(msg)
+def process(change):
+    logging.info(change)
+
+    if change['meta']['domain'] == 'canary':
         return
 
     try:

rc_abusefilter.py

@@ -117,9 +117,7 @@ def tz2int(timetz):
                 'parsedcomment': log['parsedcomment']
             }
 
-            data = json.dumps(change)
-            data = data.encode('utf-8')
-            process(data)
+            process(change)
 
         if len(res['query']['logevents']) > 0:
             timestamp = res['query']['logevents'][-1]['timestamp']

rc_abuselog.py

@@ -196,9 +196,7 @@ def tz2int(timetz):
 
             newlog['timestamp'] = tz2int(newlog['timestamp'])
 
-            data = json.dumps(newlog)
-            data = data.encode('utf-8')
-            process(data)
+            process(newlog)
 
         if len(res["query"]["abuselog"]) > 0:
             timestamp = res["query"]["abuselog"][-1]["timestamp"]

rc_newusers_autocreate.py

@@ -127,9 +127,7 @@ def tz2int(timetz):
                 'parsedcomment': log['parsedcomment']
             }
 
-            data = json.dumps(change)
-            data = data.encode('utf-8')
-            process(data)
+            process(change)
 
         if len(res['query']['logevents']) > 0:
             timestamp = res['query']['logevents'][-1]['timestamp']

rc_stream.py

@@ -1,19 +1,23 @@
 # -*- coding: utf-8 -*-
 import argparse
+import json
 import logging
+import multiprocessing
 import os
 import socket
 import time
 import traceback
+from datetime import datetime, timezone
 
-from Monitor import Monitor, MonitorLogHandler
-from rc_config import SOCKET_HOST, SOCKET_PORT
 from sseclient import SSEClient as EventSource
-from rc import process
 
+from Monitor import Monitor, MonitorLogHandler
+from rc import process
+from rc_config import SOCKET_HOST, SOCKET_PORT
 
 parser = argparse.ArgumentParser()
 parser.add_argument('--sleep', type=int, default=60)
+parser.add_argument('--timeout', type=int, default=900)
 parser.add_argument('-d', '--debug', action='store_const', dest='loglevel', const=logging.DEBUG)
 parser.add_argument('-v', '--verbose', action='store_const', dest='loglevel', const=logging.INFO)
 parser.set_defaults(loglevel=logging.WARNING)
@@ -29,7 +33,7 @@
 
 os.environ['TZ'] = 'UTC'
 
-url = 'https://stream.wikimedia.org/v2/stream/recentchange'
+URL = 'https://stream.wikimedia.org/v2/stream/recentchange'
 
 sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
 sock.connect((SOCKET_HOST, SOCKET_PORT))
@@ -39,23 +43,42 @@
     'User-Agent': M.wp_user_agent,
 }
 
-errorWaitTime = 1
-while True:
-    try:
-        for event in EventSource(url, headers=headers):
-            if event.event == 'message':
-                if len(event.data) == 0:
-                    continue
 
-                logging.debug(event.data)
+def main(lastOffset):
+    fullUrl = URL
+    if lastOffset.value > 0:
+        fullUrl += '?since={}'.format(datetime.fromtimestamp(lastOffset.value, tz=timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"))
+    for event in EventSource(fullUrl, headers=headers):
+        if event.event == 'message':
+            if len(event.data) == 0:
+                continue
+
+            logging.debug(event.data)
+
+            try:
+                change = json.loads(event.data)
+            except json.decoder.JSONDecodeError as e:
+                msg = 'UnicodeDecodeError: {}. {}'.format(e, event.data)
+                logging.error(msg)
+                continue
+            process(change)
+            lastOffset.value = int(change['timestamp'])
 
-                noError = True
 
-                data = event.data.encode('utf-8')
-                process(data)
+lastOffset = multiprocessing.Value('i', 0)
+while True:
+    try:
+        print('Last-Event-ID', lastOffset.value)
+        p = multiprocessing.Process(target=main, args=(lastOffset,))
+        p.start()
+        p.join(timeout=args.timeout)
+
+        if p.is_alive():
+            p.terminate()
+            p.join()
 
     except Exception as e:
         traceback.print_exc()
         M.error(traceback.format_exc(), noRaise=True)
 
-    time.sleep(args.sleep)
+        time.sleep(args.sleep)