【Python】FTPクライアント自作

File Transfer Protocol（FTP）は、インターネット上でファイルを転送するための標準プロトコルです。FTPソフトは、このプロトコルを利用してファイル転送を行うためのクライアントアプリケーションでありウェブサイトの開発やデータ転送において欠かせないツールです。

FTPでファイルを転送する際は、自分のパソコンから「FTPサーバー」を介してデータの送信を行います。ちなみにFTP上では、操作者のことを「FTPクライアント」と呼びます。

FTPクライアントが直接Webサーバーに接続するわけではなく、それと連携しているFTPサーバーでやり取りが行われます。

前回、ターミナルエミュレータを自作した際に利用した「paramiko」を使用します。
前回の記事はこちらから

【Python】Tera Termの様なターミナルアプリを自作 Tera Term（テラターム）とはWindows用ターミナルエミュレータです。 SSH・telnet・シリアルの各通信プロトコルに対...

作成するアプリケーションの概要

今回作成するアプリケーションは、FTP、FTPS、SFTPプロトコルに対応したファイル転送クライアントで、ユーザーがリモートサーバーとのファイルの転送や管理を行うためのGUIツールを作ります。

以下はアプリケーションの主要なコンポーネントと実装する機能について説明します。

主要コンポーネント

メインウィンドウ

ローカルファイルリスト: ユーザーのローカルファイルシステムを表示。
リモートファイルリスト: リモートサーバー上のファイルを表示。
ブックマークリスト: よく使うリモートディレクトリを保存するためのリスト。

メニューバー

ファイルメニュー: 接続、終了などの基本操作を提供。
ブックマークメニュー: ブックマークの追加機能を提供。

プレビューウィンドウ

ローカルファイルやリモートファイルの内容を表示するためのサブウィンドウ。

ファイル操作

ダウンロード: リモートファイルをローカルにダウンロード。
アップロード: ローカルファイルをリモートサーバーにアップロード。
フォルダー作成: リモートサーバー上で新しいフォルダーを作成。
フォルダー削除: リモートサーバー上のフォルダーを削除。

セッション管理

接続履歴とブックマークを保存し、再利用できるようにする。

作り方

以下はアプリケーションの作成手順になります。

ライブラリのインストール、インポート

必要なライブラリ（tkinter, paramiko, ftplib, os, json, など）をインストールし、インポートします。

アプリケーションクラスの作成

FTPClientAppクラスを作成し、tk.Tkから継承します。
__init__メソッドでウィンドウの設定や初期化を行います。

GUIの構築

メインウィンドウに必要なウィジェット（リストボックス、ラベル、メニューバーなど）を追加します。
create_menusメソッドでメニューバーを作成します。
update_local_listとupdate_remote_listメソッドでローカルとリモートのファイルリストを更新します。

ファイル操作機能の実装

ファイルの転送、フォルダーの作成・削除などの操作を実装します。
transfer_files, create_folder_async, delete_folder_asyncなどのメソッドでファイルの転送やフォルダー操作を行います。

セッションとブックマークの管理

load_session_history, save_session_history, add_bookmark, save_bookmarksなどのメソッドでセッションの履歴やブックマークの保存・読み込みを行います。

プレビュー機能の追加

PreviewWindowクラスを作成し、ファイルの内容を表示するウィンドウを作成します。

操作方法

アプリケーションが起動したら、メニューバーから接続設定を行い、ローカルとリモートのファイルリストを操作できます。
ダウンロードやアップロード、フォルダーの作成・削除などの操作は、コンテキストメニューやボタンから行います。

想定される用途

ファイル転送: ユーザーがリモートサーバーとローカルコンピュータ間でファイルを簡単に転送できるようにします。FTP、FTPS、SFTPプロトコルに対応しているため、さまざまなサーバーと接続できます。
ファイル管理: リモートサーバー上のファイルやフォルダーの作成、削除、リスト表示などの操作を行います。
セッション管理: 接続履歴やブックマーク機能により、よく使用するサーバーやディレクトリに簡単にアクセスできます。
ファイルプレビュー: リモートサーバーやローカルのファイル内容を直接アプリケーション内で確認できます。

