add method to get ip information
This commit is contained in:
@@ -12,6 +12,13 @@ class ContactPojo:
|
||||
mail: str
|
||||
store: str
|
||||
ip_country: str
|
||||
ip_address: str
|
||||
isp: str = None
|
||||
|
||||
def __repr__(self):
|
||||
return "phone:{}, passport:{}, last_name:{}, first_name:{}, mail:{}, store:{}, ip_country:{}, ip_address:{},isp:{}".format(
|
||||
self.phone, self.passport, self.last_name, self.first_name, self.mail, self.store, self.ip_country,
|
||||
self.ip_address, self.isp)
|
||||
|
||||
def __init__(self, phone_number: str, passport_number: str, last_name: str, first_name: str, mail: str, store: str,
|
||||
ip_country="FR"):
|
||||
|
||||
@@ -0,0 +1,74 @@
|
||||
import json
|
||||
from time import sleep
|
||||
from typing import Union
|
||||
|
||||
import requests
|
||||
import xlsxwriter
|
||||
|
||||
from src.db.mongo_manager import MONGO_STORE_MANAGER
|
||||
from src.pojo.contact_pojo import ContactPojo
|
||||
from src.pojo.ip.ip_location import IPLocationInfo
|
||||
|
||||
|
||||
def get_contact_list() -> list:
|
||||
_successful_items = MONGO_STORE_MANAGER.get_all_successful_items_for_yesterday()
|
||||
_contact_list = []
|
||||
for item in _successful_items:
|
||||
if item.url_validated:
|
||||
_contact = ContactPojo(first_name=item.first_name, last_name=item.last_name, mail=item.mail,
|
||||
phone_number=item.phone, passport_number=item.passport, store=item.store)
|
||||
_contact.ip_address = item.ip_address
|
||||
_contact_list.append(_contact)
|
||||
return _contact_list
|
||||
|
||||
|
||||
def get_info_from_ip(ip_address) -> Union[None, IPLocationInfo]:
|
||||
_ip_info = None
|
||||
try:
|
||||
response = requests.get("http://ip-api.com/json/" + ip_address)
|
||||
response.raise_for_status()
|
||||
ip_in_dict = json.loads(response.text)
|
||||
_ip_info = IPLocationInfo(ip_address, ip_in_dict['country'], ip_in_dict['city'], ip_in_dict['isp'])
|
||||
except Exception as err:
|
||||
print(err)
|
||||
return _ip_info
|
||||
|
||||
|
||||
def write_contact_with_ip_info_to_file(contact_list):
|
||||
row = 0
|
||||
col = 0
|
||||
# Create a workbook and add a worksheet.
|
||||
workbook = xlsxwriter.Workbook('ip_info_{}.xlsx'.format(len(contact_list)))
|
||||
header_data = ['name', 'email', 'isp', 'ip_address']
|
||||
worksheet = workbook.add_worksheet()
|
||||
header_format = workbook.add_format({'bold': True})
|
||||
|
||||
for col_num, data in enumerate(header_data):
|
||||
worksheet.write(row, col_num, data, header_format)
|
||||
row = row + 1
|
||||
for info in contact_list:
|
||||
# Iterate over the data and write it out row by row.
|
||||
worksheet.write(row, col, "{} {}".format(info.last_name, info.first_name))
|
||||
worksheet.write(row, col + 1, info.mail)
|
||||
worksheet.write(row, col + 2, info.isp)
|
||||
worksheet.write(row, col + 3, info.ip_address)
|
||||
row += 1
|
||||
workbook.close()
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
# ip_in_dict = get_info_from_ip("83.137.1.206")
|
||||
# print(ip_in_dict)
|
||||
items_to_save = []
|
||||
# print(len(get_contact_list()))
|
||||
for item in get_contact_list():
|
||||
# print(item)
|
||||
_ip_info = get_info_from_ip(item.ip_address)
|
||||
if _ip_info is None:
|
||||
continue
|
||||
item.isp = _ip_info.isp
|
||||
item.ip_country = _ip_info.country
|
||||
print(item)
|
||||
items_to_save.append(item)
|
||||
sleep(3)
|
||||
write_contact_with_ip_info_to_file(items_to_save)
|
||||
Reference in New Issue
Block a user