2026年8月31日 星期一

Python一下:用jieba整理水井村地方故事與關鍵詞

《Python一下:從風土資料到智慧生活》第 32 篇

用jieba整理水井村地方故事與關鍵詞

讓「水井三寶、姻緣花、柴井車、智慧養殖」不再被切散,也不讓演算法取代社區自己的說法。

CH14 斷詞處理jieba地方故事
學習目標
完成本篇後,你能解釋中文斷詞的必要性,使用jieba精確模式處理繁體中文,建立水井村自訂詞典與停用詞,統計詞頻、搜尋上下文、擷取TF-IDF關鍵詞,建立可人工審查的結果檔,並說明口述故事的同意、去識別、台語轉寫及地方詮釋限制。

一、電腦為何不知道「姻緣花」是一個詞?

英文常以空白分隔單字,中文句子通常沒有天然空白。「水井三寶包含烏龜、白馬與姻緣花」若逐字處理,就無法直接知道「水井三寶」和「姻緣花」是地方概念。斷詞的工作,就是提出一種詞語邊界。

原文可能結果問題
水井三寶水井/三寶地方品牌詞被拆開
姻緣花姻緣/花地方文化名稱失去完整性
魚菜共生淨水魚菜/共生/淨水可能需要保留「魚菜共生」
HUB 8735 Ultra英數與空白混合需另外設計技術詞正規化
斷詞沒有唯一答案:適合搜尋的切法、適合關鍵詞統計的切法,以及社區認同的地方詞,可能不同。結果必須依任務評估,不能把工具輸出當成地方語言的標準答案。

二、安裝jieba並記錄維護風險

import subprocess
import sys


subprocess.run(
    [sys.executable, "-m", "pip", "install", "jieba"],
    check=True,
)

截至本篇核對時,PyPI顯示的最新版仍是2020年發布的0.42.1。它適合教學與既有專案評估,但正式採用前應重新檢查維護狀態、Python相容性、授權、安全與替代工具,不能因為安裝成功就直接長期部署。

三、第一個繁體中文斷詞

import jieba


text = "水井三寶包含烏龜、白馬與姻緣花。"
tokens = list(jieba.cut(text, cut_all=False, HMM=True))

print(tokens)
print("/".join(tokens))

cut_all=False是精確模式,較適合一般文字分析;HMM可嘗試辨識未登錄詞,但不保證地方詞一定正確。

四、三種模式用途不同

sentence = "水井村推動智慧養殖與魚菜共生淨水。"

precise = list(jieba.cut(sentence, cut_all=False))
full = list(jieba.cut(sentence, cut_all=True))
search = list(jieba.cut_for_search(sentence))

print("精確模式:", precise)
print("全模式:", full)
print("搜尋模式:", search)
模式適合限制
精確模式詞頻、一般分析仍可能切錯地方詞
全模式觀察可能詞語重疊多,不能直接計算一般詞頻
搜尋引擎模式建立搜尋索引長詞會再次切分,詞數不可和精確模式混比

五、使用獨立Tokenizer,避免全域設定互相干擾

import jieba


tokenizer = jieba.Tokenizer()
for word in [
    "水井三寶",
    "姻緣花",
    "柴井車",
    "魚菜共生",
    "智慧養殖",
    "工作蝦",
]:
    tokenizer.add_word(word, freq=100000)

text = "水井三寶與柴井車承載地方記憶。"
print(list(tokenizer.cut(text, cut_all=False)))

在多人專案或測試中,獨立Tokenizer比修改全域 jieba.dt更容易控制。詞頻100000只是確保教學示例優先成詞,不代表真實語料頻率。

六、繁體中文詞典與地方詞典分工

jieba專案提供較大的 extra_dict/dict.txt.big,可作為繁體中文候選詞典;實際專案應把經審核的詞典檔納入版本管理,記錄來源、授權與雜湊,不在程式執行時臨時從不明網址下載。

from pathlib import Path
import jieba


def build_tokenizer(base_dictionary=None, user_dictionary=None):
    tokenizer = jieba.Tokenizer(
        dictionary=str(base_dictionary) if base_dictionary else None
    )
    if user_dictionary:
        path = Path(user_dictionary)
        if not path.is_file():
            raise FileNotFoundError(path)
        tokenizer.load_userdict(str(path))
    return tokenizer


# 專案備妥詞典後再啟用:
# tokenizer = build_tokenizer(
#     "data/dict.txt.big", "data/shuijing_words.txt"
# )

七、自訂詞典格式

自訂詞典每行為「詞語 詞頻 詞性」,詞頻與詞性可省略,檔案使用UTF-8。建議由社區與課程團隊共同審查:

水井三寶 100000 nz
姻緣花 100000 nz
柴井車 100000 nz
工作蝦 100000 nz
魚菜共生 100000 nz
智慧養殖 100000 nz
風頭水尾 100000 nz

nz只是常見自訂專名標記,詞性標註並非本篇主要目標。地方詞典還應另有說明表,記錄詞義、別稱、使用者、例句、審核者與版本日期。

八、先做最小文字清理

import re
import unicodedata


def normalize_text(text):
    text = unicodedata.normalize("NFKC", str(text))
    text = text.replace("\u3000", " ")
    text = re.sub(r"[ \t]+", " ", text)
    text = re.sub(r"\n{3,}", "\n\n", text)
    return text.strip()


sample = "水井村 推動  智慧養殖。"
print(normalize_text(sample))

NFKC會統一部分全形與相容字元,但也可能改變原始書寫形式。口述史原稿應另外保存;分析使用的是衍生副本,並記錄清理規則。

九、停用詞:刪掉常見詞也可能刪掉地方語氣

STOPWORDS = {
    "的", "了", "在", "和", "與", "是",
    "有", "也", "就", "我們", "一個",
}


def meaningful_tokens(text, tokenizer, stopwords=STOPWORDS):
    for token in tokenizer.cut(normalize_text(text)):
        token = token.strip()
        if not token:
            continue
        if token in stopwords:
            continue
        if not re.search(r"[\u3400-\u9fffA-Za-z0-9]", token):
            continue
        yield token


story = "我們在水井村用智慧養殖守護工作蝦。"
print(list(meaningful_tokens(story, tokenizer)))

停用詞必須依任務制定。長者口述中的「阮、咱、彼時、做伙」可能承載語氣、群體關係與台語特色,不應直接套用一般中文停用詞刪除。

十、用Counter統計地方詞頻

from collections import Counter


stories = [
    "水井村推動智慧養殖,也用魚菜共生淨水。",
    "水井三寶包含烏龜、白馬與姻緣花。",
    "姻緣花承載水井村的地方記憶與祝福。",
]

counter = Counter()
for story in stories:
    counter.update(meaningful_tokens(story, tokenizer))

for word, count in counter.most_common(10):
    print(word, count)

詞頻高表示在這批語料出現多,不等於對社區最重要。一次只被長者說出的罕見詞,可能反而是珍貴文化線索。

十一、同一詞在幾篇故事出現?

document_frequency = Counter()

for story in stories:
    unique_words = set(
        meaningful_tokens(story, tokenizer)
    )
    document_frequency.update(unique_words)

for word, documents in document_frequency.most_common(10):
    print(word, "出現在", documents, "篇")

詞頻計算總出現次數;文件頻率計算出現於幾篇文件。兩者回答不同問題,不能混稱「熱門程度」。

十二、顯示關鍵詞所在原句

def find_contexts(stories, keyword):
    results = []
    for index, story in enumerate(stories, start=1):
        sentences = re.split(r"(?<=[。!?])", story)
        for sentence in sentences:
            sentence = sentence.strip()
            if keyword in sentence:
                results.append({
                    "story": index,
                    "sentence": sentence,
                })
    return results


for item in find_contexts(stories, "姻緣花"):
    print(item)

詞頻表應能回到原句,才能讓社區夥伴確認用法與意義。公開結果若含可識別人物或私密經驗,仍需授權與去識別。

十三、TF-IDF擷取關鍵詞

import jieba.analyse


corpus_text = "\n".join(stories)
keywords = jieba.analyse.extract_tags(
    corpus_text,
    topK=8,
    withWeight=True,
)

for word, weight in keywords:
    print(word, round(weight, 4))

TF-IDF權重不是機率,也不是社區重要性分數。jieba預設IDF來自其內建語料,不一定代表臺灣農漁村或水井村;地方分析可建立經審查的領域語料與IDF,但需要足夠文件及清楚方法。

十四、讓TF-IDF使用自訂Tokenizer的地方詞

jieba.analyse預設使用全域斷詞器,可能和前面的獨立Tokenizer設定不同。簡單教材可在分析前把地方詞加入全域詞典,但要在單一初始化函式集中管理:

DOMAIN_WORDS = [
    "水井三寶", "姻緣花", "柴井車",
    "魚菜共生", "智慧養殖", "工作蝦",
]


def configure_global_jieba():
    for word in DOMAIN_WORDS:
        jieba.add_word(word, freq=100000, tag="nz")


configure_global_jieba()
keywords = jieba.analyse.extract_tags(
    "\n".join(stories), topK=8, withWeight=True
)
print(keywords)

測試要先重新初始化或固定執行順序,避免全域狀態讓測試互相污染。大型系統可評估更易注入Tokenizer的分析設計。

十五、比較詞頻與TF-IDF,不急著選「唯一答案」

frequency_top = [
    word for word, count in counter.most_common(8)
]
tfidf_top = [word for word, weight in keywords]

comparison = {
    "both": sorted(set(frequency_top) & set(tfidf_top)),
    "frequency_only": [
        word for word in frequency_top if word not in tfidf_top
    ],
    "tfidf_only": [
        word for word in tfidf_top if word not in frequency_top
    ],
}

print(comparison)

共同出現的詞可優先檢視;只在一邊出現的詞也不能直接丟掉。最終主題命名應回到原句、訪談脈絡與社區共同討論。

十六、為地方詞典建立回歸測試

CASES = {
    "水井三寶包含姻緣花": {"水井三寶", "姻緣花"},
    "魚菜共生協助淨水": {"魚菜共生"},
    "智慧養殖記錄工作蝦": {"智慧養殖", "工作蝦"},
}


def test_domain_dictionary(tokenizer):
    failures = []
    for sentence, expected in CASES.items():
        actual = set(tokenizer.cut(sentence, cut_all=False))
        missing = expected - actual
        if missing:
            failures.append({
                "sentence": sentence,
                "missing": sorted(missing),
                "actual": sorted(actual),
            })
    return failures


failures = test_domain_dictionary(tokenizer)
assert not failures, failures
print("地方詞典測試通過")

每次更新jieba、基礎詞典或地方詞典都執行測試,避免原本正確的地方詞突然被拆開。

十七、輸出可審查的JSON,而非只有詞雲

import json
from datetime import datetime, timezone


report = {
    "created_at": datetime.now(timezone.utc).isoformat(),
    "method": "jieba精確模式+自訂地方詞+詞頻與TF-IDF",
    "documents": len(stories),
    "top_frequency": counter.most_common(10),
    "top_tfidf": [
        {"word": word, "weight": weight}
        for word, weight in keywords
    ],
    "review_status": "待社區人工審查",
}

print(json.dumps(report, ensure_ascii=False, indent=2))

詞雲視覺吸引人,但會省略方法、分母與上下文。研究或USR成果應保留版本、語料範圍、清理規則、詞典、停用詞、程式與人工審查紀錄。

十八、台語口述故事要多一層語言尊重

階段應注意
語音轉文字台語辨識可能誤轉人名、地名、農產與料理
轉寫選擇漢字、台羅或混合書寫需保留原則與原始錄音
jieba斷詞主要針對中文文字,不等於台語語言分析工具
修訂由說話者或熟悉地方語言的人確認,不擅自「改成標準中文」
公開確認用途、範圍、署名/匿名、撤回及保存期限

十九、個資去識別不能只靠正規表示式

import re


