Prod Release v0.1 Merge pull request #13 from Jackiebibili/develop

Prod Release v0.1

.gitignore

@@ -2,6 +2,7 @@
 /**/js/node_modules
 /**/js/antibot-simulation.js
 /**/tmp
+/**/.DS_Store
 .vscode
 docker-compose.yml
 entrypoint.sh

apps/pushnotification/constants.py

@@ -0,0 +1,7 @@
+from ...secret import MAILER_PW
+
+port = 465  # For starttls
+smtp_server = "smtp.gmail.com"
+sender_email = "noreply.ticketmasterbestseat@gmail.com"
+subject = "Message from Ticketmaster Ticket Tracker"
+app_password = MAILER_PW

apps/pushnotification/msg_formatter.py

@@ -0,0 +1,100 @@
+from datetime import datetime
+from ..ticketscraping.models.pick import Pick
+
+def price_formatter(price):
+   price_str = ''
+   if type(price) is float or int:
+      price_str = "$" + "{:.2f}".format(price)
+   elif type(price) is str:
+      price_str = price
+   return price_str
+
+def decimal_to_percent(num: float):
+   return "{:.2f}".format(num*100) + "%"
+
+def format_date(date: datetime):
+   return date.isoformat()
+
+def default_formatter(s: str):
+   return s
+
+def format_seat_columns(cols):
+   if type(cols) is str:
+      return cols
+   elif type(cols) is list:
+      return "(" + ",".join(cols) + ")"
+   return '-'
+
+def apply_format(s, formatter)->str:
+   return formatter(s)
+
+def apply(values: list, formatters: list, delimiter="\t"):
+   if len(values) != len(formatters):
+      raise Exception('values and formatters must have the same length')
+   s = []
+   for i in range(len(values)):
+      s.append(apply_format(values[i], formatters[i]))
+   return delimiter.join(s)
+
+def format_full_seat(seat: dict, delimiter="\t"):
+   price = seat.get("price", "n/a")
+   section = seat.get("section", "n/a")
+   row = seat.get("row", "n/a")
+   seat_columns = seat.get("seat_columns", "n/a")
+   last_modified = seat.get("last_modified", "n/a")
+   return apply(
+      [price, section, row, seat_columns, last_modified],
+      [price_formatter, default_formatter, default_formatter,
+       format_seat_columns, format_date],
+      delimiter)
+
+def format_price_only_seat(seat: dict, delimiter="\t"):
+   price = seat.get("price", "n/a")
+   last_modified = seat.get("last_modified", "n/a")
+   return apply([price, last_modified], [price_formatter, format_date], delimiter)
+
+def format_seat(seat: dict, price_only=False, delimiter="\t"):
+   if price_only:
+      return format_price_only_seat(seat, delimiter)
+   else:
+      return format_full_seat(seat, delimiter)
+
+def format_seats(seats: list, price_only=False, delimiter="\t"):
+   return "\n".join([format_seat(seat, price_only, delimiter) for seat in seats])
+
+
+def format_entire_mail(pick: Pick, target_price: int, percentile: float, rank: int, num_total: int, top_history_seats: list, same_seats: list):
+    """
+        structure of message:
+         1. greetings
+         2. attributes of new seats
+         3. top 3 comparable history seats
+         4. exact same seats if possible
+         5. signature
+   """
+    p1 = (
+        f"Hi!"
+    )
+    p2 = (
+        f"Congratulations! Ticket tracker reminds you that your ticket subscription request with target price {price_formatter(target_price)} "
+        f"found better budget seats (price, section, row, seats) at ({format_full_seat(vars(pick), delimiter=', ')}). "
+        f"{decimal_to_percent(percentile)} of all comparable seats in the history are better than the newly found seats, that is, "
+        f"they rank no.{rank} out of {num_total} comparable seats in the history."
+    )
+    p3 = (
+        f"You can compare to history seats that are better than the newly found seats:"
+        f"{chr(10)}"
+        f"{format_seats(top_history_seats, price_only=False)}"
+    ) if len(top_history_seats) > 0 else ""
+    p4 = (
+        f"The newly found seats have history prices:"
+        f"{chr(10)}"
+        f"{format_seats(same_seats, price_only=True)}"
+    ) if len(same_seats) > 0 else ""
+    p5 = (
+        f"Bests,"
+        f"{chr(10)}"
+        f"Ticketmaster Ticket Tracker"
+    )
+    paras = list(filter(lambda p: len(p) > 0, [p1, p2, p3, p4, p5]))
+    return "\n\n".join(paras)

apps/pushnotification/smtp.py

@@ -0,0 +1,44 @@
+from smtplib import SMTP_SSL
+from ssl import create_default_context
+from email.message import EmailMessage
+from . import constants
+
+
+def init_server():
+    context = create_default_context()
+    server = SMTP_SSL(constants.smtp_server, constants.port, context=context)
+    return server
+
+
+def server_login(server: SMTP_SSL, password: str):
+    return server.login(constants.sender_email, password)
+
+
+def server_send_email(server: SMTP_SSL, receiver_emails: list[str], message: str):
+    em = EmailMessage()
+    em['From'] = constants.sender_email
+    em['To'] = receiver_emails
+    em['subject'] = constants.subject
+
+    em.set_content(message)
+    return server.sendmail(constants.sender_email, receiver_emails, em.as_string())
+
+
+def send_email(receiver_emails: list[str], messages: list[str]):
+    if len(messages) == 0:
+        return
+    # print(messages[0])
+    try:
+        err = server_send_email(server, receiver_emails, messages[0])
+        if err is not None:
+            raise Exception('could not send email to the receiver')
+    except Exception as ex:
+        print(ex)
+
+
+server = init_server()
+
+
+def auth_server():
+    global server
+    server_login(server, constants.app_password)

apps/startup/apps.py

@@ -1,14 +1,39 @@
 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()
 
 
 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 apps.pushnotification.smtp import send_email, auth_server
+from apps.ticketscraping.constants import SERVICE_LOCALHOST, MAIL_RECEIVER_PORT
+
+def run():
+   # start itself
+   auth_server()
+   receiver = ReceiverProcess(send_email, 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()