Merge pull request #15 from Jackiebibili/schedulers

add async task scheduler and mail scheduler and synchronize startup process

Author: FrankQixiangGao

apps/startup/apps.py

@@ -1,14 +1,40 @@
 from django.apps import AppConfig
-from ..ticketscraping.scraping import start
 from datetime import datetime
 from threading import Thread
+from multiprocessing import Process
+
+
+def run_prepare():
+    # import module inside the child process to prevent execution in the parent process
+    print(
+        f"ticket scraping service started at {datetime.now().strftime('%d/%m/%Y %H:%M:%S')}")
+
+    # start sender socket
+    from apps.ticketscraping.schedulers.async_tasks_scheduler import async_tasks_scheduler
+    conn_thread = Thread(target=async_tasks_scheduler.connect)
+    conn_thread.start()
+    # wait for async tasks handler to connect
+    conn_thread.join()
+
+    # start itself (scraping)
+    from apps.ticketscraping.scraping import start
+    start()
+
+
+def run():
+    # starter
+    p = Process(target=run_prepare, daemon=True)
+    p.start()
+    # start receiver socket
+    from apps.ticketscraping.connection.asyn_tasks_receiver import run
+    conn_process = Process(target=run)
+    conn_process.start()
+    conn_process.join()
 
 
 class MyAppConfig(AppConfig):
    name = "apps.startup"
    verbose_name = "start tmtracker"
 
    def ready(self):
-      print(
-          f"server started at {datetime.now().strftime('%d/%m/%Y %H:%M:%S')}")
-      Thread(target=start).start()
+        run()

apps/storage/base.py

@@ -30,12 +30,18 @@ def find_one_and_update__(coll: collection.Collection, filter: dict, update=dict
 def find_one_and_delete__(coll: collection.Collection, filter: dict):
    return coll.find_one_and_delete(filter)
 
-def find_one__(coll: collection.Collection, filter: dict, projection):
-   return coll.find_one(filter=filter, projection=projection)
+def find_one__(coll: collection.Collection, filter: dict, projection, **kwargs):
+   return coll.find_one(filter=filter, projection=projection, **kwargs)
 
 def find_many__(coll: collection.Collection, filter: dict, projection, **kwargs):
    return coll.find(filter=filter, projection=projection, **kwargs)
 
+def count_docs__(coll: collection.Collection, filter: dict):
+   return coll.count_documents(filter=filter)
+
+def estimated_count_docs__(coll: collection.Collection):
+   return coll.estimated_document_count()
+
 def watch__(coll: collection.Collection, **kwargs):
    return coll.watch(**kwargs)
 

apps/storage/query.py

@@ -0,0 +1,16 @@
+from .storage import *
+import pymongo
+
+# find the max value in a collection
+def find_max(collection_name, filter: dict, sort_key: str, db_name="tickets"):
+   sort_seq = [(sort_key, pymongo.DESCENDING)]
+   return find_one(collection_name, filter, db_name=db_name, sort=sort_seq)
+
+# find the min value in a collection
+def find_min(collection_name, filter: dict, sort_key: str, db_name="tickets"):
+   sort_seq = [(sort_key, pymongo.ASCENDING)]
+   return find_one(collection_name, filter, db_name=db_name, sort=sort_seq)
+
+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)

apps/storage/storage.py

@@ -51,17 +51,29 @@ def find_one_and_delete(collection_name, filter: dict, db_name="tickets"):
    return find_one_and_delete__(coll, filter)
 
 # find one
-def find_one(collection_name, filter: dict, projection=None, db_name="tickets"):
+def find_one(collection_name, filter: dict, projection=None, db_name="tickets", **kwargs):
    db = get_db_handle(db_name)
    coll = db[collection_name]
-   return find_one__(coll, filter, projection)
+   return find_one__(coll, filter, projection, **kwargs)
 
 # find many
 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))
 
+# count with filter
+def count_docs(collection_name, filter: dict, db_name="tickets"):
+   db = get_db_handle(db_name)
+   coll = db[collection_name]
+   return count_docs__(coll, filter)
+
+# count all docs in a collection
+def estimated_count_docs(collection_name, db_name="tickets"):
+   db = get_db_handle(db_name)
+   coll = db[collection_name]
+   return estimated_count_docs__(coll)
+
 # watch changes
 def watch(collection_name, db_name="tickets", **kwargs):
    db = get_db_handle(db_name)

apps/ticketscraping/connection/asyn_tasks_receiver.py

