1131101
-拆分伺服端程式,變成模組
可以將這段程式碼拆分成幾個模組,以提高可讀性和可維護性。以下是拆分的建議:
這部分負責處理客戶端的選擇和檔案選擇。
python
複製程式碼
def select_client_and_file(self):
selected_items = self.tree.selection()
if not selected_items:
messagebox.showwarning("警告", "請先選擇至少一個 client 電腦")
return
client_ips = [self.tree.item(item, 'values')[0] for item in selected_items]
file_paths = filedialog.askopenfilenames(title="選擇檔案")
if file_paths:
self.status_label.config(text="目前傳輸狀態:正在準備傳送檔案...")
self.prepare_file_transfer_list(client_ips, file_paths)
這部分負責準備檔案傳輸的清單。
python
複製程式碼
def prepare_file_transfer_list(self, client_ips, file_paths):
transfer_list = "\n".join(file_paths)
for client_ip in client_ips:
print(f"準備傳輸檔案到 {client_ip}:\n{transfer_list}")
threading.Thread(target=self.send_file_list, args=(client_ip, file_paths), daemon=True).start()
這部分負責具體的檔案傳輸邏輯。
python
複製程式碼
def send_file_list(self, client_ip, file_paths):
self.transfer_button.config(state=tk.DISABLED)
self.broadcast_button.config(state=tk.DISABLED)
file_list_str = ",".join(file_paths)
try:
for idx, file_path in enumerate(file_paths):
self.server_socket.sendto(file_list_str.encode(), (client_ip, 8500))
print(f"檔案傳輸清單已傳送至 {client_ip}")
time.sleep(2)
client_count = len(self.tree.selection())
udp_sender_path = self.get_udp_sender_path()
for file_path in file_paths:
if os.path.exists(file_path):
file_size = os.path.getsize(file_path)
self.status_label.config(text=f"目前傳輸檔案:{os.path.basename(file_path)},大小:{file_size // (1024 * 1024)} MB")
command = f"{udp_sender_path} --file \"{file_path}\" --port 8500 --min-receivers {client_count} --nokbd --autostart 5"
subprocess.run(command, shell=True)
self.progress_bar['value'] = idx + 1
self.root.update_idletasks()
self.progress_bar['maximum'] = len(file_paths)
self.status_label.config(text="目前傳輸狀態:傳輸完成!")
messagebox.showinfo("傳輸完成", "檔案傳輸已完成!")
except Exception as e:
self.handle_transfer_error(e)
finally:
self.transfer_button.config(state=tk.NORMAL)
self.broadcast_button.config(state=tk.NORMAL)
def get_udp_sender_path(self):
if getattr(sys, 'frozen', False):
current_dir = os.path.dirname(sys.executable)
else:
current_dir = os.getcwd()
udp_sender_path = os.path.join(current_dir, "udp-sender.exe")
return udp_sender_path if os.path.exists(udp_sender_path) else r"C:\Windows\System32\udp-sender.exe"
def handle_transfer_error(self, e):
messagebox.showerror("錯誤", f"傳送檔案時發生錯誤: {e}")
print(f"傳送檔案清單時發生錯誤: {e}")
self.status_label.config(text="傳輸失敗!")
messagebox.showinfo("傳輸失敗!", "傳輸失敗!")
這樣拆分後,每個模組的功能更為單一且清晰,這樣更容易進行測試和維護。你可以根據需要進一步調整和擴展每個模組的功能。
"3. 檔案傳輸模組",還能再拆分嗎?
將 send_file_list 函數拆分成更小的函數,每個函數專注於一個特定的任務。
python
複製程式碼
def send_file_list(self, client_ip, file_paths):
self.disable_buttons()
file_list_str = ",".join(file_paths)
try:
self.send_file_list_to_client(client_ip, file_list_str)
time.sleep(2) # 等待傳送清單的時間
self.transfer_files(client_ip, file_paths)
self.complete_transfer()
except Exception as e:
self.handle_transfer_error(e)
finally:
self.enable_buttons()
專門處理檔案清單傳送的邏輯。
python
複製程式碼
def send_file_list_to_client(self, client_ip, file_list_str):
self.server_socket.sendto(file_list_str.encode(), (client_ip, 8500))
print(f"檔案傳輸清單已傳送至 {client_ip}")
專門處理檔案的實際傳送。
python
複製程式碼
def transfer_files(self, client_ip, file_paths):
client_count = len(self.tree.selection())
udp_sender_path = self.get_udp_sender_path()
for idx, file_path in enumerate(file_paths):
if os.path.exists(file_path):
self.update_transfer_status(file_path)
command = f"{udp_sender_path} --file \"{file_path}\" --port 8500 --min-receivers {client_count} --nokbd --autostart 5"
subprocess.run(command, shell=True)
self.update_progress(idx + 1)
更新 UI 狀態的邏輯單獨處理。
python
複製程式碼
def update_transfer_status(self, file_path):
file_size = os.path.getsize(file_path)
self.status_label.config(text=f"目前傳輸檔案:{os.path.basename(file_path)},大小:{file_size // (1024 * 1024)} MB")
單獨處理進度條更新。
python
複製程式碼
def update_progress(self, value):
self.progress_bar['value'] = value
self.root.update_idletasks()
處理傳輸完成的邏輯。
python
複製程式碼
def complete_transfer(self):
self.progress_bar['maximum'] = self.progress_bar['value'] # 更新進度條的最大值
self.status_label.config(text="目前傳輸狀態:傳輸完成!")
messagebox.showinfo("傳輸完成", "檔案傳輸已完成!")
專門處理按鈕的啟用與禁用。
python
複製程式碼
def disable_buttons(self):
self.transfer_button.config(state=tk.DISABLED)
self.broadcast_button.config(state=tk.DISABLED)
def enable_buttons(self):
self.transfer_button.config(state=tk.NORMAL)
self.broadcast_button.config(state=tk.NORMAL)
以下是整合後的類別結構範例:
python
複製程式碼
class FileTransferApp:
def select_client_and_file(self):
selected_items = self.tree.selection()
if not selected_items:
messagebox.showwarning("警告", "請先選擇至少一個 client 電腦")
return
client_ips = [self.tree.item(item, 'values')[0] for item in selected_items]
file_paths = filedialog.askopenfilenames(title="選擇檔案")
if file_paths:
self.status_label.config(text="目前傳輸狀態:正在準備傳送檔案...")
self.prepare_file_transfer_list(client_ips, file_paths)
def prepare_file_transfer_list(self, client_ips, file_paths):
for client_ip in client_ips:
print(f"準備傳輸檔案到 {client_ip}:\n{file_paths}")
threading.Thread(target=self.send_file_list, args=(client_ip, file_paths), daemon=True).start()
def send_file_list(self, client_ip, file_paths):
self.disable_buttons()
file_list_str = ",".join(file_paths)
try:
self.send_file_list_to_client(client_ip, file_list_str)
time.sleep(2) # 等待傳送清單的時間
self.transfer_files(client_ip, file_paths)
self.complete_transfer()
except Exception as e:
self.handle_transfer_error(e)
finally:
self.enable_buttons()
def send_file_list_to_client(self, client_ip, file_list_str):
self.server_socket.sendto(file_list_str.encode(), (client_ip, 8500))
print(f"檔案傳輸清單已傳送至 {client_ip}")
def transfer_files(self, client_ip, file_paths):
client_count = len(self.tree.selection())
udp_sender_path = self.get_udp_sender_path()
for idx, file_path in enumerate(file_paths):
if os.path.exists(file_path):
self.update_transfer_status(file_path)
command = f"{udp_sender_path} --file \"{file_path}\" --port 8500 --min-receivers {client_count} --nokbd --autostart 5"
subprocess.run(command, shell=True)
self.update_progress(idx + 1)
def update_transfer_status(self, file_path):
file_size = os.path.getsize(file_path)
self.status_label.config(text=f"目前傳輸檔案:{os.path.basename(file_path)},大小:{file_size // (1024 * 1024)} MB")
def update_progress(self, value):
self.progress_bar['value'] = value
self.root.update_idletasks()
def complete_transfer(self):
self.progress_bar['maximum'] = self.progress_bar['value']
self.status_label.config(text="目前傳輸狀態:傳輸完成!")
messagebox.showinfo("傳輸完成", "檔案傳輸已完成!")
def disable_buttons(self):
self.transfer_button.config(state=tk.DISABLED)
self.broadcast_button.config(state=tk.DISABLED)
def enable_buttons(self):
self.transfer_button.config(state=tk.NORMAL)
self.broadcast_button.config(state=tk.NORMAL)
def handle_transfer_error(self, e):
messagebox.showerror("錯誤", f"傳送檔案時發生錯誤: {e}")
print(f"傳送檔案清單時發生錯誤: {e}")
self.status_label.config(text="傳輸失敗!")
messagebox.showinfo("傳輸失敗!", "傳輸失敗!")
def get_udp_sender_path(self):
if getattr(sys, 'frozen', False):
current_dir = os.path.dirname(sys.executable)
else:
current_dir = os.getcwd()
udp_sender_path = os.path.join(current_dir, "udp-sender.exe")
return udp_sender_path if os.path.exists(udp_sender_path) else r"C:\Windows\System32\udp-sender.exe"
確保所有的 self. 語句都正確,因為這些方法都是類別的成員方法。
根據實際情況,調整函數的引數和返回值。
測試每個功能以確保整合後的程式碼能夠正常運行。
這樣整合後的程式碼將保持原有的功能,同時結構更清晰,便於維護和擴展。
1131002-upd-cast-test10y