add subscription watch and apis

Author: Jack Li

.gitignore

@@ -2,3 +2,4 @@
 /**/js/node_modules
 /**/js/antibot-simulation.js
 /**/tmp
+.vscode

startupApp/apps.py

@@ -1,6 +1,7 @@
 from django.apps import AppConfig
 from ticketscraping.scraping import start
 from datetime import datetime
+from threading import Thread
 
 
 class MyAppConfig(AppConfig):
@@ -11,4 +12,4 @@ def ready(self):
       print(
           f"server started at {datetime.now().strftime('%d/%m/%Y %H:%M:%S')}")
       print("=== database connection is established ===")
-      start()
+      Thread(target=start).start()

storage/base.py

@@ -15,8 +15,18 @@ def delete_many__(coll: collection.Collection, filter: dict):
    return coll.delete_many(filter)
 
 def find_one_and_replace__(coll: collection.Collection, filter: dict, new_doc: dict):
+   new_doc.update(gen_additional_attrs())
    return coll.find_one_and_replace(filter, new_doc)
 
+def find_one_and_update__(coll: collection.Collection, filter: dict, update=dict):
+   meta_attrs = get_meta_attrs()
+   key = "$set"
+   if key in update:
+      update[key].update(meta_attrs)
+   else:
+      update[key] = meta_attrs
+   return coll.find_one_and_update(filter, update)
+
 def find_one_and_delete__(coll: collection.Collection, filter: dict):
    return coll.find_one_and_delete(filter)
 
@@ -26,8 +36,16 @@ def find_one__(coll: collection.Collection, filter: dict, projection):
 def find_many__(coll: collection.Collection, filter: dict, projection, **kwargs):
    return coll.find(filter=filter, projection=projection, **kwargs)
 
+def watch__(coll: collection.Collection, **kwargs):
+   return coll.watch(**kwargs)
+
 def gen_additional_attrs():
    return {
        "_id": str(uuid4()),
+       **get_meta_attrs()
+   }
+
+def get_meta_attrs():
+   return {
        "last_modified": datetime.today().replace(microsecond=0)
    }

storage/storage.py

