remove unused codes
This commit is contained in:
+10
-73
@@ -1,27 +1,22 @@
|
||||
import datetime
|
||||
import logging
|
||||
import random
|
||||
import re
|
||||
import string
|
||||
import threading
|
||||
import time
|
||||
from typing import Union
|
||||
|
||||
from gsmmodem import GsmModem
|
||||
from playwright.sync_api import sync_playwright
|
||||
|
||||
import params
|
||||
from params import PROXY_SERVER, PROXY_PASSWORD
|
||||
from pojo.ReserveResultPojo import ReserveResultPojo, PublishType
|
||||
from pojo.contact_pojo import ContactPojo
|
||||
from pojo.serial_modem import SerialModem
|
||||
from utils.excel_reader import ExcelHelper
|
||||
from utils.operator import Operator, check_operator
|
||||
|
||||
RDV_URL = "https://rendezvousparis.hermes.com/client/register"
|
||||
# RDV_URL = "https://rendezvousparis.hermes.com/client/register"
|
||||
|
||||
# RDV_URL = "file:///Users/lpan/Downloads/test_appointment.html"
|
||||
# RDV_URL = "https://api.ipify.org"
|
||||
RDV_URL = "https://api.ipify.org"
|
||||
# RDV_URL ="https://bot.sannysoft.com/"
|
||||
otp_value = None
|
||||
user_agent_list = ExcelHelper().read_user_agent_list()
|
||||
@@ -47,75 +42,24 @@ class Tls(threading.local):
|
||||
class CommandorPage:
|
||||
tls = Tls()
|
||||
|
||||
def __init__(self, serial_modem: SerialModem, slot_position, sim_position, store_type=0):
|
||||
def __init__(self, contact: ContactPojo, store_type=0):
|
||||
self.otp_value = None
|
||||
self.logger = logging.getLogger("CommandorPage")
|
||||
self.is_finished = False
|
||||
self.current_gsm_modem = serial_modem
|
||||
self.slot_position = slot_position
|
||||
self.sim_position = sim_position
|
||||
self.contact = serial_modem.contact
|
||||
self.contact = contact
|
||||
self.store_type = store_type
|
||||
|
||||
def on_success(self, result: ReserveResultPojo):
|
||||
self.logger.info("on_success called.")
|
||||
self.logger.info(result)
|
||||
params.oracle_log_sender.send_appoint_result(result)
|
||||
self.current_gsm_modem.modem.close()
|
||||
self.is_finished = True
|
||||
|
||||
def handle_sms(self, sms):
|
||||
self.logger.info(
|
||||
u'== SMS message received ==\nFrom: {0}\nTime: {1}\nMessage:\n{2}\n'.format(sms.number, sms.time, sms.text))
|
||||
# extract the otp number
|
||||
date = str(sms.time)[0:10]
|
||||
params.oracle_log_sender.send_sms_reception_log(sms.number, sms.text, self.current_gsm_modem.ccid)
|
||||
if date == str(datetime.date.today()):
|
||||
self.logger.info("this sms is for today")
|
||||
if "rendez-vous" in sms.text or "appointment" in sms.text:
|
||||
self.logger.info("try to extract the otp")
|
||||
pattern = r'\d{6,8}'
|
||||
# if re.match(pattern, sms.text):
|
||||
match = re.search(pattern, sms.text)
|
||||
otp = match.group(0)
|
||||
self.logger.info("otp is " + otp)
|
||||
self.otp_value = otp
|
||||
self.logger.info("will set thread event")
|
||||
self.thread_event.set()
|
||||
self.is_finished = True
|
||||
else:
|
||||
self.logger.info("The sms is not for RDV")
|
||||
else:
|
||||
self.logger.info("The sms is not for today")
|
||||
|
||||
def set_up_sms_listener(self):
|
||||
if check_operator(self.current_gsm_modem.ccid) == Operator.LYCAMOBILE:
|
||||
# lycamobile
|
||||
self.current_gsm_modem.modem.deleteMultipleStoredSms(memory="SM")
|
||||
self.current_gsm_modem.modem.smsReceivedCallback = self.handle_sms
|
||||
self.is_finished = False
|
||||
self.current_gsm_modem.smsTextMode = False
|
||||
self.logger.info('Waiting for SMS message, for phone number ' + str(self.current_gsm_modem.phone_number))
|
||||
listen_at = time.time()
|
||||
while not self.is_finished:
|
||||
time.sleep(2)
|
||||
# check whether timeout
|
||||
now = time.time()
|
||||
if (listen_at + OTP_TIMEOUT) < now:
|
||||
self.logger.info("time out for {}, switch to next contact".format(self.current_gsm_modem.phone_number))
|
||||
# save the contact in timeout
|
||||
self.timeout_occurred()
|
||||
self.current_gsm_modem.modem.close()
|
||||
return
|
||||
return
|
||||
|
||||
def timeout_occurred(self):
|
||||
params.firebase_store_manager.save_timeout_contact(self.current_gsm_modem.contact)
|
||||
params.oracle_log_sender.send_timeout_log(self.current_gsm_modem)
|
||||
params.oracle_log_sender.send_timeout_log(self.contact)
|
||||
self.logger.info("will close timeout modem")
|
||||
self.thread_event.set()
|
||||
self.current_gsm_modem.modem.close()
|
||||
self.reset_air_plan_mode()
|
||||
self.termine()
|
||||
|
||||
def _run(self, e: threading.Event, proxy):
|
||||
print("will start browser")
|
||||
@@ -143,7 +87,6 @@ class CommandorPage:
|
||||
# wait for sms_code field
|
||||
# self.clickOnValidBtn()
|
||||
self.thread_event = e
|
||||
self.set_up_sms_listener()
|
||||
otp_input = self.page.locator(OTP_FIELD_ID)
|
||||
otp_input.wait_for(state='visible', timeout=TIME_OUT)
|
||||
event_is_set = e.wait()
|
||||
@@ -163,7 +106,7 @@ class CommandorPage:
|
||||
self.publish_message_to_queue(self.contact, PublishType.SUCCESS, self.page.url)
|
||||
else:
|
||||
self.logger.info("timeout")
|
||||
self.reset_air_plan_mode()
|
||||
self.termine()
|
||||
|
||||
def start_browser(self, proxy, pwright, device) -> Union[str, None]:
|
||||
try:
|
||||
@@ -300,7 +243,7 @@ class CommandorPage:
|
||||
time.sleep(get_random_wait_time())
|
||||
self.page.fill(OTP_FIELD_ID, otp)
|
||||
|
||||
def reset_air_plan_mode(self):
|
||||
def termine(self):
|
||||
self.logger.info("will close browser")
|
||||
self.browser.close()
|
||||
|
||||
@@ -311,8 +254,6 @@ class CommandorPage:
|
||||
firstName=contact.first_name, lastName=contact.last_name, email=contact.mail,
|
||||
passport=contact.passport, ccid=contact.ccid)
|
||||
result.id = id
|
||||
result.sim_position = self.sim_position
|
||||
result.slot_position = self.slot_position
|
||||
params.firebase_store_manager.save(result)
|
||||
if status is PublishType.SUCCESS:
|
||||
self.on_success(result)
|
||||
@@ -353,12 +294,8 @@ def launch_page():
|
||||
contact = ContactPojo(phone_number="+33758912245", passport_number=passport_number, last_name="XU",
|
||||
first_name="xingzhen",
|
||||
mail="ColbyPatel653@gmail.com", ccid="", position=0)
|
||||
modem = GsmModem("/dev/tty.usbmodem1422307")
|
||||
serial_modem = SerialModem(modem)
|
||||
serial_modem.contact = contact
|
||||
serial_modem.phone_number = contact.phone
|
||||
page = CommandorPage(serial_modem, slot_position=1, sim_position=1, store_type=1)
|
||||
return page.start_page(proxy)
|
||||
page = CommandorPage(contact, store_type=1)
|
||||
return page.start_page(None)
|
||||
|
||||
|
||||
def wait_for_otp(event: threading.Event, commandor: CommandorPage):
|
||||
|
||||
Reference in New Issue
Block a user