[python] [linux] 第1次將工作自動化就上手

點閱: 33

專案說明

公司有一支用 pascal(delphi) 寫的程式,功能是給使用者手動更新一張健保署的 excel 到資料庫的程式,由於 delphi 只支援 BIG5 的編碼,在處理那張資料表的時候,有時候會出現亂碼而無法順利地匯入資料庫中。因此最近被交辦用 python 來處理這項工作任務。

看了一下原本的程式碼後,發現任務很單純,也沒什麼需要人工處理的例外,大致的過程是在 linux server 使用 crontab 排程執行 python 程式,並把執行結果 mail 給自己即可,就以此文來記錄工作的內容。

任務分解

先來分解這項任務所需要的功能:

  1. 自動化下載與解壓縮
  2. 讀取檔案,進行資料處理
  3. 資料庫操作
  4. 程式記錄檔(log)
    • 自訂 logger
    • 同一 logger 在不同 module 間共用
  5. 自動化執行
    • 排程執行
    • email 通知

以下逐步紀錄工作的過程

任務說明

1. 自動化下載與解壓縮

對於網站的解析,最常用到的套件就是 requestsbeautifulsoup 了,在 terminal 下輸入指令安裝:

pip install requests
pip install beautifulsoup4

beautifulsoup 在解析網站的時候有幾種方法,詳細說明在此,這邊我選用 lxml 的方法,因此也要在 terminal 中安裝 lxml 的解析引擎。

pip install lxml

安裝完成後,就可以開始進行 web 的操作了。我的目標是,到網站中找到目標 excel 存放的位置,並讀取最後更新日期,來判斷是否需要更新資料表。當然也需要用到

以下是程式碼

import requests
from bs4 import BeautifulSoup
import re

web = 'https://my_web.index'
soup = BeautifulSoup(web_html, 'lxml')  # 使用 lxml 的方法去解析網站
target_list = str(soup.select('li.annex ul')[0])  # soup.select 可找出 CSS 底下的 tag
target_str = re.search(r'健保特約醫療院所名冊壓縮檔\(([^\)]+)\)', target_list)[0]  # 用 re 拿到我要的資料表以及更新日期
target_date = re.search(r'([\d.]+)', target_str)[0]  # 用 re 取出日期
date_ad = str(int(target_date.split('.')[0]) + 1911) + target_date.split('.')[1] + target_date.split('.')[2]  # 把民國年轉為西元年

下載好之後,透過 zipfile 這個 package 來完成解壓縮

import requests
from zipfile import ZipFile

url = 'https://my_web.index/my_file.zip'
r = requests.get(url)  # 把 url 解析出來
path = './input/'
zip_name = 'download_nhi.zip'

# download and save file.
with open(path + zip_name, 'wb') as f:
    f.write(r.content)  # 把 r.content 寫到 f 中

# unzip .zip file and get the exact .txt file.
with ZipFile(path + zip_name, 'r') as z:
    z.extractall(path)  # 解壓縮檔案
for file in os.listdir(path):
    if fnmatch.fnmatch(file, '*.txt'):
        file_name = file  # 把 .txt 的檔案命名為 file_name 讓我後續使用

2. 讀取檔案,進行資料處理

這段的任務很單純,就是把 file_name 讀進來後,將欄位以及值的空格去掉,用到的是處理資料最常用的 pandas 以及 re 來移除空白。

import pandas as pd
import re

pattern = re.compile(r'\s+')  # 把空白移除的 pattern

data = pd.read_csv(path + file_name, sep=',', encoding='UTF-16')
data = data.applymap(lambda x: pattern.sub('', x) if isinstance(x, str) else x)  # remove whitespaces in df if type of column is str
data.columns = data.columns.str.strip()  # remove whitespaces from two side of column names.

# db processing
data_dict = data.to_dict(orient='records')  # 把 pd.DF 轉成 dictionary ,可以加速 insert into DB 的速度
db_process(db, data_dict)

3. 資料庫操作

這邊介紹的是 db_process.py ,主要是透過 sqlalchemy 來操作。

from sqlalchemy import create_engine, Column, String, Table
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

engine = create_engine(get_db_string(db_str))
Base = declarative_base()
Base.metadata.bind = engine
Base.metadata.schema = 'my_schema'

# Creating the session and execute insert statement
session = sessionmaker(bind=engine)
session = session()
table = Table('sysclinc', Base.metadata, autoload=True)  # use autoload to reflect db and get table

session.execute('table.delete()')  # 先把舊資料清掉
session.commit()
session.execute(table.insert(), data_dict)  # 將新的資料塞入 DB
session.commit()
session.close()

4. 程式記錄檔

python 中最多人用的紀錄套件,非 logging 莫屬了,參考這篇文章後,設置了專屬於自己的 logger , logger 可以自己設定要記錄的等級,如下:

等級 數值
CRITICAL 50
ERROR 40
WARNING 30
INFO 20
DEBUG 10
NOTSET 0

完整程式碼如下:

import logging
import os
from datetime import datetime
from pytz import timezone

# 基本設定:時區、路徑、檔名
tz = timezone('Asia/Taipei')
dir_path = './log/'  # 設定 logs 目錄
filename = datetime.strftime(datetime.now().astimezone(tz), '%Y%m%d_%H%M%S') + '.log'  # 設定檔名

# 開始產生自己的 logger
def create_logger():
    logging.captureWarnings(True)  # 捕捉 py waring message
    formatter = logging.Formatter('[%(asctime)s] [%(levelname)s] [%(message)s]')  # 設定 logger 每次回傳的格式為 [時間] [level] [訊息]
    logging.Formatter.converter = lambda *args: datetime.now(tz=tz).timetuple()  # 將每次 logger 記錄下的時間資訊強制轉為 tz 的時區
    my_logger = logging.getLogger('my_logger')  # 將自己的 logger 命名為 my_logger ,讓程式的其他地方都可以使用同樣的 logger
    my_logger.setLevel(logging.INFO)  # 設定我的 logger 要記錄 INFO 以上的錯誤

    # 若不存在目錄則新建
    if not os.path.exists(dir_path):
        os.makedirs(dir_path)

    # file handler: 設定 logger 儲存的檔案格式
    fileHandler = logging.FileHandler(dir_path + filename, 'w', 'utf-8')
    fileHandler.setFormatter(formatter)
    my_logger.addHandler(fileHandler)

    # console handler: 設定 logger 將訊息丟到 console 的格式
    consoleHandler = logging.StreamHandler()
    consoleHandler.setLevel(logging.DEBUG)
    consoleHandler.setFormatter(formatter)
    my_logger.addHandler(consoleHandler)

    return my_logger

設置完成後,在其他 module 中,我只要設定 logger = logging.getLogger('my_logger') 後,就可以在不同的 module 中都使用同樣設定的 logger ,太棒了。

5. 自動化執行

這段落可再細分為命令列參數設定與 linux server 的自動化布署。

5.1 命令列參數設定

自動化執行會在一台 linux server 上進行,因為時區的關係,所以要強制指定時區

因為要透過 terminal 執行 python ,需要把整支程式設計為可下參數的執行模式,這樣才能在自動化作業中指定部分設定,也方便我布署(例如我可以透過 python main.py -db db 這條命令中調整 db 為 production/dev/test 等資料庫),這個可透過 argparse 來完成。

import argparse
def parse_arguments():
    parser = argparse.ArgumentParser()  # init ArgumentParser
    parser.add_argument('-db', '--database', type=str, choices=['p', 'u', 't'],
                        help='指定要操作的資料庫,p: production, u: uat, t: test')  # --database 為全名,外部要取用時必須透過此全名才可訪問
    args = parser.parse_args()
    return args

設定好 argument 後,就可以在執行的時候把值取出來,並透過 if else 來控制參數與程式行為:

if __name__ == '__main__':
    args = parse_arguments()
    if args.database not in ('p', 'd', 't'):
        raise Exception('資料庫參數 (-db) 必須指定為下列其一: p, d, t')
    else:
        print(f"參數正確,將使用下列參數執行: args.db: {args.database}")
    logger = create_logger()
    logger.info('開始執行')
    try:
        main(args.database)
    except Exception as e:
        logger.exception('錯誤訊息: ')

這樣設定好之後,日後就可以在命令列中執行此程式:

python main.py -db p  # 執行程式並操作 production 資料庫
python main.py -db d  # 執行程式並操作 dev 資料庫
python main.py -db t  # 執行程式並操作 test 資料庫

linux server 的自動化布署

linux 的排程工作是透過 crontab 來執行的,我們可以透過 crontab -h 來查看指令如何下:

usage:  crontab [-u user] file
        crontab [ -u user ] [ -i ] { -e | -l | -r }
        -e      (edit user's crontab)
        -l      (list user's crontab)
        -r      (delete user's crontab)
        -i      (prompt before deleting user's crontab)

要新增一個例行性的排程工作,就從 crontab -e 開始吧 (輸入之後會進入編輯器,每一行皆為一個例行性工作,可用 # 為工作下註解)

crontab -e

# 時間的輸入範例
# ┌───────────── 分鐘   (0 - 59)
# │ ┌─────────── 小時   (0 - 23)
# │ │ ┌───────── 日     (1 - 31)
# │ │ │ ┌─────── 月     (1 - 12)
# │ │ │ │ ┌───── 星期幾 (0 - 7,0 是週日,6 是週六,7 也是週日)
# │ │ │ │ │
# * * * * * /path/to/command

# 特殊的時間輸入範例
# @reboot       每次重新開機之後,執行一次。
# @yearly       每年執行一次,亦即 0 0 1 1 *。
# @annually     每年執行一次,亦即 0 0 1 1 *。
# @monthly      每月執行一次,亦即 0 0 1 * *。
# @weekly       每週執行一次,亦即 0 0 * * 0。
# @daily        每天執行一次,亦即 0 0 * * *。
# @hourly       每小時執行一次,亦即 0 * * * *。

MAILTO=my_email@my_hostname # 當我把工作執行完畢後,寄信到我的信箱中
50 15 * * * bash /home/project_path/crontab.sh  # 這行的意思是:在每天機器時間 15:50 的時候,以 bash 執行我路徑底下的 crontab.sh

為了讓 linux 變成可以發信的 mail server ,還需要安裝套件,讓 linux 變成只寄不收信的 mail server,詳細的設定請參考這篇文章,而安裝 mali server 的指令如下:

sudo apt-get install mailutils

至於 crontab.sh 裡面的內容,就是我把我要 server 執行的命令集,每一行就代表一個命令。內容就是移動到專案目錄底下 -> 啟動虛擬環境 -> 在虛擬環境下執行 python -> 離開虛擬環境 -> 完成工作

# crontab.sh
cd /home/project_path  # 移動到專案目錄
source /home/project_path/venv/bin/activate  # 啟動虛擬環境
python main.py -db d  # 透過命令列執行程式,並下參數指定要操作的資料庫
deactivate  # 離開虛擬環境
echo 0  # 回傳 0(可省略)

結語

以上,記錄了如何在 linux server 中,讓 server 把每日執行(crontab)的工作(main.py arguments)結果寄給自己(mail to)的工作流程,算是自己完成了第一個全自動的任務,不過還有幾點尚待改進:

  1. 判斷是否要更新目前是根據網站的日期與 server 執行程式當下的日期相比,若網站日期 >= server 日期則啟動程式更新。

    這樣的比法沒辦法每次執行後就完成更新,若A次完成更新但A+1次時出錯了未更新,則要待A+2次時網站有更新後我的資料庫才會更新。比較好的作法應該是在資料庫的表中新增一欄 "最後更新日期" ,這樣我每天執行時就拿網路上的日期跟我資料庫的最後更新日期比較,很簡便的就可以判斷出我是否需要更新資料庫了。

  2. 程式碼還不夠簡潔明瞭

    在 main() 裡面,應把 web_parser 獨立出來(web_parser.py),把 unzip 的功能與 data processing 合再一起且獨立出來為(data_processor.py)

About the Author

發佈留言

發佈留言必須填寫的電子郵件地址不會公開。 必填欄位標示為 *

Related Posts