最新国产好看的视频,伊人天堂AV在线,国产Aaaaaa视频,蜜臀视频在线观看一区,人妻av色图,密臀久久久精品影片,青青视频免费观看毛片,久草在线观看视,国产三级精品色情在线

Python+Tkinter編寫一個(gè)批量IP地址歸屬地查詢

 更新時(shí)間:2025年11月11日 08:57:35   作者:Atlas Shepherd  
這篇文章主要為大家詳細(xì)介紹了Python如何結(jié)合Tkinter編寫一個(gè)批量IP地址歸屬地查詢,文中的示例代碼講解詳細(xì),感興趣的小伙伴可以跟隨小編一起學(xué)習(xí)一下

這是一個(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)文章

最新評(píng)論

科尔| 莫力| 连山| 陕西省| 迭部县| 榆林市| 清河县| 招远市| 连州市| 西林县| 大埔县| 册亨县| 秦安县| 石首市| 定南县| 桃园市| 淮阳县| 西乌| 安塞县| 嘉黎县| 文成县| 贵德县| 榆社县| 松江区| 泰州市| 丰宁| 阿坝| 蒙阴县| 连平县| 青岛市| 屏山县| 阿坝县| 龙岩市| 确山县| 和静县| 澜沧| 孟津县| 蓬溪县| 中阳县| 阿城市| 靖安县|