使用Python編寫解壓縮工具


不管是儲存或是傳輸用途,壓縮檔一直是非常常見的檔案格式。一般而言,許多系統與軟體都提供解壓縮功能,我們也可以嘗試著自己製作一個解壓縮工具

自己製作一個解壓縮工具?使用情境為何?

一般而言,許多系統與軟體都提供解壓縮功能,想要解壓縮可說是非常輕易達成。但試想一些情況,若今天有一堆壓縮檔案需要進行解壓縮呢?可能有一些工具一時之間就無法起到很好的作用了。

所以除了磨練自己寫程式的能力之外,我們也能藉此機會,自己客製化具有解壓縮功能的小工具,以解決某些特殊問題。

一堆壓縮檔案需要解壓縮?何種情況下會有這樣的需求?

基本上之所以會有一堆壓縮檔案,可能是為了儲存以及分享用途,所以對檔案進行分門別類後,個別進行壓縮。

哪天誰需要使用這些資料時,就會遇到這樣的解壓縮問題。

舉個例子,在台灣期貨交易所網站中,提供多種商品的詳細資料下載,可能以天、週或是年為單位,部分檔案格式是zip壓縮檔,像是期貨前30個交易日期貨每筆成交資料期貨前30個交易日期貨價差委託成交概況表期貨前30個交易日期貨價差每筆成交資料選擇權前30個交易日選擇權每筆成交資料等等,像這些以zip壓縮檔形式,並且以天為單位做區分的檔案,一次下載下來,若能夠一次解完壓縮,可說是幫個大忙了。


例子

假設目前有以下這樣的壓縮檔案,我們可以看到總共有22個檔案,平均一個壓縮檔1MB:


而全部解壓縮後,發現資料大小整個爆增,直接來到了1.2GB:


雖在預料之內,但這能讓我們意識到,像金融交易這樣的大量固定格式資料而言,壓縮的效果是非常好的。

而這樣的每日資料要全部解壓縮,不使用一些特殊技巧可能會把自己累得半死,何況若要對資料進行分析,還沒開始就一股無力感都上來了~

所以我們會需要像以下這樣的小工具,即便介面非常簡單:

確立目標

所以,我們在此訂立一個目標,這個使用Python實作出的解壓縮工具,應該要具有將指定目標資料夾內的所有壓縮檔皆進行解壓縮,並輸出到預設的輸出目錄

此外,即便指定目標資料夾內,還有子資料夾,也會依照原定目錄結構,依序遍歷尋找所有壓縮檔進行解壓縮,輸出後,目錄結構不變。

使用的Python環境

本次介紹使用的Python版本為3.6.5

使用到的模組都是內建模組,也不需要額外安裝套件。

作業系統環境為Windows 10。

開始使用Python撰寫解壓縮工具

為了實現解壓縮功能,我們需要引入zipfile模組。

import zipfile

此外,我們會需要一些路徑管理功能的模組,其中我們只需要使用pathlib模組中的Path Class。

from pathlib import Path
import os

模組參考連結:

再來,我們設定一個Global variable,用來儲存解壓縮目的地路徑。

destination_dir = './unzip_file'

再來是設定壓縮檔資料夾路徑,這邊有加上檢查資料夾是否存在的機制,可以依個人需求進行其他調整:

zip_dir = ''
while not os.path.exists(zip_dir):
if zip_dir != '':
print('路徑不存在或錯誤...')
zip_dir = input('請輸入zip壓縮檔所在目錄 : ')
zip_dir = zip_dir.replace('\\', '/')

其中,zip_dir變數之所以額外進行路徑分隔符的取代,是因為不同作業系統的路徑分隔符號不同,為了通用,一律取代為'/'

此外,為了避免目的資料夾不存在,導致解壓縮過程出錯,我們添加一個建立資料夾函式,若資料夾已經存在,則不進行任何動作(exist_ok參數設定為True時將不會產生exception):

os.makedirs(destination_dir, exist_ok=True)

為了提示已開始解壓縮,輸出一行工作開始的訊息。

print("資料解壓縮...")

當所有前置步驟都完成後,便是要進行主要功能實作。

主要功能實現

首先我們會需要遍尋所有可能的壓縮檔,以zip檔為例,我們使用到了pathlib模組中的Path Class。

Path Class會需要一個引數,為目標路徑,其會為這個目標路徑,實體化一個PosixPath或者WindowsPath物件,有了這樣的物件,便可以呼叫相對應且相當好用的函式。其中兩種物件會根據執行的作業環境,或者說作業系統而決定哪種會被實體化。

