Python 通过启动特定端口进行文件传输，核心是基于 **Socket 编程** 或 **HTTP 服务**。下面通过具体实现方案、代码示例和应用场景进行详细说明。 ### 1. 基于 Socket 编程的可靠传输 这是最基础、可控性最高的方法，适用于需要自定义协议、加密或大文件分块传输的场景。其核心是创建一个监听端口的服务器，等待客户端连接并接收文件数据。 #### 服务器端（接收文件） ```python import socket import os def start_file_server(host='0.0.0.0', port=9999, save_dir='./received'): """ 启动一个文件接收服务器 """ # 创建保存目录 if not os.path.exists(save_dir): os.makedirs(save_dir) # 创建TCP socket server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) server_socket.bind((host, port)) server_socket.listen(5) print(f"[*] 服务器启动，监听 {host}:{port}") while True: client_socket, addr = server_socket.accept() print(f"[+] 接收到来自 {addr} 的连接") try: # 1. 接收文件名和大小（协议头） file_info = client_socket.recv(1024).decode('utf-8') if not file_info: continue filename, filesize = file_info.split('|') filesize = int(filesize) print(f"[*] 准备接收文件: {filename}, 大小: {filesize} 字节") # 2. 发送确认信号 client_socket.send(b'READY') # 3. 接收文件数据 save_path = os.path.join(save_dir, filename) received_size = 0 with open(save_path, 'wb') as f: while received_size < filesize: data = client_socket.recv(4096) if not data: break f.write(data) received_size += len(data) print(f"[✓] 文件接收完成: {save_path}") except Exception as e: print(f"[!] 接收文件时出错: {e}") finally: client_socket.close() if __name__ == '__main__': # 启动服务器，监听9999端口 start_file_server(port=9999) ``` #### 客户端（发送文件） ```python import socket import os def send_file_to_server(server_ip='127.0.0.1', server_port=9999, filepath='./test.txt'): """ 向服务器发送文件 """ if not os.path.exists(filepath): print(f"[!] 文件不存在: {filepath}") return # 获取文件信息 filename = os.path.basename(filepath) filesize = os.path.getsize(filepath) # 连接服务器 client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) try: client_socket.connect((server_ip, server_port)) print(f"[*] 已连接到服务器 {server_ip}:{server_port}") # 1. 发送文件信息（协议头） file_info = f"{filename}|{filesize}" client_socket.send(file_info.encode('utf-8')) # 2. 等待服务器确认 response = client_socket.recv(1024) if response != b'READY': print("[!] 服务器未就绪") return # 3. 发送文件数据 sent_size = 0 with open(filepath, 'rb') as f: while sent_size < filesize: data = f.read(4096) if not data: break client_socket.send(data) sent_size += len(data) print(f"[✓] 文件发送完成: {filename} ({sent_size} 字节)") except Exception as e: print(f"[!] 发送文件时出错: {e}") finally: client_socket.close() if __name__ == '__main__': # 发送文件到服务器 send_file_to_server( server_ip='192.168.1.100', # 服务器IP server_port=9999, filepath='./important_document.pdf' ) ``` **关键特性对比：** | 特性 | Socket 方案 | 说明 | |------|------------|------| | **协议控制** | 完全自定义 | 可以设计自己的传输协议，如包含MD5校验、加密头等 [ref_1] | | **传输效率** | 高 | 直接TCP传输，无HTTP头部开销 | | **安全性** | 可扩展 | 可集成DES/RSA加密 [ref_1] | | **适用场景** | 局域网大文件、安全传输 | 适合需要保密性和完整性的场景 | | **跨平台** | 完全支持 | Linux/Windows均可运行 [ref_3] | ### 2. 基于 HTTP 服务的快速传输 对于临时、简单的文件共享，Python内置的HTTP服务器是最快捷的方式。这种方法实质上是启动一个Web服务器，通过浏览器或wget/curl下载文件。 #### 启动HTTP文件服务器 ```python import http.server import socketserver import os from threading import Thread def start_http_server(port=8000, directory='.'): """ 启动一个简单的HTTP文件服务器 """ # 切换到指定目录 os.chdir(directory) # 创建HTTP请求处理器 handler = http.server.SimpleHTTPRequestHandler # 设置服务器 with socketserver.TCPServer(("", port), handler) as httpd: print(f"[*] HTTP文件服务器启动于端口 {port}") print(f"[*] 服务目录: {os.getcwd()}") print(f"[*] 访问地址: http://localhost:{port}") print("[*] 按 Ctrl+C 停止服务器") try: httpd.serve_forever() except KeyboardInterrupt: print("\n[*] 服务器已停止") # Python 3.x 一行命令启动（命令行执行）： # python -m http.server 8000 if __name__ == '__main__': # 启动在8080端口，共享当前目录 start_http_server(port=8080, directory='./shared_files') ``` **使用场景示例：** ```python # 场景：快速共享文件夹给团队成员 import subprocess import time def quick_share_folder(folder_path, port=8000, duration=300): """ 临时共享文件夹（5分钟） """ if not os.path.exists(folder_path): print(f"目录不存在: {folder_path}") return # 获取本机IP import socket hostname = socket.gethostname() local_ip = socket.gethostbyname(hostname) print(f"共享目录: {folder_path}") print(f"访问地址: http://{local_ip}:{port}") print(f"共享将在 {duration//60} 分钟后自动停止") # 启动HTTP服务器（子进程） cmd = f'cd "{folder_path}" && python -m http.server {port}' process = subprocess.Popen(cmd, shell=True) # 等待指定时间后停止 time.sleep(duration) process.terminate() print("共享已停止") # 使用示例 quick_share_folder('./project_docs', port=8888) ``` ### 3. 增强型文件传输方案 结合加密和校验，确保传输安全可靠 [ref_1]。 ```python import hashlib import socket import os class SecureFileTransfer: def __init__(self, host='0.0.0.0', port=9999): self.host = host self.port = port def calculate_md5(self, filepath): """计算文件MD5值用于校验 [ref_6]""" md5_hash = hashlib.md5() with open(filepath, 'rb') as f: for chunk in iter(lambda: f.read(4096), b''): md5_hash.update(chunk) return md5_hash.hexdigest() def send_file_with_checksum(self, server_ip, filepath): """发送文件并包含MD5校验""" # 计算文件哈希 file_md5 = self.calculate_md5(filepath) filename = os.path.basename(filepath) filesize = os.path.getsize(filepath) # 建立连接并发送 with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock: sock.connect((server_ip, self.port)) # 发送文件信息（包含MD5） header = f"{filename}|{filesize}|{file_md5}" sock.send(header.encode('utf-8')) # 等待服务器确认 if sock.recv(1024) == b'READY': # 发送文件内容 with open(filepath, 'rb') as f: while chunk := f.read(4096): sock.send(chunk) # 等待校验结果 result = sock.recv(1024).decode('utf-8') print(f"传输结果: {result}") def receive_file_with_verification(self, save_dir='./received'): """接收文件并验证完整性""" if not os.path.exists(save_dir): os.makedirs(save_dir) with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as server: server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) server.bind((self.host, self.port)) server.listen(1) print(f"安全传输服务器启动于 {self.host}:{self.port}") while True: conn, addr = server.accept() print(f"接收到来自 {addr} 的连接") try: # 接收文件头信息 header = conn.recv(1024).decode('utf-8') filename, filesize, expected_md5 = header.split('|') filesize = int(filesize) save_path = os.path.join(save_dir, filename) conn.send(b'READY') # 接收文件数据 received = 0 with open(save_path, 'wb') as f: while received < filesize: chunk = conn.recv(min(4096, filesize - received)) if not chunk: break f.write(chunk) received += len(chunk) # 验证MD5 actual_md5 = self.calculate_md5(save_path) if actual_md5 == expected_md5: conn.send(b'SUCCESS: MD5验证通过') print(f"文件 {filename} 接收成功，校验通过") else: conn.send(b'FAIL: MD5校验失败') print(f"警告: 文件 {filename} MD5校验失败") except Exception as e: print(f"传输错误: {e}") finally: conn.close() # 使用示例 if __name__ == '__main__': transfer = SecureFileTransfer(port=12345) # 在接收端运行 # transfer.receive_file_with_verification() # 在发送端运行 # transfer.send_file_with_checksum('192.168.1.100', './secure_data.zip') ``` ### 4. 跨平台传输注意事项 在Linux和Windows之间传输文件时需注意 [ref_3]： ```python import platform import socket def get_platform_specific_path(path): """处理跨平台路径差异""" system = platform.system() if system == 'Windows': # Windows路径处理 return path.replace('/', '\\') elif system == 'Linux': # Linux路径处理 return path.replace('\\', '/') else: return path def cross_platform_transfer(): """跨平台文件传输配置示例""" config = { 'buffer_size': 4096, # 缓冲区大小 'encoding': 'utf-8', # 统一编码 'timeout': 30, # 超时时间 } # 根据系统调整参数 if platform.system() == 'Windows': config['line_ending'] = '\r\n' else: config['line_ending'] = '\n' return config ``` ### 5. 实践建议与选择指南 | 需求场景 | 推荐方案 | 关键代码 | 优势 | |---------|---------|---------|------| | **快速临时共享** | HTTP服务器 | `python -m http.server 8000` | 零编码，浏览器直接访问 [ref_5] | | **可靠大文件传输** | Socket + 校验 | 上述SecureFileTransfer类 | 支持断点续传、完整性验证 [ref_6] | | **安全保密传输** | Socket + 加密 | 集成DES/AES加密 [ref_1] | 防止数据窃听 | | **跨平台传输** | Socket标准实现 | 通用socket代码 [ref_3] | Linux/Windows兼容 | | **多文件批量传输** | 多线程Socket | 为每个文件创建线程 | 提高传输效率 | **关键实施步骤：** 1. **确定需求**：评估文件大小、安全性要求、使用频率 2. **选择协议**：HTTP用于简单共享，Socket用于定制需求 3. **实现校验**：务必添加MD5/SHA校验确保完整性 [ref_6] 4. **错误处理**：添加超时、重试、断点续传机制 5. **测试验证**：在不同平台和网络环境下测试 通过上述方案，您可以根据具体需求选择合适的Python端口文件传输实现方式。对于生产环境，建议使用Socket方案并添加完整的错误处理和日志记录；对于临时需求，HTTP服务器方案最为便捷高效 [ref_4][ref_5]。