FTPクライアントソフトウェアの基本的な機能を備えた、ユーザーフレンドリーなツールとして作成したいと思います。

基本的な機能のみを実装した簡易的なものになります。
更なる拡張を行い、実用に耐え得るアプリケーションに育てて下さい。

実装

「my_ftp_client.py」という名前でファイルを作成して下さい。

import tkinter as tk
from tkinter import ttk, messagebox, filedialog, simpledialog, scrolledtext
import paramiko
import os
import threading
import json
import zipfile
import tarfile
from ftplib import FTP, FTP_TLS
from queue import Queue
import time
import logging

# ログの設定
logging.basicConfig(level=logging.DEBUG)
paramiko.util.log_to_file('paramiko.log')

class FTPClientApp(tk.Tk):
    def __init__(self):
        super().__init__()
        self.title("My FTP Client App")
        self.geometry("800x600")

        self.session_history = []
        self.bookmarks = []
        self.load_session_history()
        self.load_bookmarks()

        self.status_label = ttk.Label(self, text="接続なし", relief=tk.SUNKEN, anchor=tk.W)
        self.status_label.pack(side=tk.BOTTOM, fill=tk.X)

        self.main_frame = ttk.Frame(self)
        self.main_frame.pack(fill=tk.BOTH, expand=True)

        self.local_listbox = tk.Listbox(self.main_frame, selectmode=tk.EXTENDED)
        self.remote_listbox = tk.Listbox(self.main_frame, selectmode=tk.EXTENDED)
        self.local_listbox.pack(side=tk.LEFT, fill=tk.BOTH, expand=True)
        self.remote_listbox.pack(side=tk.RIGHT, fill=tk.BOTH, expand=True)

        self.bookmark_listbox = tk.Listbox(self.main_frame)
        self.bookmark_listbox.pack(side=tk.LEFT, fill=tk.Y)
        self.bookmark_listbox.bind('<Double-Button-1>', self.open_bookmark)

        self.create_menus()
        self.update_local_list()

    def create_menus(self):
        menubar = tk.Menu(self)

        file_menu = tk.Menu(menubar, tearoff=0)
        file_menu.add_command(label="接続", command=self.open_connect_window)
        file_menu.add_command(label="終了", command=self.quit)
        menubar.add_cascade(label="ファイル", menu=file_menu)

        bookmark_menu = tk.Menu(menubar, tearoff=0)
        bookmark_menu.add_command(label="ブックマークに追加", command=self.add_bookmark)
        menubar.add_cascade(label="ブックマーク", menu=bookmark_menu)

        self.config(menu=menubar)

    def open_connect_window(self):
        ConnectWindow(self)

    def update_local_list(self, directory=None):
        if directory is None:
            directory = os.getcwd()

        self.local_listbox.delete(0, tk.END)
        for item in os.listdir(directory):
            self.local_listbox.insert(tk.END, item)

        self.local_listbox.bind('<Double-Button-1>', self.preview_local_file)
        self.local_listbox.bind('<Button-3>', self.show_context_menu)

    def preview_local_file(self, event):
        selected_file = self.local_listbox.get(self.local_listbox.curselection())
        if os.path.isfile(selected_file):
            with open(selected_file, 'r') as f:
                PreviewWindow(self, f.read(), selected_file)

    def preview_remote_file(self, remote_path):
        with self.sftp.file(remote_path, 'r') as file:
            content = file.read().decode('utf-8')
            PreviewWindow(self, content, remote_path)

    def update_remote_list(self):
        if hasattr(self, 'sftp'):
            remote_dir = self.sftp.getcwd()
            self.remote_listbox.delete(0, tk.END)
            for item in self.sftp.listdir(remote_dir):
                self.remote_listbox.insert(tk.END, item)
            self.remote_listbox.bind('<Double-Button-1>', self.preview_remote_file_async)
            self.remote_listbox.bind('<Button-3>', self.show_context_menu)

    def show_context_menu(self, event):
        context_menu = tk.Menu(self, tearoff=0)
        context_menu.add_command(label="ダウンロード", command=self.download_selected_files)
        context_menu.add_command(label="アップロード", command=self.upload_selected_files)
        context_menu.add_command(label="フォルダ作成", command=self.create_folder_async)
        context_menu.add_command(label="削除", command=self.delete_folder_async)
        context_menu.post(event.x_root, event.y_root)

    def download_selected_files(self):
        selected_files = [self.remote_listbox.get(i) for i in self.remote_listbox.curselection()]
        self.transfer_files(selected_files, download=True)

    def upload_selected_files(self):
        selected_files = [self.local_listbox.get(i) for i in self.local_listbox.curselection()]
        self.transfer_files(selected_files, download=False)

    def transfer_files(self, files, download=True):
        queue = Queue()
        for file in files:
            queue.put(file)

        for _ in range(min(3, len(files))):  # 並列に3つのファイルを転送
            threading.Thread(target=self._transfer_file_worker, args=(queue, download)).start()

    def _transfer_file_worker(self, queue, download):
        while not queue.empty():
            file = queue.get()
            try:
                if download:
                    self.download_file(file)
                else:
                    self.upload_file(file)
            except Exception as e:
                self.log(f"転送エラー: {e}")
            queue.task_done()

    def download_file(self, remote_file):
        local_path = os.path.join(os.getcwd(), remote_file)
        self.sftp.get(remote_file, local_path)
        self.update_local_list()
        self.log(f"ダウンロード完了: {remote_file}")

    def upload_file(self, local_file):
        remote_path = self.sftp.getcwd()
        self.sftp.put(local_file, os.path.join(remote_path, local_file))
        self.update_remote_list()
        self.log(f"アップロード完了: {local_file}")

    def create_folder_async(self):
        threading.Thread(target=self.create_folder).start()

    def create_folder(self):
        folder_name = simpledialog.askstring("フォルダ作成", "新しいフォルダ名を入力してください:")
        if folder_name:
            self.sftp.mkdir(folder_name)
            self.update_remote_list()
            self.log(f"フォルダ作成: {folder_name}")

    def delete_folder_async(self):
        threading.Thread(target=self.delete_folder).start()

    def delete_folder(self):
        selected_folder = self.remote_listbox.get(tk.ACTIVE)
        if messagebox.askyesno("削除", f"{selected_folder} を削除しますか？"):
            self.sftp.rmdir(selected_folder)
            self.update_remote_list()
            self.log(f"フォルダ削除: {selected_folder}")

    def add_bookmark(self):
        current_dir = self.sftp.getcwd() if hasattr(self, 'sftp') else os.getcwd()
        bookmark_name = simpledialog.askstring("ブックマーク", "ブックマーク名を入力してください:")
        if bookmark_name:
            self.bookmarks.append((bookmark_name, current_dir))
            self.save_bookmarks()
            self.update_bookmark_list()
            self.log(f"ブックマーク追加: {bookmark_name}")

    def update_bookmark_list(self):
        self.bookmark_listbox.delete(0, tk.END)
        for name, path in self.bookmarks:
            self.bookmark_listbox.insert(tk.END, name)

    def open_bookmark(self, event):
        selected_index = self.bookmark_listbox.curselection()
        if selected_index:
            _, path = self.bookmarks[selected_index[0]]
            if hasattr(self, 'sftp'):
                self.sftp.chdir(path)
                self.update_remote_list()
            else:
                os.chdir(path)
                self.update_local_list()
            self.log(f"ブックマークオープン: {path}")

    def save_session_history(self):
        with open("session_history.json", "w") as f:
            json.dump(self.session_history, f)

    def load_session_history(self):
        if os.path.exists("session_history.json"):
            with open("session_history.json", "r") as f:
                self.session_history = json.load(f)

    def save_bookmarks(self):
        with open("bookmarks.json", "w") as f:
            json.dump(self.bookmarks, f)

    def load_bookmarks(self):
        if os.path.exists("bookmarks.json"):
            with open("bookmarks.json", "r") as f:
                self.bookmarks = json.load(f)

    def log(self, message):
        with open("app.log", "a") as f:
            f.write(f"{time.strftime('%Y-%m-%d %H:%M:%S')} - {message}\n")
        print(message)  # コンソールにもログを出力

    def update_status(self, status):
        self.status_label.config(text=status)

    def compress_file(self, file_path):
        zip_path = f"{file_path}.zip"
        with zipfile.ZipFile(zip_path, 'w') as zipf:
            zipf.write(file_path, os.path.basename(file_path))
        return zip_path

    def decompress_file(self, zip_path):
        with zipfile.ZipFile(zip_path, 'r') as zipf:
            zipf.extractall(os.path.dirname(zip_path))

