あらすじ
ローカルPCでロギングしたデータをSSDに格納する、という作業が定常的に発生していました。
その作業には、以下二点の問題点がありました。
１．作業そのものは単純である一方、細かいルールがあり、そのルールを知っている人以外は作業し辛い
ルール例：
・SSDにデータを格納する際のフォルダ名は日付にする
・ローカルPCにデータを残したくないのでカット＆ペーストする
２．１のルールを知っていても作業自体が面倒臭い
上記問題を解決すべく、Pythonで自動化しようと考えました。
備考・Python 3.6.6
挙動
以下フローにて動作します。
ソース
extract_data_to_SSD.py
import os
import re
import sys
import time
import shutil
import datetime
import threading
import traceback
def disp_copying_file_num(fnum, despath): # task1:ファイル移動の進捗状況をコンソールに出力
if os.path.exists(despath): # ファイル移動先のフォルダが既に存在している場合
despath = despath + 'tmp' # 移動先フォルダ名をとりあえず別名で設定
acfnum = 1 # 既にコピーしたファイルの数
fflg = 1 # 最初のファイルをコピーしている場合を表すフラグ
time.sleep(2)
while acfnum < fnum: # 全てのファイルをコピーしていない場合、コピー中のファイル番号をコピー終了までコンソール出力
if len(os.listdir(despath)) > acfnum or fflg: # コピー中のファイルの数が増えた場合
print(str(fnum) + " ファイルのうち " + str(len(os.listdir(despath))) + " つ目をコピー中 … ")
fflg = 0 # 最初のファイルをコピー中の場合のみ、それをコンソールに出力
acfnum = len(os.listdir(despath)) # コピー中のファイル数を更新
time.sleep(0.2)
def move_files_with_folder(file_from, file_to): # task2:ファイルを移動
shutil.copytree(file_from, file_to) # ファイルをコピー
shutil.rmtree(file_from) # 移動元のファイルを削除
def move_only_files(file_from_pc, file_from_ssd, file_to): # task2:ファイルを移動
files = os.listdir(file_from_pc) # コピー対象ファイル一覧を更新
shutil.copytree(file_from_pc, file_from_ssd) # ファイルを一旦、SSD上にtmpフォルダとしてコピー
for i in range(len(files)): # tmpフォルダ内のデータを１つずつ移動
if not os.path.exists(file_to + '/' + files[i]): # 同じ名前のファイルが移動先にない場合
shutil.move(file_from_ssd + '/' + files[i], file_to) # ファイルを移動
else: # 同じ名前のファイルが移動先にあった場合
print('「' + files[i] + '」は既にファイル移動先フォルダに格納されています。')
break
shutil.rmtree(file_from_ssd) # 移動元のファイルを削除
shutil.rmtree(file_from_pc) # コピー元のファイルを削除
def get_ssd_path(): # PCに接続したSSDのパスを取得
drv = ['D:/', 'E:/', 'F:/', 'G:/']
ssd = ['SSD_1', 'SSD_2', 'SSD_3', 'SSD_4'] # SSDを特定するためのファイルなど
for i in range(len(ssd)):
for j in range(len(drv)):
if os.path.exists(drv[j] + ssd[i]):
return drv[j]
return 0
def main():
try:
opath = "C:/Users/○○/Desktop/" # コピー元のパス
dpath = get_ssd_path() # コピー先のパス
if dpath: # SSDが接続されている場合
header = 'soft_' # ファイル名のうちのバージョン番号以外の文字列
files = os.listdir(opath) # コピー対象ファイル一覧
filevers = [0] * len(files) # list型の初期化
# コピー対象フォルダのバージョン番号のみを抽出
for i in range(len(files)):
# 対象フォルダ郡以外のフォルダを除く
if len(files[i]) >= len(header) + 4:
# バージョン番号の箇所にアルファベットが入ってしまっているものを除く
if not bool(re.search('[a-zA-Z_]+', files[i][len(header):len(header) + 4])):
# コピー対象フォルダのバージョン番号のみを格納
filevers[i] = int(files[i][len(header):len(header) + 4])
# バージョン番号が一番大きい(＝新しい)フォルダをコピー対象として設定
ofilepath = opath + header + str(max(filevers)) + '/logging data'
date = str(datetime.date.today()).replace('-', '')
dfilepath = dpath + date # コピー先のパス設定
# マルチスレッド設定 task1:ファイル移動の進捗状況をコンソールに出力 task2:ファイルを移動
task1 = threading.Thread(target=disp_copying_file_num, args=([len(os.listdir(ofilepath)), dfilepath]))
if os.listdir(ofilepath): # 移動するファイルが存在する場合
if os.path.exists(dfilepath): # コピー先に今日日付のファイルが存在する（＝今日2回目以降のコピーする）場合
print('既に存在するフォルダ「' + date + '」にデータを移動します。\n')
task2 = threading.Thread(target=move_only_files, args=([ofilepath, dfilepath + 'tmp', dfilepath]))
task1.start()
task2.start()
else: # コピー先に今日日付のファイルが存在しない（＝今日はじめてコピーする）場合
print('フォルダ「' + date + '」を作成してデータを移動します。\n')
task2 = threading.Thread(target=move_files_with_folder, args=([ofilepath, dfilepath]))
task1.start()
task2.start()
task1.join()
task2.join()
time.sleep(2)
os.mkdir(opath + header + str(max(filevers)) + '/logging data') # SSDに移動して消滅したフォルダを復元
print('\nファイルの移動が完了しました。\n')
else:
print('移動するファイルがありません。\n')
else:
print('SSDがパソコンに接続されていません。\n')
except Exception as e:
allerr = ''
msg = traceback.format_exc()
tmp = msg
for i in range(msg.count('", line ')): # エラー発生個所を抽出
startnum = tmp.find('", line ', len('", line '), len(tmp)) # 'i'個目の'line'があるところをスタートに設定
tmp = tmp[startnum:len(tmp)] # 'i'個目の'line'から最後までを抽出
errnum = tmp[len('", line '):tmp.find(', in ')] # 'i'個目の'エラー発生行数'を抽出
allerr = allerr + errnum # エラー発生行数を追加
if not i == (msg.count('", line ') - 1): # エラー行数間をカンマで区切る
allerr = allerr + ', '
errtitle = str(type(e)).replace('class ', '') # エラー概要
errdtl = str(e.args).replace('("', '').replace('",)', '') # エラー詳細
print('エラーが発生しました。開発者までご連絡下さい。\nTEL: 090-0000-0000\n')
print(' エラー発生箇所：' + allerr)
print(' エラーメッセージ：' + errtitle + ' ' + errdtl)
print('\n')
os.system('PAUSE') # 処理の最後にコンソールを停止
if __name__ == "__main__":
main()
困ったところ（解決済み）
・移動するファイルが複数ある場合、全体のうちいくつまでコピーしたか、分かるようにしたい
→ マルチスレッドを使用し、片方でファイル移動、もう片方で移動先のフォルダをモニタしてコピー中のファイル数をコンソールに出力することで対応しました。
・ファイルの移動に失敗した場合、移動しようとしたデータが消失してしまわないか不安
→ 一応、ファイルの移動は厳密にはコピー＆元ファイル削除、で対応しました。（データ移動先のSSDのファイルシステムがNTFSではなくFATであった場合も考慮に入れて）
・エラーなどのトラブルにすぐ対応できるようにしたい
→ try/exceptの例外処理で全体を覆って、エラー発生時はエラー発生場所・内容・開発者の連絡先がコンソールに出力されるようにしました。
参考にさせていただいた情報
特に下記サイト様にお世話になりました。ありがとうございます。
| 内容 | リンク先 |
|---|---|
| 並列処理（マルチスレッド） | https://qiita.com/__init__/items/74b36eba31ccbc0364ed#%E3%83%87%E3%83%BC%E3%83%A2%E3%83%B3%E3%82%B9%E3%83%AC%E3%83%83%E3%83%89 |
| tkinterを他スレッドで使用する際の注意点 | https://qiita.com/shiracamus/items/cd1d5f2d8fabd4e8a366 |
| エラーが発生した行番号の取得 | https://uyaroom.com/try-except/ |
最後に
コメントやご指摘を下さる方々、本当にありがとうございます。本記事に関しても、改善点や間違い等のご指摘をいただけますと、枕を濡らして喜びに咽びます。