use multi-thread to wait sms
This commit is contained in:
+1
-3
@@ -1,15 +1,13 @@
|
||||
import asyncio
|
||||
import random
|
||||
import threading
|
||||
from concurrent.futures import ThreadPoolExecutor
|
||||
from enum import Enum
|
||||
from typing import Union
|
||||
|
||||
from playwright.async_api import async_playwright
|
||||
from playwright.sync_api import sync_playwright
|
||||
|
||||
import params
|
||||
from commandor_page import get_random_id_number_for_proxy
|
||||
from workers.commandor_page import get_random_id_number_for_proxy
|
||||
from logs.LogSender import TYPE_EVENT_CHECK_RESULTS, LOG_SUBJECT_EVENT
|
||||
from pojo.ReserveResultPojo import ReserveResultPojo
|
||||
from utils.excel_reader import ExcelHelper
|
||||
|
||||
+1
-1
@@ -4,7 +4,7 @@ from dotenv import load_dotenv
|
||||
load_dotenv()
|
||||
|
||||
LOG_SOURCE = os.getenv("LOG_SOURCE")
|
||||
|
||||
SMS_TIMEOUT = 60
|
||||
BAUDRATE = 115200
|
||||
|
||||
ROOT_DIR = os.path.dirname(os.path.abspath(__file__))
|
||||
|
||||
+2
-2
@@ -67,8 +67,8 @@ class LogSender:
|
||||
)
|
||||
return response
|
||||
|
||||
def send_sms_reception_log(self, phone, sms_text):
|
||||
msg = "phone:{}, sms:{}".format(phone, sms_text)
|
||||
def send_sms_reception_log(self, phone, sms_text, ccid):
|
||||
msg = "from:{}, sms:{}, ccid{}".format(phone, sms_text, ccid)
|
||||
self.send_log(msg=msg, subject=LOG_SUBJECT_SMS, type=TYPE_SMS_RECEIVED)
|
||||
|
||||
def send_timeout_log(self, serial_modem: SerialModem):
|
||||
|
||||
@@ -1,25 +1,20 @@
|
||||
import datetime
|
||||
import logging
|
||||
import random
|
||||
import re
|
||||
import sys
|
||||
import time
|
||||
from concurrent.futures import ThreadPoolExecutor
|
||||
from threading import Event
|
||||
from typing import Union
|
||||
|
||||
from gsmmodem import GsmModem
|
||||
|
||||
import params
|
||||
from commandor_page import CommandorPage
|
||||
from workers.commandor_page import CommandorPage
|
||||
from logs.AppLogging import init_logger
|
||||
from modems.ModemPool import ModemPool
|
||||
from modems.card_pool import CardPool
|
||||
from params import MODEM_POOL_PORTS, CARD_POOL_PORT, firebase_store_manager, oracle_log_sender
|
||||
from pojo.ReserveResultPojo import ReserveResultPojo
|
||||
from params import MODEM_POOL_PORTS, CARD_POOL_PORT, oracle_log_sender
|
||||
from pojo.serial_modem import SerialModem
|
||||
from utils.excel_reader import ExcelHelper
|
||||
from utils.operator import check_operator, Operator
|
||||
|
||||
OTP_TIMEOUT = 240
|
||||
current_gsm_modem = None
|
||||
@@ -66,8 +61,8 @@ def init_modems() -> list:
|
||||
|
||||
|
||||
def start_book():
|
||||
start_slot_number = 13
|
||||
end_slot_number = 13
|
||||
start_slot_number = 14
|
||||
end_slot_number = 15
|
||||
slot_list = list(range(start_slot_number, end_slot_number + 1))
|
||||
random.shuffle(slot_list)
|
||||
for i in slot_list:
|
||||
@@ -85,7 +80,7 @@ def start_book():
|
||||
contacts = excel_reader.read_contacts()
|
||||
global current_sim_position
|
||||
current_sim_position = 0
|
||||
with ThreadPoolExecutor(max_workers=2) as executor:
|
||||
with ThreadPoolExecutor(max_workers=5) as executor:
|
||||
for modem in modem_list:
|
||||
current_sim_position = current_sim_position + 1
|
||||
try:
|
||||
|
||||
+30
-51
@@ -1,19 +1,21 @@
|
||||
import logging
|
||||
import sys
|
||||
import time
|
||||
from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor
|
||||
from typing import Union
|
||||
|
||||
from gsmmodem import GsmModem
|
||||
|
||||
import params
|
||||
from definitions import SMS_TIMEOUT
|
||||
from logs.AppLogging import init_logger
|
||||
from modems.ModemPool import ModemPool
|
||||
from modems.card_pool import CardPool
|
||||
from params import MODEM_POOL_PORTS, CARD_POOL_PORT
|
||||
from pojo.serial_modem import SerialModem
|
||||
from utils.excel_reader import ExcelHelper
|
||||
from workers.wait_sms_worker import WaitSmsWorker
|
||||
|
||||
OTP_TIMEOUT = 60
|
||||
thread_event = None
|
||||
current_gsm_modem = None
|
||||
card_pool = CardPool(CARD_POOL_PORT)
|
||||
@@ -21,7 +23,6 @@ init_logger()
|
||||
logger = logging.getLogger()
|
||||
logger.addHandler(logging.StreamHandler(stream=sys.stdout))
|
||||
|
||||
|
||||
def get_devices_ports() -> list:
|
||||
return MODEM_POOL_PORTS
|
||||
|
||||
@@ -42,39 +43,6 @@ def timeout_occurred(serial_modem: SerialModem):
|
||||
serial_modem.modem.close()
|
||||
|
||||
|
||||
def start_to_handle_sms(serial_modem: SerialModem):
|
||||
global current_gsm_modem
|
||||
current_gsm_modem = serial_modem.modem
|
||||
try:
|
||||
current_gsm_modem.deleteMultipleStoredSms(memory="SM")
|
||||
except Exception as error:
|
||||
print(error)
|
||||
|
||||
serial_modem.modem.smsReceivedCallback = handle_sms
|
||||
serial_modem.modem.smsTextMode = False
|
||||
logger.info('Waiting for SMS message, for phone number ' + str(serial_modem.phone_number))
|
||||
# input("Press Enter to continue...")
|
||||
# return
|
||||
listen_at = time.time()
|
||||
while True:
|
||||
time.sleep(2)
|
||||
# check whether timeout
|
||||
now = time.time()
|
||||
if (listen_at + OTP_TIMEOUT) < now:
|
||||
logger.info("time out for {}, switch to next contact".format(serial_modem.phone_number))
|
||||
# save the contact in timeout
|
||||
timeout_occurred(serial_modem)
|
||||
current_gsm_modem.close()
|
||||
return
|
||||
|
||||
|
||||
def handle_sms(sms):
|
||||
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
|
||||
params.oracle_log_sender.send_sms_reception_log(sms.number, sms.text)
|
||||
|
||||
|
||||
def init_modems() -> list:
|
||||
modems = []
|
||||
for port in get_devices_ports():
|
||||
@@ -89,7 +57,7 @@ def start_waiting_sms():
|
||||
slot_number = 1
|
||||
slot_sum = 21
|
||||
slot_list = list(range(slot_number, slot_sum + 1))
|
||||
for i in slot_list:
|
||||
for i in reversed(slot_list):
|
||||
card_pool.reset()
|
||||
logger.info("will switch to " + str(i))
|
||||
card_pool.switch_to_slot(i)
|
||||
@@ -99,21 +67,32 @@ def start_waiting_sms():
|
||||
# read the contact, and merge the 2 objects together
|
||||
excel_reader = ExcelHelper()
|
||||
contacts = excel_reader.read_contacts()
|
||||
for modem in modem_list:
|
||||
try:
|
||||
# get contact for current modem
|
||||
modem.get_ccid()
|
||||
contact = [contact for contact in contacts if
|
||||
contact.ccid.replace("F", "") == modem.ccid.replace("F", "")]
|
||||
if len(contact) > 0:
|
||||
modem.phone_number = contact[0].phone
|
||||
modem.contact = contact[0]
|
||||
start_to_handle_sms(modem)
|
||||
else:
|
||||
print("contact not found, skip")
|
||||
except Exception as error:
|
||||
print(error)
|
||||
continue
|
||||
with ThreadPoolExecutor(max_workers=len(MODEM_POOL_PORTS)) as executor:
|
||||
for modem in modem_list:
|
||||
try:
|
||||
modem.get_ccid()
|
||||
# find the contact with ccid
|
||||
contact = [contact for contact in contacts if
|
||||
contact.ccid.replace("F", "") == modem.ccid.replace("F", "")]
|
||||
if len(contact) > 0:
|
||||
modem.phone_number = contact[0].phone
|
||||
modem.contact = contact[0]
|
||||
if modem.contact:
|
||||
logger.info("contact found for this ccid")
|
||||
commandor = WaitSmsWorker(modem)
|
||||
# start the task in thread
|
||||
executor.submit(commandor.run())
|
||||
except Exception as error:
|
||||
print(error)
|
||||
continue
|
||||
listen_at = time.time()
|
||||
now = time.time()
|
||||
while (listen_at + SMS_TIMEOUT + 5) > now:
|
||||
now = time.time()
|
||||
print("sleep for 2 s")
|
||||
time.sleep(2)
|
||||
print("will call continue")
|
||||
continue
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
|
||||
@@ -0,0 +1,49 @@
|
||||
import logging
|
||||
import threading
|
||||
import time
|
||||
|
||||
import params
|
||||
from definitions import SMS_TIMEOUT
|
||||
from pojo.serial_modem import SerialModem
|
||||
|
||||
|
||||
class WaitSmsWorker:
|
||||
|
||||
def __init__(self, serial_modem: SerialModem):
|
||||
self.serial_modem = serial_modem
|
||||
self.logger = logging.getLogger("WaitSmsWorker")
|
||||
|
||||
def handle_sms(self, sms):
|
||||
self.logger.info(
|
||||
"sms received for phone:{}(ccid:{})".format(self.serial_modem.phone_number, self.serial_modem.ccid))
|
||||
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
|
||||
params.oracle_log_sender.send_sms_reception_log(sms.number, sms.text, self.serial_modem.ccid)
|
||||
|
||||
def run(self):
|
||||
t = threading.Thread(target=self.start_wait)
|
||||
t.start()
|
||||
# t.join()
|
||||
|
||||
def start_wait(self):
|
||||
try:
|
||||
self.serial_modem.modem.deleteMultipleStoredSms(memory="SM")
|
||||
except Exception as error:
|
||||
print(error)
|
||||
|
||||
self.serial_modem.modem.smsReceivedCallback = self.handle_sms
|
||||
self.serial_modem.modem.smsTextMode = False
|
||||
self.logger.info('Waiting for SMS message, for phone number ' + str(self.serial_modem.phone_number))
|
||||
listen_at = time.time()
|
||||
# threading.current_thread().join(self.OTP_TIMEOUT)
|
||||
while True:
|
||||
# self.logger.info("sleep for 2s in thread({})".format(threading.currentThread().name))
|
||||
time.sleep(5)
|
||||
# check whether timeout
|
||||
now = time.time()
|
||||
if (listen_at + SMS_TIMEOUT) < now:
|
||||
self.logger.info("time out for {}, switch to next contact".format(self.serial_modem.phone_number))
|
||||
# save the contact in timeout
|
||||
self.serial_modem.modem.close()
|
||||
return
|
||||
Reference in New Issue
Block a user