class ConnectWindow(tk.Toplevel):
    def __init__(self, parent):
        super().__init__(parent)
        self.parent = parent
        self.title("接続設定")
        self.geometry("400x600")

        self.protocol_var = tk.StringVar(value="FTP")
        self.create_widgets()

    def create_widgets(self):
        self.protocol_label = ttk.Label(self, text="プロトコル:")
        self.protocol_label.pack(pady=5)
        self.protocol_combobox = ttk.Combobox(self, textvariable=self.protocol_var, values=["FTP", "FTPS", "SFTP"])
        self.protocol_combobox.pack(pady=5)

        self.host_label = ttk.Label(self, text="ホスト:")
        self.host_label.pack(pady=5)
        self.host_entry = ttk.Entry(self)
        self.host_entry.pack(pady=5)

        self.port_label = ttk.Label(self, text="ポート:")
        self.port_label.pack(pady=5)
        self.port_entry = ttk.Entry(self)
        self.port_entry.pack(pady=5)

        self.user_label = ttk.Label(self, text="ユーザー名:")
        self.user_label.pack(pady=5)
        self.user_entry = ttk.Entry(self)
        self.user_entry.pack(pady=5)

        self.pass_label = ttk.Label(self, text="パスワード:")
        self.pass_label.pack(pady=5)
        self.pass_entry = ttk.Entry(self, show='*')
        self.pass_entry.pack(pady=5)

        self.key_label = ttk.Label(self, text="秘密鍵ファイル:")
        self.key_label.pack(pady=5)
        self.key_button = ttk.Button(self, text="選択", command=self.select_key_file)
        self.key_button.pack(pady=5)
        self.key_path_label = ttk.Label(self, text="")
        self.key_path_label.pack(pady=5)

        self.key_pass_label = ttk.Label(self, text="鍵パスフレーズ:")
        self.key_pass_label.pack(pady=5)
        self.key_pass_entry = ttk.Entry(self, show='*')
        self.key_pass_entry.pack(pady=5)

        self.connect_button = ttk.Button(self, text="接続", command=self.connect)
        self.connect_button.pack(pady=20)

    def select_key_file(self):
        file_path = filedialog.askopenfilename(filetypes=[("PEM Files", "*.pem"), ("All Files", "*.*")])
        if file_path:
            self.key_path_label.config(text=file_path)
            self.key_file = file_path

    def connect(self):
        protocol = self.protocol_var.get()
        host = self.host_entry.get()
        port = int(self.port_entry.get())
        user = self.user_entry.get()
        password = self.pass_entry.get()
        key_file = getattr(self, 'key_file', None)
        key_passphrase = self.key_pass_entry.get() if hasattr(self, 'key_file') else None

        if protocol == "FTP":
            self.connect_ftp(host, port, user, password)
        elif protocol == "FTPS":
            self.connect_ftps(host, port, user, password)
        elif protocol == "SFTP":
            self.connect_sftp(host, port, user, password, key_file, key_passphrase)

    def connect_ftp(self, host, port, user, password):
        try:
            self.ftp = FTP()
            self.ftp.connect(host, port)
            self.ftp.login(user, password)
            self.parent.update_status("FTP接続成功")
            self.destroy()
        except Exception as e:
            messagebox.showerror("接続エラー", str(e))

    def connect_ftps(self, host, port, user, password):
        try:
            self.ftps = FTP_TLS()
            self.ftps.connect(host, port)
            self.ftps.login(user, password)
            self.ftps.prot_p()  # データの暗号化を有効にする
            self.parent.update_status("FTPS接続成功")
            self.destroy()
        except Exception as e:
            messagebox.showerror("接続エラー", str(e))

     def connect_sftp(self, host, port, user, password, key_file, key_passphrase):
        try:
            transport = paramiko.Transport((host, port))
            if key_file:
                # 秘密鍵の読み込み
                private_key = paramiko.RSAKey.from_private_key_file(key_file, password=key_passphrase)
                transport.connect(username=user, pkey=private_key)
            else:
                transport.connect(username=user, password=password)
            self.sftp = paramiko.SFTPClient.from_transport(transport)
            self.parent.update_status("SFTP接続成功")
            self.destroy()
        except paramiko.AuthenticationException as e:
            messagebox.showerror("接続エラー", f"認証エラー: {str(e)}")
        except paramiko.SSHException as e:
            messagebox.showerror("接続エラー", f"SSHエラー: {str(e)}")
        except Exception as e:
            messagebox.showerror("接続エラー", f"SFTP接続エラー: {str(e)}")
            
    def __init__(self, parent):
        super().__init__(parent)
        self.parent = parent
        self.title("接続設定")
        self.geometry("400x500")

        self.protocol_var = tk.StringVar(value="FTP")
        self.create_widgets()

    def create_widgets(self):
        self.protocol_label = ttk.Label(self, text="プロトコル:")
        self.protocol_label.pack(pady=5)
        self.protocol_combobox = ttk.Combobox(self, textvariable=self.protocol_var, values=["FTP", "FTPS", "SFTP"])
        self.protocol_combobox.pack(pady=5)

        self.host_label = ttk.Label(self, text="ホスト:")
        self.host_label.pack(pady=5)
        self.host_entry = ttk.Entry(self)
        self.host_entry.pack(pady=5)

        self.port_label = ttk.Label(self, text="ポート:")
        self.port_label.pack(pady=5)
        self.port_entry = ttk.Entry(self)
        self.port_entry.pack(pady=5)

        self.user_label = ttk.Label(self, text="ユーザー名:")
        self.user_label.pack(pady=5)
        self.user_entry = ttk.Entry(self)
        self.user_entry.pack(pady=5)

        self.pass_label = ttk.Label(self, text="パスワード:")
        self.pass_label.pack(pady=5)
        self.pass_entry = ttk.Entry(self, show='*')
        self.pass_entry.pack(pady=5)

        self.key_label = ttk.Label(self, text="秘密鍵ファイル:")
        self.key_label.pack(pady=5)
        self.key_button = ttk.Button(self, text="選択", command=self.select_key_file)
        self.key_button.pack(pady=5)
        self.key_path_label = ttk.Label(self, text="")
        self.key_path_label.pack(pady=5)

        self.connect_button = ttk.Button(self, text="接続", command=self.connect)
        self.connect_button.pack(pady=20)

    def select_key_file(self):
        file_path = filedialog.askopenfilename(filetypes=[("PEM Files", "*.pem"), ("All Files", "*.*")])
        if file_path:
            self.key_path_label.config(text=file_path)
            self.key_file = file_path

    def connect(self):
        protocol = self.protocol_var.get()
        host = self.host_entry.get()
        port = int(self.port_entry.get())
        user = self.user_entry.get()
        password = self.pass_entry.get()
        key_file = getattr(self, 'key_file', None)

        if protocol == "FTP":
            self.connect_ftp(host, port, user, password)
        elif protocol == "FTPS":
            self.connect_ftps(host, port, user, password)
        elif protocol == "SFTP":
            self.connect_sftp(host, port, user, password, key_file)

    def connect_ftp(self, host, port, user, password):
        try:
            self.ftp = FTP()
            self.ftp.connect(host, port)
            self.ftp.login(user, password)
            self.parent.update_status("FTP接続成功")
            self.destroy()
        except Exception as e:
            messagebox.showerror("接続エラー", str(e))

    def connect_ftps(self, host, port, user, password):
        try:
            self.ftps = FTP_TLS()
            self.ftps.connect(host, port)
            self.ftps.login(user, password)
            self.ftps.prot_p()  # データの暗号化を有効にする
            self.parent.update_status("FTPS接続成功")
            self.destroy()
        except Exception as e:
            messagebox.showerror("接続エラー", str(e))

    def connect_sftp(self, host, port, user, password, key_file):
        try:
            transport = paramiko.Transport((host, port))
            if key_file:
                private_key = paramiko.RSAKey.from_private_key_file(key_file)
                transport.connect(username=user, pkey=private_key)
            else:
                transport.connect(username=user, password=password)
            self.sftp = paramiko.SFTPClient.from_transport(transport)
            self.parent.update_status("SFTP接続成功")
            self.destroy()
        except Exception as e:
            messagebox.showerror("接続エラー", str(e))

class PreviewWindow(tk.Toplevel):
    def __init__(self, parent, content, title):
        super().__init__(parent)
        self.title(f"プレビュー: {title}")
        self.geometry("600x400")

        self.text_widget = scrolledtext.ScrolledText(self)
        self.text_widget.pack(fill=tk.BOTH, expand=True)
        self.text_widget.insert(tk.END, content)
        self.text_widget.config(state=tk.DISABLED)

if __name__ == "__main__":
    app = FTPClientApp()
    app.mainloop()

長くて複雑に見えますが、殆どが画面を作成しているコードで形を覚えると簡単なコードです。

次の項で重要なポイントを解説していきます。

解説

tkinterを使ってGUIのFTPクライアントアプリケーションを作成しました。FTP、FTPS、SFTPのプロトコルをサポートし、ファイルの転送やフォルダーの作成・削除などの操作を実装してあります。

インポート部分

import tkinter as tk
from tkinter import ttk, messagebox, filedialog, simpledialog, scrolledtext
import paramiko
import os
import threading
import json
import zipfile
import tarfile
from ftplib import FTP, FTP_TLS
from queue import Queue
import time
import logging

tkinterとそのサブモジュールは、GUIアプリケーションを作成するために使用されます。
paramikoはSFTP（SSH File Transfer Protocol）接続を管理するためのライブラリです。
osはファイルやディレクトリの操作を行います。
threadingは並行処理を管理します。
jsonはJSONデータの読み書きを行います。
zipfileとtarfileはファイルの圧縮と解凍に使用します。
ftplibはFTPとFTPS接続を管理します。
queueはスレッド間でのタスクキュー管理に使用します。
timeは時間に関連する機能を提供します。
loggingはログの記録を行います。

`FTPClientApp`クラス

このクラスは、主なアプリケーションのGUIを作成し、ファイルの転送や接続設定などを管理します。

コンストラクタと初期設定

class FTPClientApp(tk.Tk):
    def __init__(self):
        super().__init__()
        self.title("My FTP Client App")
        self.geometry("800x600")

        self.session_history = []
        self.bookmarks = []
        self.load_session_history()
        self.load_bookmarks()

super().__init__()で親クラスの初期化を行い、アプリケーションウィンドウを設定します。
session_historyとbookmarksは、それぞれセッションの履歴とブックマークを管理するリストです。
load_session_history()とload_bookmarks()メソッドで、アプリケーション開始時に保存された履歴やブックマークを読み込みます。

GUIの要素

        self.status_label = ttk.Label(self, text="接続なし", relief=tk.SUNKEN, anchor=tk.W)
        self.status_label.pack(side=tk.BOTTOM, fill=tk.X)

        self.main_frame = ttk.Frame(self)
        self.main_frame.pack(fill=tk.BOTH, expand=True)

        self.local_listbox = tk.Listbox(self.main_frame, selectmode=tk.EXTENDED)
        self.remote_listbox = tk.Listbox(self.main_frame, selectmode=tk.EXTENDED)
        self.local_listbox.pack(side=tk.LEFT, fill=tk.BOTH, expand=True)
        self.remote_listbox.pack(side=tk.RIGHT, fill=tk.BOTH, expand=True)

        self.bookmark_listbox = tk.Listbox(self.main_frame)
        self.bookmark_listbox.pack(side=tk.LEFT, fill=tk.Y)
        self.bookmark_listbox.bind('<Double-Button-1>', self.open_bookmark)

        self.create_menus()
        self.update_local_list()

status_labelは、現在の接続状態を表示します。
main_frameは、ローカルファイル、リモートファイル、およびブックマークのリストボックスを含むメインフレームです。
local_listboxとremote_listboxは、ローカルとリモートのファイルリストを表示します。
bookmark_listboxは、ブックマークを表示し、ダブルクリックでブックマークを開く機能を提供します。

メニューの作成

    def create_menus(self):
        menubar = tk.Menu(self)

        file_menu = tk.Menu(menubar, tearoff=0)
        file_menu.add_command(label="接続", command=self.open_connect_window)
        file_menu.add_command(label="終了", command=self.quit)
        menubar.add_cascade(label="ファイル", menu=file_menu)

        bookmark_menu = tk.Menu(menubar, tearoff=0)
        bookmark_menu.add_command(label="ブックマークに追加", command=self.add_bookmark)
        menubar.add_cascade(label="ブックマーク", menu=bookmark_menu)

        self.config(menu=menubar)

メニューバーを作成し、ファイルメニューとブックマークメニューを追加します。

ローカルファイルリストの更新

    def update_local_list(self, directory=None):
        if directory is None:
            directory = os.getcwd()

        self.local_listbox.delete(0, tk.END)
        for item in os.listdir(directory):
            self.local_listbox.insert(tk.END, item)

        self.local_listbox.bind('<Double-Button-1>', self.preview_local_file)
        self.local_listbox.bind('<Button-3>', self.show_context_menu)

ローカルのファイルリストを更新し、ファイルのプレビューやコンテキストメニューの表示を設定します。

ファイルのプレビュー

    def preview_local_file(self, event):
        selected_file = self.local_listbox.get(self.local_listbox.curselection())
        if os.path.isfile(selected_file):
            with open(selected_file, 'r') as f:
                PreviewWindow(self, f.read(), selected_file)

選択されたローカルファイルの内容をプレビューウィンドウで表示します。

リモートファイルリストの更新

    def update_remote_list(self):
        if hasattr(self, 'sftp'):
            remote_dir = self.sftp.getcwd()
            self.remote_listbox.delete(0, tk.END)
            for item in self.sftp.listdir(remote_dir):
                self.remote_listbox.insert(tk.END, item)
            self.remote_listbox.bind('<Double-Button-1>', self.preview_remote_file_async)
            self.remote_listbox.bind('<Button-3>', self.show_context_menu)

リモートディレクトリのファイルリストを更新し、リモートファイルのプレビューやコンテキストメニューの表示を設定します。

コンテキストメニューの表示

    def show_context_menu(self, event):
        context_menu = tk.Menu(self, tearoff=0)
        context_menu.add_command(label="ダウンロード", command=self.download_selected_files)
        context_menu.add_command(label="アップロード", command=self.upload_selected_files)
        context_menu.add_command(label="フォルダ作成", command=self.create_folder_async)
        context_menu.add_command(label="削除", command=self.delete_folder_async)
        context_menu.post(event.x_root, event.y_root)

右クリックで表示されるコンテキストメニューを作成し、ファイルやフォルダーの操作を提供します。

ファイルの転送

    def transfer_files(self, files, download=True):
        queue = Queue()
        for file in files:
            queue.put(file)

        for _ in range(min(3, len(files))):  # 並列に3つのファイルを転送
            threading.Thread(target=self._transfer_file_worker, args=(queue, download)).start()

ファイルの転送を非同期で実行するために、スレッドを使用してファイルの転送作業を行います。

フォルダーの作成と削除

    def create_folder_async(self):
        threading.Thread(target=self.create_folder).start()

    def create_folder(self):
        folder_name = simpledialog.askstring("フォルダ作成", "新しいフォルダ名を入力してください:")
        if folder_name:
            self.sftp.mkdir(folder_name)
            self.update_remote_list()
            self.log(f"フォルダ作成: {folder_name}")

    def delete_folder_async(self):
        threading.Thread(target=self.delete_folder).start()

    def delete_folder(self):
        selected_folder = self.remote_listbox.get(tk.ACTIVE)
        if messagebox.askyesno("削除", f"{selected_folder} を削除しますか？"):
            self.sftp.rmdir(selected_folder)
            self.update_remote_list()
            self.log(f"フォルダ削除: {selected_folder}")

フォルダーの作成と削除を非同期で実行するためにスレッドを使用します。

ブックマークの管理

    def add_bookmark(self):
        current_dir = self.sftp.getcwd() if hasattr(self, 'sftp') else os.getcwd()
        bookmark_name = simpledialog.askstring("ブックマーク", "ブックマーク名を入力してください:")
        if bookmark_name:
            self.bookmarks.append((bookmark_name, current_dir))
            self.save_bookmarks()
            self.update_bookmark_list()
            self.log(f"ブックマーク追加: {bookmark_name}")

    def load_bookmarks(self):
        if os.path.exists("bookmarks.json"):
            with open("bookmarks.json", "r") as f:
                self.bookmarks = json.load(f)

    def save_bookmarks(self):
        with open("bookmarks.json", "w") as f:
            json.dump(self.bookmarks, f)

    def update_bookmark_list(self):
        self.bookmark_listbox.delete(0, tk.END)
        for bookmark in self.bookmarks:
            self.bookmark_listbox.insert(tk.END, bookmark[0])

ブックマークの追加、読み込み、保存、リストの更新を行います。

`PreviewWindow`クラス

このクラスは、ファイルの内容を表示するプレビューウィンドウを作成します。

class PreviewWindow(tk.Toplevel):
    def __init__(self, parent, content, filename):
        super().__init__(parent)
        self.title(f"Preview: {filename}")
        self.text = scrolledtext.ScrolledText(self, wrap=tk.WORD)
        self.text.insert(tk.END, content)
        self.text.pack(fill=tk.BOTH, expand=True)
        self.protocol("WM_DELETE_WINDOW", self.on_close)

    def on_close(self):
        self.destroy()

Toplevelウィンドウでファイルの内容を表示し、スクロール可能なテキストボックスを使用しています。

アプリケーションの拡張アイディア

簡易的な実装のみなので、次の様な機能を実装すると更に便利なツールになるかと思います。

非同期処理の改善: 転送やファイル操作を行う際のスレッド処理をさらに効率的にするために、非同期I/Oやキュー管理を強化することができます。
GUIの改善: より洗練されたデザインや、ユーザーのフィードバックに基づいた機能の追加を検討することができます。例えば、進行状況バーや通知機能を追加することが考えられます。
セキュリティの強化: 接続情報や認証情報の保護、ファイアウォールやプロキシの設定など、セキュリティ対策を強化することができます。
デバッグとログ: アプリケーションのログをより詳細に記録し、デバッグ情報を適切に管理することで、問題の診断と修正を容易にすることができます。

ご要望があれば、実装方法などを説明します。

まとめ

今回は前回の「Tera Term クローン自作」に続いて、実際に私が使用しているFTPクライアントソフトをイメージしながら実装しました。

Pythonで実装しているので、コードの可読性が高いのとライブラリが豊富で簡単なコードで高度な処理が実装できる分、学習用途としては優れているかと思います。

実運用に耐え得るレベルに実装するとなると、C言語などの様なパフォーマンスが求められる様になるので、その辺りをご自身で考えて実装してみて下さい。

Linux Python