212 lines
8.4 KiB
Python
Executable File
212 lines
8.4 KiB
Python
Executable File
import json
|
|
import random
|
|
import string
|
|
|
|
import pandas as pandas
|
|
import validators as validators
|
|
import xlsxwriter
|
|
|
|
from src.db.mongo_manager import MONGO_STORE_MANAGER
|
|
from src.pojo.contact_pojo import ContactPojo
|
|
from src.pojo.mail.mail_pojo import MailAddress
|
|
from src.utils.contacts.generate_random_passport_id import get_random_passport_id_number
|
|
|
|
fr_phone_number_prefix = ['73', '74', '75', '76', '77', '78', '79']
|
|
chinnese_number_prefix = ['13', '15', '18']
|
|
|
|
|
|
def read_links_to_click(file_path):
|
|
links_info_in_json = pandas.read_excel(file_path).to_json(orient='records')
|
|
# print(links_info_in_json)
|
|
for item in json.loads(links_info_in_json):
|
|
link = item['link']
|
|
if validators.url(link):
|
|
print(link)
|
|
MONGO_STORE_MANAGER.save_links_to_validate(link, "")
|
|
|
|
else:
|
|
print("error on link " + link)
|
|
|
|
|
|
def read_contacts(file_name) -> list:
|
|
print("read file " + file_name)
|
|
contact_list_in_json = pandas.read_excel(file_name).to_json(orient='records')
|
|
contact_dict_list = json.loads(contact_list_in_json)
|
|
contact_list = []
|
|
for contact_dict in contact_dict_list:
|
|
if contact_dict['name']:
|
|
raw_name = contact_dict['name'].strip()
|
|
name = raw_name.split(' ')
|
|
last_name = name[0]
|
|
if len(name) == 2:
|
|
first_name = name[-1]
|
|
else:
|
|
first_name = ''.join(name[1:len(name)])
|
|
ip_country = "FR"
|
|
if contact_dict.get('ip_country') is not None:
|
|
ip_country = contact_dict['ip_country']
|
|
store = "random"
|
|
if contact_dict.get('store') is not None:
|
|
store = contact_dict['store']
|
|
ua = ""
|
|
|
|
if contact_dict.get('ua') is not None:
|
|
ua = contact_dict['ua']
|
|
|
|
contact = ContactPojo(phone_number=contact_dict['phone'],
|
|
last_name=last_name,
|
|
first_name=first_name,
|
|
passport_number=contact_dict['passport'],
|
|
mail=contact_dict['email'], store=store)
|
|
if contact_dict.get('serial') is not None:
|
|
serial = contact_dict['serial']
|
|
contact.serial = serial
|
|
contact.ip_country = ip_country
|
|
contact.ua = ua
|
|
contact_list.append(contact)
|
|
return contact_list
|
|
|
|
|
|
class ExcelHelper:
|
|
|
|
def __init__(self):
|
|
self._df = pandas.Series()
|
|
|
|
def read_user_agens(self) -> list:
|
|
user_agent_in_json = pandas.read_excel(
|
|
"/Users/lpan/Documents/workspace/appointment_tool/docs/mobile_user_agent_list.xlsx").to_json(
|
|
orient='records')
|
|
user_agent_dict_list = json.loads(user_agent_in_json)
|
|
user_agent_list = []
|
|
for user_agent_dict in user_agent_dict_list:
|
|
user_agent_str = user_agent_dict['user_agent']
|
|
if 'Mozilla' in user_agent_str:
|
|
if 'Android 5.1' in user_agent_str:
|
|
if 'Mac OS' not in user_agent_str:
|
|
user_agent_list.append(user_agent_dict['user_agent'])
|
|
print(user_agent_list)
|
|
|
|
def check_contact_list(self, file_name):
|
|
contact_list = read_contacts(file_name)
|
|
for contact in contact_list:
|
|
if contact.first_name is None or len(contact.first_name) == 0:
|
|
print("error in firstName for " + contact.mail)
|
|
if contact.last_name is None or len(contact.last_name) == 0:
|
|
print("error in last_name for " + contact.mail)
|
|
if contact.phone is None or len(contact.phone) == 0:
|
|
print("error in phone for " + contact.mail)
|
|
if contact.passport is None or len(contact.passport) == 0:
|
|
print("error in passport_number for " + contact.mail)
|
|
if contact.mail is None or len(contact.mail) == 0:
|
|
print("error in mail for " + contact.phone_number)
|
|
|
|
def read_mails_and_pwd(self,
|
|
file_name='/Users/lpan/Desktop/163.xlsx'):
|
|
contact_list = []
|
|
mail_list_in_json = pandas.read_excel(file_name).to_json(orient='records')
|
|
contact_dict_list = json.loads(mail_list_in_json)
|
|
for contact_dict in contact_dict_list:
|
|
if contact_dict['email']:
|
|
mail = contact_dict['email'].strip()
|
|
pwd = contact_dict['password']
|
|
contact = MailAddress(mail, pwd)
|
|
contact_list.append(contact)
|
|
return contact_list
|
|
|
|
def read_names(self, file_name) -> list:
|
|
contact_list_in_json = pandas.read_excel(file_name).to_json(orient='records')
|
|
contact_dict_list = json.loads(contact_list_in_json)
|
|
contact_list = []
|
|
count = 2
|
|
for contact_dict in contact_dict_list:
|
|
if contact_dict['name']:
|
|
raw_name = contact_dict['name'].strip()
|
|
name = raw_name.split(' ')
|
|
if len(name) == 1:
|
|
name = raw_name.split('\xa0')
|
|
if len(name) == 1:
|
|
print("error in " + str(name))
|
|
last_name = name[0]
|
|
if len(name) == 2:
|
|
first_name = name[-1]
|
|
else:
|
|
first_name = ''.join(name[1:len(name)])
|
|
|
|
contact = ContactPojo(phone_number="",
|
|
last_name=last_name,
|
|
first_name=first_name,
|
|
passport_number="",
|
|
mail="")
|
|
|
|
if len(first_name) == 0:
|
|
print("first_name is empty: position:" + str(count))
|
|
print(name)
|
|
if len(last_name) == 0:
|
|
print("last_name is empty: position:" + str(count))
|
|
count = count + 1
|
|
contact_list.append(contact)
|
|
|
|
return contact_list
|
|
|
|
def read_email_pojo(self, file_name) -> list:
|
|
email_info_in_json = pandas.read_excel(file_name).to_json(orient='records')
|
|
contact_dict_list = json.loads(email_info_in_json)
|
|
contact_list = []
|
|
count = 0
|
|
for contact_dict in contact_dict_list:
|
|
if contact_dict['email']:
|
|
email = contact_dict['email'].strip()
|
|
password = contact_dict['code']
|
|
email_destinaire = MailAddress(email, password)
|
|
count = count + 1
|
|
contact_list.append(email_destinaire)
|
|
|
|
return contact_list
|
|
|
|
|
|
def get_random_fr_phone_numbers():
|
|
length = 7 # number of characters in the string.
|
|
ran = ''.join(random.choices(string.digits, k=length))
|
|
_phone_number = random.choice(fr_phone_number_prefix) + str(ran)
|
|
return _phone_number
|
|
|
|
|
|
def get_random_cn_phone_numbers():
|
|
length = 8 # number of characters in the string.
|
|
ran = ''.join(random.choices(string.digits, k=length))
|
|
_phone_number = random.choice(fr_phone_number_prefix) + str(ran)
|
|
prefix = random.choice(chinnese_number_prefix)
|
|
return prefix + _phone_number
|
|
|
|
|
|
def get_random_id_number() -> str:
|
|
# write_the_valid_profiles_to_excel()
|
|
S = 8 # number of characters in the string.
|
|
# call random.choices() string module to find the string in Uppercase + numeric data.
|
|
ran = ''.join(random.choices(string.digits, k=S))
|
|
print("The randomly generated string is : 94" + str(ran)) # print the random data
|
|
return ran
|
|
|
|
|
|
def save_mails_to_db():
|
|
excel_reader = ExcelHelper()
|
|
emails = excel_reader.read_email_pojo("~/Downloads/邮箱及密码.xlsx")
|
|
print(emails)
|
|
for mail in emails:
|
|
MONGO_STORE_MANAGER.save_destinary_emails(mail)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
# excel_reader = ExcelHelper()
|
|
# contacts = excel_reader.read_names("/Users/lpan/Downloads/gmail_10.xlsx")
|
|
# print(contacts)
|
|
# write_new_contacts_to_excel(valid_contacts=contacts)
|
|
# excel_reader = ExcelHelper()
|
|
# excel_reader.check_contact_list("/Users/lpan/Desktop/contact_email_valid.xlsx")
|
|
# read_links_to_click("/Users/lpan/Downloads/links.xlsx")
|
|
save_mails_to_db()
|
|
# for mail in excel_reader.read_mails_and_pwd():
|
|
# MONGO_STORE_MANAGER.insert_email(mail)
|
|
# for i in range(1, 300):
|
|
# print(get_random_fr_phone_numbers())
|