def flag_possible_sensitive_text(text):
    patterns = {
        "phone": r"(?<!\d)09\d{8}(?!\d)",
        "email": r"[\w.+-]+@[\w.-]+\.[A-Za-z]{2,}",
        "address_hint": r"(?:路|街|巷|弄|號)",
    }
    return {
        name: bool(re.search(pattern, text))
        for name, pattern in patterns.items()
    }


print(flag_possible_sensitive_text(
    "這是一段不含聯絡方式的示範故事。"
))

這只能標記部分可能風險,無法辨識暱稱、親屬關係、罕見事件或組合資訊。去識別必須人工審查;即使移除姓名,故事仍可能讓社區成員辨識當事人。

二十、課堂挑戰

挑戰A|基礎:以5段匿名地方故事建立詞頻表,列出有效詞數、文件頻率及每個前10詞的原句。
挑戰B|進階:建立20個水井村地方詞的詞典與至少10個回歸測試案例,記錄新增前後斷詞差異。
挑戰C|USR場域:請長者或社區夥伴檢視演算法選出的關鍵詞,標記「認同、需改名、不宜公開、演算法漏掉」,比較機器結果與地方觀點。

二十一、用AI協助整理,但不能代替受訪者確認

請擔任地方故事斷詞與關鍵詞分析助教。
語料來自已同意教學分析並完成去識別的社區故事。
請先檢查繁體中文、台語轉寫、地方專名、同義詞、停用詞與個資風險,
再協助比較詞頻、文件頻率與TF-IDF結果。
不得把演算法關鍵詞稱為「社區最重要的文化」,
不得擅自改寫長者原意,也不要要求提供真實姓名或未授權逐字稿。
請輸出待人工確認清單、原句索引與地方詞典修訂建議。

二十二、延伸閱讀

Python一下:用NumPy讀懂智慧養殖感測資料的數學基礎

《Python一下:從風土資料到智慧生活》第 31 篇

用NumPy讀懂智慧養殖感測資料的數學基礎

把一串水溫、pH與溶氧讀值,轉成可計算、可比較、也知道限制的資料判讀。

CH14 數學相關NumPy智慧養殖
學習目標
完成本篇後,你能建立與檢查NumPy陣列、使用索引與布林遮罩、理解向量化與broadcasting、沿正確axis計算統計量、處理NaN、計算移動平均與變化率、建立異常候選遮罩,並對相關係數與標準化結果作負責任解讀。

一、為什麼智慧養殖需要NumPy?

Python串列可以保存數值,但NumPy的 ndarray 能以一致資料型別組成一維或多維陣列,並進行向量化運算。它是pandas、SciPy、影像處理與許多機器學習工具的數值基礎。

資料問題NumPy工具仍需人工確認
每小時水溫的平均與波動mean、std、min、max探頭是否校正、取樣是否完整
設備離線造成缺值NaN與nan系列函式缺值原因與補傳紀錄
短期趨勢太雜亂移動平均、diff平滑是否掩蓋突發事件
某筆數值特別不同IQR、z-score候選設備錯誤還是真實場域變化
本篇資料為教學示範:數值與門檻不代表特定魚種、蝦種、池型或季節的正式養殖標準。任何巡查或控制策略,都應由場域養殖者與專業人員共同確認。

二、安裝與確認NumPy

import subprocess
import sys


subprocess.run(
    [sys.executable, "-m", "pip", "install", "NumPy"],
    check=True,
)

請在虛擬環境安裝,並記錄課程驗證版本。程式匯入時慣例使用 import numpy as np

三、第一個水溫陣列

import numpy as np


temperature = np.array(
    [27.2, 27.5, 28.0, 28.4, 29.1, 29.4],
    dtype=np.float64,
)

print("資料:", temperature)
print("形狀:", temperature.shape)
print("維度:", temperature.ndim)
print("筆數:", temperature.size)
print("型別:", temperature.dtype)

shape=(6,)代表包含6個元素的一維陣列。感測資料通常使用浮點型別;若需要缺值NaN,也必須使用可表示NaN的型別。

四、索引與切片:查看特定時段

print("第一筆:", temperature[0])
print("最後一筆:", temperature[-1])
print("第2到第4筆:", temperature[1:4])

morning = temperature[:3].copy()
morning[0] = 99.0

print("複製後修改:", morning)
print("原始資料不變:", temperature)

基本切片常得到原陣列的view,修改時可能影響原資料。當你需要獨立資料時明確使用 .copy()

五、向量化:一次換算全部感測值

temperature_f = temperature * 9 / 5 + 32
change_from_first = temperature - temperature[0]

print("華氏:", np.round(temperature_f, 1))
print("相對第一筆變化:", change_from_first)

不必逐筆寫for迴圈,算術運算會套用到每個元素。這種向量化表達通常更簡潔,也能利用NumPy底層實作。

六、Broadcasting:每個感測器套用不同校正值

raw = np.array([
    [27.2, 7.3, 5.8],
    [27.5, 7.4, 5.6],
    [28.0, 7.2, 5.4],
    [28.4, 7.1, 5.2],
])

# 三欄依序是水溫、pH、溶氧
offset = np.array([0.2, -0.1, 0.0])
calibrated = raw + offset

print(calibrated)

形狀為 (4,3) 的資料可和形狀 (3,) 的校正值broadcasting:每一列都套用三個欄位校正。若形狀無法相容,NumPy會報錯;不能靠猜測哪個方向被套用。

七、理解axis:沿時間還是沿指標?

column_mean = calibrated.mean(axis=0)
row_mean = calibrated.mean(axis=1)

print("各指標跨時間平均:", column_mean)
print("每個時間點跨欄平均:", row_mean)

axis=0壓縮列,得到每一欄跨時間的平均;axis=1壓縮欄。但水溫、pH、溶氧單位不同,直接算跨欄平均通常沒有合理物理意義。程式算得出來,不代表應該這樣算。

八、基本統計:中心與離散程度

summary = {
    "count": temperature.size,
    "mean": float(np.mean(temperature)),
    "median": float(np.median(temperature)),
    "minimum": float(np.min(temperature)),
    "maximum": float(np.max(temperature)),
    "std_population": float(np.std(temperature, ddof=0)),
    "std_sample": float(np.std(temperature, ddof=1)),
}

for name, value in summary.items():
    print(name, value)

ddof=0計算這批資料本身的母體標準差;ddof=1常用於由樣本估計母體變異。選哪一個取決於問題,不是固定答案。

九、NaN:未知不等於0

temperature_with_missing = np.array(
    [27.2, 27.5, np.nan, 28.4, 29.1, 29.4],
    dtype=float,
)

missing_mask = np.isnan(temperature_with_missing)
valid_count = np.count_nonzero(~missing_mask)

print("缺值位置:", missing_mask)
print("有效筆數:", valid_count)
print("忽略NaN平均:", np.nanmean(temperature_with_missing))
print("忽略NaN標準差:", np.nanstd(
    temperature_with_missing, ddof=1
))

np.mean()遇到NaN通常會得到NaN;np.nanmean()忽略NaN,但不能假裝缺值不存在。報告平均時要同時呈現有效筆數、缺值率與時間範圍。

十、資料值與品質標記分開

values = np.array([27.2, 27.5, 99.0, np.nan, 28.4])
quality = np.array([
    "valid", "valid", "invalid", "unknown", "valid"
])

usable_mask = (quality == "valid") & np.isfinite(values)
usable_values = values[usable_mask]

print("可用遮罩:", usable_mask)
print("可納入一般統計:", usable_values)
print("平均:", usable_values.mean())

invalid保留原始數值供追查,但不納入一般統計;unknown通常搭配NaN。只靠NaN無法表達「為何不可用」,品質標記仍要保留。

十一、布林遮罩:找出教材門檻外讀值

low, high = 26.0, 30.0
valid = np.isfinite(temperature_with_missing)
outside = valid & (
    (temperature_with_missing < low)
    | (temperature_with_missing > high)
)

print("門檻外位置:", np.flatnonzero(outside))
print("門檻外數值:", temperature_with_missing[outside])

26~30°C只是程式示範。門檻必須附上物種、成長階段、場域、季節、資料來源、核定者與版本日期,不能把網路上找到的一個數字直接寫進控制系統。

十二、差分:觀察相鄰時間的變化

change = np.diff(temperature)
rapid_change = np.abs(change) >= 0.6

print("相鄰變化:", change)
print("快速變化區段:", np.flatnonzero(rapid_change))

np.diff()的結果比原陣列少一筆,第0個差分代表原始第0筆到第1筆。若取樣間隔不固定,還要除以實際時間差,不能直接比較。

十三、移動平均:平滑短期波動

from numpy.lib.stride_tricks import sliding_window_view


def moving_average(values, window):
    values = np.asarray(values, dtype=float)
    if window < 1 or window > values.size:
        raise ValueError("window超出資料範圍")
    windows = sliding_window_view(values, window_shape=window)
    return np.nanmean(windows, axis=-1)


smoothed = moving_average(temperature_with_missing, 3)
print(smoothed)

3筆移動平均輸出會少2筆,時間應對齊視窗尾端或中心並清楚說明。平滑可以看趨勢,也可能遮住短暫但重要的急變;原始資料不可被覆蓋。

十四、IQR:找統計上的異常候選

candidate_data = np.array(
    [27.1, 27.2, 27.3, 27.4, 27.5, 31.8],
    dtype=float,
)

q1, q3 = np.quantile(candidate_data, [0.25, 0.75])
iqr = q3 - q1
lower = q1 - 1.5 * iqr
upper = q3 + 1.5 * iqr
candidate_mask = (
    (candidate_data < lower) | (candidate_data > upper)
)

print("IQR範圍:", lower, upper)
print("異常候選:", candidate_data[candidate_mask])

IQR規則只是探索方法。資料有日夜週期、季節變化、餵食或換水事件時,整批混算可能把正常情境標成異常。

十五、z-score:先檢查樣本量與標準差

def z_scores(values):
    values = np.asarray(values, dtype=float)
    valid = values[np.isfinite(values)]
    if valid.size < 3:
        raise ValueError("有效資料不足")
    mean = valid.mean()
    std = valid.std(ddof=1)
    if np.isclose(std, 0.0):
        raise ValueError("標準差接近0,無法計算z-score")
    return (values - mean) / std


scores = z_scores(candidate_data)
print(np.round(scores, 2))

z-score適合近似穩定分布的探索,不應機械式把 |z|>3當成通用場域告警。樣本小、分布偏斜或有時間趨勢時尤其要小心。

十六、標準化:不同量尺可比較,但單位意義消失

sensor_matrix = np.array([
    [27.2, 7.3, 5.8],
    [27.5, 7.4, 5.6],
    [28.0, 7.2, 5.4],
    [28.4, 7.1, 5.2],
])

mean = sensor_matrix.mean(axis=0, keepdims=True)
std = sensor_matrix.std(axis=0, ddof=1, keepdims=True)
if np.any(np.isclose(std, 0.0)):
    raise ValueError("至少一欄沒有變化,無法標準化")

standardized = (sensor_matrix - mean) / std
print(np.round(standardized, 2))

keepdims=True保留形狀為 (1,3),可安全broadcast回原矩陣。標準化結果沒有°C、pH或mg/L的原單位,不能拿來直接對照養殖門檻。

十七、相關係數:相關不代表因果

temperature_pair = np.array(
    [27.2, 27.5, 28.0, 28.4, 29.1, np.nan]
)
dissolved_oxygen = np.array(
    [5.8, 5.6, 5.4, 5.2, 4.9, 5.0]
)

pair_mask = (
    np.isfinite(temperature_pair)
    & np.isfinite(dissolved_oxygen)
)
if np.count_nonzero(pair_mask) < 3:
    raise ValueError("成對有效資料不足")

correlation = np.corrcoef(
    temperature_pair[pair_mask],
    dissolved_oxygen[pair_mask],
)[0, 1]
print("Pearson相關係數:", correlation)

高度正相關或負相關都不證明因果。共同時間趨勢、日夜循環、曝氣、餵食與感測器位置都可能影響結果;要回到實驗設計及場域紀錄。

十八、綜合函式:產生可追溯摘要

def sensor_summary(values, quality):
    values = np.asarray(values, dtype=float)
    quality = np.asarray(quality)
    if values.shape != quality.shape:
        raise ValueError("values與quality形狀不同")

    usable = (quality == "valid") & np.isfinite(values)
    selected = values[usable]
    if selected.size == 0:
        return {
            "total": int(values.size),
            "valid": 0,
            "missing_or_invalid": int(values.size),
            "mean": None,
            "minimum": None,
            "maximum": None,
        }

    return {
        "total": int(values.size),
        "valid": int(selected.size),
        "missing_or_invalid": int(values.size - selected.size),
        "mean": float(selected.mean()),
        "minimum": float(selected.min()),
        "maximum": float(selected.max()),
    }


print(sensor_summary(values, quality))

轉成Python的int與float,較方便輸出JSON。摘要仍應附感測器代碼、單位、起訖時間、取樣間隔、品質規則版本及產生時間。

十九、不要犯的五個錯誤

錯誤後果改善
把缺值補成0平均與告警失真保留NaN及quality
混合不同單位直接平均得到沒有物理意義的數字依欄位與單位分別統計
忘記axis方向跨時間與跨指標算反每步印shape並用小資料驗證
平滑後覆蓋原始值失去稽核與急變資訊原始值與衍生值分開保存
把統計候選當成場域異常錯誤巡查或控制結合校正、事件及專業判讀

二十、課堂挑戰

挑戰A|基礎:建立24筆每小時水溫資料,計算平均、中位數、最大值、最小值、有效筆數與缺值率。
挑戰B|進階:比較3筆與5筆移動平均,標示輸出對應的時間點,說明哪一種保留急變較多。
挑戰C|USR場域:與養殖者共同挑選一次真實事件,將感測值、品質標記、設備校正、天候與人工巡查紀錄排在同一時間軸,解釋統計方法在哪裡可能誤判。

二十一、用AI協助檢查數學,而不是代替場域判斷

請擔任NumPy與智慧養殖資料分析助教。
資料包含水溫、pH、溶氧、時間與quality標記。
請先確認每個陣列的shape、dtype、axis、單位、缺值與取樣間隔,
再檢查平均、標準差、移動平均、IQR與相關係數程式。
不得把NaN自動補成0,也不得把統計異常直接稱為養殖異常。
請為每個結果列出:數學意義、資料前提、場域限制、
最小測試資料,以及需要養殖者確認的問題。

二十二、延伸閱讀

Python一下:用Pillow打造水井三寶地方圖卡處理工具

《Python一下:從風土資料到智慧生活》第 30 篇

用Pillow打造水井三寶地方圖卡處理工具

讓烏龜、白馬與姻緣花的照片,變成尺寸一致、文字清楚、可追溯來源的地方圖卡。

CH14 圖片處理Pillow水井三寶
學習目標
完成本篇後,你能安全開啟與驗證圖片、依EXIF修正方向、比較thumbnail與fit、加入繁體中文標題及半透明浮水印、正確輸出JPEG/PNG、批次產生圖卡與處理紀錄,並說明肖像權、著作權、位置資訊及AI生成標示。

一、地方圖卡不是「把字貼上去」而已

同一張水井三寶照片可能來自手機、相機、社區長者或AI創作,具有不同方向、尺寸、色彩模式、檔案格式及使用權。圖卡工具至少要回答四件事:

問題程式處理USR責任
圖片能否安全讀取?限制大小、辨識格式、捕捉錯誤不接受不明來源大檔直接進系統
版型是否一致?轉向、縮放、裁切與色彩模式不扭曲人物、工藝或地方物件
文字是否清楚?字型、文字框、對比與安全邊界名稱與故事先由社區確認
能否合法公開?輸出時移除不必要metadata確認授權、肖像同意及AI標示

二、安裝Pillow:安裝名稱與匯入名稱不同

import subprocess
import sys


subprocess.run(
    [sys.executable, "-m", "pip", "install", "Pillow"],
    check=True,
)

pip安裝名稱是Pillow,程式匯入名稱仍是PIL。請在專案虛擬環境安裝,並把實際驗證版本記錄在依賴檔中。

三、讀取圖片基本資訊

from pathlib import Path
from PIL import Image


source = Path("images/turtle.jpg")

with Image.open(source) as image:
    print("格式:", image.format)
    print("尺寸:", image.size)
    print("色彩模式:", image.mode)
    print("影格數:", getattr(image, "n_frames", 1))

Image.open()會延遲讀取像素,因此要使用with確保檔案關閉。不要只依副檔名判定內容;Pillow會由檔案內容辨識格式。

四、不可信圖片先驗證,再重新開啟

import warnings
from pathlib import Path
from PIL import Image, UnidentifiedImageError


MAX_FILE_BYTES = 15 * 1024 * 1024
ALLOWED_FORMATS = {"JPEG", "PNG", "WEBP"}


def validate_image(path):
    path = Path(path)
    if not path.is_file():
        raise FileNotFoundError(path)
    if path.stat().st_size > MAX_FILE_BYTES:
        raise ValueError("檔案超過15 MB限制")

    with warnings.catch_warnings():
        warnings.simplefilter("error", Image.DecompressionBombWarning)
        try:
            with Image.open(path) as image:
                detected_format = image.format
                if detected_format not in ALLOWED_FORMATS:
                    raise ValueError("不支援的圖片格式")
                image.verify()
        except UnidentifiedImageError as error:
            raise ValueError("無法辨識圖片") from error

    return detected_format

verify()檢查後必須重新開啟圖片,不能繼續用同一物件處理像素。檔案大小與像素量是兩種不同風險;Pillow會對異常巨大像素數提出decompression bomb警告或錯誤,不應為方便直接關閉保護。

五、手機照片方向錯誤:套用EXIF Orientation

from PIL import Image, ImageOps


def open_oriented_rgb(path):
    validate_image(path)
    with Image.open(path) as image:
        oriented = ImageOps.exif_transpose(image)
        return oriented.convert("RGB")


image = open_oriented_rgb("images/turtle.jpg")
print(image.size, image.mode)

ImageOps.exif_transpose()依EXIF方向旋轉或鏡射像素,並處理方向資訊。若略過這一步,手機上直立的照片在輸出後可能橫躺。

六、thumbnail:完整保留、放進指定範圍

from pathlib import Path
from PIL import Image


image = open_oriented_rgb("images/turtle.jpg")
thumbnail = image.copy()
thumbnail.thumbnail(
    (800, 800),
    resample=Image.Resampling.LANCZOS,
)
Path("output").mkdir(parents=True, exist_ok=True)
thumbnail.save("output/turtle_thumbnail.jpg", quality=88)

print(thumbnail.size)

thumbnail()會原地修改圖片,保持長寬比並限制在指定框內,通常不會放大小圖。它適合完整保留照片,但輸出寬高不一定完全一致。

七、ImageOps.fit:裁成一致的社群圖卡

from PIL import Image, ImageOps


image = open_oriented_rgb("images/white_horse.jpg")
card_photo = ImageOps.fit(
    image,
    (1200, 800),
    method=Image.Resampling.LANCZOS,
    centering=(0.5, 0.45),
)
card_photo.save("output/white_horse_1200x800.jpg", quality=90)

print(card_photo.size)

fit()會等比例縮放後裁切,得到固定尺寸。centering可以調整保留重心;人物、樹木或工藝品不能一律置中,最好提供裁切預覽讓社區夥伴確認。

八、不要用resize硬拉成固定尺寸

from PIL import ImageOps, Image


def make_card_background(image, size=(1200, 1200), mode="crop"):
    if mode == "crop":
        return ImageOps.fit(
            image, size, method=Image.Resampling.LANCZOS
        )
    if mode == "contain":
        return ImageOps.pad(
            image,
            size,
            method=Image.Resampling.LANCZOS,
            color=(245, 240, 224),
        )
    raise ValueError("mode必須是crop或contain")

crop填滿版面但會裁掉邊緣;contain完整保留但可能出現留白。直接 resize((1200,1200))會改變比例,使姻緣花、烏龜或人物變形。

九、繁體中文字型要由專案明確提供

不同電腦的中文字型位置不同。最穩定做法是在取得合法再散布授權後,將指定字型作為專案資產,或由部署環境透過設定提供路徑。

import os
from pathlib import Path
from PIL import ImageFont


def load_font(size):
    font_path = Path(os.environ["SHUIJING_FONT_PATH"])
    if not font_path.is_file():
        raise FileNotFoundError("找不到指定中文字型")
    return ImageFont.truetype(str(font_path), size=size)


title_font = load_font(64)

不要假設Arial、微軟正黑體或Noto一定存在;也不要未確認授權就把系統字型複製到公開專案。

十、使用textbbox量字,再畫文字底板

from PIL import ImageDraw


def draw_title(image, title, font, margin=48):
    draw = ImageDraw.Draw(image, "RGBA")
    box = draw.textbbox((0, 0), title, font=font)
    text_width = box[2] - box[0]
    text_height = box[3] - box[1]

    x = margin
    y = image.height - text_height - margin * 2
    panel = (
        x - 20,
        y - 16,
        min(x + text_width + 20, image.width - margin),
        y + text_height + 20,
    )
    draw.rounded_rectangle(
        panel, radius=18, fill=(13, 66, 55, 205)
    )
    draw.text(
        (x, y), title, font=font,
        fill=(255, 255, 255, 255),
        stroke_width=1,
        stroke_fill=(0, 0, 0, 180),
    )


card = make_card_background(
    open_oriented_rgb("images/marriage_flower.jpg")
)
draw_title(card, "水井三寶|姻緣花", load_font(64))

textbbox()可先取得文字範圍,再決定底板大小。長標題仍可能超出畫面,下一節加入自動縮小。

十一、讓長標題自動縮小

from PIL import ImageDraw


def fit_font(draw, text, max_width, start=72, minimum=28):
    for size in range(start, minimum - 1, -2):
        font = load_font(size)
        left, top, right, bottom = draw.textbbox(
            (0, 0), text, font=font
        )
        if right - left <= max_width:
            return font
    raise ValueError("標題太長,請編輯文字或改用多行版型")

自動縮小仍應設最低可讀字級。若再縮就看不清楚,應改寫標題或設計多行文字,而不是讓字擠成一條。

十二、加入半透明來源標示

from PIL import Image, ImageDraw


def add_credit(image, text, font):
    base = image.convert("RGBA")
    layer = Image.new("RGBA", base.size, (0, 0, 0, 0))
    draw = ImageDraw.Draw(layer)
    box = draw.textbbox((0, 0), text, font=font)
    width = box[2] - box[0]
    height = box[3] - box[1]
    position = (
        base.width - width - 30,
        24,
    )
    draw.text(
        position, text, font=font,
        fill=(255, 255, 255, 190),
        stroke_width=2,
        stroke_fill=(0, 0, 0, 150),
    )
    return Image.alpha_composite(base, layer)


credited = add_credit(
    card, "影像來源:經社區授權", load_font(28)
)

浮水印不能取代授權。文字要依真實來源填寫;AI生成或經AI大幅修改的圖片,也應依使用情境清楚標示。

十三、輸出JPEG或PNG前先處理色彩模式

from pathlib import Path


def save_for_web(image, destination):
    destination = Path(destination)
    destination.parent.mkdir(parents=True, exist_ok=True)
    suffix = destination.suffix.lower()

    if suffix in {".jpg", ".jpeg"}:
        image.convert("RGB").save(
            destination,
            format="JPEG",
            quality=88,
            optimize=True,
        )
    elif suffix == ".png":
        image.convert("RGBA").save(
            destination,
            format="PNG",
            optimize=True,
        )
    else:
        raise ValueError("輸出格式只允許JPEG或PNG")


save_for_web(credited, "output/marriage_flower_card.jpg")

JPEG不支援透明度,RGBA直接存JPEG會出錯;PNG適合透明圖示,但照片可能較大。重新建立輸出檔可避免原照片中的GPS等EXIF跟著公開,但若你主動傳入exif或其他metadata則另當別論。

十四、完成單張水井三寶圖卡函式

from PIL import ImageDraw


def create_local_card(
    source, destination, title, credit,
    size=(1200, 1200), crop_mode="crop"
):
    image = open_oriented_rgb(source)
    card = make_card_background(image, size, crop_mode)

    draw = ImageDraw.Draw(card, "RGBA")
    title_font = fit_font(
        draw, title, max_width=size[0] - 120
    )
    draw_title(card, title, title_font, margin=48)
    card = add_credit(card, credit, load_font(28))
    save_for_web(card, destination)

    return {
        "source": str(source),
        "destination": str(destination),
        "title": title,
        "size": size,
    }


result = create_local_card(
    "images/turtle.jpg",
    "output/turtle_card.jpg",
    "水井三寶|烏龜",
    "影像來源:經社區授權",
)
print(result)

十五、批次處理三張圖卡

JOBS = [
    {
        "source": "images/turtle.jpg",
        "destination": "output/turtle_card.jpg",
        "title": "水井三寶|烏龜",
    },
    {
        "source": "images/white_horse.jpg",
        "destination": "output/white_horse_card.jpg",
        "title": "水井三寶|白馬",
    },
    {
        "source": "images/marriage_flower.jpg",
        "destination": "output/marriage_flower_card.jpg",
        "title": "水井三寶|姻緣花",
    },
]

records = []
for job in JOBS:
    try:
        record = create_local_card(
            **job,
            credit="影像來源:經社區授權",
        )
        record["status"] = "success"
    except Exception as error:
        record = {
            "source": job["source"],
            "status": "failed",
            "error": type(error).__name__,
        }
    records.append(record)

for record in records:
    print(record)

批次工具不應因一張壞圖讓所有工作中止,但錯誤紀錄不要包含個資、秘密路徑或完整內部例外內容。

十六、用雜湊與尺寸驗證輸出

import hashlib
from pathlib import Path
from PIL import Image


def inspect_output(path, expected_size=(1200, 1200)):
    path = Path(path)
    digest = hashlib.sha256(path.read_bytes()).hexdigest()
    with Image.open(path) as image:
        image.load()
        if image.size != expected_size:
            raise ValueError("輸出尺寸不正確")
        return {
            "path": str(path),
            "format": image.format,
            "mode": image.mode,
            "size": image.size,
            "bytes": path.stat().st_size,
            "sha256": digest,
        }


print(inspect_output("output/turtle_card.jpg"))

雜湊可協助確認檔案是否變更,不能證明圖片內容正確。仍要人工檢查裁切、文字、對比、地方名稱與來源標示。

十七、發布前的人文與倫理檢查

檢查發布前要確認
著作權攝影者、插畫者或權利人是否同意使用及改作?
肖像與隱私可辨識人物是否同意?是否含門牌、車牌或位置資訊?
地方詮釋「水井三寶」名稱與故事是否經社區夥伴確認?
AI透明AI生成、補圖或大幅修改是否清楚標示?
可近用性網站是否提供替代文字,而非只把文字畫進圖片?
原圖要另外保管:批次程式輸出到獨立資料夾,不覆蓋社區提供的原始照片。若未取得公開同意,處理完成也不代表可以上網發布。

十八、課堂挑戰

挑戰A|基礎:用同一張照片分別產生thumbnail、crop與contain三種版本,說明各自適合的情境。
挑戰B|進階:讓長標題自動換成兩行,確保不低於32px,並為文字底板保留左右安全距離。
挑戰C|USR場域:和社區夥伴共同建立圖卡metadata表,記錄作品名、來源、授權範圍、人物同意、AI使用、審核者及下架日期。

十九、用AI協助檢查,但不要交出未授權照片

請擔任Python Pillow程式碼審查助教。
我要把社區授權照片製作成1200×1200地方圖卡。
請檢查:不可信圖片驗證、解壓縮炸彈、EXIF方向、
等比例縮放、裁切重心、中文字型授權、文字可讀性、
JPEG/PNG模式、metadata移除、錯誤處理與不覆蓋原圖。
不要要求我上傳未取得同意的人像或含GPS的原始照片。
請提供最小測試案例與人工審查清單。

二十、延伸閱讀

Python一下:第三方模組怎麼選?從需求、授權到安全更新

《Python一下:從風土資料到智慧生活》第 29 篇

第三方模組怎麼選?從需求、授權到安全更新

會安裝套件只是起點;真正的能力,是知道為何選、如何驗證、何時更新與怎麼退回。

CH14 第三方模組虛擬環境供應鏈安全
學習目標
完成本篇後,你能區分標準函式庫、第三方發行套件與匯入模組;從功能、維護、授權、安全及部署條件評估套件;使用venv隔離環境;建立可重現的依賴清單;進行基本弱點稽核,並為USR場域制定安全更新與回復流程。

一、先問「需要什麼」,再問「裝哪一個」

水井村智慧生活專案可能需要網頁請求、影像處理、資料分析、資料庫、斷詞與硬體通訊。每多一個第三方套件,就多一份功能,也多一份版本、授權、弱點與維護責任。

USR需求先檢查可能使用
呼叫YouBike或場域API標準urllib是否足夠?是否需Session、重試?requests、httpx
處理水井三寶圖卡只需縮圖,還是要進階影像運算?Pillow、OpenCV
整理感測資料內建csv能否完成?資料量多大?pandas、NumPy
存取MySQL直接SQL還是需要ORM?PyMySQL、SQLAlchemy
中文文字處理語言、詞典與授權是否合適?依任務比較斷詞工具
不要只看下載量:熱門不等於適合。需要一起看官方文件、最近維護狀態、Python版本支援、授權、已知弱點、相依套件、安裝大小、硬體架構及是否能在PythonAnywhere、Raspberry Pi或Colab運作。

二、Package、Distribution與Module不是同一件事

日常口語都稱「套件」,但安裝名稱與匯入名稱可能不同。pip安裝的是發行套件(distribution package),Python程式匯入的是模組或import package。

import importlib.metadata


distribution_name = "Pillow"
print(importlib.metadata.version(distribution_name))

# 安裝名稱是Pillow,匯入名稱卻是PIL
from PIL import Image

print(Image.__name__)

因此不能只用 import名稱 == pip名稱 猜測。應以套件官方文件及PyPI專案頁為準,並小心拼字相近的惡意套件。

三、七項評估表:先評分再安裝

面向檢查問題證據
需求適配能否用最少功能完成任務?最小原型與測試
維護狀態是否仍支援所用Python版本?官方文件、發行紀錄
授權教學、商用、再散布是否允許?LICENSE與專案metadata
安全是否有未修補已知弱點?安全公告與依賴稽核
可部署性平台、CPU、記憶體與系統函式庫是否相容?乾淨環境安裝測試
資料治理資料是否會外傳?預設遙測為何?隱私政策與網路測試
可替換性若停止維護,能否退出?包裝介面與回復計畫

四、用Python建立簡易評估紀錄

from dataclasses import dataclass


@dataclass(frozen=True)
class PackageReview:
    name: str
    need_fit: int
    maintenance: int
    license_fit: int
    security: int
    deployability: int
    evidence: str

    @property
    def total(self):
        scores = (
            self.need_fit,
            self.maintenance,
            self.license_fit,
            self.security,
            self.deployability,
        )
        if any(score not in range(1, 6) for score in scores):
            raise ValueError("各項評分必須是1到5")
        return sum(scores)


review = PackageReview(
    name="示範套件",
    need_fit=5,
    maintenance=4,
    license_fit=4,
    security=4,
    deployability=5,
    evidence="官方文件、最小原型、授權檔與測試報告",
)
print(review.total)

分數不能取代判斷;若授權不合、存在無法接受的弱點或會把敏感資料送出,即使總分高也不應採用。

五、每個專案都建立自己的虛擬環境

PyPA建議使用虛擬環境管理第三方套件,避免不同專案互相干擾。Windows與macOS/Linux啟用方式不同:

$ py -m venv .venv
$ .venv\Scripts\activate
$ python -m pip install --upgrade pip

# macOS或Linux
$ python3 -m venv .venv
$ source .venv/bin/activate
$ python -m pip install --upgrade pip

在專案根目錄建立 .venv,並把它加入.gitignore。不要把整個虛擬環境提交到版本控制;應提交依賴宣告,讓環境可以重建。

六、確認目前真的在虛擬環境

import sys
from pathlib import Path


in_virtual_environment = sys.prefix != sys.base_prefix
print("Python:", Path(sys.executable))
print("虛擬環境:", in_virtual_environment)

啟用只是方便shell找到正確Python;關鍵是執行的直譯器路徑。Thonny、VS Code、排程或網站服務也要指定同一環境。

七、用python -m pip避免裝錯地方

import subprocess
import sys


result = subprocess.run(
    [sys.executable, "-m", "pip", "--version"],
    check=True,
    capture_output=True,
    text=True,
)
print(result.stdout.strip())

python -m pip可明確使用目前Python所屬的pip,減少電腦同時安裝多個Python時的混亂。

八、不要無條件安裝「最新版」

開發初期可指定一段相容範圍,正式部署則需要經測試的解析結果。以下只是格式示範,不宣稱這些版本永遠合適:

requests>=2.0,<3.0
Pillow>=10.0,<13.0
SQLAlchemy>=2.0,<3.0
PyMySQL>=1.1,<2.0

版本範圍要依支援政策與測試調整。對教室教材,可以提供一份已驗證版本;對正式系統,升級應經過測試、稽核、備份與回復流程。

九、requirements.txt與pip freeze

$ python -m pip install -r requirements.txt
$ python -m pip freeze > requirements-lock.txt
$ python -m pip check

pip freeze記錄目前環境所有已安裝版本,適合製作環境快照,但可能把未使用的套件也收入。較成熟的專案會分開維護「直接需求」與「解析後鎖定結果」。

十、用metadata查看版本、授權與依賴

from importlib.metadata import metadata, requires, version


def package_summary(distribution_name):
    info = metadata(distribution_name)
    return {
        "name": info.get("Name"),
        "version": version(distribution_name),
        "license_expression": info.get("License-Expression"),
        "license": info.get("License"),
        "home_page": info.get("Home-page"),
        "dependencies": requires(distribution_name) or [],
    }


print(package_summary("pip"))

Metadata可以協助盤點,但授權欄位可能缺漏或表達方式不同。真正採用前仍要讀專案LICENSE、例外條款與所包含資料或模型的授權。

十一、產生最小軟體物料清單

import importlib.metadata
import json
from datetime import datetime, timezone


packages = sorted(
    (
        {
            "name": item.metadata.get("Name", item.name),
            "version": item.version,
        }
        for item in importlib.metadata.distributions()
    ),
    key=lambda item: item["name"].lower(),
)

inventory = {
    "created_at": datetime.now(timezone.utc).isoformat(),
    "python": __import__("sys").version,
    "packages": packages,
}
print(json.dumps(inventory, ensure_ascii=False, indent=2))

這是教學用盤點,不是完整標準SBOM。至少要知道部署了哪些元件與版本,發生安全公告時才能判斷是否受影響。

十二、檢查已知弱點,但不要誤解結果

PyPA的pip-audit可掃描Python環境中套件的已知弱點。它不是原始碼掃描器,也不能保證發現所有風險;掃描前仍要把套件視為會被解析與處理的外部輸入。

$ python -m pip install pip-audit
$ python -m pip_audit
$ python -m pip_audit -r requirements-lock.txt
稽核不等於自動升級:先確認弱點是否影響實際使用路徑、修正版是否相容,再於測試環境更新並跑完整測試。若暫時無法更新,應記錄理由、降低暴露面、設定期限與負責人,而不是永久忽略。

十三、雜湊可以強化可重現安裝

pip支援hash-checking mode,requirements中的套件可附上允許的檔案雜湊,再用 --require-hashes 驗證。雜湊能確認下載檔案符合鎖定內容,但不能證明套件本身沒有惡意程式或設計缺陷。

$ python -m pip install \
    --require-hashes \
    -r requirements-hashed.txt

十四、授權不是只有「免費」與「付費」

檢查項目教學與USR要問的問題
程式碼授權可否修改、散布、商用?需否保留聲明?
資料/模型授權圖片、詞典、模型權重是否另有條款?
Copyleft義務與自有程式結合或部署服務時有何義務?
專利與商標程式碼授權是否同時處理專利或名稱使用?
隱私與服務條款雲端API是否保存、再利用或跨境傳輸資料?

遇到正式商用、再散布或複雜授權時,應由具資格的法務或授權專業人員確認,不能只靠AI摘要下結論。

十五、先做煙霧測試,再讓套件進場域

from importlib import import_module
from importlib.metadata import version


EXPECTED = {
    "requests": "requests",
    "Pillow": "PIL",
    "PyMySQL": "pymysql",
    "SQLAlchemy": "sqlalchemy",
}


def smoke_test(distribution_name):
    module_name = EXPECTED[distribution_name]
    module = import_module(module_name)
    return {
        "distribution": distribution_name,
        "version": version(distribution_name),
        "imported_as": module.__name__,
    }


for name in EXPECTED:
    try:
        print(smoke_test(name))
    except Exception as error:
        print(name, "FAIL", type(error).__name__)

煙霧測試只能確認基本匯入。還要測USR實際流程,例如圖片能否正確縮圖、MySQL能否回滾、Raspberry Pi能否安裝、斷線後資料能否補傳。

十六、用介面包住第三方套件

不要讓整個專案到處直接依賴某套件。把外部功能包在小型介面後,較容易測試、替換與集中處理錯誤:

class ImageThumbnailer:
    def __init__(self, image_module):
        self.image_module = image_module

    def create(self, source, destination, size=(800, 800)):
        with self.image_module.open(source) as image:
            image.thumbnail(size)
            image.save(destination)


from PIL import Image

thumbnailer = ImageThumbnailer(Image)

這也讓測試時能傳入替身物件,不必每次真的處理大型圖片或呼叫外部服務。

十七、安全更新的六步驟

  1. 盤點:確認直接與間接依賴、Python與作業系統版本。
  2. 閱讀:查看官方發行說明、安全公告與破壞性變更。
  3. 備份:保存可回復的程式版本、設定與資料庫備份。
  4. 測試:在乾淨環境重建,執行單元、整合與場域流程測試。
  5. 分段部署:先測試站或少量設備,觀察Log、效能與錯誤率。
  6. 記錄與回復:留下版本、日期、負責人、驗證結果及退版條件。
from dataclasses import dataclass
from datetime import date


@dataclass
class UpgradeRecord:
    package: str
    old_version: str
    new_version: str
    tested_on: date
    tests_passed: bool
    rollback_tag: str
    reviewer: str


record = UpgradeRecord(
    package="示範套件",
    old_version="1.0",
    new_version="1.1",
    tested_on=date.today(),
    tests_passed=True,
    rollback_tag="before-demo-upgrade",
    reviewer="課程小組",
)
print(record)

十八、USR場域的特別風險

場域常見限制選套件時的因應
養殖池邊緣設備網路不穩、ARM平台、儲存有限先測離線、wheel支援與資源用量
長者與社區資料語音、照片可能含個資優先本機處理,確認外傳與保存規則
學生共同開發環境不一致、帳密誤傳虛擬環境、鎖定檔、Secrets與程式審查
長期USR維運學生畢業、套件停更文件化、介面隔離、替代方案與交接

十九、課堂挑戰

挑戰A|基礎:選擇一個目前課程用到的第三方套件,完成七項評估表,所有判斷都附官方來源。
挑戰B|進階:在全新虛擬環境依鎖定清單重建專案,執行pip check、弱點稽核及三項煙霧測試,記錄結果。
挑戰C|USR場域:為Raspberry Pi智慧養殖閘道器設計「更新、觀察、退版」流程,包含斷網、套件無ARM wheel及資料不得遺失三種情境。

二十、用AI當評估助教,但證據要回到官方來源

請擔任Python第三方套件評估助教。
我的任務、Python版本、作業系統與部署硬體如下:〔請填寫〕。
請先問清楚必要功能,再比較不超過3個候選套件。
評估需求適配、官方維護、Python支援、授權、安全公告、
間接依賴、安裝大小、ARM與離線部署、資料是否外傳。
每項必須指向官方文件、PyPI metadata或專案LICENSE,
不以下載量或AI印象直接下結論。
最後提出最小驗證、更新與退版計畫。

二十一、延伸閱讀

Python一下:用SQLAlchemy與PyMySQL重構智慧養殖資料存取層

《Python一下:從風土資料到智慧生活》第 28 篇

用SQLAlchemy與PyMySQL重構智慧養殖資料存取層

讓資料表成為Python物件,讓交易有清楚邊界,也讓SQLite與MySQL共用大部分程式。

CH13 關聯式資料庫SQLAlchemy 2.xORM
學習目標
完成本篇後,你能說明Engine、Model、Session與Repository的責任;使用SQLAlchemy 2.x宣告關聯模型;以ORM新增、查詢及更新資料;正確控制交易與連線池;並以SQLite進行快速測試、以PyMySQL連接MySQL。

一、為何要增加SQLAlchemy這一層?

直接使用PyMySQL完全可行,但當SQL散落在API、排程、分析腳本與管理頁面中,欄名修改、交易處理及測試會逐漸困難。SQLAlchemy提供一致的資料存取介面;ORM則把資料表的列映射成Python物件。

元件主要責任水井村例子
Engine管理資料庫方言、連線與連線池MySQL正式環境、SQLite課堂測試
Model描述資料表、欄位與關聯Site、Sensor、Reading
Session追蹤物件及管理一個工作單元新增一次巡查紀錄並提交
Repository集中常用資料存取規則查最新水溫、建立場域
ORM不是不用學SQL:SQLAlchemy最後仍會產生SQL。若不了解JOIN、索引、NULL與交易,就可能寫出「看起來很Python、實際查詢很慢」的程式。

二、安裝SQLAlchemy與PyMySQL

import subprocess
import sys


subprocess.run(
    [
        sys.executable, "-m", "pip", "install",
        "SQLAlchemy>=2.0,<3.0", "PyMySQL>=1.1,<2.0",
    ],
    check=True,
)

實際專案應使用虛擬環境並鎖定經測試的版本。套件升級前先跑測試,不在正式環境臨時更新。

三、建立資料庫網址,但不把密碼寫進程式

import os
from sqlalchemy import URL


def mysql_url_from_environment():
    password = os.getenv("SHUIJING_DB_PASSWORD", "")
    if not password:
        raise RuntimeError("尚未設定SHUIJING_DB_PASSWORD")

    return URL.create(
        drivername="mysql+pymysql",
        username=os.getenv("SHUIJING_DB_USER", "shuijing_app"),
        password=password,
        host=os.getenv("SHUIJING_DB_HOST", "127.0.0.1"),
        port=int(os.getenv("SHUIJING_DB_PORT", "3306")),
        database=os.getenv("SHUIJING_DB_NAME", "shuijing_demo"),
        query={"charset": "utf8mb4"},
    )

URL.create()會處理密碼中的特殊字元。不要print完整URL,因為它可能包含憑證;Colab請使用Secrets,GitHub則使用平台的Secret設定。

四、建立Engine與連線池

from sqlalchemy import create_engine


engine = create_engine(
    mysql_url_from_environment(),
    pool_pre_ping=True,
    pool_recycle=1800,
    pool_size=5,
    max_overflow=5,
    pool_timeout=10,
    echo=False,
)

print(engine.url.render_as_string(hide_password=True))

pool_pre_ping可在借出連線前檢查是否仍有效;連線池大小必須配合MySQL上限、網站程序數與實際流量計算,不可每個程序都任意設很大。

五、先宣告共同Model基底

from sqlalchemy.orm import DeclarativeBase


class Base(DeclarativeBase):
    pass

本篇採SQLAlchemy 2.x的型別化宣告方式。每個Model都繼承Base,後續可由metadata取得所有資料表描述。

六、建立Site、Sensor與Reading模型

from __future__ import annotations

from datetime import datetime
from typing import Optional

from sqlalchemy import (
    BigInteger, Boolean, DateTime, Float, ForeignKey,
    Index, Integer, String, UniqueConstraint,
)
from sqlalchemy.orm import Mapped, mapped_column, relationship


ID_TYPE = BigInteger().with_variant(Integer, "sqlite")


class Site(Base):
    __tablename__ = "sites"

    site_id: Mapped[int] = mapped_column(
        ID_TYPE, primary_key=True, autoincrement=True
    )
    site_code: Mapped[str] = mapped_column(
        String(30), unique=True, nullable=False
    )
    display_name: Mapped[str] = mapped_column(String(100))
    active: Mapped[bool] = mapped_column(Boolean, default=True)
    sensors: Mapped[list[Sensor]] = relationship(
        back_populates="site"
    )


class Sensor(Base):
    __tablename__ = "sensors"

    sensor_id: Mapped[int] = mapped_column(
        ID_TYPE, primary_key=True, autoincrement=True
    )
    sensor_code: Mapped[str] = mapped_column(
        String(50), unique=True, nullable=False
    )
    site_id: Mapped[int] = mapped_column(
        ForeignKey("sites.site_id", ondelete="RESTRICT")
    )
    kind: Mapped[str] = mapped_column(String(30))
    unit: Mapped[str] = mapped_column(String(20))
    site: Mapped[Site] = relationship(back_populates="sensors")
    readings: Mapped[list[Reading]] = relationship(
        back_populates="sensor"
    )


class Reading(Base):
    __tablename__ = "readings"
    __table_args__ = (
        UniqueConstraint(
            "sensor_id", "observed_at",
            name="uq_sensor_time",
        ),
        Index("idx_readings_time", "observed_at"),
    )

    reading_id: Mapped[int] = mapped_column(
        ID_TYPE, primary_key=True, autoincrement=True
    )
    sensor_id: Mapped[int] = mapped_column(
        ForeignKey("sensors.sensor_id", ondelete="RESTRICT")
    )
    observed_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=False)
    )
    value: Mapped[Optional[float]] = mapped_column(Float)
    quality: Mapped[str] = mapped_column(String(10))
    note: Mapped[str] = mapped_column(String(255), default="")
    sensor: Mapped[Sensor] = relationship(
        back_populates="readings"
    )

ID_TYPE讓MySQL使用BIGINT,SQLite測試則使用可正確連結ROWID自動編號的INTEGER。Python型別提示協助編輯器及靜態檢查,但不會自動完成所有資料驗證;quality允許值、時間規則及數值合理性仍應在應用層和資料庫層共同約束。

七、建立資料表:教材可以,正式系統要遷移

Base.metadata.create_all(engine)

create_all()適合初學、測試及全新資料庫,但不會安全地替既有資料表改欄位。正式系統應使用Alembic建立可審查、可追蹤、可回復的結構遷移版本。

八、用Session新增一座場域

from sqlalchemy.orm import Session


with Session(engine) as session:
    with session.begin():
        site = Site(
            site_code="SITE-01",
            display_name="示範池A",
        )
        session.add(site)

print("場域新增完成")

session.begin()區塊正常結束時提交,發生例外時回滾。離開Session後不要任意依賴尚未載入的關聯資料。

九、一次建立場域、感測器與讀值

from datetime import datetime


with Session(engine) as session:
    with session.begin():
        site = Site(
            site_code="SITE-02",
            display_name="示範池B",
        )
        sensor = Sensor(
            sensor_code="TEMP-02",
            kind="temperature",
            unit="°C",
        )
        sensor.readings.append(
            Reading(
                observed_at=datetime(2026, 8, 31, 9, 0),
                value=29.1,
                quality="valid",
            )
        )
        site.sensors.append(sensor)
        session.add(site)

ORM會依關聯順序寫入三張表。教材時間為示例;正式專案應統一以UTC或明確規則寫入,且不可把無時區時間與有時區時間混用。

十、用select()查詢,而不是舊式Query

from sqlalchemy import select


statement = (
    select(Site)
    .where(Site.active.is_(True))
    .order_by(Site.site_code)
)

with Session(engine) as session:
    sites = session.scalars(statement).all()
    for site in sites:
        print(site.site_code, site.display_name)

select(Site)回傳Site物件;如果選擇個別欄位,結果形態會不同。先確認自己需要物件、欄位列,還是單一純量。

十一、JOIN查詢最新有效水溫

statement = (
    select(
        Site.site_code,
        Site.display_name,
        Sensor.sensor_code,
        Reading.observed_at,
        Reading.value,
        Sensor.unit,
    )
    .join(Site.sensors)
    .join(Sensor.readings)
    .where(
        Site.site_code == "SITE-02",
        Sensor.kind == "temperature",
        Reading.quality == "valid",
    )
    .order_by(
        Reading.observed_at.desc(),
        Reading.reading_id.desc(),
    )
    .limit(1)
)

with Session(engine) as session:
    latest = session.execute(statement).mappings().first()
    print(dict(latest) if latest else "尚無有效資料")

SQLAlchemy會把比較值轉成參數,而不是直接拼進SQL。若需要動態欄位或排序,仍必須使用程式內允許清單。

十二、彙總各場域的有效資料

from sqlalchemy import and_, func


statement = (
    select(
        Site.site_code,
        func.count(Reading.reading_id).label("valid_count"),
        func.round(func.avg(Reading.value), 2).label("average"),
    )
    .outerjoin(
        Sensor,
        and_(
            Sensor.site_id == Site.site_id,
            Sensor.kind == "temperature",
        ),
    )
    .outerjoin(
        Reading,
        and_(
            Reading.sensor_id == Sensor.sensor_id,
            Reading.quality == "valid",
        ),
    )
    .group_by(Site.site_id, Site.site_code)
    .order_by(Site.site_code)
)

with Session(engine) as session:
    for row in session.execute(statement).mappings():
        print(dict(row))

把有效資料條件放在LEFT JOIN的ON子句,才能保留沒有有效資料的場域。若放到WHERE,查詢可能悄悄變成只剩有資料的場域。

十三、避免N+1查詢

逐一讀取每座場域的sensors,可能額外送出很多SQL。需要一起使用關聯時,可明確預載:

from sqlalchemy.orm import selectinload


statement = (
    select(Site)
    .options(selectinload(Site.sensors))
    .order_by(Site.site_code)
)

with Session(engine) as session:
    sites = session.scalars(statement).all()
    for site in sites:
        print(site.site_code, len(site.sensors))

selectinload通常以第二個IN查詢載入集合,避免每座場域各查一次。不要為了省查詢而預載所有歷史讀值,資料量可能非常大。

十四、建立輸入驗證函式

from datetime import datetime


def build_reading(observed_at, value, quality, note=""):
    if quality not in {"valid", "unknown", "invalid"}:
        raise ValueError("不支援的資料品質")
    if not isinstance(observed_at, datetime):
        raise TypeError("observed_at必須是datetime")
    if quality == "unknown":
        value = None
    elif value is None:
        raise ValueError("valid或invalid必須保留原始值")
    else:
        value = float(value)

    return Reading(
        observed_at=observed_at,
        value=value,
        quality=quality,
        note=str(note)[:255],
    )
資料品質先於圖表:unknown表示沒有可信數值,invalid表示保留原始值但不宜納入一般統計。不要把兩者都轉成0,否則平均值與告警會被扭曲。

十五、Repository:集中常用存取規則

class SiteRepository:
    def __init__(self, session):
        self.session = session

    def get_by_code(self, site_code):
        statement = select(Site).where(
            Site.site_code == site_code
        )
        return self.session.scalar(statement)

    def add(self, site_code, display_name):
        if self.get_by_code(site_code) is not None:
            raise ValueError("場域代碼已存在")
        site = Site(
            site_code=site_code,
            display_name=display_name,
        )
        self.session.add(site)
        return site


with Session(engine) as session:
    with session.begin():
        repository = SiteRepository(session)
        repository.add("SITE-03", "示範池C")

Repository不應偷偷commit;由外層服務決定交易邊界,才能把多個Repository操作包在同一交易中。

十六、服務層:把場域規則與資料存取分開

def register_sensor(session, site_code, sensor_code, kind, unit):
    repository = SiteRepository(session)
    site = repository.get_by_code(site_code)
    if site is None:
        raise ValueError("場域不存在")

    duplicate = session.scalar(
        select(Sensor).where(Sensor.sensor_code == sensor_code)
    )
    if duplicate is not None:
        raise ValueError("感測器代碼已存在")

    sensor = Sensor(
        sensor_code=sensor_code,
        kind=kind,
        unit=unit,
        site=site,
    )
    session.add(sensor)
    return sensor


with Session(engine) as session:
    with session.begin():
        register_sensor(
            session, "SITE-03", "TEMP-03",
            "temperature", "°C",
        )

十七、用SQLite快速測試相同Model

from sqlalchemy import create_engine
from sqlalchemy.orm import Session


test_engine = create_engine("sqlite+pysqlite:///:memory:")
Base.metadata.create_all(test_engine)

with Session(test_engine) as session:
    with session.begin():
        repository = SiteRepository(session)
        repository.add("TEST-01", "測試場域")

with Session(test_engine) as session:
    saved = session.scalar(
        select(Site).where(Site.site_code == "TEST-01")
    )
    assert saved is not None
    assert saved.display_name == "測試場域"

print("測試通過")

SQLite測試快速,但不能完全代表MySQL:型別、排序規則、鎖定、時區、函式與並行行為可能不同。因此還需要少量MySQL整合測試。

十八、Session常見錯誤

錯誤改善方式
全站共用一個Session每個請求或工作單元建立並關閉Session
Repository自行commit由服務層統一決定提交或回滾
離開Session後才讀延遲關聯在Session內預載或轉為輸出資料
直接回傳ORM物件給外部API選擇必要欄位,避免洩漏內部資料
把echo=True留在正式環境使用受控Log,避免敏感參數外洩

十九、USR資料治理:Model不等於公開資料

Model可能包含場域內部欄位,但學生、養殖戶、公開儀表板與研究資料集需要不同視圖。應在服務層或API輸出層做角色授權與欄位選擇,使用匿名場域代碼,避免輸出姓名、電話、精確座標、設備Token與連線資訊。

def public_site_summary(site, latest_value=None):
    return {
        "site_code": site.site_code,
        "display_name": site.display_name,
        "latest_temperature": latest_value,
    }


# 不輸出內部主鍵、帳號、精確位置或設備憑證

二十、課堂挑戰

挑戰A|基礎:為Reading增加received_at欄位,說明observed_at與received_at的差別。
挑戰B|進階:建立ReadingRepository,完成「批次新增」及「查每支感測器最新有效讀值」,但不要在Repository內commit。
挑戰C|USR場域:設計公開版、養殖戶版及維運版三種輸出資料,逐欄說明誰能看、為何需要,以及保存多久。

二十一、用AI協助審查ORM,而不是盲目產碼

請擔任SQLAlchemy 2.x程式碼審查助教。
系統使用MySQL、PyMySQL,模型為Site、Sensor、Reading。
請檢查:關聯方向、外鍵、唯一鍵、NULL、索引、時區、
Session生命週期、交易邊界、N+1查詢、連線池與敏感資料輸出。
不要改用舊式session.query,也不要要求真實帳密。
請把問題分成「會造成資料錯誤」「效能風險」「維護性」三類,
每項提供最小修正範例與可執行測試。

Python一下:從SQLite走向MySQL——建立多人共用的智慧養殖資料庫

《Python一下:從風土資料到智慧生活》第 27 篇

從SQLite走向MySQL——建立多人共用的智慧養殖資料庫

讓場域設備、學生團隊與管理平台安全共用資料,而不是共用一個資料庫檔案。

CH13 關聯式資料庫MySQLPyMySQL
學習目標
完成本篇後,你能比較SQLite與MySQL的適用情境,建立UTF-8資料庫及最小權限帳號,使用環境變數與PyMySQL安全連線,以交易完成批次寫入、查詢與更新,並說明遠端部署、備份及個資保護的基本原則。

一、SQLite很好,為何還需要MySQL?

SQLite把整個資料庫放在單一檔案中,適合個人練習、離線程式、小型裝置及單機原型。當多座智慧養殖場域、Django平台與學生團隊需要同時存取,伺服器型資料庫較容易集中管理帳號、連線、權限、交易與備份。

比較SQLiteMySQL
架構程式直接讀寫檔案用戶端連線至資料庫伺服器
安裝Python內建sqlite3需安裝、啟動及維護服務
多人並行讀取方便,寫入並行有限適合多使用者與網路服務
帳號權限主要依賴檔案權限可依帳號、主機與資料庫授權
適合本系列課堂練習、離線採集多人共用平台與正式場域
不是資料越多就一定要換MySQL:是否遷移應看同時使用者、寫入頻率、權限、維運能力、可用性及備份需求。若SQLite已能安全滿足需求,不必為了「看起來專業」增加系統複雜度。

二、準備MySQL與Python驅動程式

先由教師或系統管理者安裝受支援版本的MySQL Server。Python端使用PyMySQL:

import subprocess
import sys


subprocess.run(
    [sys.executable, "-m", "pip", "install", "PyMySQL"],
    check=True,
)

在終端機直接執行時,也可輸入 python -m pip install PyMySQL。教室與正式專案應使用虛擬環境,並將套件版本記錄於requirements.txt。

三、由管理者建立資料庫與專用帳號

以下SQL應由有權限的管理者在MySQL工具中執行;請替換密碼,且不要直接開放給整個網際網路。

ADMIN_SQL = """
CREATE DATABASE IF NOT EXISTS shuijing_demo
  CHARACTER SET utf8mb4
  COLLATE utf8mb4_0900_ai_ci;

CREATE USER IF NOT EXISTS
  'shuijing_app'@'10.0.0.%'
  IDENTIFIED BY '請換成長且唯一的密碼';

GRANT SELECT, INSERT, UPDATE, DELETE
  ON shuijing_demo.*
  TO 'shuijing_app'@'10.0.0.%';

FLUSH PRIVILEGES;
"""

print(ADMIN_SQL)

10.0.0.%只是私有網段示例,必須依實際網路縮小允許來源。應另外設計遷移帳號負責CREATE與ALTER;平常執行的應用程式帳號不需要DROP、GRANT等高權限。

四、把連線設定放在環境變數

不要把密碼寫進Python、Colab、GitHub或部落格。不同作業系統的設定方式不同,程式只負責讀取:

import os


DB_CONFIG = {
    "host": os.getenv("SHUIJING_DB_HOST", "127.0.0.1"),
    "port": int(os.getenv("SHUIJING_DB_PORT", "3306")),
    "user": os.getenv("SHUIJING_DB_USER", "shuijing_app"),
    "password": os.getenv("SHUIJING_DB_PASSWORD", ""),
    "database": os.getenv("SHUIJING_DB_NAME", "shuijing_demo"),
}

if not DB_CONFIG["password"]:
    raise RuntimeError("尚未設定SHUIJING_DB_PASSWORD")
Google Colab提醒:不要在公開Notebook中顯示密碼。可使用Colab的Secrets功能讀取密鑰;資料庫也不應直接暴露3306連接埠給所有來源。教學可使用校內測試網段、VPN、SSH通道或受控雲端環境。

五、建立安全連線函式

import pymysql


def connect_database(config):
    return pymysql.connect(
        host=config["host"],
        port=config["port"],
        user=config["user"],
        password=config["password"],
        database=config["database"],
        charset="utf8mb4",
        cursorclass=pymysql.cursors.DictCursor,
        autocommit=False,
        connect_timeout=10,
        read_timeout=10,
        write_timeout=10,
    )


connection = connect_database(DB_CONFIG)
with connection.cursor() as cursor:
    cursor.execute("SELECT VERSION() AS version")
    print(cursor.fetchone())

連線逾時避免程式無限等待;DictCursor讓查詢結果以欄名存取。正式遠端連線應依部署環境驗證TLS憑證,不應只為方便而關閉驗證。

六、建立三張InnoDB關聯表

SCHEMA_SQL = [
    """
    CREATE TABLE IF NOT EXISTS sites (
      site_id BIGINT UNSIGNED AUTO_INCREMENT PRIMARY KEY,
      site_code VARCHAR(30) NOT NULL UNIQUE,
      display_name VARCHAR(100) NOT NULL,
      active BOOLEAN NOT NULL DEFAULT TRUE,
      created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP
    ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
    """,
    """
    CREATE TABLE IF NOT EXISTS sensors (
      sensor_id BIGINT UNSIGNED AUTO_INCREMENT PRIMARY KEY,
      sensor_code VARCHAR(50) NOT NULL UNIQUE,
      site_id BIGINT UNSIGNED NOT NULL,
      kind VARCHAR(30) NOT NULL,
      unit VARCHAR(20) NOT NULL,
      CONSTRAINT fk_sensors_site
        FOREIGN KEY (site_id) REFERENCES sites(site_id)
        ON UPDATE CASCADE ON DELETE RESTRICT
    ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
    """,
    """
    CREATE TABLE IF NOT EXISTS readings (
      reading_id BIGINT UNSIGNED AUTO_INCREMENT PRIMARY KEY,
      sensor_id BIGINT UNSIGNED NOT NULL,
      observed_at DATETIME(6) NOT NULL,
      value DOUBLE NULL,
      quality ENUM('valid','unknown','invalid') NOT NULL,
      note VARCHAR(255) NOT NULL DEFAULT '',
      received_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
      UNIQUE KEY uq_sensor_time (sensor_id, observed_at),
      KEY idx_readings_time (observed_at),
      CONSTRAINT fk_readings_sensor
        FOREIGN KEY (sensor_id) REFERENCES sensors(sensor_id)
        ON DELETE RESTRICT
    ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
    """,
]

try:
    with connection.cursor() as cursor:
        for statement in SCHEMA_SQL:
            cursor.execute(statement)
    connection.commit()
except Exception:
    connection.rollback()
    raise

使用InnoDB才能可靠支援交易與外鍵。時間欄位的時區策略要先約定:常見做法是以UTC寫入 DATETIME,應用程式顯示時再轉成臺灣時間,避免夏令時間或跨區部署造成歧義。

七、SQLite與MySQL最容易混淆的差異

用途SQLitePyMySQL/MySQL
參數佔位符?%s
自動編號INTEGER PRIMARY KEYAUTO_INCREMENT
真假值通常以0與1儲存BOOLEAN實際為TINYINT(1)
新增或忽略ON CONFLICT DO NOTHINGINSERT IGNORE或ON DUPLICATE KEY UPDATE
列資料sqlite3.RowDictCursor

八、用參數化SQL新增場域

def add_site(connection, site_code, display_name):
    sql = """
        INSERT INTO sites (site_code, display_name)
        VALUES (%s, %s)
    """
    try:
        with connection.cursor() as cursor:
            cursor.execute(sql, (site_code, display_name))
        connection.commit()
    except pymysql.MySQLError:
        connection.rollback()
        raise


add_site(connection, "SITE-01", "示範池A")

即使PyMySQL的佔位符長得像字串格式化,也不能自己使用 % 或f-string拼接SQL;參數必須交給 execute() 的第二個參數處理。

九、重複執行也安全的Upsert

def upsert_site(connection, site_code, display_name):
    sql = """
        INSERT INTO sites (site_code, display_name)
        VALUES (%s, %s) AS new
        ON DUPLICATE KEY UPDATE
          display_name = new.display_name
    """
    try:
        with connection.cursor() as cursor:
            cursor.execute(sql, (site_code, display_name))
        connection.commit()
    except pymysql.MySQLError:
        connection.rollback()
        raise


upsert_site(connection, "SITE-01", "示範池A")

Upsert可讓同步工作重跑而不產生重複場域,但更新哪些欄位仍需明確規劃,避免新資料不小心覆蓋人工校正內容。

十、交易:場域與感測器一起成功

def create_site_with_sensor(
    connection, site_code, display_name,
    sensor_code, kind, unit
):
    try:
        with connection.cursor() as cursor:
            cursor.execute(
                """
                INSERT INTO sites (site_code, display_name)
                VALUES (%s, %s)
                """,
                (site_code, display_name),
            )
            site_id = cursor.lastrowid
            cursor.execute(
                """
                INSERT INTO sensors
                  (sensor_code, site_id, kind, unit)
                VALUES (%s, %s, %s, %s)
                """,
                (sensor_code, site_id, kind, unit),
            )
        connection.commit()
    except pymysql.MySQLError:
        connection.rollback()
        raise

任何一步失敗都回滾,避免只留下沒有設備的半套資料。不要在尚未完成整個工作單元前提早commit。

十一、批次寫入感測資料

def insert_readings(connection, sensor_id, items):
    sql = """
        INSERT INTO readings
          (sensor_id, observed_at, value, quality, note)
        VALUES (%s, %s, %s, %s, %s)
        ON DUPLICATE KEY UPDATE
          value = VALUES(value),
          quality = VALUES(quality),
          note = VALUES(note)
    """
    rows = [
        (
            sensor_id,
            item["observed_at"],
            item.get("value"),
            item["quality"],
            item.get("note", ""),
        )
        for item in items
    ]
    try:
        with connection.cursor() as cursor:
            cursor.executemany(sql, rows)
        connection.commit()
    except pymysql.MySQLError:
        connection.rollback()
        raise
先驗證再入庫:程式仍應檢查時間格式、資料型別、quality允許值與合理範圍。超出範圍的原始數值可標成invalid保留,以利追查;設備離線則用NULL加unknown,不要偽裝成0。

十二、JOIN查出最新有效水溫

def latest_temperature(connection, site_code):
    sql = """
        SELECT s.site_code,
               s.display_name,
               e.sensor_code,
               r.observed_at,
               r.value,
               e.unit
        FROM sites AS s
        JOIN sensors AS e ON e.site_id = s.site_id
        JOIN readings AS r ON r.sensor_id = e.sensor_id
        WHERE s.site_code = %s
          AND e.kind = 'temperature'
          AND r.quality = 'valid'
        ORDER BY r.observed_at DESC, r.reading_id DESC
        LIMIT 1
    """
    with connection.cursor() as cursor:
        cursor.execute(sql, (site_code,))
        return cursor.fetchone()


print(latest_temperature(connection, "SITE-01"))

十三、鎖定資料後更新:避免互相覆蓋

def deactivate_site(connection, site_code):
    try:
        with connection.cursor() as cursor:
            cursor.execute(
                """
                SELECT site_id, active
                FROM sites
                WHERE site_code = %s
                FOR UPDATE
                """,
                (site_code,),
            )
            site = cursor.fetchone()
            if site is None:
                raise ValueError("找不到場域")
            cursor.execute(
                "UPDATE sites SET active = FALSE WHERE site_id = %s",
                (site["site_id"],),
            )
        connection.commit()
    except Exception:
        connection.rollback()
        raise

FOR UPDATE需置於交易中,適合「先讀再改」且不能被他人同時改動的流程。鎖定範圍與交易時間應盡量縮小,以免造成等待或死結。

十四、連線生命週期與錯誤紀錄

import logging


logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("shuijing-db")


def run_job(config):
    connection = None
    try:
        connection = connect_database(config)
        with connection.cursor() as cursor:
            cursor.execute("SELECT 1 AS healthy")
            logger.info("database health=%s", cursor.fetchone()["healthy"])
    except pymysql.MySQLError as error:
        logger.error("資料庫作業失敗:%s", type(error).__name__)
        raise
    finally:
        if connection is not None:
            connection.close()

正式Log不要輸出密碼、完整連線字串、Token、住戶姓名或精確座標。網站服務通常使用連線池;每次請求都新建連線會浪費資源,下篇SQLAlchemy會再處理。

十五、從SQLite搬到MySQL的正確流程

  1. 盤點SQLite結構、型別、外鍵、索引、NULL及重複資料。
  2. 先在MySQL測試環境建立結構,不直接動正式系統。
  3. 以Python分批讀取SQLite、驗證、轉換時間與寫入MySQL。
  4. 比對每表筆數、抽樣內容、外鍵及彙總結果。
  5. 安排短暫停止寫入或雙寫切換,保留可回復方案。
  6. 確認應用程式穩定後,再依保存政策封存舊資料。
def compare_counts(sqlite_connection, mysql_connection, table):
    allowed = {"sites", "sensors", "readings"}
    if table not in allowed:
        raise ValueError("不允許的資料表")

    sqlite_count = sqlite_connection.execute(
        f"SELECT COUNT(*) FROM {table}"
    ).fetchone()[0]
    with mysql_connection.cursor() as cursor:
        cursor.execute(f"SELECT COUNT(*) AS total FROM {table}")
        mysql_count = cursor.fetchone()["total"]
    return sqlite_count, mysql_count

表名不能使用一般參數佔位符,所以只能從程式內的允許清單選取。筆數相等只是第一關,還要比對內容、關聯與統計結果。

十六、備份不是複製資料夾

MySQL運作中不能只複製資料目錄。應使用資料庫提供的邏輯備份或經驗證的實體備份機制,安排自動化、異地保存、加密、保存期限及定期還原演練。

BACKUP_CHECKLIST = [
    "備份範圍包含結構、資料、觸發器與必要帳號設定",
    "備份檔加密,存取權限與正式資料一致或更嚴格",
    "至少保存一份於不同故障範圍",
    "記錄備份時間、版本、雜湊值與執行結果",
    "定期在隔離環境實際還原並驗證查詢",
]

for number, item in enumerate(BACKUP_CHECKLIST, start=1):
    print(number, item)

十七、水井村USR的資料治理界線

做法原因
教材使用匿名場域代碼避免揭露養殖戶身分與精確位置
設備只寫入指定表格降低憑證外洩造成的損害
管理者、教師、學生分開帳號便於停權、稽核與責任釐清
平台不直接回傳所有原始資料依角色與任務提供最少資料
保留品質標記與規則版本避免把設備異常誤認為養殖異常

十八、課堂挑戰

挑戰A|基礎:建立一個只能SELECT指定資料表的唯讀帳號,說明它和應用程式讀寫帳號的差別。
挑戰B|進階:設計Python匯入程式,將1000筆資料分批寫入,每批失敗時只回滾該批,並記錄可重跑的錯誤清單。
挑戰C|USR場域:畫出「感測器→閘道器→API→MySQL→儀表板」的資料流,為每一段標示傳輸加密、身分驗證、最小權限與斷線補傳策略。

十九、用AI協助檢查遷移設計

請擔任MySQL資料庫助教,檢查SQLite移轉到MySQL的設計。
資料表為sites、sensors、readings,使用Python與PyMySQL。
請逐項檢查:資料型別、utf8mb4、時區、NULL、外鍵、
唯一鍵、索引、參數化SQL、交易、最小權限、TLS、備份與回復。
不要要求我貼出密碼或真實個資。
請將「一定要改」「建議改進」「需要場域確認」分成三類,
並為每一項提供可驗證方法。

Python一下:從SELECT到JOIN——用SQL回答智慧養殖問題

《Python一下:從風土資料到智慧生活》第 26 篇

從SELECT到JOIN——用SQL回答智慧養殖問題

讓資料庫不只是存資料,而能回答「哪座池需要優先巡查?」

CH13 關聯式資料庫SQL查詢智慧養殖
學習目標
完成本篇後,你能使用SELECT、WHERE、ORDER BY、LIMIT、GROUP BY與HAVING查詢資料;利用INNER JOIN及LEFT JOIN連接場域、感測器與讀值;以參數化SQL產生巡查清單,並說明「異常資料」不等於「養殖異常」。

一、先把場域問題翻成資料問題

場域想知道的事SQL工具
最近有哪些有效讀值?SELECT、WHERE、ORDER BY、LIMIT
各場域平均水溫是多少?GROUP BY、AVG、COUNT
哪些場域目前沒有資料?LEFT JOIN、IS NULL
哪些讀值超出教材示範門檻?JOIN、CASE、參數化條件
重要觀念:SQL查到的是「符合規則的紀錄」,不是自動完成專業診斷。感測器誤差、校正狀態、物種、成長階段、天候與養殖戶經驗都會影響判讀;系統應協助巡查,不應直接取代人的決策。

二、建立可重複練習的記憶體資料庫

以下資料全部為匿名教材資料。使用 :memory: 可在記憶體中練習,程式結束後不留下檔案。

import sqlite3


connection = sqlite3.connect(":memory:")
connection.row_factory = sqlite3.Row
connection.execute("PRAGMA foreign_keys = ON")

connection.executescript("""
CREATE TABLE sites (
    site_id INTEGER PRIMARY KEY,
    site_code TEXT NOT NULL UNIQUE,
    display_name TEXT NOT NULL
);
CREATE TABLE sensors (
    sensor_id INTEGER PRIMARY KEY,
    sensor_code TEXT NOT NULL UNIQUE,
    site_id INTEGER NOT NULL,
    kind TEXT NOT NULL,
    unit TEXT NOT NULL,
    FOREIGN KEY (site_id) REFERENCES sites(site_id)
);
CREATE TABLE readings (
    reading_id INTEGER PRIMARY KEY,
    sensor_id INTEGER NOT NULL,
    observed_at TEXT NOT NULL,
    value REAL,
    quality TEXT NOT NULL,
    FOREIGN KEY (sensor_id) REFERENCES sensors(sensor_id)
);
""")

三、放入水井村智慧養殖示範資料

sites = [
    (1, "SITE-01", "示範池A"),
    (2, "SITE-02", "示範池B"),
    (3, "SITE-03", "示範池C"),
]
sensors = [
    (1, "TEMP-01", 1, "temperature", "°C"),
    (2, "TEMP-02", 2, "temperature", "°C"),
    (3, "PH-01", 1, "ph", "pH"),
]
readings = [
    (1, 1, "2026-08-31T08:00:00+08:00", 27.2, "valid"),
    (2, 1, "2026-08-31T09:00:00+08:00", 29.4, "valid"),
    (3, 2, "2026-08-31T08:00:00+08:00", 30.1, "valid"),
    (4, 2, "2026-08-31T09:00:00+08:00", None, "unknown"),
    (5, 3, "2026-08-31T09:00:00+08:00", 7.4, "valid"),
]

with connection:
    connection.executemany(
        "INSERT INTO sites VALUES (?, ?, ?)", sites
    )
    connection.executemany(
        "INSERT INTO sensors VALUES (?, ?, ?, ?, ?)", sensors
    )
    connection.executemany(
        "INSERT INTO readings VALUES (?, ?, ?, ?, ?)", readings
    )

四、SELECT:決定要看哪些欄位

rows = connection.execute("""
    SELECT sensor_id, observed_at, value, quality
    FROM readings
""").fetchall()

for row in rows:
    print(dict(row))

教學初期可用 SELECT * 探索資料;正式查詢最好明確列出欄位,避免表格修改後輸出跟著意外改變,也減少不必要的資料傳輸。

五、WHERE:只留下可用的水溫資料

rows = connection.execute("""
    SELECT observed_at, value
    FROM readings
    WHERE sensor_id = ?
      AND quality = ?
      AND value IS NOT NULL
""", (1, "valid")).fetchall()

for row in rows:
    print(row["observed_at"], row["value"])

SQL判斷空值要用 IS NULLIS NOT NULL,不能用 = NULL。查詢值仍採參數化方式傳入。

六、ORDER BY與LIMIT:找最新紀錄

latest = connection.execute("""
    SELECT observed_at, value, quality
    FROM readings
    WHERE sensor_id = ?
    ORDER BY observed_at DESC, reading_id DESC
    LIMIT 1
""", (1,)).fetchone()

print(dict(latest) if latest else "尚無資料")

本例時間使用含時區的ISO 8601字串,格式一致時可正確排序。真實系統宜統一以UTC儲存,再於顯示時轉為臺灣時間。

七、計算欄位與CASE:把數值轉成巡查提示

low, high = 26.0, 30.0
rows = connection.execute("""
    SELECT observed_at, value,
           CASE
             WHEN quality != 'valid' THEN '資料待確認'
             WHEN value < ? THEN '低於示範門檻'
             WHEN value > ? THEN '高於示範門檻'
             ELSE '示範範圍內'
           END AS status
    FROM readings
    WHERE sensor_id = ?
    ORDER BY observed_at
""", (low, high, 1)).fetchall()

for row in rows:
    print(row["observed_at"], row["value"], row["status"])

26~30°C僅為SQL教學用門檻,不代表任何物種或場域的正式養殖標準。正式規則應由養殖專業者確認,並保留版本、適用條件及修改紀錄。

八、GROUP BY:各場域有幾筆有效水溫?

rows = connection.execute("""
    SELECT s.site_code,
           COUNT(r.reading_id) AS valid_count,
           ROUND(AVG(r.value), 2) AS average_temperature,
           MIN(r.value) AS minimum_temperature,
           MAX(r.value) AS maximum_temperature
    FROM sites AS s
    LEFT JOIN sensors AS e
      ON e.site_id = s.site_id
     AND e.kind = 'temperature'
    LEFT JOIN readings AS r
      ON r.sensor_id = e.sensor_id
     AND r.quality = 'valid'
    GROUP BY s.site_id, s.site_code
    ORDER BY s.site_code
""").fetchall()

for row in rows:
    print(dict(row))

COUNT(r.reading_id) 不計NULL,因此沒有有效讀值的示範池C會得到0;其平均值仍為NULL,而不是被誤寫成0°C。

九、HAVING:篩選彙總後的結果

rows = connection.execute("""
    SELECT sensor_id,
           COUNT(*) AS sample_count,
           ROUND(AVG(value), 2) AS average_value
    FROM readings
    WHERE quality = 'valid'
    GROUP BY sensor_id
    HAVING COUNT(*) >= ?
    ORDER BY sensor_id
""", (2,)).fetchall()

for row in rows:
    print(dict(row))

WHERE 在分組前篩選每一列;HAVING 在分組後篩選統計結果。

十、INNER JOIN:把代碼翻成場域資訊

rows = connection.execute("""
    SELECT s.display_name,
           e.sensor_code,
           e.kind,
           r.observed_at,
           r.value,
           e.unit
    FROM readings AS r
    JOIN sensors AS e ON e.sensor_id = r.sensor_id
    JOIN sites AS s ON s.site_id = e.site_id
    WHERE r.quality = 'valid'
    ORDER BY r.observed_at DESC, s.site_code
""").fetchall()

for row in rows:
    print(dict(row))

INNER JOIN只保留兩邊都對得上的紀錄,適合查看已產生讀值的感測器。

十一、LEFT JOIN:找出沒有資料的場域

rows = connection.execute("""
    SELECT s.site_code, s.display_name
    FROM sites AS s
    LEFT JOIN sensors AS e ON e.site_id = s.site_id
    LEFT JOIN readings AS r ON r.sensor_id = e.sensor_id
    GROUP BY s.site_id, s.site_code, s.display_name
    HAVING COUNT(r.reading_id) = 0
    ORDER BY s.site_code
""").fetchall()

for row in rows:
    print(row["site_code"], row["display_name"])

若改用INNER JOIN,完全沒有感測器或讀值的場域會消失,反而看不見最需要確認的缺漏。

十二、子查詢:取得每支感測器的最新讀值

rows = connection.execute("""
    SELECT e.sensor_code, r.observed_at, r.value, r.quality
    FROM sensors AS e
    JOIN readings AS r ON r.sensor_id = e.sensor_id
    WHERE r.reading_id = (
        SELECT r2.reading_id
        FROM readings AS r2
        WHERE r2.sensor_id = e.sensor_id
        ORDER BY r2.observed_at DESC, r2.reading_id DESC
        LIMIT 1
    )
    ORDER BY e.sensor_code
""").fetchall()

for row in rows:
    print(dict(row))

用reading_id作第二排序條件,可在時間相同時得到穩定結果。若資料量很大,可再學習視窗函式與複合索引。

十三、綜合應用:產生優先巡查清單

def patrol_list(connection, low, high):
    return connection.execute("""
        SELECT s.site_code,
               s.display_name,
               e.sensor_code,
               r.observed_at,
               r.value,
               r.quality,
               CASE
                 WHEN r.quality != 'valid' THEN 1
                 WHEN r.value < ? OR r.value > ? THEN 2
                 ELSE 3
               END AS priority
        FROM readings AS r
        JOIN sensors AS e ON e.sensor_id = r.sensor_id
        JOIN sites AS s ON s.site_id = e.site_id
        WHERE e.kind = 'temperature'
          AND r.observed_at = (
              SELECT MAX(r2.observed_at)
              FROM readings AS r2
              WHERE r2.sensor_id = e.sensor_id
          )
        ORDER BY priority, s.site_code
    """, (low, high)).fetchall()


for item in patrol_list(connection, 26.0, 30.0):
    print(dict(item))
結果怎麼讀?優先級1表示資料未知或品質待確認,應先檢查供電、網路、探頭與校正;優先級2才是有效數值超出教材門檻。兩者需要不同的巡查行動。

十四、安全地讓使用者選擇排序

SQL參數只能替代「資料值」,不能直接替代欄名。動態排序應使用固定允許清單。

ALLOWED_SORTS = {
    "time": "r.observed_at DESC",
    "value": "r.value DESC",
    "site": "s.site_code ASC",
}


def query_readings(connection, sort_name="time"):
    order_sql = ALLOWED_SORTS.get(sort_name)
    if order_sql is None:
        raise ValueError("不支援的排序方式")

    sql = f"""
        SELECT s.site_code, r.observed_at, r.value
        FROM readings AS r
        JOIN sensors AS e ON e.sensor_id = r.sensor_id
        JOIN sites AS s ON s.site_id = e.site_id
        WHERE r.quality = ?
        ORDER BY {order_sql}
    """
    return connection.execute(sql, ("valid",)).fetchall()


print([dict(row) for row in query_readings(connection, "site")])

這裡的f-string只插入程式內已核准的SQL片段;外部輸入仍不能直接放進SQL。

十五、看懂查詢計畫與索引

connection.execute("""
    CREATE INDEX idx_readings_sensor_time
    ON readings(sensor_id, observed_at DESC)
""")

plan = connection.execute("""
    EXPLAIN QUERY PLAN
    SELECT reading_id, value
    FROM readings
    WHERE sensor_id = ?
    ORDER BY observed_at DESC
    LIMIT 1
""", (1,)).fetchall()

for step in plan:
    print(tuple(step))

索引應依常用查詢建立,不是越多越好;每個索引都會占空間,也增加新增與更新資料的成本。

十六、課堂挑戰

挑戰A|基礎:查出所有 quality='unknown' 的紀錄,依時間由新到舊排列。
挑戰B|進階:列出每座場域的感測器數量,沒有設備的場域也必須顯示為0。
挑戰C|USR場域:和養殖者共同定義「設備待確認、數值待確認、正常追蹤」三類,不記錄個資,並讓規則附上來源、適用物種與版本日期。

十七、用AI協助,但要自己驗證

請擔任SQL學習助教。資料表為sites、sensors、readings。
請先用一句話說明查詢邏輯,再提供SQLite相容SQL。
所有外部資料值必須參數化;不得虛構欄位。
我的問題是:找出每座場域最新一筆有效水溫,
沒有有效資料的場域也要顯示。
最後請提供兩筆可驗證LEFT JOIN行為的最小測試資料。

收到AI產生的SQL後,至少檢查欄名、JOIN條件、NULL處理、重複列、參數化及測試資料;能執行不代表答案一定正確。