在前一篇博客文章 《使用 Python 編寫腳本並發布》 中,我介紹了如何使用 Python 進行腳本編程,說實話這是我在嘗試 Python 進行網站和網絡編程之后首次使用 Python 進行腳本編程,前面也說過之前雖然使用 Bash 構建過一些腳本,但是由於我對 Bash 不熟練,對它的使用都僅限於最基礎的命令行操作,僅僅是比 alias 別名操作稍微簡單一點。上次介紹的腳本是如何添加命令行參數以及將現有的操作流程用一個腳本簡單化,這一次介紹的腳本是一個非常實用而且經過優化的文件變動事件監視腳本。
P1 Python 腳本:文件變動檢測
在廖雪峰 Python 教程實戰部分的 Day 13 - 提升開發效率 中,他給我們介紹了一種用於提升開發效率的方法:
- 首先執行我們需要的命令
- 監聽當前目錄,並判斷變動文件的后綴名,若后綴名為 .py,則觸發回調函數
- 回調函數觸發后,自動重新啟動命令
流程很清楚,實現起來也很簡答,廖雪峰利用 Python 的 subprocess 和第三方庫 watchdog 分別實現了重啟命令和監聽當前目錄的文件變動情況。大概 70 余行代碼就能完成這樣一個簡單且實用的腳本。
在我編程的過程中,經常需要用到這樣一個監控文件變動並自動重新執行預設命令的操作,比如我在編寫 SSPYMGR 這個網站程序時經常要用到文件改動后自動重啟服務器的操作,或者我經常需要在改動某些文件后自動上傳到虛擬機上。當我有這些需求時,我之前的做法就是將上面廖雪峰介紹的腳本復制到我要監視的文件夾中,然后直接修改腳本里面的命令參數,這樣做很直接,但是很繁瑣。
我要做的是:將上面的簡單腳本進行優化,使得可以通過命令行參數對腳本的行為進行設置。主要的優化目標有:
- 可以預設命令,並且該命令可以帶參數
- 可以設置監聽的目錄,並且設置是否遞歸監聽子目錄
- 可以設置監聽的文件后綴名,並設置可以排除在監聽范圍內的文件名
- 增加保存參數功能,並且能夠讀取保存的配置文件
P2 優化腳本
為了實現上面這些目標,就像我們在上一篇博客那樣,用 argparse 庫來對復雜的命令行參數進行解析,這一次我們換一種代碼的組織方式,將命令行參數的解析和配置文件封裝到類中,然后通過實例化類對象解析參數,然后將配置寫入到字典中,程序執行流程以指定的配置文件為主:
若指定了要讀取的配置文件,則將配置文件中的內容作為配置,忽略掉其他選項。指定配置文件主要可以簡化命令行的參數輸入過程。若沒有指定讀取的配置文件,則以命令行中其他的選項為配置。
在 monitor.py 這個腳本中我將配置和命令行參數讀取封裝到類 Configuration 中:
class Configuration(object):
_DEFAULT_LOC = _CONFIG_DIR / "monitor_default.json"
def __init__(self):
self.config = {}
self._addArgs()
def readConfig(self, file: Path):
pass
def _addArgs(self):
pass
def parseArgs(self):
pass
監聽目錄
接下來就要用到第三方庫 watchdog 來監聽指定的目錄及指定事件觸發時的操作了。事件處理器要用到 watchdog.events.FileSystemEventHandler,我們用繼承的方式處理事件:
from watchdog.events import FileSystemEventHandler
class MyFileSystemEventHander(FileSystemEventHandler):
def __init__(self, fn, config: Configuration):
super(MyFileSystemEventHander, self).__init__()
self.restart = fn
self.config = config
self.last = time.time()
構造事件處理器時需要傳入回調函數和配置對象,接下來定義事件處理函數,這里會監聽目標文件夾中所有的文件事件 on_any_event,但是該事件會在保存文件時觸發兩次,因此需要對它做一個防抖處理,防抖處理就是判斷兩次事件觸發的時間間隔是否超過預設值,若兩次事件時間間隔過短,則忽略第二次事件。
以下時事件處理的代碼:
class MyFileSystemEventHander(FileSystemEventHandler):
def on_any_event(self, event):
# for debounce
cur = time.time()
if cur - self.last < 0.25:
return
self.last = cur
ext_able = False
src = Path(event.src_path)
if src.name not in self.config["exclude"]:
for ext in self.config["mon_ext"]:
if src.suffix == ext:
ext_able = True
break
if ext_able:
logger.info('File changed: {}'.format(src))
self.restart()
上面的防抖處理時以第一次事件為准,忽略掉之后一段時間內的其它事件,這樣做更方便。
還有另一種復雜但更合理的處理方式,即事件觸發時不立即調用處理函數,延遲一段時間,在該段時間內若有其他事件發生,則以新事件為准,重新計算延遲時間,超過時間后再執行事件處理的代碼。
第二種處理方式更合理。打個比方,我在很短的時間內先后保存了兩個不同的文件 A 和 B,用第一種方式,程序重啟后只會重新加載 A 文件而 B 文件的改動很可能被忽略掉了;而用第二種方式 A 文件改動后程序並不會立即重新加載,而 B 文件的改動會被監聽到,最終就是在延遲一段時間后程序會重新加載 A 和 B 這兩個文件。
自動重啟程序
自動重啟程序時依靠 subprocess.Popen 對象實現的,啟動的時候實例化一個 Popen 對象,停止程序時調用它的 kill() 方法;重啟就是先 kill 再重新實例化。這個過程用 NewProcess 類進行封裝:
import sys
from watchdog.observers import Observer
import subprocess
class NewProcess(object):
def __init__(self, config: dict):
self.process = None
self.config = config
self.command = self.config["cmd_args"][:]
self.command[0:0] = self.config["cmd"]
self.args = ' '.join(self.command)
然后還需要用到 watchdog.observers.Observer,用來監聽目錄,並且通知處理器進行處理:
class NewProcess(object):
def start_watch(self):
observer = Observer()
observer.schedule(MyFileSystemEventHander(self._restart, self.config),
path=self.config["mon_dir"],
recursive=self.config["recursive"]
)
observer.start()
logger.info('Watching directory: {}'.format(self.config["mon_dir"]))
self._start()
try:
while True:
time.sleep(0.5)
except KeyboardInterrupt:
observer.stop()
observer.join()
腳本就完成了。可以在命令行中嘗試一下,輸入 bf_monitor -c echo -a test 可以看到類似的輸出:

它還有些缺陷,不能在 -a 后面添加的參數里帶有 - 前綴:bf_monitor -c echo -a -test 是不允許的:

為了解決這個問題,只有在 -c 后面將這些命令用引號包裹起來,bf_monitor -c "python -V" :

關於 watchdog 的詳細使用或者 API,請參閱其 官方文檔.
P3 python 國際化 i18n
到目前為止,我已經用 Python 做了兩個腳本:bf_gitrepo 和 bf_monitor,並且我給他們都加上了命令行幫助信息,但是它們的幫助信息都是英文,我們要把這些信息翻譯成中文。翻譯工作主要依靠 Python 的 gettext 模塊和第三方的 pybabel 模塊。
事實上,國際化只要嘗試一遍流程之后就很簡單了,我第一次使用 pybabel 時,大部分時間都是在提取可翻譯文本上,之后做 monitor.py 腳本的翻譯時就輕車熟路,完成的很快,只在翻譯上花了點時間。
在 brifuture-facilities 中,我將 gettext 模塊簡單的封裝了一下,程序會在腳本的同級目錄下尋找 locale 文件夾中的 .mo 文件,然后替換腳本中的文本:
LANGUAGE_DIR = (Path(__file__).parent / "locale").resolve()
import gettext
def initGetText(domain="myfacilities") -> gettext.gettext:
gettext.bindtextdomain(domain, LANGUAGE_DIR)
gettext.textdomain(domain)
gettext.find(domain, "locale", languages=["zh_CN", "en_US"])
return gettext.gettext
一般會將 gettext.gettext 以其他的名稱導入到 Python 程序中,如 from gettext import gettext as _,由於之前我習慣用 Qt 翻譯方法 tr,所以我將 gettext.gettext 用別名 tr 代替。在程序中要替換文本的位置用 tr 方法包裹起來:
parser.add_argument("-d", "--directory", help=tr("The directory to monitor, . by default."))
然后我們需要配置 babel,要讀取的只有 python 文件(如果你要讀取其他文件,可以看看 [文檔](http://babel.pocoo.org/en/latest/):
# file: babel.cfg
# Extraction from Python source files
[python: **.py]
keywrods = tr
文本查找
接下來使用 pybabel 程序進行文本查找,我們只用查找 monitor.py 文件:
# pybabel extract -F ./babel.cfg -o ./bffacilities/locale/{}.pot -k tr ./bffacilities/{}.py
pybabel extract -F ./babel.cfg -o ./bffacilities/locale/monitor.pot -k tr ./bffacilities/monitor.py
盡管前面的配置文件中指定了關鍵字為 tr,但我在使用中發現調用 extract 子命令時最好還是加上選項 -k tr,保證能夠提取出文本。
查看 bffacilities/locale 目錄下,應該有 monitor.pot 文件,里面有很多的 msgid、msgstr。這個文件就保存了所有要翻譯的文本。當程序中的文本更新后,重新調用上面的命令再次提取文本即可。
文本翻譯
然后我們要對提取出來的文本進行翻譯,如果是初次翻譯要使用 init 子命令,但若是更新翻譯就不是用 init 子命令而是用 update 子命令了:
# pybabel init -i ./bffacilities/locale/{}.pot -d ./bffacilities/locale/ -l zh_CN -D {}
pybabel init -i ./bffacilities/locale/monitor.pot -d ./bffacilities/locale/ -l zh_CN -D monitor
# pybabel update -i ./bffacilities/locale/{}.pot -d ./bffacilities/locale/ -l zh_CN -D {}
pybabel update -i ./bffacilities/locale/monitor.pot -d ./bffacilities/locale/ -l zh_CN -D monitor
之后我們就可以開始翻譯了,在 ./bffacilities/locale/zh_CN/LC_MESSAGES/ 目錄下找到 monitor.po 文件,用編輯器打開,或者用 Poedit 打開,在 msgid 對應的 msgstr 下面填入文本即可。注意有些語句的 msgid 可能會跨多行,不用管它直接翻譯就行。
翻譯完成后對其進行打包:
# pybabel compile -d ./locale/ -D {}
pybabel compile -d ./bffacilities/locale/ -D monitor
查看翻譯效果
制作完翻譯文件后,來看看腳本幫助信息是不是輸出中文了,先檢查一下 locale 的輸出:

查看 bf_monitor 的幫助信息:

修改 locale,LANG=en_US.UTF-8 && LANGUAGE=en_US,locale 輸出變為:

再看看 bf_monitor 的幫助信息:

修改 setup.py
最后我們要將寫好的程序打包,為了防止在打包過程中丟失翻譯文件,我們要將 setup 參數中的 zip_safe 改為 false: zip_safe=False,
然后在 setup.py 的同級目錄下添加 MANIFEST.in 文件,內容如下:
recursive-include bffacilities/locale *
global-exclude *.pyc
最后上傳到 pypi 上面即可通過 pip 下載安裝。
P4 小結
之前使用 Nodejs 時,我用 Node 編寫過一個文件變動檢測的腳本,但是現在我找不到之前的那篇博客了,文件變動檢測的 Python 腳本和 Node.JS 腳本原理都是一樣的,都是通過監聽文件事件,然后執行回調函數。
另外通過這次的翻譯過程我掌握了如何國際化 Python 程序,之前我做 Qt 程序時對 Qt 的翻譯流程比較清楚,轉用 Python 程序后發現其實國際化的流程都很類似,在文件中查找調用翻譯函數,提取之后用軟件或編輯器進行翻譯。最后轉換成程序可以直接讀取的格式(可能這樣做能夠提高程序的效率吧)。
這段程序的代碼可以在 github 上找到,你也可以看到整個項目的源代碼。如果你覺得這篇文章對你有所幫助或者你認為這篇文章還不錯,就給我點個贊吧,感謝你的支持。
參考
python的國際化gettext模塊
Flask-Babel 簡介
The Invent with Python Blog
http://babel.pocoo.org/en/latest/messages.html
