基于Python實現(xiàn)局域網(wǎng)內(nèi)Windows桌面文件傳輸

更新時間：2025年11月23日 14:12:52 作者：燕鵬01

這篇文章介紹了如何使用Python實現(xiàn)一個局域網(wǎng)文件傳輸系統(tǒng),包括發(fā)送端和接收端的代碼示例,發(fā)送端和接收端都需要在同一局域網(wǎng)內(nèi)運行,并且確保防火墻允許通信,圖形界面版本需要安裝tkinter庫,需要的朋友可以參考下

發(fā)送端代碼 (sender.py)

import socket
import os
import tqdm
import argparse

def send_file(host, port, file_path):
    """
    向指定主機(jī)發(fā)送文件
    """
    # 獲取文件大小
    file_size = os.path.getsize(file_path)
    filename = os.path.basename(file_path)
    
    # 創(chuàng)建socket對象
    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    
    print(f"[+] 正在連接到 {host}:{port}")
    s.connect((host, port))
    print("[+] 連接成功")
    
    # 發(fā)送文件名和文件大小
    file_info = f"{filename}|{file_size}"
    s.send(file_info.encode())
    
    # 等待確認(rèn)
    response = s.recv(1024).decode()
    if response != "READY":
        print("[-] 接收端未準(zhǔn)備就緒")
        s.close()
        return
    
    # 發(fā)送文件數(shù)據(jù)
    progress = tqdm.tqdm(range(file_size), f"發(fā)送 {filename}", unit="B", unit_scale=True, unit_divisor=1024)
    
    with open(file_path, "rb") as f:
        while True:
            # 讀取文件數(shù)據(jù)
            bytes_read = f.read(4096)
            if not bytes_read:
                # 文件傳輸完成
                break
            # 發(fā)送數(shù)據(jù)
            s.sendall(bytes_read)
            # 更新進(jìn)度條
            progress.update(len(bytes_read))
    
    progress.close()
    print("[+] 文件發(fā)送完成")
    s.close()

if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="文件發(fā)送端")
    parser.add_argument("host", help="目標(biāo)主機(jī)IP地址")
    parser.add_argument("file", help="要發(fā)送的文件路徑")
    parser.add_argument("-p", "--port", type=int, default=8888, help="端口號 (默認(rèn): 8888)")
    
    args = parser.parse_args()
    
    if not os.path.exists(args.file):
        print(f"[-] 文件 {args.file} 不存在")
        exit(1)
    
    send_file(args.host, args.port, args.file)

接收端代碼 (receiver.py)

import socket
import os
import tqdm

def start_server(host, port, save_dir):
    """
    啟動文件接收服務(wù)器
    """
    # 創(chuàng)建socket對象
    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    
    # 綁定地址和端口
    s.bind((host, port))
    
    # 開始監(jiān)聽
    s.listen(5)
    print(f"[+] 服務(wù)器正在 {host}:{port} 上監(jiān)聽...")
    
    while True:
        # 接受連接
        client_socket, address = s.accept()
        print(f"[+] 收到來自 {address} 的連接")
        
        # 接收文件信息
        file_info = client_socket.recv(1024).decode()
        filename, file_size = file_info.split("|")
        file_size = int(file_size)
        
        print(f"[+] 接收文件: {filename} ({file_size} 字節(jié))")
        
        # 確認(rèn)準(zhǔn)備接收
        client_socket.send("READY".encode())
        
        # 確保保存目錄存在
        if not os.path.exists(save_dir):
            os.makedirs(save_dir)
        
        # 文件保存路徑
        file_path = os.path.join(save_dir, filename)
        
        # 接收文件數(shù)據(jù)
        progress = tqdm.tqdm(range(file_size), f"接收 {filename}", unit="B", unit_scale=True, unit_divisor=1024)
        
        with open(file_path, "wb") as f:
            received_bytes = 0
            while received_bytes < file_size:
                # 接收數(shù)據(jù)
                bytes_read = client_socket.recv(4096)
                if not bytes_read:
                    break
                # 寫入文件
                f.write(bytes_read)
                received_bytes += len(bytes_read)
                # 更新進(jìn)度條
                progress.update(len(bytes_read))
        
        progress.close()
        print(f"[+] 文件已保存到: {file_path}")
        client_socket.close()

if __name__ == "__main__":
    import argparse
    
    parser = argparse.ArgumentParser(description="文件接收端")
    parser.add_argument("-H", "--host", default="0.0.0.0", help="監(jiān)聽地址 (默認(rèn): 0.0.0.0)")
    parser.add_argument("-p", "--port", type=int, default=8888, help="監(jiān)聽端口 (默認(rèn): 8888)")
    parser.add_argument("-d", "--dir", default="received_files", help="文件保存目錄 (默認(rèn): received_files)")
    
    args = parser.parse_args()
    
    start_server(args.host, args.port, args.dir)

圖形界面版本 (可選)

如果你想要圖形界面，可以使用以下代碼：

import tkinter as tk
from tkinter import filedialog, messagebox, ttk
import threading
import socket
import os
import tqdm