@@ -0,0 +1,35 @@
+# start sockets
+from threading import Thread
+from multiprocessing import Process
+from apps.ticketscraping.connection.receiver_process import ReceiverProcess
+from apps.ticketscraping.constants import SERVICE_LOCALHOST, ASYNC_TASKS_RECEIVER_PORT
+
+
+def run_prepare():
+   # start receiver socket
+   from apps.ticketscraping.connection.mail_receiver import run
+   conn_process = Process(target=run, daemon=True)
+   conn_process.start()
+
+   # start sender socket
+   from apps.ticketscraping.schedulers.mail_scheduler import mail_scheduler
+   conn_thread = Thread(target=mail_scheduler.connect)
+   conn_thread.start()
+   # wait for mailer to connect
+   conn_thread.join()
+
+   # start itself
+   from apps.ticketscraping.tasks.asynchronous import run_async_tasks
+   receiver = ReceiverProcess(run_async_tasks, SERVICE_LOCALHOST, ASYNC_TASKS_RECEIVER_PORT)
+   receiver.connect()
+   receiver.serve_forever()
+
+
+def run():
+   # starter
+   p = Process(target=run_prepare)
+   p.start()
+
+
+if __name__ == '__main__':
+   run()

apps/ticketscraping/connection/mail_receiver.py

@@ -0,0 +1,13 @@
+from apps.ticketscraping.connection.receiver_process import ReceiverProcess
+# from ..tasks.asynchronous import run_async_tasks
+from apps.ticketscraping.constants import SERVICE_LOCALHOST, MAIL_RECEIVER_PORT
+
+def run():
+   # start itself
+   receiver = ReceiverProcess(lambda x: print(
+       x), SERVICE_LOCALHOST, MAIL_RECEIVER_PORT)
+   receiver.connect()
+   receiver.serve_forever()
+
+if __name__ == '__main__':
+   run()

apps/ticketscraping/connection/receiver.py

@@ -0,0 +1,24 @@
+from multiprocessing.connection import Client
+from threading import Semaphore
+
+class Receiver:
+   def __init__(self, hostname: str, port: int):
+      self.lock = Semaphore(1)
+      self.hostname = hostname
+      self.port = port
+      self.conn = None
+
+   def connect(self):
+      self.conn = Client(address=(self.hostname, self.port,))
+
+   def recv(self):
+      if self.conn is None:
+         raise Exception('connection is not established')
+      self.lock.acquire()
+      res = self.conn.recv()
+      self.lock.release()
+      return res
+
+   def __del__(self):
+      if self.conn is not None: self.conn.close()
+

apps/ticketscraping/connection/receiver_process.py

@@ -0,0 +1,11 @@
+from .receiver import Receiver
+
+class ReceiverProcess(Receiver):
+   def __init__(self, action, hostname: str, port: int):
+      super().__init__(hostname, port)
+      self.action = action
+
+   def serve_forever(self):
+      while True:
+         res = self.recv()
+         self.action(*res)

apps/ticketscraping/connection/sender.py

@@ -0,0 +1,26 @@
+from multiprocessing.connection import Listener
+from threading import Semaphore
+
+class Sender:
+   def __init__(self, hostname: str, port: int):
+      self.lock = Semaphore(1)
+      self.hostname = hostname
+      self.port = port
+      self.conn = None
+
+   def connect(self):
+      listener = Listener(address=(self.hostname, self.port))
+      self.conn = listener.accept()
+      print("conn accepted ", self.port)
+
+   def send(self, *args):
+      if self.conn is None:
+         raise Exception('connection is not established')
+      self.lock.acquire()
+      self.conn.send(args)
+      self.lock.release()
+      return True
+
+   def __del__(self):
+      if self.conn is not None:
+         self.conn.close()

apps/ticketscraping/constants.py

@@ -1,5 +1,10 @@
 from uuid import uuid4
 
+# services - async action handlers
+ASYNC_TASKS_RECEIVER_PORT = 8100
+MAIL_RECEIVER_PORT = 8200
+SERVICE_LOCALHOST = 'localhost'
+
 ANTIBOT_JS_CODE_URL = "https://epsf.ticketmaster.com/eps-d"
 TOKEN_INTERROGATION_URL = "https://epsf.ticketmaster.com/eps-d?d=www.ticketmaster.com"
 
@@ -17,16 +22,39 @@ def get_top_picks_url(
     "BEST_AVAILABLE_SEATS": "best-available-seats",
     "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.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',