@@ -36,6 +36,12 @@ def find_one_and_replace(collection_name, filter: dict, new_doc: dict, db_name="
    coll = db[collection_name]
    return find_one_and_replace__(coll, filter, new_doc)
 
+# find one and update
+def find_one_and_update(collection_name, filter: dict, update, db_name="tickets"):
+   db = get_db_handle(db_name)
+   coll = db[collection_name]
+   return find_one_and_update__(coll, filter, update)
+
 # find one and delete
 def find_one_and_delete(collection_name, filter: dict, db_name="tickets"):
    db = get_db_handle(db_name)
@@ -52,4 +58,11 @@ def find_one(collection_name, filter: dict, projection=None, db_name="tickets"):
 def find_many(collection_name, filter: dict, projection=None, db_name="tickets", **kwargs):
    db = get_db_handle(db_name)
    coll = db[collection_name]
-   return list(find_many__(coll, filter, projection, **kwargs))
+   return list(find_many__(coll, filter, projection, **kwargs))
+
+# watch changes
+def watch(collection_name, db_name="tickets", **kwargs):
+   db = get_db_handle(db_name)
+   coll = db[collection_name]
+   return watch__(coll, **kwargs)
+

ticketscraping/scraping.py

@@ -1,26 +1,28 @@
 import sched
 import time
+import ctypes
 import random
 import requests
-from threading import Thread
+import threading
 from . import constants
 from threading import Semaphore
 from .prepare_reese84token import getReese84Token
 from storage.storage import *
 
+
 class Reese84TokenUpdating():
     def __init__(self):
         self.is_running = False
         self.reese84_token = {}
         self.token_semaphore = Semaphore(0)
         self.scheduler = sched.scheduler(time.time, time.sleep)
-    
+
     def initialize_reese84_token(self):
         """
         This method should not be called directly.
         """
         self.reese84_token = getReese84Token()
-        self.token_semaphore.release() # produce a new token
+        self.token_semaphore.release()  # produce a new token
         self.scheduler.enter(self.reese84_token['renewInSec'] -
                              constants.TOKEN_RENEW_SEC_OFFSET, constants.TOKEN_RENEW_PRIORITY, self.renew_reese84_token)
 
@@ -29,24 +31,26 @@ def renew_reese84_token(self):
         This method should not be called directly.
         """
         print("renewing token")
-        self.token_semaphore.acquire() # invalidate a token
+        self.token_semaphore.acquire()  # invalidate a token
         self.reese84_token = getReese84Token()
         self.token_semaphore.release()
         self.scheduler.enter(self.reese84_token['renewInSec'] -
                              constants.TOKEN_RENEW_SEC_OFFSET, constants.TOKEN_RENEW_PRIORITY, self.renew_reese84_token)
-    
+
     def start(self):
         # if the scheduler is already started - do nothing
-        if self.is_running: return
+        if self.is_running:
+            return
         self.is_running = True
         self.initialize_reese84_token()
         self.scheduler.run()
-    
 
 
-class TicketScraping():
+class TicketScraping(threading.Thread):
     def __init__(self, token_generator: Reese84TokenUpdating, event_id, subscribe_id, num_seats=2, price_range=(0, 200)):
+        threading.Thread.__init__(self)
         self.is_running = False
+        self.is_stopping = False
         self.event_id = event_id
         self.subscribe_id = subscribe_id
         self.num_seats = num_seats
@@ -56,6 +60,19 @@ def __init__(self, token_generator: Reese84TokenUpdating, event_id, subscribe_id
         self.initialDelay = random.randint(
             1, constants.TICKET_SCRAPING_INTERVAL)
 
+    def flag_for_termination(self):
+        # cancel all scheduled jobs
+        list(map(self.scheduler.cancel, self.scheduler.queue))
+        # raise exception to terminate the thread
+        thread_id = self.get_id()
+        res = ctypes.pythonapi.PyThreadState_SetAsyncExc(thread_id,
+                                                         ctypes.py_object(SystemExit))
+        print(
+            f"Ticket scraping with subscription id={self.subscribe_id} marked for termination.")
+        if res > 1:
+            ctypes.pythonapi.PyThreadState_SetAsyncExc(thread_id, 0)
+            print(f'Failed to terminate the thread with id={thread_id}')
+
     def ticket_scraping(self):
         if self.token_gen.token_semaphore._value <= 0:
             # retry after a delay
@@ -71,34 +88,89 @@ def ticket_scraping(self):
         # print(res.json())
         print("Got the ticket info from TM. /", res.status_code)
         self.scheduler.enter(constants.TICKET_SCRAPING_INTERVAL,
-                  constants.TICKET_SCRAPING_PRIORITY, self.ticket_scraping)
+                             constants.TICKET_SCRAPING_PRIORITY, self.ticket_scraping)
 
-    def start(self):
-        # if the scheduler is already started - do nothing
-        if self.is_running:
-            return
-        self.is_running = True
-        self.ticket_scraping()
-        # randomize start time to scatter out event of API fetching
-        time.sleep(self.initialDelay) 
-        self.scheduler.run()
+    def get_id(self):
+        # returns id of the respective thread
+        if hasattr(self, '_thread_id'):
+            return self._thread_id
+        for id, thread in threading._active.items():
+            if thread is self:
+                return id
+
+    def run(self):
+        try:
+            # if the scheduler is already started - do nothing
+            if self.is_running:
+                return
+            self.is_running = True
+            self.is_stopping = False
+            self.ticket_scraping()
+            # randomize start time to scatter out event of API fetching
+            time.sleep(self.initialDelay)
+            self.scheduler.run()
+        finally:
+            print(
+                f"Ticket scraping with subscription id={self.subscribe_id} has been terminated.")
 
 
 def start():
     # reese84 token renewing thread
     reese_token_gen = Reese84TokenUpdating()
-    serverThread_reese = Thread(target=reese_token_gen.start)
+    serverThread_reese = threading.Thread(target=reese_token_gen.start)
     serverThread_reese.start()
 
     # ticket scraping threads
-    scraping_list = []
-    events = find_many(constants.DATABASE["EVENTS"], {})
-
+    scraping_list = dict()
+    events = find_many(constants.DATABASE["EVENTS"], {
+                       '$or': [{'markPaused': {'$exists': False}}, {'markPaused': False}]})
     for evt in events:
-        ticket_scraping = TicketScraping(reese_token_gen, evt["tm_event_id"], evt["_id"], evt["ticket_num"], evt["price_range"])
+        ticket_scraping = TicketScraping(
+            reese_token_gen, evt["tm_event_id"], evt["_id"], evt["ticket_num"], evt["price_range"])
         print(ticket_scraping.initialDelay, "s")
-        serverThread_ticket_scraping = Thread(target=ticket_scraping.start)
-        scraping_list.append(serverThread_ticket_scraping)
-
-    for scraping_thread in scraping_list:
+        scraping_list[ticket_scraping.subscribe_id] = ticket_scraping
+    for scraping_thread in scraping_list.values():
         scraping_thread.start()
+
+    # listen for changes in ticket scraping subscriptions
+    while(True):
+        with watch(constants.DATABASE["EVENTS"], pipeline=[{'$match': {'operationType': {'$in': ['delete', 'insert', 'replace', 'update']}}}], full_document='updateLookup') as stream:
+            for change in stream:
+                if change['operationType'] == "delete":
+                    # stop the thread
+                    doc_id = change['documentKey']['_id']
+                    if doc_id in scraping_list:
+                        scraping_obj = scraping_list[doc_id]
+                        scraping_obj.flag_for_termination()
+                        del scraping_list[doc_id]
+                elif change['operationType'] == "insert":
+                    # spawn a thread to do scraping operations
+                    full_doc = change['fullDocument']
+                    ticket_scraping = TicketScraping(
+                        reese_token_gen, full_doc["tm_event_id"], full_doc["_id"], full_doc["ticket_num"], full_doc["price_range"])
+                    print(ticket_scraping.initialDelay, "s")
+                    scraping_list[ticket_scraping.subscribe_id] = ticket_scraping
+                    ticket_scraping.start()
+                else:
+                    # replace or update - pause or resume ticket scraping
+                    full_doc = change['fullDocument']
+                    doc_id = full_doc['_id']
+                    if not 'markPaused' in full_doc:
+                        print("'markPaused' flag is unset, skip processing.")
+                        break
+                    if full_doc['markPaused'] == True:
+                        # pause scraping
+                        if doc_id in scraping_list:
+                            scraping_obj = scraping_list[doc_id]
+                            scraping_obj.flag_for_termination()
+                            del scraping_list[doc_id]
+                    else:
+                        # resume scraping if currently paused
+                        if doc_id not in scraping_list:
+                            ticket_scraping = TicketScraping(
+                                reese_token_gen, full_doc["tm_event_id"], full_doc["_id"], full_doc["ticket_num"], full_doc["price_range"])
+                            print(ticket_scraping.initialDelay, "s")
+                            scraping_list[ticket_scraping.subscribe_id] = ticket_scraping
+                            ticket_scraping.start()
+                # display current number of ticket scraping
+                print(f"{len(scraping_list)} ticket scraping threads now.")

trackerapi/urls.py

@@ -2,5 +2,6 @@
 from . import views
 
 urlpatterns = [
-   path('subscribe/', views.subscribe_tm_event_price_tracking)
+   path('subscribe/', views.subscribe_tm_event_price_tracking),
+   path('unsubscribe/', views.unsubscribe_tm_event_price_tracking)
 ]

trackerapi/views.py

@@ -1,6 +1,6 @@
 from django.shortcuts import render
 from django.http import HttpResponse, HttpRequest
-from storage.storage import insert_one
+from storage.storage import insert_one, find_one_and_update
 import ticketscraping.constants as constants
 import json
 
@@ -21,3 +21,16 @@ def subscribe_tm_event_price_tracking(req: HttpRequest):
       }
       insert_one(constants.DATABASE['EVENTS'], doc)
       return HttpResponse('OK', status=200)
+
+def unsubscribe_tm_event_price_tracking(req: HttpRequest):
+   if req.method == 'POST':
+      body = json.loads(req.body)
+      # validation
+      id = body['subscription_id']
+      if not id:
+         return HttpResponse('Request is invalid.', status=400)
+      res = find_one_and_update(constants.DATABASE['EVENTS'], {"_id": id}, {'$set': {'markPaused': True, }})
+      if res:
+         return HttpResponse('OK', status=200)
+      else:
+         return HttpResponse('Subscription id not found.', status=400)