Merge pull request #2 from AvocadoStyle/infra/message_infrastructure_…

…creation

#1 Will use this branch as the message infrastruture branch

main.py

@@ -14,16 +14,19 @@
  user_name = T["user_name"]
  password = T["password"]
  # ok = OkCupidSelenium(user_name, password, PATH)
- # contain = "*משחק אותה שלא מתלהב בכלל* מה קורה?"
+ contain = "*משחק אותה שלא מתלהב בכלל* מה קורה?"
  # contain = "*משחק אותה שלא מתלהב* מה קורה?"
- contain = "את ממש מוכרת לי"
+ # contain = "את ממש מוכרת לי"
  people = 50000
  # gets the cache data path of the user, And if there isn't it will retrieve `None`, And it's valid behavior.
+ # such as the path: Chrome\User Data\Profile 1
  usr_cached_data = T.get('user_cache_data')
- ok = OkCupidSelenium(user_name, password, driver_path, okcupid_url, '')
+ ok = OkCupidSelenium(
+ user_name=user_name, password=password, driver_path=driver_path, url=okcupid_url,
+ user_cached_data=usr_cached_data)
  people_max_qty_to_msg = 40
  while True:
- ok.message(contain=contain, people=people_max_qty_to_msg)
- # ok.like_action.like_quantity(quantity=10)
- # people_max_qty_to_msg = 10
+ message_again = ok.message_action.message(msg_to_send=contain)
+ # if not message_again:
+ #  ok.like_action.like_quantity(quantity=10)
  time.sleep(2)

model/action/browser_action.py

@@ -8,39 +8,64 @@ class BrowserAction:
  def __init__(self, browser_driver):
  self.browser_driver = browser_driver
 
- def goto_discover_swipe_section(self):
+ def goto_discover_swipe_section(self, break_condition=False):
  """
- Will go to the Main Page.
+ Will go to the Main Page (Where the New Profiles Likes or Dislikes resides).
+ :exception TimeoutException: Will handle the exception by refresh the driver and close the open widgets and will
+ try once again the current procedure. The break condition is when we're trying to execute this procedure once
+ again and it failes with `TimeoutException` it will
  """
  try:
  search = WebDriverWait(self.browser_driver.driver, 20).until(
  EC.presence_of_element_located((By.XPATH, "/html/body/div[1]/main/nav/div/span[1]/a[4]"))
  )
  search.click()
+ print(f"[debug] Navigate to discover swipe section")
  except TimeoutException as e:
- print("not found the element")
+ print(f"[error] not found the element")
  self.browser_driver.driver_refresh()
  self.browser_driver.driver_close_widgets_escape()
- self.goto_discover_swipe_section()
+ if not break_condition:
+ print(f"[error] will try again to run the procedure")
+ self.goto_discover_swipe_section(True)
+ print(f"[error] couldn't execute the procedure - somthing bad happened")
  except Exception as e:
  raise Exception(f"unfamiliar exception {e}")
 
  def goto_likes_section(self):
  """
- Will go to the Main Page.
+ Will go to likes Section in the main navbar
  """
  try:
  search = WebDriverWait(self.browser_driver.driver, 20).until(
- EC.presence_of_element_located((By.XPATH, "/html/body/div[1]/main/nav/div/span[1]/a[4]"))
+ EC.presence_of_element_located((By.XPATH, "/html/body/div[1]/main/nav/div/span[1]/a[5]"))
  )
  search.click()
  except TimeoutException as e:
- print("not found the element")
+ print("[error] not found the element")
  self.browser_driver.driver_refresh()
  self.browser_driver.driver_close_widgets_escape()
  self.goto_discover_swipe_section()
  except Exception as e:
- raise Exception(f"unfamiliar exception {e}")
+ raise Exception(f"[error] unfamiliar exception {e}")
 
  def goto_you_like_section(self):
- """"""
+ """
+ Inside `Likes` section in the main NAVBAR will direct into `You Like` - The previous likes that the user
+ have been like in the past.
+ """
+ try:
+ self.goto_likes_section()
+ search = WebDriverWait(self.browser_driver.driver, 20).until(
+ EC.presence_of_element_located(
+ (By.XPATH, "/html/body/div[1]/main/div/div[2]/section/div/nav/div/span[3]/a"))
+ )
+ search.click()
+ print(f"[debug] Navigate to `You Like` section")
+ except TimeoutException as e:
+ print("[error] not found the element")
+ self.browser_driver.driver_refresh()
+ self.browser_driver.driver_close_widgets_escape()
+ self.goto_discover_swipe_section()
+ except Exception as e:
+ raise Exception(f"[error] unfamiliar exception {e}")

model/action/like_action.py

@@ -8,12 +8,13 @@
 
 from model.Utilities.operation_utils import sleep_decorator
 from model.action.browser_action import BrowserAction