這邊提個題外話,Posix,全稱為Portable Operating System Interface,叫做可移植作業系統介面,是一系列的IEEE標準,其定義了application programming interface (API),並附帶command line shells以及utility interfaces。之所以定義這樣的API,主要是為了不同作業系統之間軟體的相容。有了API,我們可以不用直接調用System call等內核核心指令,透過應用程式介面,達成人機互動,方便與電腦的溝通。
而最一開始作為基礎的作業系統為Unix,主要原因是自由的,也就是vendor-neutral,這也是The Open Group的主要目標。相關參考資料如下: 

由於執行的作業系統為Windows,它幫我實體化了WindowsPath物件。這個物件有一個glob()函式,其可以使用*號來為路徑做囊括表示,其功用相當於正規表達式中定義的*號字元。

若使用"*.副檔名",可用來代表任何檔名的指定副檔名。

若使用"**",則可代表中間任何路徑。比如以下的程式碼,可以遍尋找到最底端的所有zip檔。以樹來想像,便是找到所有符合條件的葉節點,或者說終端節點。

# Path(zip_dir) => <class 'pathlib.WindowsPath'>
list(Path(zip_dir).glob('**/*.zip'))

這邊要提一下,glob()函式將會回傳一個generator物件,因其可迭代的特性,我們另外轉型為list結構,方便之後取用。

generator object Path.glob at some address

到這裡,我們可以得到以下這樣的資料,list中的元素也都會是WindowsPath物件,每一個都代表已經找到的zip檔路徑。

[WindowsPath('1.zip'), ..., WindowsPath('n.zip')]

因此,我們可以使用for … in ... 迴圈進行元素拜訪,取出的元素因為是WindowsPath物件,而我們到這個步驟只需要路徑,我們透過str()進行轉型,並且為了通用,一律將路徑分隔符進行替代。

for zip_file in list(Path(zip_dir).glob('**/*.zip')):
zip_file = str(zip_file).replace('\\', '/')

print('\t' + zip_file)

針對每個元素,我們會使用zipfile模組中的ZipFile()函式建立ZipFile物件實例,而由於ZipFile物件是一個context manager,因此我們可以使用with關鍵字為我們處理一部份的檔案管理,當任務結束便會自動進行關閉檔案的動作。

ZipFile()函式的第一引數file需要檔案路徑、file-like object或者path-like object,所以也可以是平常存取檔案常用的open()函式的回傳結果(_io.TextIOWrapper)。

mode引數可選擇是否輸入,有read file =>'r'以及write file =>'w'(truncate and write)、'x'(exclusively create and write)、'a'(append)。

另外,在這邊我們會需要決定最終的解壓縮路徑,為保持原結構,我們使用到了destination_dir變數、zip_dir變數以及zip_file變數。

例子:

destination_dir => 'D:/a/b/c'

zip_dir => 'E:/d/e/f'

zip_file => 'E:/d/e/f/g/h.zip'

output => 'D:/a/b/c/f/g'

for zip_file in list(Path(zip_dir).glob('**/*.zip')):
zip_file = str(zip_file).replace('\\', '/')

print('\t' + zip_file)
with zipfile.ZipFile(zip_file, mode='r') as f_zip:
zip_file = zip_file[:zip_file.rfind('/')]

output_dir = os.path.join(destination_dir, zip_file.replace(zip_dir[:zip_dir.rfind('/') + 1], ''))
print(output_dir)

到目前為止,解壓縮的所有要素都備齊了。

這邊要另外提zip_dir[:zip_dir.rfind('/') + 1]這段程式碼,其實可以在進入迴圈之前就決定好,畢竟其輸出結果在整個迴圈的執行階段內並不會再更動了,多少可以加快效率。

為了解壓縮全部檔案,我們可以使用到ZipFile物件所提供的extractall()函式,其第一引數path需要輸出的目的地目錄。而若壓縮檔有密碼加密,則可以使用pwd引數進行指定,一般是以bytes型態給定。

for zip_file in list(Path(zip_dir).glob('**/*.zip')):
zip_file = str(zip_file).replace('\\', '/')

print('\t' + zip_file)
with zipfile.ZipFile(zip_file, mode='r') as f_zip:
zip_file = zip_file[:zip_file.rfind('/')]

output_dir = os.path.join(destination_dir, zip_file.replace(zip_dir[:zip_dir.rfind('/') + 1], ''))
print(output_dir)

f_zip.extractall(output_dir, pwd=b"")

最後,所有工作結束時,輸出提示訊息。

print("解壓縮完成!")
input("PRESS ANYKEY TO LEAVE.")

參考執行結果:




結論

事實上,若在一個自動化系統中,無論是做交易策略的,或是純粹進行資料分析的,若碰到需要額外需求,就像取得的資料是經過壓縮,取用時會需要解壓縮這樣,可能的重複工作任務,納入自動化一部份是非常值得的,何況可能還需進行檔案管理,對不同資料進行資料夾分類,指派到指定資料夾。因此,對這樣解壓縮的小任務,進行客製化,除了是一個小的編寫程式練習,也是一種實際應用。


沒有留言:

張貼留言