add async task scheduler and mail scheduler

Author: Jackiebibili

apps/startup/apps.py

@@ -11,4 +11,5 @@ class MyAppConfig(AppConfig):
    def ready(self):
       print(
           f"server started at {datetime.now().strftime('%d/%m/%Y %H:%M:%S')}")
+      # start scraping
       Thread(target=start).start()

apps/storage/query.py

@@ -13,4 +13,4 @@ def find_min(collection_name, filter: dict, sort_key: str, db_name="tickets"):
 
 def find_many_ascending_order(collection_name, filter: dict, sort_key: str, db_name="tickets"):
    sort_seq = [(sort_key, pymongo.ASCENDING)]
-   return find_many(collection_name, filter, db_name=db_name, sort=sort_seq)
+   return find_many(collection_name, filter, db_name=db_name, sort=sort_seq)

apps/ticketscraping/constants.py

@@ -1,3 +1,4 @@
+from functools import reduce
 from uuid import uuid4
 
 ANTIBOT_JS_CODE_URL = "https://epsf.ticketmaster.com/eps-d"
@@ -18,20 +19,38 @@ def get_top_picks_url(
     "BEST_HISTORY_SEATS": "best-history-seats"
 }
 
+SUBSCRIBE_REQUEST_PROPS = {
+    'NAME': 'name',
+    'TARGET_PRICE': 'target_price',
+    'TOLERANCE': 'tolerance',
+    'TICKET_NUM': 'ticket_num',
+    'TM_EVENT_ID': 'tm_event_id'
+}
+
+def filter_obj_from_attrs(obj, atts: dict[str,str]):
+    res = {}
+    for key in atts.values():
+        if key in obj:
+            res[key] = obj[key]
+    return res
+
+
 # metric thresholds
+MINIMUM_HISTORY_DATA = 3
 PERCENT_OF_CHANGE = 0.5
-PERCENTILE_HISTORY_PRICES = 0.5
+PERCENTILE_HISTORY_PRICES = 0.25
+ALERT_SEATS_MAX_COUNT = 3
 
 def get_top_picks_header(): return {
     **BASIC_REQ_HEADER,
     "tmps-correlation-id": str(uuid4())
 }
 
-def get_top_picks_query_params(qty, priceInterval): return {
+def get_top_picks_query_params(qty: int, target_price: int, tolerance: int): return {
     'show': 'places maxQuantity sections',
     'mode': 'primary:ppsectionrow resale:ga_areas platinum:all',
     'qty': qty,
-    'q': f"and(not(\'accessible\'),any(listprices,$and(gte(@,{priceInterval[0]}),lte(@,{priceInterval[1]}))))",
+    'q': f"and(not(\'accessible\'),any(listprices,$and(gte(@,{target_price - tolerance}),lte(@,{target_price + tolerance}))))",
     'includeStandard': 'true',
     'includeResale': 'true',
     'includePlatinumInventoryType': 'false',

apps/ticketscraping/prepare_reese84token.py

@@ -6,7 +6,7 @@
 from . import constants
 
 
-def getReese84Token():
+def getReese84Token()->tuple[str, int]:
     def readFileContentToString(filename):
         f = open(filename, 'r')
         content = f.read()
@@ -19,7 +19,7 @@ def readFileContentToString(filename):
     # trim the code to the function that is only used
     match_obj = re.search(constants.FN_MATCHING_REGEX, antibot_js_code_full)
     if not match_obj:
-        return None
+        raise Exception('reese84 manufacture fails')
     start, end = match_obj.span()
     antibot_js_code_trim = antibot_js_code_full[start:end]
 
@@ -51,4 +51,5 @@ def readFileContentToString(filename):
     # invoke the get token api to get the reese84 token
     token_json_res = requests.post(
         constants.TOKEN_INTERROGATION_URL, headers=constants.BASIC_REQ_HEADER, json=token)
-    return token_json_res.json()
+    json_obj = token_json_res.json()
+    return json_obj['token'], json_obj['renewInSec']

apps/ticketscraping/schedulers/async_task_scheduler.py

@@ -0,0 +1,5 @@
+from .scheduler_process import SchedulerProcess
+
+print(__name__)
+async_tasks_scheduler = SchedulerProcess()
+async_tasks_scheduler.start()

apps/ticketscraping/schedulers/async_tasks_child_process.py

@@ -0,0 +1,12 @@
+import threading
+from multiprocessing import Queue
+from ..tasks.util.scheduler import Scheduler
+from ..tasks.asynchronous import run_async_tasks
+
+def child_process(queue: Queue):
+   scheduler = Scheduler('asynchronous tasks scheduler', run_async_tasks)
+   sched_thread = threading.Thread(target=scheduler.run)
+   sched_thread.start()
+   while True:
+      received_params = queue.get()
+      scheduler.enter(*received_params)

apps/ticketscraping/schedulers/mail_scheduler.py

@@ -0,0 +1,5 @@
+# from .scheduler_process import SchedulerProcess
+# from .mail_scheduler_child_process import child_process
+
+# mail_scheduler = SchedulerProcess(child_process)
+# mail_scheduler.start()

apps/ticketscraping/schedulers/mail_scheduler_child_process.py

@@ -0,0 +1,12 @@
+import threading
+from multiprocessing import Queue
+from ..tasks.util.scheduler import Scheduler
+
+def child_process(queue: Queue):
+   #TODO
+   scheduler = Scheduler('mail scheduler', lambda x: x)
+   sched_thread = threading.Thread(target=scheduler.run)
+   sched_thread.start()
+   while True:
+      received_params = queue.get()
+      scheduler.enter(*received_params)

apps/ticketscraping/schedulers/scheduler_process.py

@@ -0,0 +1,25 @@
+import threading
+from multiprocessing import Process, Queue
+from .async_tasks_child_process import child_process
+
+class SchedulerProcess:
+   def __init__(self):
+      self.queue = Queue()
+      self.started = False
+      self.starter_thread = None
+
+   def generate_child_process(self):
+      cp = Process(target=child_process, args=(self.queue,))
+      cp.start()
+      self.started = True
+      cp.join()
+
+   def start(self):
+      self.starter_thread = threading.Thread(
+          target=self.generate_child_process)
+      self.starter_thread.start()
+
+   def produce(self, *args):
+      if not self.started:
+         return
+      self.queue.put(args)

apps/ticketscraping/scraping.py

@@ -14,17 +14,18 @@
 class Reese84TokenUpdating():
     def __init__(self):
         self.is_running = False
-        self.reese84_token = {}
+        self.reese84_token = ''
+        self.reese84_renewInSec = 0
         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.reese84_token, self.reese84_renewInSec = getReese84Token()
         self.token_semaphore.release()  # produce a new token
-        self.scheduler.enter(self.reese84_token['renewInSec'] -
+        self.scheduler.enter(self.reese84_renewInSec -
                              constants.TOKEN_RENEW_SEC_OFFSET, constants.TOKEN_RENEW_PRIORITY, self.renew_reese84_token)
 
     def renew_reese84_token(self):
@@ -35,7 +36,7 @@ def renew_reese84_token(self):
         self.token_semaphore.acquire()  # invalidate a token
         self.reese84_token = getReese84Token()
         self.token_semaphore.release()
-        self.scheduler.enter(self.reese84_token['renewInSec'] -
+        self.scheduler.enter(self.reese84_renewInSec -
                              constants.TOKEN_RENEW_SEC_OFFSET, constants.TOKEN_RENEW_PRIORITY, self.renew_reese84_token)
 
     def start(self):
@@ -48,14 +49,15 @@ def start(self):
 
 
 class TicketScraping(threading.Thread):
-    def __init__(self, token_generator: Reese84TokenUpdating, event_id, subscribe_id, num_seats=2, price_range=(0, 200)):
+    def __init__(self, token_generator: Reese84TokenUpdating, event_id: str, subscribe_id: str, num_seats: int, target_price: int, tolerance: int):
         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
-        self.price_range = price_range
+        self.target_price = target_price
+        self.tolerance = tolerance
         self.token_gen = token_generator
         self.scheduler = sched.scheduler(time.time, time.sleep)
         self.initialDelay = random.randint(
@@ -83,17 +85,17 @@ def ticket_scraping(self):
         # scrape the top-picks from ticketmaster
         top_picks_url = constants.get_top_picks_url(self.event_id)
         top_picks_q_params = constants.get_top_picks_query_params(
-            self.num_seats, self.price_range)
+            self.num_seats, self.target_price, self.tolerance)
         top_picks_header = constants.get_top_picks_header()
         res = requests.get(top_picks_url, headers=top_picks_header, params=top_picks_q_params,
-                           cookies=dict(reese84=self.token_gen.reese84_token['token']))
-        # print(res.json())
+                           cookies=dict(reese84=self.token_gen.reese84_token))  # type: ignore
 
         # prune and format the received picks
         picks_obj = format_seats(res.json(), self.subscribe_id)
 
         # periodic task: update collections best_available_seats and best_history_seats
-        run_periodic_task(picks_obj, self.subscribe_id)
+        # and automatically spawn async tasks
+        run_periodic_task(picks_obj, self.subscribe_id, self.target_price)
 
         print("Got the ticket info from TM. /", res.status_code)
         self.scheduler.enter(constants.TICKET_SCRAPING_INTERVAL,
@@ -102,8 +104,8 @@ def ticket_scraping(self):
     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():
+            return self._thread_id  # type: ignore
+        for id, thread in threading._active.items():  # type: ignore
             if thread is self:
                 return id
 
@@ -135,7 +137,7 @@ def start():
                        '$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"])
+            reese_token_gen, evt["tm_event_id"], evt["_id"], evt["ticket_num"], evt["target_price"], evt["tolerance"])
         print(ticket_scraping.initialDelay, "s")
         scraping_list[ticket_scraping.subscribe_id] = ticket_scraping
     for scraping_thread in scraping_list.values():
@@ -156,7 +158,7 @@ def start():
                     # 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"])
+                        reese_token_gen, full_doc["tm_event_id"], full_doc["_id"], full_doc["ticket_num"], full_doc["target_price"], full_doc["tolerance"])
                     print(ticket_scraping.initialDelay, "s")
                     scraping_list[ticket_scraping.subscribe_id] = ticket_scraping
                     ticket_scraping.start()
@@ -177,7 +179,7 @@ def start():
                         # 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"])
+                                reese_token_gen, full_doc["tm_event_id"], full_doc["_id"], full_doc["ticket_num"], full_doc["target_price"], full_doc["tolerance"])
                             print(ticket_scraping.initialDelay, "s")
                             scraping_list[ticket_scraping.subscribe_id] = ticket_scraping
                             ticket_scraping.start()