基于Python实现局域网内Windows桌面文件传输
作者：燕鹏01
这篇文章介绍了如何使用Python实现一个局域网文件传输系统,包括发送端和接收端的代码示例,发送端和接收端都需要在同一局域网内运行,并且确保防火墙允许通信,图形界面版本需要安装tkinter库,需要的朋友可以参考下
使用Python实现的局域网文件传输系统，包含发送端和接收端。
发送端代码 (sender.py)
import socket
import os
import tqdm
import argparse
def send_file(host, port, file_path):
"""
向指定主机发送文件
"""
# 获取文件大小
file_size = os.path.getsize(file_path)
filename = os.path.basename(file_path)
# 创建socket对象
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
print(f"[+] 正在连接到 {host}:{port}")
s.connect((host, port))
print("[+] 连接成功")
# 发送文件名和文件大小
file_info = f"{filename}|{file_size}"
s.send(file_info.encode())
# 等待确认
response = s.recv(1024).decode()
if response != "READY":
print("[-] 接收端未准备就绪")
s.close()
return
# 发送文件数据
progress = tqdm.tqdm(range(file_size), f"发送 {filename}", unit="B", unit_scale=True, unit_divisor=1024)
with open(file_path, "rb") as f:
while True:
# 读取文件数据
bytes_read = f.read(4096)
if not bytes_read:
# 文件传输完成
break
# 发送数据
s.sendall(bytes_read)
# 更新进度条
progress.update(len(bytes_read))
progress.close()
print("[+] 文件发送完成")
s.close()
if __name__ == "__main__":
parser = argparse.ArgumentParser(description="文件发送端")
parser.add_argument("host", help="目标主机IP地址")
parser.add_argument("file", help="要发送的文件路径")
parser.add_argument("-p", "--port", type=int, default=8888, help="端口号 (默认: 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):
"""
启动文件接收服务器
"""
# 创建socket对象
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
# 绑定地址和端口
s.bind((host, port))
# 开始监听
s.listen(5)
print(f"[+] 服务器正在 {host}:{port} 上监听...")
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} 字节)")
# 确认准备接收
client_socket.send("READY".encode())
# 确保保存目录存在
if not os.path.exists(save_dir):
os.makedirs(save_dir)
# 文件保存路径
file_path = os.path.join(save_dir, filename)
# 接收文件数据
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:
# 接收数据
bytes_read = client_socket.recv(4096)
if not bytes_read:
break
# 写入文件
f.write(bytes_read)
received_bytes += len(bytes_read)
# 更新进度条
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="监听地址 (默认: 0.0.0.0)")
parser.add_argument("-p", "--port", type=int, default=8888, help="监听端口 (默认: 8888)")
parser.add_argument("-d", "--dir", default="received_files", help="文件保存目录 (默认: 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("局域网文件传输工具")
self.root.geometry("500x400")
self.setup_ui()
def setup_ui(self):
# 发送文件框架
send_frame = ttk.LabelFrame(self.root, text="发送文件", padding=10)
send_frame.pack(fill="x", padx=10, pady=5)
ttk.Label(send_frame, text="目标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="发送文件", 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="监听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="启动接收服务", 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("错误", "请填写目标IP和文件路径")
return
if not os.path.exists(file_path):
messagebox.showerror("错误", "文件不存在")
return
# 在新线程中发送文件
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("[-] 接收端未准备就绪")
s.close()
return
self.log(f"[+] 开始发送文件: {filename} ({file_size} 字节)")
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"[+] 发送进度: {progress:.1f}%")
self.log("[+] 文件发送完成")
s.close()
except Exception as e:
self.log(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("错误", "请填写监听IP和保存目录")
return
self.server_running = True
self.start_server_btn.config(text="停止接收服务")
# 在新线程中启动服务器
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"[+] 接收服务已启动，监听 {host}:{port}")
def stop_server(self):
self.server_running = False
self.start_server_btn.config(text="启动接收服务")
self.log("[+] 接收服务已停止")
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) # 设置超时以便定期检查server_running状态
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"[-] 服务器错误: {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} 字节)")
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"[+] 接收进度: {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
- 在发送端运行：
python sender.py 192.168.1.100 file.txt -p 8888
图形界面版本
- 安装依赖：
pip install tqdm
- 运行图形界面：
python file_transfer_gui.py
注意事项
- 确保发送端和接收端在同一个局域网内
- 确保防火墙允许程序通过指定端口通信
- 大文件传输可能需要较长时间
- 图形界面版本需要安装tkinter（通常Python自带）
这个实现提供了基本的文件传输功能，可以根据需要进行扩展，比如添加加密、压缩、断点续传等功能。
到此这篇关于基于Python实现局域网内Windows桌面文件传输的文章就介绍到这了,更多相关Python Windows文件传输内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家！