+from model.action.operation_action import OperationAction
+from model.driver.browser_driver import BrowserDriver
 
 
-class Like:
- def __init__(self, browser_driver):
- self.browser_driver = browser_driver
- self.browser_action = BrowserAction(self.browser_driver)
+class Like(OperationAction):
+ def __init__(self, browser_driver: BrowserDriver, browser_action: BrowserAction):
+ super().__init__(browser_driver=browser_driver, browser_action=browser_action)
 
  def close_widget_superlike_suggest(self):
  try:

model/action/message_action.py

@@ -8,17 +8,170 @@
 
 from model.Utilities.operation_utils import sleep_decorator
 from model.action.browser_action import BrowserAction
+from model.action.operation_action import OperationAction
 
 
-class Message:
- def __init__(self, browser_driver):
- self.browser_driver = browser_driver
- self.browser_action = BrowserAction(self.browser_driver)
+class Message(OperationAction):
+ def __init__(self, browser_driver, browser_action):
+ super().__init__(browser_driver=browser_driver, browser_action=browser_action)
+ self.MAX_COLUMN = 4
+ self.MAX_ROW = 4
 
- def message(self):
+ @sleep_decorator(prefix_duration=2, postfix_duration=2)
+ def press_on_liked_profile(self, row=1, col=1): # TODO: copy it to `browser_action` BrowserAction class maybe
  """
- Message From the people that have been liked in the past.
+ Inside The previous likes you can press on profiles, There are 4 profiles per row. We'll use MAX row qty and
+ MAX column qty to verify the user didn't insert higher value.
+ The row and column will start from `1` which means it's the first profile.
  """
- pass
+ try:
+ if row > self.MAX_ROW or col > self.MAX_COLUMN:
+ raise ValueError(f"The row: {row} and column: {col} are outofbounds")
+ search = WebDriverWait(self.browser_driver.driver, 50).until(
+ EC.presence_of_element_located(
+ (By.XPATH, f"/html/body/div[1]/main/div/div[2]/div/main/div/div/div/div[2]/div[{row}]/div[{col}]"))
+ )
+ search.click()
+ print(f"[debug] pressed on Liked profile")
+ except Exception as e:
+ error_msg = "Cannot press on the profile inside Likes -> You Like"
+ print(f"[error] {error_msg}")
+ raise Exception(f"{error_msg}, {e}")
 
+ @sleep_decorator(prefix_duration=2, postfix_duration=2)
+ def press_message_button(self):
+ """
+ Inside the profile Press the message button
+ :exception: Not found the MESSAGE button
+ """
+ try:
+ search4 = WebDriverWait(self.browser_driver.driver, 10).until(
+ EC.presence_of_element_located(
+ (By.XPATH, "/html/body/div[1]/main/div/div[3]/div[1]/div/div/div[3]/div/button[2]/div")))
+ search4.click()
+ print(f"[debug] pressed on message button")
+ except Exception as e:
+ error_msg = f"Cannot press on the first profile! In the 'You Like' section."
+ print(f"[error] {error_msg}")
+ raise Exception(f"{error_msg}, {e}")
+ @sleep_decorator(prefix_duration=2, postfix_duration=2)
+ def press_pass_button(self):
+ """
+ Will press the pass button and skip the profile immediately
+ :exception: Not found the PASS button
+ """
+ try:
+ search4 = WebDriverWait(self.browser_driver.driver, 10).until(
+ EC.presence_of_element_located(
+ (By.XPATH, "/html/body/div[1]/main/div/div[3]/div[1]/div/div/div[3]/div/button[1]/div")))
+ search4.click()
+ print(f"[deubg] Pressed on the pass button to pass the profile, Will direct to the `You Like` section")
+ except Exception as e:
+ error_msg = f"Cannot press on the first profile! In the 'You Like' section."
+ print(f"[error] {error_msg}")
+ raise Exception(f"{error_msg}, {e}")
+
+ @sleep_decorator(prefix_duration=2, postfix_duration=2)
+ def press_exit_from_message_text_area(self):
+ try:
+ print(f"[debug] try to press `X` exit from the message text box with XPath x1")
+ search = WebDriverWait(self.browser_driver.driver, 5).until(
+ EC.presence_of_element_located(
+ (By.XPATH,
+ "/html/body/div[1]/main/div/div[1]/div[2]/div[2]/div/div/div/div/div/div[1]/div[1]/button")))
+ search.click()
+ except Exception as e:
+ print(f"[debug] try to press `X` exit from the message text box with XPath x2")
+ try:
+ search = WebDriverWait(self.browser_driver.driver, 5).until(
+ EC.presence_of_element_located(
+ (By.XPATH,
+ "/html/body/div[1]/main/div/div[1]/div[2]/div[2]/div/div/div/div/div/div[1]/button")))
+ search.click()
+ except Exception as e:
+ error_msg = f"Cannot press on the X button and exit the message text area box"
+ print(f"[error] {error_msg}")
+ raise Exception(f"{error_msg}, {e}")
+
+ def insert_message_content_into_textarea(self, msg_to_send: str):
+ """
+ Inside the Message Box Will Send a message
+ :exception: Raise exception when we've already sent a message, or the message box of the other user is full
+ or other disturb.
+ """
+ try:
+ search5 = WebDriverWait(self.browser_driver.driver, 10).until(
+ EC.presence_of_element_located(
+ (By.XPATH, "/html/body/div[1]/main/div/div[1]/div[2]/div[2]/div/div/div/div/div/div[2]/textarea")))
+ time.sleep(1)
+ search5.send_keys(msg_to_send)
+ time.sleep(2)
+ print(f"[debug] sent keys into the text area {msg_to_send}")
+ # if their message box is full or already sent msg or other disturb
+ except Exception as e:
+ error_msg = f"Seems like the message box is full or already sent a msg or other disturb"
+ print(f"[error] {error_msg}")
+ raise Exception(f"{error_msg}, {e}")
+
+ def send_message(self):
+ """
+ After the content inserted to the textarea will press the button and send the message
+ """
+ try:
+ search = WebDriverWait(self.browser_driver.driver, 20).until(
+ EC.presence_of_element_located(
+ (By.XPATH, "/html/body/div[1]/main/div/div[1]/div[2]/div[2]/div/div/div/div/div/div[3]/button")))
+ search.click()
+ print(f"[debug] inside the message box that opened, "
+ f"pressed Send Message button to send the content inside the text area")
+ except Exception as e:
+ error_msg = f"Cannot send the message."
+ print(f"[error] {error_msg}")
+ raise Exception(f"{error_msg}, {e}")
+
+
+ @sleep_decorator(prefix_duration=1, postfix_duration=1)
+ def message(self, msg_to_send: str):
+ """
+ Message From the people that have been liked in the past. 
+ """
+ try:
+ # start with refresh
+ self.browser_driver.driver_refresh()
+ # navigate to the main discover page
+ self.browser_action.goto_discover_swipe_section()
+ # navigate to Likes -> You Like section
+ self.browser_action.goto_you_like_section()
+ # press on the first profile that appears with column 0 and row 0 by default
+ time.sleep(5)
+ try:
+ print(f"[info] try to press on liked profile x1")
+ self.press_on_liked_profile()
+ except Exception as e:
+ print(f"[info] try to press on liked profile x2")
+ self.press_on_liked_profile()
+ # press on message button box
+ self.press_message_button()
+ # try to send message content keys
+ try:
+ self.insert_message_content_into_textarea(msg_to_send=msg_to_send)
+ self.send_message()
+ self.press_exit_from_message_text_area()
+ except Exception as e:
+ # Exception occur, will pass the profile
+ self.press_exit_from_message_text_area()
+ # self.press_pass_button() # TODO: pass profile only if the inbox of the user is full with '⚠️' sign.
+ self.browser_driver.driver_refresh()
+ return False
+ self.browser_action.goto_you_like_section()
+ self.browser_driver.driver_refresh()
+ return True
+
+ except Exception as e:
+ raise Exception(f"Exception occur {e}")
 
