Python 實現批量查詢IP並解析為歸屬地

一、背景：
 最近工作中做了一個小功能，目的是為了分析注冊用戶區域分布和訂單的區域分布情況。所以需要將其對應的IP信息解析為歸屬地，並同步每天同步更新。
線上跑起來效率還是有優化的空間，優化的方向：在調用IP查詢API過程可以調整為多線程並行解析IP。后續會更新這方便的調整。
技術： Pyhton3
 postgreSQL
 env配置文件

附加信息：iP地址查詢(iP138官方企業版)：https://market.aliyun.com/products/56928004/cmapi015606.html#sku=yuncode960600002
 .可提供免費的IP查詢API.

二、實現思路： 1、 讀取數據庫IP信息
 2、 調用第三方IP解析API進行解析
 3、 將解析歸屬地信息存入數據庫
三、幾點說明： 1、環境信息等參數配置
 2、日志輸出
 3、異常處理： 數據庫連接異常
 請求連接查詢IP的URL異常：HTTP ERROR 503
 4、json,字典，數組等類型數據輸入輸出
 5、分頁查詢並批量解析
 5.功能實現很簡單，所以就沒有做詳細的介紹了。詳情可直接看完整代碼，有詳細的備注。

四、步驟簡單介紹：
 針對實現思路的3個步驟寫了3個函數，彼此調用執行。
 函數：
 def get_ip_info(table_name):
 def get_ip_area(table_name):
 def ip_write_db(table_name)：
 調用：
 ip_write_db("h_user_stat")
 
五、關鍵代碼說明：

語法：urllib.request.urlopen(url, data=None, [timeout, ]*, cafile=None, capath=None, cadefault=False, context=None)

 # 對從數據庫表中出查詢的IP進行解析
       querys = 'callback&datatype=jsonp&ip=' + get_ip
       bodys = {}
       url = host + path + '?' + querys
       request = urllib.request.Request(url)
       request.add_header('Authorization', 'APPCODE ' + appcode)

       # 連接url時可能會出現 ERROR: HTTP Error 503: Service Unavailable
       try: 
         response = urllib.request.urlopen(request)
       except Exception as e:
         logging.error(e) # 輸出異常日志信息 
         time.sleep(5)
         response = urllib.request.urlopen(request)
       finally:
         content = response.read()
         ip_area = content.decode('utf8')
         ip_area = json.loads(ip_area)['data'] # json類型轉字典類型並取'data'健值
         arr.append([get_ip, ip_area]) # 將結果集存於二元數組

說明：從數據庫分頁查詢固定數量的IP存入數組，並遍歷該數組並將解析后的地區信息data健值存於二元數組中。

 

六、Python代碼實現如下:

# 導入psycopg2包
import psycopg2, time,datetime,sys
import json
import urllib, urllib.request
import os
import configparser
import logging
                     # purpose: 連接數據庫讀取表IP信息
def get_ip_info(table_name):
    # 全局變量作用局部作用域
    global pagesize # 每頁查詢數據條數
    global rows_count

    # 測試1
    starttime_1 = time.time()
    # 建立游標，用來執行數據庫操作
    cur = conn.cursor()
    # 執行SQL命令
    cur.execute("SELECT remote_ip FROM (select remote_ip,min(created_at) from " + table_name + " group by remote_ip) h1 where remote_ip is not null and remote_ip <> '' and  not exists (select 1 from d_ip_area_mapping h2 where h1.remote_ip = h2.remote_ip) limit " + str(pagesize) + ";")


    # 獲取結果集條數
    rows_count = cur.rowcount
   
    # print('解析用戶IP的總數：' + str(rows_count))

     # 當有未解析的用戶的IP，返回元組，否則退出程序
    if rows_count > 0:
      # 獲取SELECT返回的元組
      rows =  cur.fetchall()        # all rows in table

      for row in rows:
          tuple = rows

      conn.commit()
      # 關閉游標
      cur.close()

    else:
        tuple = []
    logging.info('每頁查詢秒數：' + str(time.time() - starttime_1))
    return tuple
    # 調用解析函數


def get_ip_area(table_name):
  # 內包含用戶ID和IP的數組的元組
  tuple = get_ip_info(table_name)  

  # 測試2
  starttime_2 = time.time()
  host = 'http://ali.ip138.com'
  path = '/ip/'
  method = 'GET'
  appcode = '917058e6d7c84104b7cab9819de54b6e'
  arr = []
  for row in tuple:

       get_ip = row[0]
       #get_user = "".join(str(row))
       #get_user = row[0]

            # 對從數據庫表中出查詢的IP進行解析
       querys = 'callback&datatype=jsonp&ip=' + get_ip
       bodys = {}
       url = host + path + '?' + querys
       request = urllib.request.Request(url)
       request.add_header('Authorization', 'APPCODE ' + appcode)

       # 連接url時可能會出現 ERROR: HTTP Error 503: Service Unavailable
       try: 
         response = urllib.request.urlopen(request)
       except Exception as e:
         logging.error(e) # 輸出異常日志信息 
         time.sleep(5)
         response = urllib.request.urlopen(request)
       finally:
         content = response.read()
         ip_area = content.decode('utf8')
         ip_area = json.loads(ip_area)['data'] # json類型轉字典類型並取'data'健值
         arr.append([get_ip, ip_area]) # 將結果集存於二元數組
  logging.info('每頁解析秒數：' + str(time.time() - starttime_2))
  return  arr


