panleicim/appointment_tool
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

remove unused codes

Browse Source
This commit is contained in:
Lei PAN
2022-05-10 19:23:30 +02:00
parent de5bcb2698
commit a9a569a314
8 changed files with 41 additions and 390 deletions
Download Patch File Download Diff File Expand all files Collapse all files
error/SIMError.py
-8
View File
@@ -1,8 +0,0 @@
from enum import Enum
class SIMError(Enum):
SIM_DISABLED = "SIM_DISABLED"
CME_ERROR = "CME_ERROR"
STORAGE_ERROR = "STORAGE_ERROR"
TIMEOUT = "TIMEOUT"
main.py
+38 -63
View File
@@ -10,18 +10,13 @@ from gsmmodem import GsmModem
import params
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, oracle_log_sender
from params import oracle_log_sender
from pojo.serial_modem import SerialModem
from utils.excel_reader import ExcelHelper
OTP_TIMEOUT = 240
current_gsm_modem = None
card_pool = CardPool(CARD_POOL_PORT)
# used to save the current slot position
current_card_pool_slot = 1
current_sim_position = 1
init_logger()
logger = logging.getLogger()
@@ -51,62 +46,42 @@ def create_modem_for_port(port: str) -> Union[SerialModem, None]:
return serial_modem
def init_modems() -> list:
modems = []
for port in params.MODEM_POOL_PORTS:
serial_modem = create_modem_for_port(port)
if serial_modem:
modems.append(serial_modem)
return modems
def start_book(start_slot_number, end_slot_number, store=0):
slot_list = list(range(start_slot_number, end_slot_number + 1))
random.shuffle(slot_list)
for i in slot_list:
random.shuffle(params.MODEM_POOL_PORTS)
card_pool.reset()
logger.info("will switch to " + str(i))
global current_card_pool_slot
current_card_pool_slot = i
card_pool.switch_to_slot(i)
modem_pool = ModemPool(MODEM_POOL_PORTS)
modem_pool.reset_all_modems()
modem_list = init_modems()
# read the contact, and contact the 2 objects together
excel_reader = ExcelHelper()
contacts = excel_reader.read_contacts()
global current_sim_position
current_sim_position = 0
with ThreadPoolExecutor(max_workers=10) as executor:
for modem in modem_list:
current_sim_position = current_sim_position + 1
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]
else:
logger.info("contact not found for this ccid:{}".format(modem.ccid))
error_msg = "slot({}):sim({}):ccid({})".format(i, current_sim_position, modem.ccid)
oracle_log_sender.send_contact_not_found(error_msg)
modem.modem.close()
continue
if modem.contact:
logger.info("contact found for this ccid")
signal = modem.modem.signalStrength
logger.info("信号强度: " + str(signal))
proxy = get_proxy(modem.phone_number)
commandor = CommandorPage(modem, sim_position=current_sim_position,
slot_position=current_card_pool_slot, store_type=store)
# start the task in thread
executor.submit(commandor.start_page, proxy)
except Exception as error:
print(error)
continue
def start_book(start_number, end_number, store=0):
# read the contact, and contact the 2 objects together
excel_reader = ExcelHelper()
contacts = excel_reader.read_contacts()[start_number - 1: end_number - 1]
print(contacts)
# for i in contacts_range:
# logger.info("will switch to contact {}".format(i))
# with ThreadPoolExecutor(max_workers=10) as executor:
# for modem in modem_list:
# current_sim_position = current_sim_position + 1
# 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]
# else:
# logger.info("contact not found for this ccid:{}".format(modem.ccid))
# error_msg = "slot({}):sim({}):ccid({})".format(i, current_sim_position, modem.ccid)
# oracle_log_sender.send_contact_not_found(error_msg)
# modem.modem.close()
# continue
# if modem.contact:
# logger.info("contact found for this ccid")
# signal = modem.modem.signalStrength
# logger.info("信号强度: " + str(signal))
# proxy = get_proxy(modem.phone_number)
# commandor = CommandorPage(modem, sim_position=current_sim_position,
# slot_position=current_card_pool_slot, store_type=store)
# # start the task in thread
# executor.submit(commandor.start_page, proxy)
# except Exception as error:
# print(error)
# continue
def get_proxy(phone_number):
@@ -123,4 +98,4 @@ def get_proxy(phone_number):
if __name__ == '__main__':
# 修改起始行，结束行, 第三个参数store等于0的时候是随机，传入1的时候是总店
start_book(17, 17, store=0)
start_book(2, 18, store=0)
modems/ModemPool.py
-139
View File
@@ -1,139 +0,0 @@
import logging
import re
import time
import serial
from definitions import BAUDRATE
from error.SIMError import SIMError
from logs.LogSender import LOG_APPOINTMENT_SUCCESS, SUBJECT_SIM_INFO
from params import firebase_store_manager, oracle_log_sender
from pojo.SimInfoPojo import SimInfoPojo
from utils.excel_reader import ExcelHelper
from utils.operator import check_operator, Operator
class ModemPool:
phone_number_position = 10
TAG = "ModemPool"
def __init__(self, port_list: list):
self._port_list = port_list
self._serial_list = []
self._excel_helper = ExcelHelper()
self.contacts = self._excel_helper.read_contacts("/contact_all.xlsx")
self._log_sender = oracle_log_sender
self._db_manager = firebase_store_manager
self.logger = logging.getLogger(self.TAG)
for port in self._port_list:
ser = serial.Serial(port, BAUDRATE, timeout=1)
self._serial_list.append(ser)
def reset_all_modems(self):
self.logger.info("will reset modem pool")
for ser in self._serial_list:
#等待4秒
time.sleep(4)
self._send_command("AT+CFUN=1,1\r", ser, 10)
#等待15秒,加载新的SIM卡
time.sleep(15)
def _generate_error_msg(self, slot_position, index, error: SIMError):
msg = "slot({}) SIM({}), error:{}".format(slot_position, index + 1,
error.value)
self._log_sender.send_log(msg, source=self.TAG, subject=SUBJECT_SIM_INFO, type=error.value)
return msg
def get_raw_phone_number(self, slot_position):
for index, ser in enumerate(self._serial_list):
sim_position = index + 1
position = (slot_position - 1) * len(self._port_list) + sim_position
# unlock sim
unlock_cmd = 'AT+CPIN="{0}\r"'.format("0000")
self._send_command(unlock_cmd, ser, 30)
cmd = "AT+CCID\r"
response = str(self._send_command(cmd, ser))
ccid_group = re.search("[0-9F]+", response)
ccid = ccid_group.group(0)
operator = check_operator(ccid)
if operator == Operator.SFR or operator == Operator.CHINA_TELECOM:
contact = [contact for contact in self.contacts if
contact.ccid.replace("F", "") == ccid.replace("F", "")]
if len(contact) > 0:
phone_number = contact[0].phone
self._db_manager.save_sim_info(
SimInfoPojo(phone=str(phone_number), ccid=ccid, position=position, sim_position=sim_position,
slot_position=slot_position, operator=operator.value))
else:
error_msg = "slot({}),sim({})".format(slot_position, sim_position)
oracle_log_sender.send_contact_not_found(error_msg)
else:
self.logger.info("will get phone number for slot({}) SIM({}), port:{}".format(slot_position, index + 1, ser.port))
if not self._select_sim_storage(ser):
self.logger.info(self._generate_error_msg(slot_position, index, SIMError.STORAGE_ERROR))
continue
msg = self._execut_USSD_cmd("AT+CUSD=1, *132#\r", ser)
if "Unfortunately" in str(msg):
self.logger.info(self._generate_error_msg(slot_position, index, SIMError.SIM_DISABLED))
continue
elif "CME ERROR" in str(msg):
self.logger.info(self._generate_error_msg(slot_position, index, SIMError.CME_ERROR))
continue
elif len(msg) == 0:
self.logger.info(self._generate_error_msg(slot_position, index, SIMError.TIMEOUT))
continue
# find phone number
match = re.search(r'33\d{9}', str(msg))
phone_number = match.group(0)
print("phone is " + phone_number)
if phone_number:
self._db_manager.save_sim_info(
SimInfoPojo(phone=phone_number, ccid=ccid, position=position, slot_position=slot_position,
sim_position=sim_position, operator=operator.value))
self._log_sender.send_log(phone_number, source=self.TAG, subject=SUBJECT_SIM_INFO,
type=LOG_APPOINTMENT_SUCCESS)
def _select_sim_storage(self, ser) -> bool:
# use SIM Card storage
cmd_sm = "AT+CPBS=\"SM\"\r"
result = self._send_command(cmd_sm, ser)
if "ERROR" in str(result):
return False
else:
return True
def check_response(self, msg:str) -> bool:
print("msg is "+msg)
# if 'NOT READY' in msg:
# return False
if 'OK' in msg:
return True
else:
return False
def _send_command(self, cmd: str, ser, wait_time_in_s: int = 0) -> bytes:
ser.write(cmd.encode())
msg = None
try:
msg = ser.read(100)
count = 0
while not self.check_response(str(msg)) and count < wait_time_in_s:
time.sleep(1)
count = count + 1
msg = ser.read(100)
except Exception as exc:
print(exc)
print(msg)
return msg
def _execut_USSD_cmd(self, cmd, ser) -> bytes:
# the timeout for ussd command can be 120 s in mac
return self._send_command(cmd, ser, 120)
def close(self):
pass
# for serial in self._serial_list:
# serial.close()
modems/card_pool.py
-50
View File
@@ -1,50 +0,0 @@
import logging
import serial
import params
from logs.LogSender import LOG_SUBJECT_EVENT, TYPE_EVENT_CHANGE_SLOT
BAUDRATE = 115200
class CardPool:
def __init__(self, port_list):
self.logger = logging.getLogger("CardPool")
self._serial_list = []
for port in port_list:
self._serial_list.append(serial.Serial(port, BAUDRATE, timeout=1))
def _send_command(self, serial, cmd: str) -> bytes:
print("send command {}".format(cmd))
serial.write(cmd.encode())
msg = serial.read(100)
self.logger.info(msg)
return msg
# info: after reset, we need to restart modem pool
def reset(self):
for serial in self._serial_list:
self._send_command(serial, "AT+NEXT00\r")
def switch_to_next(self):
for serial in self._serial_list:
self._send_command(serial, "AT+NEXT11\r")
def switch_to_slot(self, slot_number: int):
params.oracle_log_sender.send_log(msg="换到" + str(slot_number) + "行", source=LOG_SUBJECT_EVENT,
subject=LOG_SUBJECT_EVENT,
type=TYPE_EVENT_CHANGE_SLOT)
if slot_number < 10:
for serial in self._serial_list:
self._send_command(serial, "AT+SWIT00-000{}\r".format(slot_number))
else:
for serial in self._serial_list:
self._send_command(serial, "AT+SWIT00-00{}\r".format(slot_number))
# not work for the pool
def find_current_slot(self):
for serial in self._serial_list:
self._send_command(serial, "AT+USIM\r")
params.py
-38
View File
@@ -1,44 +1,6 @@
from db.DbManager import DataManager
from logs.LogSender import LogSender
MODEM_POOL_PORTS = [
# modem 1
"/dev/tty.usbmodem1422101",
"/dev/tty.usbmodem1422103",
"/dev/tty.usbmodem1422105",
"/dev/tty.usbmodem1422107",
"/dev/tty.usbmodem1422201",
"/dev/tty.usbmodem1422203",
"/dev/tty.usbmodem1422205",
"/dev/tty.usbmodem1422207",
"/dev/tty.usbmodem1422301",
"/dev/tty.usbmodem1422303",
# "/dev/tty.usbmodem1422305",
"/dev/tty.usbmodem1422307",
"/dev/tty.usbmodem1422401",
"/dev/tty.usbmodem1422403",
"/dev/tty.usbmodem1422405",
"/dev/tty.usbmodem1422407",
# modem2
"/dev/tty.usbmodem1423101",
"/dev/tty.usbmodem1423103",
"/dev/tty.usbmodem1423105",
"/dev/tty.usbmodem1423107",
"/dev/tty.usbmodem1423201",
"/dev/tty.usbmodem1423203",
"/dev/tty.usbmodem1423205",
"/dev/tty.usbmodem1423207",
# "/dev/tty.usbmodem1423301",
## "/dev/tty.usbmodem1423303",
# "/dev/tty.usbmodem1423305",
# "/dev/tty.usbmodem1423307",
# "/dev/tty.usbmodem1423401",
# "/dev/tty.usbmodem1423403",
# "/dev/tty.usbmodem1423405",
# "/dev/tty.usbmodem1423407",
]
CARD_POOL_PORT = ["/dev/tty.usbmodem1423501", "/dev/tty.usbmodem1422601"]
firebase_store_manager = DataManager()
oracle_log_sender = LogSender()
requirements.txt
+3 -3
View File
@@ -3,10 +3,10 @@ firebase_admin==5.2.0
pandas==1.4.1
playwright==1.20.1
pyserial==3.5
python_gsmmodem_new==0.13.0
pydotenv==0.0.7
dataclasses~=0.6
oci~=2.54.1
XlsxWriter~=3.0.3
SQLAlchemy~=1.4.29
boto3~=1.21.13
boto3~=1.21.13
openpyxl==3.0.9
reset_all_sim_card.py
-29
View File
@@ -1,29 +0,0 @@
import logging
import sys
import params
from modems.ModemPool import ModemPool
from logs.AppLogging import init_logger
from logs.LogSender import LOG_SUBJECT_EVENT, TYPE_EVENT_RESET_ALL_SIM_CARDS
from main import card_pool
#
def read_all_the_phone_number():
params.oracle_log_sender.send_log(msg="SIM卡自检开始", subject=LOG_SUBJECT_EVENT, type=TYPE_EVENT_RESET_ALL_SIM_CARDS)
start_slot_number = 1
end_slot_number = 32
params.firebase_store_manager.clear_all_sim_info()
for i in range(start_slot_number, end_slot_number + 1):
card_pool.reset()
logger.info("will switch to " + str(i))
card_pool.switch_to_slot(i)
modem_pool = ModemPool(params.MODEM_POOL_PORTS)
modem_pool.reset_all_modems()
modem_pool.get_raw_phone_number(i)
if __name__ == '__main__':
init_logger()
logger = logging.getLogger()
logger.addHandler(logging.StreamHandler(stream=sys.stdout))
read_all_the_phone_number()
workers/commandor.py
-60
View File
@@ -1,60 +0,0 @@
import subprocess
import time
from pojo.contact_pojo import ContactPojo
PACKAGE_NAME = "com.opera.mini.android"
ACTIVITY_NAME = "MainActivity"
BROADCAST_ACTION = "com.lpa.appointement.broadcast"
class Commandor:
def __init__(self):
pass
def start_page(self, contact: ContactPojo):
# specifying an explicit component name
self.clear_app_data()
cmd = "/Users/panlei/Library/Android/sdk/platform-tools/adb shell am start -n {}/.{} --es \"first_name\" \"{}\" --es \"last_name\" \"{}\" --es \"phone\" \"{}\" --es \"email\" \"{}\" --es \"passport\" \"{}\"".format(
PACKAGE_NAME, ACTIVITY_NAME, contact.last_name, contact.first_name, "+33{}".format(contact.phone),
"{}_{}@gmail.com".format(contact.first_name, contact.last_name), contact.passport)
print("cmd is " + cmd)
subprocess.call(cmd, shell=True)
pass
def clear_app_data(self):
subprocess.call("/Users/panlei/Library/Android/sdk/platform-tools/adb shell pm clear {}".format(PACKAGE_NAME),
shell=True)
def send_otp(self, otp: str):
subprocess.call(
"/Users/panlei/Library/Android/sdk/platform-tools/adb shell am broadcast -a {} --es otp \"{}\"".format(
BROADCAST_ACTION, otp), shell=True)
pass
def reset_air_plan_mode(self):
subprocess.call(
"/Users/panlei/Library/Android/sdk/platform-tools/adb shell settings put global airplane_mode_on 1",
shell=True)
time.sleep(1)
subprocess.call(
"/Users/panlei/Library/Android/sdk/platform-tools/adb shell am broadcast -a android.intent.action.AIRPLANE_MODE",
shell=True)
time.sleep(1)
subprocess.call(
"/Users/panlei/Library/Android/sdk/platform-tools/adb shell settings put global airplane_mode_on 0",
shell=True)
time.sleep(1)
subprocess.call(
"/Users/panlei/Library/Android/sdk/platform-tools/adb shell am broadcast -a android.intent.action.AIRPLANE_MODE",
shell=True)
time.sleep(10)
if __name__ == '__main__':
commandor = Commandor()
# contact = ContactPojo("0649614591", "E24183897", "LIU", "Yusi", "AZEER", "lei-pan@outlook.com")
# commandor.start_page(contact)
commandor.reset_air_plan_mode()
# commandor.send_otp("262353")