+ def message_quantity(self, quantity):
+ print(f"----------------------- start message: quantity={quantity} -----------------------")
+ for count in range(quantity):
+ self.message()
+ print(f"message count={count+1}")

model/action/operation_action.py

@@ -0,0 +1,18 @@
+import time
+from abc import ABC
+
+from selenium.common import ElementClickInterceptedException, NoSuchElementException, TimeoutException
+from selenium.webdriver.common.by import By
+from selenium.webdriver.support.wait import WebDriverWait
+from selenium.webdriver.support import expected_conditions as EC
+
+
+from model.Utilities.operation_utils import sleep_decorator
+from model.action.browser_action import BrowserAction
+from model.driver.browser_driver import BrowserDriver
+
+
+class OperationAction(ABC):
+ def __init__(self, browser_driver: BrowserDriver, browser_action: BrowserAction):
+ self.browser_driver = browser_driver
+ self.browser_action = browser_action

model/driver/browser_driver.py

@@ -13,19 +13,27 @@ def __init__(self, driver_path, url):
  self.url_path_hanlder = UrlPath(url)
 
  def setupDriver(self, user_cached_data=None):
+ """
+ Setup the chrome driver.
+ :param user_cached_data: If there is user credentials cache data will direct to the folder which its resides.
+ """
  options = webdriver.ChromeOptions()
  if user_cached_data:
  options.add_argument(fr"user-data-dir={user_cached_data}")
  driver = webdriver.Chrome(executable_path=self.driver_path, chrome_options=options)
- driver.set_window_size(1900, 1024)
+ driver.set_window_size(1900, 1024) # TODO: fix the magic numbers.
  driver.get(self.url_path_hanlder.url)
  time.sleep(3)
  self.driver = driver
 
- @sleep_decorator(prefix_duration=1, postfix_duration=1)
+ @sleep_decorator(prefix_duration=2, postfix_duration=2)
  def driver_refresh(self):
+ """
+ Refreshes the driver/browser, Will raise an exception if there is any.
+ """
  try:
  self.driver.refresh()
+ print(f"[debug] driver refreshed")
  except Exception as e:
  raise Exception(f"Couldn't refresh the page, Check the WebDriver") from e