def ip_write_db(table_name):
   
    write_ip = get_ip_area(table_name)  # 內包含用戶ID和IP的數組的元組


    # 測試1
    starttime_3 = time.time()

     # 建立游標，用來執行數據庫操作
    cur = conn.cursor()
    for row in write_ip:
        # get_user = row[0]  # 獲取用戶ID
        get_ip = row[0]  # 獲取用戶對應的IP
        country = row[1][0]  # 獲取IP解析后的地區：國家
        province = row[1][1]  # 獲取IP解析后的地區：省
        city = row[1][2]  # 獲取IP解析后的地區：市
        isp = row[1][3]  # 獲取IP解析后的服務提供商

        # 執行SQL命令
        sql = "insert into d_ip_area_mapping(remote_ip,country,province,city,isp,created_at,updated_at,job_id) values (%s,%s,%s,%s,%s,%s,%s,%s);"
        val = [get_ip, country, province, city, isp, time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()),
               time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()),time.strftime("%Y-%m-%d",time.localtime())]
        
        cur.execute(sql, val)
        conn.commit()    
    # 關閉游標
    cur.close()
    logging.info('每頁插入秒數：' + str(time.time() - starttime_3))


# 1.程序開始執行計時
starttime = time.time()


     # 讀取配置文件環境信息 

# 項目路徑
rootDir = os.path.split(os.path.realpath(__file__))[0]


############################### config.env文件路徑  #############################################################

configFilePath = os.path.join(rootDir, 'db_udw.env')
config = configparser.ConfigParser()
config.read(configFilePath)

# 讀取數據庫環境信息
db_database = config.get('postgresql','database')
db_user = config.get('postgresql','user')
db_password = config.get('postgresql','password')
db_host = config.get('postgresql','host')
db_port = config.get('postgresql','port')

# 讀取輸出日志路徑
log = config.get('log','log_path')

# 每頁查詢數據條數
pagesize = config.get('page','pagesize') 

# 讀取解析IP條數限制
ip_num_limit = config.get('ip_num','ip_num_limit') 

# 配置輸出日志格式
logging.basicConfig(level=logging.DEBUG,#控制台打印的日志級別
                      filename='{my_log_path}/ip_analyzer.log'.format(my_log_path=log),  # 指定日志文件及路徑
                      filemode='a',##模式，有w和a，w就是寫模式，每次都會重新寫日志，覆蓋之前的日志 #a是追加模式，默認如果不寫的話，就是追加模式
                      format='%(asctime)s - %(pathname)s[line:%(lineno)d] - %(levelname)s: %(message)s'#日志格式
                      )

###############################   程序開始執行  #############################################################
try:

  # 連接到一個給定的數據庫
  conn = psycopg2.connect(database=db_database, user=db_user, password=db_password, host=db_host, port=db_port)
except Exception as e:
  logging.error(e) # 輸出連接異常日志信息

# 返回查詢行數 默認為0
rows_count = 0
 # 用戶表IP解析總數
user_ip_num = 0 
 # 訂單表IP解析總數
order_ip_num = 0 



try:

  # 解析用戶表注冊IP信息
  while user_ip_num <= eval(ip_num_limit):
     i = 1  # 循環次數
     ip_write_db("h_user_stat")
     user_ip_num = user_ip_num + rows_count*i
     i  = i + 1
     if rows_count == 0 :
         break

  # 解析訂單表下單IP信息 
  while user_ip_num <= eval(ip_num_limit):
      # 解析用戶表注冊IP信息
      i = 1  # 循環次數
      ip_write_db("h_order")
      order_ip_num = order_ip_num + rows_count*i
      i = i + 1
      if rows_count == 0 :
         break
except Exception as e:
  logging.error(e) # 輸出異常日志信息
finally:
  # 關閉數據庫連接
  conn.close()

# 2 程序結束執行計時
  endtime = time.time()
  
  # print('解析用戶IP的總數:' + str(user_ip_num))
  # print('解析訂單IP的總數:' + str(order_ip_num))
  # # 打印程序執行總耗時
  # print('解析總耗時秒數：' + str(endtime - starttime))
  logging.info('解析用戶IP的總數:' + str(user_ip_num))
  logging.info('解析訂單IP的總數:' + str(order_ip_num))
  logging.info('解析總耗時秒數：' + str(endtime - starttime))

 環境配置db_udw.envdb_udw.env 如下：

# 數據庫環境信息
[postgresql]
database = ahaschool_udw
user = admin
password = 123456
host = 127.0.0.0
port = 5432

# 設置日志文件路徑
[log]
log_path = /home/hjmrunning/bi_etl_product/scripts/log

# 每頁查詢數據條數
[page]
pagesize = 1000  

# IP解析條數限制
[ip_num]
ip_num_limit = 150000

最后

    我接觸Python時間也不是很久，實現方法可能會有疏漏。如果有什么疑問和見解，歡迎評論區交流。