class FileTransferGUI:
    def __init__(self, root):
        self.root = root
        self.root.title("局域網(wǎng)文件傳輸工具")
        self.root.geometry("500x400")
        
        self.setup_ui()
        
    def setup_ui(self):
        # 發(fā)送文件框架
        send_frame = ttk.LabelFrame(self.root, text="發(fā)送文件", padding=10)
        send_frame.pack(fill="x", padx=10, pady=5)
        
        ttk.Label(send_frame, text="目標(biāo)IP:").grid(row=0, column=0, sticky="w", pady=5)
        self.target_ip = ttk.Entry(send_frame, width=20)
        self.target_ip.grid(row=0, column=1, pady=5, padx=5)
        self.target_ip.insert(0, "192.168.")
        
        ttk.Label(send_frame, text="端口:").grid(row=0, column=2, sticky="w", pady=5, padx=(20,0))
        self.port_send = ttk.Entry(send_frame, width=10)
        self.port_send.grid(row=0, column=3, pady=5, padx=5)
        self.port_send.insert(0, "8888")
        
        ttk.Label(send_frame, text="文件路徑:").grid(row=1, column=0, sticky="w", pady=5)
        self.file_path = ttk.Entry(send_frame, width=30)
        self.file_path.grid(row=1, column=1, columnspan=2, pady=5, padx=5, sticky="we")
        
        ttk.Button(send_frame, text="瀏覽", command=self.browse_file).grid(row=1, column=3, pady=5, padx=5)
        ttk.Button(send_frame, text="發(fā)送文件", command=self.send_file).grid(row=2, column=0, columnspan=4, pady=10)
        
        # 接收文件框架
        receive_frame = ttk.LabelFrame(self.root, text="接收文件", padding=10)
        receive_frame.pack(fill="x", padx=10, pady=5)
        
        ttk.Label(receive_frame, text="監(jiān)聽IP:").grid(row=0, column=0, sticky="w", pady=5)
        self.listen_ip = ttk.Entry(receive_frame, width=20)
        self.listen_ip.grid(row=0, column=1, pady=5, padx=5)
        self.listen_ip.insert(0, "0.0.0.0")
        
        ttk.Label(receive_frame, text="端口:").grid(row=0, column=2, sticky="w", pady=5, padx=(20,0))
        self.port_receive = ttk.Entry(receive_frame, width=10)
        self.port_receive.grid(row=0, column=3, pady=5, padx=5)
        self.port_receive.insert(0, "8888")
        
        ttk.Label(receive_frame, text="保存目錄:").grid(row=1, column=0, sticky="w", pady=5)
        self.save_dir = ttk.Entry(receive_frame, width=30)
        self.save_dir.grid(row=1, column=1, columnspan=2, pady=5, padx=5, sticky="we")
        self.save_dir.insert(0, "received_files")
        
        ttk.Button(receive_frame, text="瀏覽", command=self.browse_dir).grid(row=1, column=3, pady=5, padx=5)
        
        self.start_server_btn = ttk.Button(receive_frame, text="啟動接收服務(wù)", command=self.toggle_server)
        self.start_server_btn.grid(row=2, column=0, columnspan=4, pady=10)
        
        self.server_running = False
        self.server_thread = None
        
        # 日志框架
        log_frame = ttk.LabelFrame(self.root, text="日志", padding=10)
        log_frame.pack(fill="both", expand=True, padx=10, pady=5)
        
        self.log_text = tk.Text(log_frame, height=10)
        scrollbar = ttk.Scrollbar(log_frame, orient="vertical", command=self.log_text.yview)
        self.log_text.configure(yscrollcommand=scrollbar.set)
        
        self.log_text.pack(side="left", fill="both", expand=True)
        scrollbar.pack(side="right", fill="y")
    
    def browse_file(self):
        filename = filedialog.askopenfilename()
        if filename:
            self.file_path.delete(0, tk.END)
            self.file_path.insert(0, filename)
    
    def browse_dir(self):
        directory = filedialog.askdirectory()
        if directory:
            self.save_dir.delete(0, tk.END)
            self.save_dir.insert(0, directory)
    
    def log(self, message):
        self.log_text.insert(tk.END, f"{message}\n")
        self.log_text.see(tk.END)
        self.root.update_idletasks()
    
    def send_file(self):
        host = self.target_ip.get()
        port = int(self.port_send.get())
        file_path = self.file_path.get()
        
        if not host or not file_path:
            messagebox.showerror("錯誤", "請?zhí)顚懩繕?biāo)IP和文件路徑")
            return
        
        if not os.path.exists(file_path):
            messagebox.showerror("錯誤", "文件不存在")
            return
        
        # 在新線程中發(fā)送文件
        thread = threading.Thread(target=self._send_file, args=(host, port, file_path))
        thread.daemon = True
        thread.start()
    
    def _send_file(self, host, port, file_path):
        try:
            file_size = os.path.getsize(file_path)
            filename = os.path.basename(file_path)
            
            s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            self.log(f"[+] 正在連接到 {host}:{port}")
            s.connect((host, port))
            self.log("[+] 連接成功")
            
            file_info = f"{filename}|{file_size}"
            s.send(file_info.encode())
            
            response = s.recv(1024).decode()
            if response != "READY":
                self.log("[-] 接收端未準(zhǔn)備就緒")
                s.close()
                return
            
            self.log(f"[+] 開始發(fā)送文件: {filename} ({file_size} 字節(jié))")
            
            with open(file_path, "rb") as f:
                sent_bytes = 0
                while True:
                    bytes_read = f.read(4096)
                    if not bytes_read:
                        break
                    s.sendall(bytes_read)
                    sent_bytes += len(bytes_read)
                    progress = (sent_bytes / file_size) * 100
                    self.log(f"[+] 發(fā)送進(jìn)度: {progress:.1f}%")
            
            self.log("[+] 文件發(fā)送完成")
            s.close()
        except Exception as e:
            self.log(f"[-] 發(fā)送文件時出錯: {str(e)}")
    
    def toggle_server(self):
        if not self.server_running:
            self.start_server()
        else:
            self.stop_server()
    
    def start_server(self):
        host = self.listen_ip.get()
        port = int(self.port_receive.get())
        save_dir = self.save_dir.get()
        
        if not host or not save_dir:
            messagebox.showerror("錯誤", "請?zhí)顚懕O(jiān)聽IP和保存目錄")
            return
        
        self.server_running = True
        self.start_server_btn.config(text="停止接收服務(wù)")
        
        # 在新線程中啟動服務(wù)器
        self.server_thread = threading.Thread(target=self._start_server, args=(host, port, save_dir))
        self.server_thread.daemon = True
        self.server_thread.start()
        
        self.log(f"[+] 接收服務(wù)已啟動，監(jiān)聽 {host}:{port}")
    
    def stop_server(self):
        self.server_running = False
        self.start_server_btn.config(text="啟動接收服務(wù)")
        self.log("[+] 接收服務(wù)已停止")
    
    def _start_server(self, host, port, save_dir):
        s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        s.bind((host, port))
        s.listen(5)
        s.settimeout(1)  # 設(shè)置超時以便定期檢查server_running狀態(tài)
        
        while self.server_running:
            try:
                client_socket, address = s.accept()
                self.log(f"[+] 收到來自 {address} 的連接")
                
                # 在新線程中處理客戶端連接
                client_thread = threading.Thread(
                    target=self._handle_client, 
                    args=(client_socket, address, save_dir)
                )
                client_thread.daemon = True
                client_thread.start()
            except socket.timeout:
                continue
            except Exception as e:
                if self.server_running:
                    self.log(f"[-] 服務(wù)器錯誤: {str(e)}")
        
        s.close()
    
    def _handle_client(self, client_socket, address, save_dir):
        try:
            file_info = client_socket.recv(1024).decode()
            filename, file_size = file_info.split("|")
            file_size = int(file_size)
            
            self.log(f"[+] 接收文件: {filename} ({file_size} 字節(jié))")
            
            client_socket.send("READY".encode())
            
            if not os.path.exists(save_dir):
                os.makedirs(save_dir)
            
            file_path = os.path.join(save_dir, filename)
            
            with open(file_path, "wb") as f:
                received_bytes = 0
                while received_bytes < file_size:
                    bytes_read = client_socket.recv(4096)
                    if not bytes_read:
                        break
                    f.write(bytes_read)
                    received_bytes += len(bytes_read)
                    progress = (received_bytes / file_size) * 100
                    self.log(f"[+] 接收進(jìn)度: {progress:.1f}%")
            
            self.log(f"[+] 文件已保存到: {file_path}")
            client_socket.close()
        except Exception as e:
            self.log(f"[-] 處理客戶端 {address} 時出錯: {str(e)}")

if __name__ == "__main__":
    root = tk.Tk()
    app = FileTransferGUI(root)
    root.mainloop()

使用說明

命令行版本

安裝依賴：

pip install tqdm

在接收端運行：

python receiver.py -H 0.0.0.0 -p 8888 -d received_files

在發(fā)送端運行：

python sender.py 192.168.1.100 file.txt -p 8888

圖形界面版本

安裝依賴：

pip install tqdm

運行圖形界面：

python file_transfer_gui.py

注意事項

確保發(fā)送端和接收端在同一個局域網(wǎng)內(nèi)
確保防火墻允許程序通過指定端口通信
大文件傳輸可能需要較長時間
圖形界面版本需要安裝tkinter（通常Python自帶）

這個實現(xiàn)提供了基本的文件傳輸功能，可以根據(jù)需要進(jìn)行擴(kuò)展，比如添加加密、壓縮、斷點續(xù)傳等功能。

到此這篇關(guān)于基于Python實現(xiàn)局域網(wǎng)內(nèi)Windows桌面文件傳輸?shù)奈恼戮徒榻B到這了,更多相關(guān)Python Windows文件傳輸內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: