Windowsファイルサーバ(SMB)をスクリプトから操作するとき、ネイティブにはPowerShellですが、Pythonで実装するとLinuxでも使えるので便利です。
昨今Linuxのファイラなら大抵のディストリビューションでサポートされていますし、コマンドラインだとmount cifsでマウントして操作する手もありますが、やはり自動化したいケースがあります。
Pythonだとpysmbが有名なようです。そこで今回はpysmbを使ってみます。
もくじ：
はじめに
pysmbも、みんな大好きpipでインストールできます。
1 2 3 | > pip install pysmb > python --version Python 3.6.1 |
基本はSMBConnectionクラスを使ってコネクション作成、ファイル参照・操作、コネクション破棄、という流れになるようです。
毎度長くなりますが、はじめにサンプルSmbClient.pyを記載しておきます。pysmbのAPIを組み合わせて、ファイル操作周りの機能を補助的に追加したSmbClientクラスを定義しています。例外の補足が面倒だったのでちゃんとやっていませんが、基本的には問題があれば止まるようになっています。
なお、環境はWindows 10 + Python3.6で検証しています。
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 | # coding: utf-8 import sys import os from datetime import datetime from traceback import print_exc from smb.SMBConnection import SMBConnection class SmbClient: def __init__(self, conf): # set configuration self.__username = conf["username"] self.__password = conf["password"] self.__domain = conf["domain"] self.__remote_host = conf["remote_host"] self.__remote_port = conf["remote_port"] # resolve host name import socket self.__remote_addr = socket.gethostbyname(conf["remote_host"]) self.__client = socket.gethostname() @staticmethod def sample_conf(): return {"username": os.getlogin(), "password": "P@ssw0rd", "domain": "WORKGROUP", "remote_host": "remotehost.local", "remote_port": 139} def list_dir(self, svc_name, path_, recurse=False): try: self.__list = [] session = self.__get_connection() # return object list if the path is existing remote directory if self.__check_dir(session, svc_name, path_): self.__get_files(session, svc_name, path_, recurse, True) session.close() return self.__list except: print_exc() session.close() return [] def receive_items(self, local_dir, svc_name, path_, recurse=False, relative=True): # check if local path is valid if not os.path.isdir(local_dir): print("local directory is not available ({})".format(local_dir)) return False try: self.__list = [] session = self.__get_connection() # switch for the path is directory or file if self.__check_dir(session, svc_name, path_): remote_dir = path_ * relative self.__get_files(session, svc_name, path_, recurse) else: remote_dir = os.path.dirname(path_) self.__list = [{"path": path_}] # batch download for index, item in enumerate([x["path"] for x in self.__list]): print("\t[{}/{}] retrieving {}...".format(index + 1, len(self.__list), os.path.basename(item))) self.__download(local_dir, session, svc_name, item, remote_dir) session.close() return True except: print_exc() session.close() return False def send_file(self, local_file, svc_name, remote_dir): # check if local path is valid if not os.path.isfile(local_file): print("invalid path: {}".format(local_file)) return False try: session = self.__get_connection() if self.__check_dir(session, svc_name, remote_dir): print("sending file...") with open(local_file, 'rb') as f: # store file to remote path path_ = os.path.join( remote_dir, os.path.basename(local_file)) session.storeFile(svc_name, path_, f) else: print("invalid remote path: {} - {}".format(svc_name, path_)) session.close() return True except: print_exc() session.close() return False def __check_dir(self, connection, service_name, remote_path): return connection.getAttributes(service_name, remote_path).isDirectory def __get_connection(self): # build connection connection = SMBConnection(username=self.__username, password=self.__password, my_name=self.__client, remote_name=self.__remote_host, domain=self.__domain) # open connection connection.connect(self.__remote_addr, self.__remote_port) return connection def __get_files(self, connection, service_name, base_dir, recursive=True, include_dir=False): # sub funciton for formatting data # see https://msdn.microsoft.com/en-us/library/ee878573.aspx def _convert_obj(sf, base_dir): return {"path": os.path.join(base_dir, sf.filename), "filename": sf.filename, "isDirectory": sf.isDirectory, "last_write_time": "{:%Y-%m-%d, %H:%M:%S}".format(datetime.fromtimestamp(sf.last_write_time)), "file_size": "{:,.1f} [KB]".format(sf.file_size / 1024), "file_attributes": "0x{:08x}".format(sf.file_attributes)} # iterate paths for item in connection.listPath(service_name, base_dir): # skip aliases if item.filename == "." or item.filename == "..": continue if item.isDirectory: if include_dir: self.__list.append(_convert_obj(item, base_dir)) if recursive: # call recursively for sub directory self.__get_files(connection, service_name, os.path.join(base_dir, item.filename), recursive, include_dir) else: self.__list.append(_convert_obj(item, base_dir)) def __download(self, local_base_dir, connection, service_name, remote_path, remote_base_dir=""): # calculate local path and create parent directory if not exist local_path = os.path.join( local_base_dir, os.path.relpath(remote_path, remote_base_dir)) if not os.path.isdir(os.path.dirname(local_path)): os.makedirs(os.path.dirname(local_path)) # download file from remote host with open(local_path, 'wb') as f: connection.retrieveFile(service_name, remote_path, f) |
まあ見たまんまなのですが、内部動作と落とし穴っぽいところを以降に記載していきます。
サービス名とパス
用語の話なのですが、サービス名(service_name, svc_name)やらパス(remote_path, path_)という言葉が出てきます。
サービス名はSMBサーバにアクセスしたときに、ディレクトリのアイコンではなくこんなアイコン↓で表示されるやつ（共有フォルダ）です。
(リモート)パスは残りのパスです。
つまり、UNCで書けば「\\(ホスト名)\(サービス名)\(パス)」という構造になっています。
コネクションの作成
サンプルでは、コンストラクタでコネクションの作成に必要な情報confをセットしています。
このクラス独自に定義しているものですが、SMBConnectionオブジェクトの作成に必要な項目をまとめたものと考えて下さい。
単なるディクショナリなので地道に定義してもいいのですが、既定値をスタティックメソッドで取得できるようにしています。
1 2 3 4 5 6 7 | @staticmethod def sample_conf(): return {"username": os.getlogin(), "password": "P@ssw0rd", "domain": "WORKGROUP", "remote_host": "remotehost.local", "remote_port": 139} |
この情報を使って、プライベートメソッド__get_connectionで内部的にSMBConnectionオブジェクトを作成しています。
渡す値が間違っていたり、サーバと通信できないときは例外が投げられます。
1 2 3 4 5 6 7 8 | def __get_connection(self): # build connection connection = SMBConnection(username=self.__username, password=self.__password, my_name=self.__client, remote_name=self.__remote_host, domain=self.__domain) # open connection connection.connect(self.__remote_addr, self.__remote_port) return connection |
内部的にはサーバのIPアドレスやらクライアントのホスト名が必要になるようなのですが、これはコンストラクタ内でsocketモジュールを使って解決させています。
Windowsユーザ諸兄にはお馴染みだと思いますが、それなりに歴史の長いSMBにはいくつかバリエーションがありますので、うまく行かない場合はいろいろ設定を試してみて下さい。
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 | from smbclient import SmbClient # get default config conf = SmbClient.sample_conf() # 非ドメイン環境(Workgroup)でSMB1 conf = {"username": "smbuser", "password": "P@ssw0rd", "domain": "WORKGROUP", "remote_host": "remotehost", "remote_port": 139} # ドメイン環境でSMB2 conf = {"username": "smbuser", "password": "P@ssw0rd", "domain": "mydomain", "remote_host": ".local", "remote_port": 445} # IPアドレスでも可 conf["remote_host"] = "192.168.100.211" |
ファイルの参照
ではまずサンプルの.list_dirメソッドを使って、ファイルやディレクトリの一覧を取得する例です。パスワードはgetpassモジュールを使って、実行中にコンソールから入力させています。
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 | import getpass from smbclient import SmbClient # get default config conf = SmbClient.sample_conf() # update config conf["username"] = "myuser" conf["password"] = getpass.getpass() conf["remote_host"] = "remotehost" # path to UNC '\\{remotehost}\{svc_name}\{remote_path}' svc_name = "servicename" remote_path = "path/to/remote/dir" # listing recurively for item in _client.list_dir(svc_name, remote_path, recurse=True): print(" {}".format(item["filename"])) for key in item.keys(): if key != "filename": print("\t{}: {}".format(key, item[key])) print("") |
実行結果はこんな感じ。この使い方だと、指定されたパス(ディレクトリ)にあるサブディレクトリとファイルの情報をリストとして返します。
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 | >> python test_listdir.py Password: subdir1 path: path/to/remote/dir\subdir1 isDirectory: True last_write_time: 2017-12-20, 22:00:08 file_size: 0.0 [KB] file_attributes: 0x00000010 subdir2 path: path/to/remote/dir\subdir2 isDirectory: True last_write_time: 2017-12-20, 22:00:21 file_size: 0.0 [KB] file_attributes: 0x00000010 file-2-002.txt path: path/to/remote/dir\subdir1\subdir2\file-2-002.txt isDirectory: False last_write_time: 2013-06-15, 02:48:05 file_size: 53.6 [KB] file_attributes: 0x00000020 file-000.txt path: path/to/remote/dir\file-000.ps1 isDirectory: False last_write_time: 2012-12-09, 05:06:15 file_size: 26.5 [KB] file_attributes: 0x00000020 |
以下、pysmbのAPIを呼び出している部分の解説です。
リモートパスの確認
ファイルやディレクトリの内容ではなく属性(メタ情報)を知りたい場合には、SMBConnection.getAttributes()を使います。
指定されたリモートパスが存在しない場合や、アクセス権限がなかった場合には例外を投げてきます。
ここではディレクトリか否かを判定するためにisDirectoryプロパティを参照しています。
1 2 | def __check_dir(self, connection, service_name, remote_path): return connection.getAttributes(service_name, remote_path).isDirectory |
ディレクトリ内の要素取得
SMBConnection.listPath()は、ローカルで言うところのos.listdir()みたいなやつです。ここではisDirectoryプロパティで判定して、サブディレクトリが見つかれば再帰的に取得し、ファイルであれば整形してリストに格納しています。
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 | def __get_files(self, connection, service_name, base_dir, recursive=True, include_dir=False): def _convert_obj(sf, base_dir): ... # iterate paths for item in connection.listPath(service_name, base_dir): # skip aliases if item.filename == "." or item.filename == "..": continue if item.isDirectory: if include_dir: self.__list.append(_convert_obj(item, base_dir)) if recursive: # call recursively for sub directory self.__get_files(connection, service_name, os.path.join(base_dir, item.filename), recursive, include_dir) else: self.__list.append(_convert_obj(item, base_dir)) |
SharedFileインスタンスの解釈
SMBConnection.listPath()はSharedFileインスタンスを返してきます。
ここでは全て無理やり文字列にフォーマットしていますが、フォーマットのやり方を見れば各プロパティの元の型がわかると思います。
1 2 3 4 5 6 7 8 9 | # sub funciton for formatting data # see https://msdn.microsoft.com/en-us/library/ee878573.aspx def _convert_obj(sf, base_dir): return {"path": os.path.join(base_dir, sf.filename), "filename": sf.filename, "isDirectory": sf.isDirectory, "last_write_time": "{:%Y-%m-%d, %H:%M:%S}".format(datetime.fromtimestamp(sf.last_write_time)), "file_size": "{:,.1f} [KB]".format(sf.file_size / 1024), "file_attributes": "0x{:08x}".format(sf.file_attributes)} |
ファイルの操作
サンプルの.receive_itemsメソッドや.send_fileメソッドの使用例はこんな感じです。
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 | import getpass from smbclient import SmbClient # get default config conf = SmbClient.sample_conf() # update config conf["username"] = "myuser" conf["password"] = getpass.getpass() conf["remote_host"] = "remotehost" # path to UNC '\\{remotehost}\{svc_name}\{remote_path}' svc_name = "servicename" remote_path = "path/to/remote/dir" # download files local_path = "path/to/local/dir" _client.receive_items(local_path, svc_name, remote_path, recurse=True) # upload files local_path = "path/to/local/file.ext" _client.send_file(local_path, svc_name, remote_path) |
ダウンロードのときは、ローカルとリモートのディレクトリをそれぞれ指定します。また、アップロードの時はローカルにあるファイルを指定し、リモートのディレクトリに同じ名前のファイルとしてコピーします。
とくにコンソール出力はないので、実行結果は省略します。
ファイルのダウンロード
一応、リモートのファイルを指定したときとディレクトリを指定したときの両方に対応できるよう挙動を変えています。ディレクトリの場合はそのリモートディレクトリ以下のファイルやサブディレクトリも対象にできます。
リモートのファイルが指定された場合にはそのまま、もしくはディレクトリが指定された場合には上記の例と同様に.__get_filesメソッドでファイルのパスを取得し、.__downloadメソッドに渡します。内部的にはローカルファイルをバイナリ書き込みモードでオープンし、SMBConnection.retrieveFile()メソッドに渡しています。
1 2 3 4 5 6 7 8 9 10 | def __download(self, local_base_dir, connection, service_name, remote_path, remote_base_dir=""): # calculate local path and create parent directory if not exist local_path = os.path.join( local_base_dir, os.path.relpath(remote_path, remote_base_dir)) if not os.path.isdir(os.path.dirname(local_path)): os.makedirs(os.path.dirname(local_path)) # download file from remote host with open(local_path, 'wb') as f: connection.retrieveFile(service_name, remote_path, f) |
なおサンプルの実装では、ファイルのみを対象にしているので、リモートにある空のディレクトリはローカル側で作成されません。
ファイルのアップロード
アップロードに関しては、あまり複雑なことをしていません。ディレクトリに対応するのが面倒だったので。ローカルのファイルをバイナリ読み込みモードで開き、SMBConnection.storeFile()メソッドに渡しています。
1 2 3 4 5 6 7 8 | session = self.__get_connection() if self.__check_dir(session, svc_name, remote_dir): print("sending file...") with open(local_file, 'rb') as f: # store file to remote path path_ = os.path.join( remote_dir, os.path.basename(local_file)) session.storeFile(svc_name, path_, f) |