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

代碼結(jié)構(gòu)分析
1. 類定義和初始化
class IPQueryTool:
def __init__(self, root):
# 設(shè)置3個備用API接口
self.apis = [
"https://ipapi.co/{}/json/",
"https://ipapi.com/ip_api.php?ip={}",
"http://ip-api.com/json/{}"
]
self.queue = Queue() # 線程安全的消息隊列2. GUI界面構(gòu)建 (setup_ui)
輸入?yún)^(qū)域: 多行文本框用于輸入IP列表
配置區(qū)域: 可調(diào)整請求間隔時間
按鈕區(qū)域: 開始查詢和清空結(jié)果按鈕
結(jié)果顯示區(qū): 帶滾動條的文本框
狀態(tài)欄: 顯示當前操作狀態(tài)
3. 核心查詢流程
多線程處理
def start_query_thread(self):
# 在新線程中執(zhí)行查詢,避免界面卡頓
thread = threading.Thread(target=self.query_ip_addresses)IP驗證和批量處理
def query_ip_addresses(self):
# 1. 獲取并驗證IP地址
# 2. 逐個查詢,包含延時控制
# 3. 統(tǒng)計成功/失敗數(shù)量帶重試機制的單個查詢
def query_single_ip_with_retry(self, ip):
# 最大重試3次
# 遇到限流(429狀態(tài)碼)時使用指數(shù)退避算法
# 失敗時自動切換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):
# 主線程定時檢查隊列,安全更新UI
# 處理不同類型消息:結(jié)果、狀態(tài)、錯誤等關(guān)鍵技術(shù)特點
健壯性設(shè)計
多重API備用: 一個API失敗自動切換下一個
請求限流控制: 可配置的請求間隔 + 隨機延遲
指數(shù)退避重試: 2^retry秒的等待時間
錯誤處理: 網(wǎng)絡(luò)超時、連接錯誤的完善處理
用戶體驗優(yōu)化
實時狀態(tài)更新: 顯示當前查詢進度
結(jié)果分格式顯示: 清晰的結(jié)果排版
輸入驗證: IP格式自動校驗
防界面卡頓: 多線程設(shè)計
使用示例
輸入格式:
8.8.8.8
114.114.114.114
202.96.134.133
輸出結(jié)果示例:
==================================================
IP地址: 8.8.8.8
國家: United States (US)
地區(qū): California
城市: Mountain View
ISP: Google LLC
時區(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端點
self.API_URL = "https://ipapi.co/{}/json/"
# 創(chuàng)建隊列用于線程間通信
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="請輸入要查詢的 IP 地址(每行一個):").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="開始查詢",
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):
"""在新線程中啟動查詢"""
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", "請輸入至少一個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"無效的IP地址格式: {ip}\n"))
if not valid_ips:
self.queue.put(("error", "沒有有效的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} 超時\n"))
except requests.exceptions.ConnectionError:
self.queue.put(("result", f"網(wǎng)絡(luò)連接錯誤,無法查詢 IP {ip}\n"))
except Exception as e:
self.queue.put(("result", f"查詢 IP {ip} 時出錯: {str(e)}\n"))
# 查詢完成
self.queue.put(("status", f"查詢完成!成功: {success_count}/{total}"))
self.queue.put(("complete", ""))
def is_valid_ip(self, ip):
"""簡單的IP地址格式驗證"""
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"國家: {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"時區(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):
"""處理隊列中的消息(在主線程中執(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("錯誤", content)
self.status_var.set("錯誤")
elif msg_type == "complete":
self.query_button.config(state=tk.NORMAL)
self.status_var.set("查詢完成")
self.root.update_idletasks()
except:
pass
# 每100毫秒檢查一次隊列
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è)置多個API端點作為備用
self.apis = [
"https://ipapi.co/{}/json/",
"https://ipapi.com/ip_api.php?ip={}",
"http://ip-api.com/json/{}"
]
self.current_api_index = 0
# 請求配置
self.delay_between_requests = 1.0 # 請求間隔(秒)
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="請輸入要查詢的 IP 地址(每行一個):").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="請求間隔(秒):").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="開始查詢",
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):
"""獲取當前API端點"""
return self.apis[self.current_api_index]
def switch_to_next_api(self):
"""切換到下一個API端點"""
self.current_api_index = (self.current_api_index + 1) % len(self.apis)
return self.get_current_api()
def start_query_thread(self):
"""在新線程中啟動查詢"""
try:
self.delay_between_requests = float(self.delay_var.get())
if self.delay_between_requests < 0.5:
messagebox.showwarning("警告", "請求間隔太短可能導致限流,建議至少0.5秒")
self.delay_between_requests = 0.5
self.delay_var.set("0.5")
except ValueError:
messagebox.showerror("錯誤", "請輸入有效的數(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", "請輸入至少一個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"無效的IP地址格式: {ip}\n"))
if not valid_ips:
self.queue.put(("error", "沒有有效的IP地址可查詢"))
return
success_count = 0
total = len(valid_ips)
api_switched = False
for i, ip in enumerate(valid_ips, 1):
# 添加請求間隔
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} 時出錯: {str(e)}\n"))
# 查詢完成
self.queue.put(("status", f"查詢完成!成功: {success_count}/{total}"))
self.queue.put(("complete", ""))
def query_single_ip_with_retry(self, ip):
"""帶重試機制的單個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端點..."))
self.switch_to_next_api()
else:
self.queue.put(("status", f"API錯誤,狀態(tài)碼: {response.status_code}"))
except requests.exceptions.Timeout:
self.queue.put(("status", f"查詢 IP {ip} 超時,重試 {retry + 1}/{self.max_retries}"))
except requests.exceptions.ConnectionError:
self.queue.put(("status", f"網(wǎng)絡(luò)連接錯誤,重試 {retry + 1}/{self.max_retries}"))
except Exception as e:
self.queue.put(("status", f"錯誤: {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"國家: {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"時區(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"國家: {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"時區(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', '未知錯誤')}\n"
def format_generic_result(self, ip, data):
"""通用結(jié)果格式化"""
result = f"{'='*50}\n"
result += f"IP地址: {ip}\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):
"""簡單的IP地址格式驗證"""
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):
"""處理隊列中的消息"""
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("錯誤", content)
self.status_var.set("錯誤")
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編寫一個批量IP地址歸屬地查詢的文章就介紹到這了,更多相關(guān)Python查詢IP地址歸屬地內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家!
相關(guān)文章
利用Python的Twisted框架實現(xiàn)webshell密碼掃描器的教程
這篇文章主要介紹了利用Python的Twisted框架實現(xiàn)webshell密碼掃描器的教程,用到了Twisted框架的異步通信機制,需要的朋友可以參考下2015-04-04
Python實現(xiàn)將txt里面多行json字符串轉(zhuǎn)成excel文件
這篇文章主要為大家詳細介紹了Python實現(xiàn)將txt里面多行json字符串轉(zhuǎn)成excel文件的相關(guān)方法,文中的示例代碼講解詳細,感興趣的小伙伴可以了解下2025-12-12
Python報mongod:?error?while?loading?shared?libraries:?l
這篇文章主要介紹的是Python報mongod:?error?while?loading?shared?libraries:?libcrypto.so.1.1的解決方法,下面文章解決過程,需要的小伙伴可以參考一下2022-02-02
python HTTPX庫實現(xiàn)同步異步請求用法示例
這篇文章主要為大家介紹了python HTTPX庫實現(xiàn)同步異步請求用法示例,有需要的朋友可以借鑒參考下,希望能夠有所幫助,祝大家多多進步,早日升職加薪2024-01-01

