Python+Tkinter編寫一個(gè)批量IP地址歸屬地查詢
這是一個(gè)基于Python Tkinter的批量IP地址歸屬地查詢工具。讓我詳細(xì)解讀代碼的各個(gè)部分:
核心功能
批量查詢IP地址的地理位置信息
支持多個(gè)備用API端點(diǎn)
具有重試機(jī)制和請(qǐng)求間隔控制
圖形化用戶界面
效果圖

代碼結(jié)構(gòu)分析
1. 類定義和初始化
class IPQueryTool:
def __init__(self, root):
# 設(shè)置3個(gè)備用API接口
self.apis = [
"https://ipapi.co/{}/json/",
"https://ipapi.com/ip_api.php?ip={}",
"http://ip-api.com/json/{}"
]
self.queue = Queue() # 線程安全的消息隊(duì)列2. GUI界面構(gòu)建 (setup_ui)
輸入?yún)^(qū)域: 多行文本框用于輸入IP列表
配置區(qū)域: 可調(diào)整請(qǐng)求間隔時(shí)間
按鈕區(qū)域: 開(kāi)始查詢和清空結(jié)果按鈕
結(jié)果顯示區(qū): 帶滾動(dòng)條的文本框
狀態(tài)欄: 顯示當(dāng)前操作狀態(tài)
3. 核心查詢流程
多線程處理
def start_query_thread(self):
# 在新線程中執(zhí)行查詢,避免界面卡頓
thread = threading.Thread(target=self.query_ip_addresses)IP驗(yàn)證和批量處理
def query_ip_addresses(self):
# 1. 獲取并驗(yàn)證IP地址
# 2. 逐個(gè)查詢,包含延時(shí)控制
# 3. 統(tǒng)計(jì)成功/失敗數(shù)量帶重試機(jī)制的單個(gè)查詢
def query_single_ip_with_retry(self, ip):
# 最大重試3次
# 遇到限流(429狀態(tài)碼)時(shí)使用指數(shù)退避算法
# 失敗時(shí)自動(dòng)切換API接口4. API響應(yīng)處理適配
支持不同API返回格式的解析:
format_ipapi_result(): 處理ipapi.co格式format_ip-api_result(): 處理ip-api.com格式format_generic_result(): 通用格式處理
5. 線程安全的消息傳遞
def process_queue(self):
# 主線程定時(shí)檢查隊(duì)列,安全更新UI
# 處理不同類型消息:結(jié)果、狀態(tài)、錯(cuò)誤等關(guān)鍵技術(shù)特點(diǎn)
健壯性設(shè)計(jì)
多重API備用: 一個(gè)API失敗自動(dòng)切換下一個(gè)
請(qǐng)求限流控制: 可配置的請(qǐng)求間隔 + 隨機(jī)延遲
指數(shù)退避重試: 2^retry秒的等待時(shí)間
錯(cuò)誤處理: 網(wǎng)絡(luò)超時(shí)、連接錯(cuò)誤的完善處理
用戶體驗(yàn)優(yōu)化
實(shí)時(shí)狀態(tài)更新: 顯示當(dāng)前查詢進(jìn)度
結(jié)果分格式顯示: 清晰的結(jié)果排版
輸入驗(yàn)證: IP格式自動(dòng)校驗(yàn)
防界面卡頓: 多線程設(shè)計(jì)
使用示例
輸入格式:
8.8.8.8
114.114.114.114
202.96.134.133
輸出結(jié)果示例:
==================================================
IP地址: 8.8.8.8
國(guó)家: United States (US)
地區(qū): California
城市: Mountain View
ISP: Google LLC
時(shí)區(qū): America/Los_Angeles
經(jīng)緯度: 37.4056, -122.0775
完整代碼
代碼一:
import tkinter as tk
from tkinter import scrolledtext, messagebox
import requests
import threading
from queue import Queue
import json
class IPQueryTool:
def __init__(self, root):
self.root = root
self.root.title("批量查詢 IP 地址歸屬地")
self.root.geometry("600x700")
# 設(shè)置API端點(diǎn)
self.API_URL = "https://ipapi.co/{}/json/"
# 創(chuàng)建隊(duì)列用于線程間通信
self.queue = Queue()
self.setup_ui()
# 定期檢查隊(duì)列
self.process_queue()
def setup_ui(self):
"""設(shè)置用戶界面"""
# 輸入?yún)^(qū)域
input_frame = tk.Frame(self.root)
input_frame.pack(pady=10, fill=tk.X, padx=10)
tk.Label(input_frame, text="請(qǐng)輸入要查詢的 IP 地址(每行一個(gè)):").pack(anchor=tk.W)
self.ip_entry = scrolledtext.ScrolledText(input_frame, height=8, width=70)
self.ip_entry.pack(pady=5)
# 按鈕區(qū)域
button_frame = tk.Frame(self.root)
button_frame.pack(pady=10)
self.query_button = tk.Button(
button_frame,
text="開(kāi)始查詢",
command=self.start_query_thread,
bg="#4CAF50",
fg="white",
width=15
)
self.query_button.pack(side=tk.LEFT, padx=5)
clear_button = tk.Button(
button_frame,
text="清空結(jié)果",
command=self.clear_results,
bg="#f44336",
fg="white",
width=15
)
clear_button.pack(side=tk.LEFT, padx=5)
# 結(jié)果區(qū)域
result_frame = tk.Frame(self.root)
result_frame.pack(pady=10, fill=tk.BOTH, expand=True, padx=10)
tk.Label(result_frame, text="查詢結(jié)果:").pack(anchor=tk.W)
self.result_text = scrolledtext.ScrolledText(result_frame, height=20, width=70)
self.result_text.pack(pady=5, fill=tk.BOTH, expand=True)
# 狀態(tài)欄
self.status_var = tk.StringVar()
self.status_var.set("就緒")
status_bar = tk.Label(self.root, textvariable=self.status_var, relief=tk.SUNKEN, anchor=tk.W)
status_bar.pack(side=tk.BOTTOM, fill=tk.X)
def start_query_thread(self):
"""在新線程中啟動(dòng)查詢"""
self.query_button.config(state=tk.DISABLED)
self.status_var.set("查詢中...")
thread = threading.Thread(target=self.query_ip_addresses)
thread.daemon = True
thread.start()
def query_ip_addresses(self):
"""查詢IP地址信息"""
ip_text = self.ip_entry.get(1.0, tk.END).strip()
ip_list = [ip.strip() for ip in ip_text.splitlines() if ip.strip()]
if not ip_list:
self.queue.put(("error", "請(qǐng)輸入至少一個(gè)IP地址"))
return
valid_ips = []
for ip in ip_list:
if self.is_valid_ip(ip):
valid_ips.append(ip)
else:
self.queue.put(("result", f"無(wú)效的IP地址格式: {ip}\n"))
if not valid_ips:
self.queue.put(("error", "沒(méi)有有效的IP地址可查詢"))
return
success_count = 0
total = len(valid_ips)
for i, ip in enumerate(valid_ips, 1):
try:
self.queue.put(("status", f"正在查詢 {i}/{total}: {ip}"))
response = requests.get(self.API_URL.format(ip), timeout=10)
if response.status_code == 200:
data = response.json()
result = self.format_result(ip, data)
self.queue.put(("result", result))
success_count += 1
else:
error_msg = f"查詢 IP {ip} 失敗,狀態(tài)碼: {response.status_code}\n"
self.queue.put(("result", error_msg))
except requests.exceptions.Timeout:
self.queue.put(("result", f"查詢 IP {ip} 超時(shí)\n"))
except requests.exceptions.ConnectionError:
self.queue.put(("result", f"網(wǎng)絡(luò)連接錯(cuò)誤,無(wú)法查詢 IP {ip}\n"))
except Exception as e:
self.queue.put(("result", f"查詢 IP {ip} 時(shí)出錯(cuò): {str(e)}\n"))
# 查詢完成
self.queue.put(("status", f"查詢完成!成功: {success_count}/{total}"))
self.queue.put(("complete", ""))
def is_valid_ip(self, ip):
"""簡(jiǎn)單的IP地址格式驗(yàn)證"""
parts = ip.split('.')
if len(parts) != 4:
return False
for part in parts:
if not part.isdigit():
return False
if not 0 <= int(part) <= 255:
return False
return True
def format_result(self, ip, data):
"""格式化查詢結(jié)果"""
result = f"{'=' * 50}\n"
result += f"IP地址: {ip}\n"
result += f"國(guó)家: {data.get('country_name', 'N/A')} ({data.get('country', 'N/A')})\n"
result += f"地區(qū): {data.get('region', 'N/A')}\n"
result += f"城市: {data.get('city', 'N/A')}\n"
result += f"ISP: {data.get('org', 'N/A')}\n"
result += f"時(shí)區(qū): {data.get('timezone', 'N/A')}\n"
result += f"經(jīng)緯度: {data.get('latitude', 'N/A')}, {data.get('longitude', 'N/A')}\n"
result += "\n"
return result
def process_queue(self):
"""處理隊(duì)列中的消息(在主線程中執(zhí)行)"""
try:
while True:
msg_type, content = self.queue.get_nowait()
if msg_type == "result":
self.result_text.insert(tk.END, content)
self.result_text.see(tk.END)
elif msg_type == "status":
self.status_var.set(content)
elif msg_type == "error":
messagebox.showerror("錯(cuò)誤", content)
self.status_var.set("錯(cuò)誤")
elif msg_type == "complete":
self.query_button.config(state=tk.NORMAL)
self.status_var.set("查詢完成")
self.root.update_idletasks()
except:
pass
# 每100毫秒檢查一次隊(duì)列
self.root.after(100, self.process_queue)
def clear_results(self):
"""清空結(jié)果文本框"""
self.result_text.delete(1.0, tk.END)
def main():
root = tk.Tk()
app = IPQueryTool(root)
root.mainloop()
if __name__ == "__main__":
main()代碼二
解讀以下代碼:
import tkinter as tk
from tkinter import scrolledtext, messagebox
import requests
import threading
from queue import Queue
import time
import random
class IPQueryTool:
def __init__(self, root):
self.root = root
self.root.title("批量查詢 IP 地址歸屬地")
self.root.geometry("600x700")
# 設(shè)置多個(gè)API端點(diǎn)作為備用
self.apis = [
"https://ipapi.co/{}/json/",
"https://ipapi.com/ip_api.php?ip={}",
"http://ip-api.com/json/{}"
]
self.current_api_index = 0
# 請(qǐng)求配置
self.delay_between_requests = 1.0 # 請(qǐng)求間隔(秒)
self.max_retries = 3 # 最大重試次數(shù)
self.queue = Queue()
self.setup_ui()
self.process_queue()
def setup_ui(self):
"""設(shè)置用戶界面"""
# 輸入?yún)^(qū)域
input_frame = tk.Frame(self.root)
input_frame.pack(pady=10, fill=tk.X, padx=10)
tk.Label(input_frame, text="請(qǐng)輸入要查詢的 IP 地址(每行一個(gè)):").pack(anchor=tk.W)
self.ip_entry = scrolledtext.ScrolledText(input_frame, height=8, width=70)
self.ip_entry.pack(pady=5)
# 配置區(qū)域
config_frame = tk.Frame(self.root)
config_frame.pack(pady=5, fill=tk.X, padx=10)
tk.Label(config_frame, text="請(qǐng)求間隔(秒):").pack(side=tk.LEFT)
self.delay_var = tk.StringVar(value="1.0")
delay_entry = tk.Entry(config_frame, textvariable=self.delay_var, width=5)
delay_entry.pack(side=tk.LEFT, padx=5)
# 按鈕區(qū)域
button_frame = tk.Frame(self.root)
button_frame.pack(pady=10)
self.query_button = tk.Button(
button_frame,
text="開(kāi)始查詢",
command=self.start_query_thread,
bg="#4CAF50",
fg="white",
width=15
)
self.query_button.pack(side=tk.LEFT, padx=5)
clear_button = tk.Button(
button_frame,
text="清空結(jié)果",
command=self.clear_results,
bg="#f44336",
fg="white",
width=15
)
clear_button.pack(side=tk.LEFT, padx=5)
# 結(jié)果區(qū)域
result_frame = tk.Frame(self.root)
result_frame.pack(pady=10, fill=tk.BOTH, expand=True, padx=10)
tk.Label(result_frame, text="查詢結(jié)果:").pack(anchor=tk.W)
self.result_text = scrolledtext.ScrolledText(result_frame, height=20, width=70)
self.result_text.pack(pady=5, fill=tk.BOTH, expand=True)
# 狀態(tài)欄
self.status_var = tk.StringVar()
self.status_var.set("就緒")
status_bar = tk.Label(self.root, textvariable=self.status_var, relief=tk.SUNKEN, anchor=tk.W)
status_bar.pack(side=tk.BOTTOM, fill=tk.X)
def get_current_api(self):
"""獲取當(dāng)前API端點(diǎn)"""
return self.apis[self.current_api_index]
def switch_to_next_api(self):
"""切換到下一個(gè)API端點(diǎn)"""
self.current_api_index = (self.current_api_index + 1) % len(self.apis)
return self.get_current_api()
def start_query_thread(self):
"""在新線程中啟動(dòng)查詢"""
try:
self.delay_between_requests = float(self.delay_var.get())
if self.delay_between_requests < 0.5:
messagebox.showwarning("警告", "請(qǐng)求間隔太短可能導(dǎo)致限流,建議至少0.5秒")
self.delay_between_requests = 0.5
self.delay_var.set("0.5")
except ValueError:
messagebox.showerror("錯(cuò)誤", "請(qǐng)輸入有效的數(shù)字")
return
self.query_button.config(state=tk.DISABLED)
self.status_var.set("查詢中...")
thread = threading.Thread(target=self.query_ip_addresses)
thread.daemon = True
thread.start()
def query_ip_addresses(self):
"""查詢IP地址信息"""
ip_text = self.ip_entry.get(1.0, tk.END).strip()
ip_list = [ip.strip() for ip in ip_text.splitlines() if ip.strip()]
if not ip_list:
self.queue.put(("error", "請(qǐng)輸入至少一個(gè)IP地址"))
return
valid_ips = []
for ip in ip_list:
if self.is_valid_ip(ip):
valid_ips.append(ip)
else:
self.queue.put(("result", f"無(wú)效的IP地址格式: {ip}\n"))
if not valid_ips:
self.queue.put(("error", "沒(méi)有有效的IP地址可查詢"))
return
success_count = 0
total = len(valid_ips)
api_switched = False
for i, ip in enumerate(valid_ips, 1):
# 添加請(qǐng)求間隔
if i > 1:
time.sleep(self.delay_between_requests + random.uniform(0, 0.5))
try:
self.queue.put(("status", f"正在查詢 {i}/{total}: {ip}"))
result = self.query_single_ip_with_retry(ip)
if result:
self.queue.put(("result", result))
success_count += 1
else:
self.queue.put(("result", f"查詢 IP {ip} 失敗\n"))
except Exception as e:
self.queue.put(("result", f"查詢 IP {ip} 時(shí)出錯(cuò): {str(e)}\n"))
# 查詢完成
self.queue.put(("status", f"查詢完成!成功: {success_count}/{total}"))
self.queue.put(("complete", ""))
def query_single_ip_with_retry(self, ip):
"""帶重試機(jī)制的單個(gè)IP查詢"""
for retry in range(self.max_retries):
try:
api_url = self.get_current_api().format(ip)
headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
}
response = requests.get(api_url, headers=headers, timeout=10)
if response.status_code == 200:
return self.process_successful_response(ip, response.json())
elif response.status_code == 429:
self.queue.put(("status", f"IP {ip} 被限流,等待重試..."))
if retry < self.max_retries - 1:
# 指數(shù)退避
wait_time = (2 ** retry) + random.uniform(0, 1)
time.sleep(wait_time)
continue
else:
self.queue.put(("status", "切換API端點(diǎn)..."))
self.switch_to_next_api()
else:
self.queue.put(("status", f"API錯(cuò)誤,狀態(tài)碼: {response.status_code}"))
except requests.exceptions.Timeout:
self.queue.put(("status", f"查詢 IP {ip} 超時(shí),重試 {retry + 1}/{self.max_retries}"))
except requests.exceptions.ConnectionError:
self.queue.put(("status", f"網(wǎng)絡(luò)連接錯(cuò)誤,重試 {retry + 1}/{self.max_retries}"))
except Exception as e:
self.queue.put(("status", f"錯(cuò)誤: {str(e)},重試 {retry + 1}/{self.max_retries}"))
if retry < self.max_retries - 1:
time.sleep(1 + retry) # 重試前等待
return None
def process_successful_response(self, ip, data):
"""處理成功的API響應(yīng)"""
# 不同API的響應(yīng)格式可能不同,需要適配
if 'ipapi.co' in self.get_current_api():
return self.format_ipapi_result(ip, data)
elif 'ip-api.com' in self.get_current_api():
return self.format_ip-api_result(ip, data)
else:
return self.format_generic_result(ip, data)
def format_ipapi_result(self, ip, data):
"""格式化ipapi.co的結(jié)果"""
result = f"{'='*50}\n"
result += f"IP地址: {ip}\n"
result += f"國(guó)家: {data.get('country_name', 'N/A')} ({data.get('country_code', 'N/A')})\n"
result += f"地區(qū): {data.get('region', 'N/A')}\n"
result += f"城市: {data.get('city', 'N/A')}\n"
result += f"ISP: {data.get('org', 'N/A')}\n"
result += f"時(shí)區(qū): {data.get('timezone', 'N/A')}\n"
result += f"經(jīng)緯度: {data.get('latitude', 'N/A')}, {data.get('longitude', 'N/A')}\n"
result += "\n"
return result
def format_ip-api_result(self, ip, data):
"""格式化ip-api.com的結(jié)果"""
if data.get('status') == 'success':
result = f"{'='*50}\n"
result += f"IP地址: {ip}\n"
result += f"國(guó)家: {data.get('country', 'N/A')}\n"
result += f"地區(qū): {data.get('regionName', 'N/A')}\n"
result += f"城市: {data.get('city', 'N/A')}\n"
result += f"ISP: {data.get('isp', 'N/A')}\n"
result += f"時(shí)區(qū): {data.get('timezone', 'N/A')}\n"
result += f"經(jīng)緯度: {data.get('lat', 'N/A')}, {data.get('lon', 'N/A')}\n"
result += "\n"
return result
else:
return f"查詢 IP {ip} 失敗: {data.get('message', '未知錯(cuò)誤')}\n"
def format_generic_result(self, ip, data):
"""通用結(jié)果格式化"""
result = f"{'='*50}\n"
result += f"IP地址: {ip}\n"
# 嘗試提取常見(jiàn)字段
mapping = {
'country': ['country', 'country_name', 'countryName'],
'region': ['region', 'regionName', 'state'],
'city': ['city'],
'isp': ['isp', 'org', 'asn'],
'timezone': ['timezone'],
}
for field, keys in mapping.items():
value = 'N/A'
for key in keys:
if key in data:
value = data[key]
break
result += f"{field}: {value}\n"
result += "\n"
return result
def is_valid_ip(self, ip):
"""簡(jiǎn)單的IP地址格式驗(yàn)證"""
parts = ip.split('.')
if len(parts) != 4:
return False
for part in parts:
if not part.isdigit():
return False
if not 0 <= int(part) <= 255:
return False
return True
def process_queue(self):
"""處理隊(duì)列中的消息"""
try:
while True:
msg_type, content = self.queue.get_nowait()
if msg_type == "result":
self.result_text.insert(tk.END, content)
self.result_text.see(tk.END)
elif msg_type == "status":
self.status_var.set(content)
elif msg_type == "error":
messagebox.showerror("錯(cuò)誤", content)
self.status_var.set("錯(cuò)誤")
elif msg_type == "complete":
self.query_button.config(state=tk.NORMAL)
self.status_var.set("查詢完成")
self.root.update_idletasks()
except:
pass
self.root.after(100, self.process_queue)
def clear_results(self):
"""清空結(jié)果文本框"""
self.result_text.delete(1.0, tk.END)
def main():
root = tk.Tk()
app = IPQueryTool(root)
root.mainloop()
if __name__ == "__main__":
main()到此這篇關(guān)于Python+Tkinter編寫一個(gè)批量IP地址歸屬地查詢的文章就介紹到這了,更多相關(guān)Python查詢IP地址歸屬地內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家!
相關(guān)文章
python中format()函數(shù)的簡(jiǎn)單使用教程
python中format函數(shù)用于字符串的格式化,接下來(lái)通過(guò)本文給大家介紹python中format()函數(shù)的簡(jiǎn)單使用教程,一起看看吧2018-03-03
利用Python的Twisted框架實(shí)現(xiàn)webshell密碼掃描器的教程
這篇文章主要介紹了利用Python的Twisted框架實(shí)現(xiàn)webshell密碼掃描器的教程,用到了Twisted框架的異步通信機(jī)制,需要的朋友可以參考下2015-04-04
Python實(shí)現(xiàn)的文本對(duì)比報(bào)告生成工具示例
這篇文章主要介紹了Python實(shí)現(xiàn)的文本對(duì)比報(bào)告生成工具,涉及Python基于difflib模塊實(shí)現(xiàn)對(duì)文本內(nèi)容進(jìn)行對(duì)比的相關(guān)操作技巧,需要的朋友可以參考下2018-05-05
Python實(shí)現(xiàn)將txt里面多行json字符串轉(zhuǎn)成excel文件
這篇文章主要為大家詳細(xì)介紹了Python實(shí)現(xiàn)將txt里面多行json字符串轉(zhuǎn)成excel文件的相關(guān)方法,文中的示例代碼講解詳細(xì),感興趣的小伙伴可以了解下2025-12-12
Python報(bào)mongod:?error?while?loading?shared?libraries:?l
這篇文章主要介紹的是Python報(bào)mongod:?error?while?loading?shared?libraries:?libcrypto.so.1.1的解決方法,下面文章解決過(guò)程,需要的小伙伴可以參考一下2022-02-02
python HTTPX庫(kù)實(shí)現(xiàn)同步異步請(qǐng)求用法示例
這篇文章主要為大家介紹了python HTTPX庫(kù)實(shí)現(xiàn)同步異步請(qǐng)求用法示例,有需要的朋友可以借鑒參考下,希望能夠有所幫助,祝大家多多進(jìn)步,早日升職加薪2024-01-01
Python訪問(wèn)純真IP數(shù)據(jù)庫(kù)腳本分享
這篇文章主要介紹了Python訪問(wèn)純真IP數(shù)據(jù)庫(kù)腳本分享,本文直接給出實(shí)現(xiàn)代碼,需要的朋友可以參考下2015-06-06

