2026年8月31日 星期一

Python一下:用CSV記錄七座養殖場巡查資料

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

用CSV記錄七座養殖場巡查資料

把一列列巡查紀錄變成可驗證、可交換、可追查的結構化資料。

CH10 CSVcsv模組智慧養殖
學習目標
完成本篇後,你能使用 csv.readerwriterDictReaderDictWriter,正確處理標題列、UTF-8、換行、逗號與引號;並完成巡查資料驗證、錯誤隔離、統計與安全匯出。

一、CSV不是「用逗號自己切」

CSV適合交換表格資料,但欄位本身可能含逗號、引號或換行,因此不應直接使用 line.split(",")。Python的 csv 模組會依CSV規則處理這些細節。

教學資料聲明:本篇使用「示範池A~G」匿名資料,數值與判斷僅供程式練習。真實場域資料可能涉及養殖戶營運、設備位置與生產資訊,公開前須取得同意並去識別;示意門檻不可直接用於設備控制。

二、用writer建立第一份CSV

import csv
from pathlib import Path

path = Path("七座場域_巡查資料.csv")
rows = [
    ["日期", "場域", "水溫", "pH", "備註"],
    ["2026-08-31", "示範池A", 27.2, 7.4, "正常"],
    ["2026-08-31", "示範池B", 28.1, 7.1, "待人工複核"],
]

with path.open("w", encoding="utf-8-sig", newline="") as file:
    writer = csv.writer(file)
    writer.writerows(rows)

print("已建立:", path)

newline="" 可避免某些平台出現多餘空白列。utf-8-sig 有助部分試算表軟體辨識UTF-8中文;純Python系統通常可使用 utf-8。讀寫雙方應先約定。

三、用reader逐列讀取

import csv
from pathlib import Path

path = Path("七座場域_巡查資料.csv")

with path.open("r", encoding="utf-8-sig", newline="") as file:
    reader = csv.reader(file)
    header = next(reader)
    print("欄位:", header)

    for row_number, row in enumerate(reader, start=2):
        print(row_number, row)

CSV讀出的內容預設都是字串,即使畫面看起來是數字,也必須明確轉換後才能計算。

四、DictWriter讓欄位更清楚

import csv
from pathlib import Path

fieldnames = ["日期", "場域", "水溫", "pH", "狀態"]
records = [
    {"日期": "2026-08-31", "場域": "示範池A",
     "水溫": 27.2, "pH": 7.4, "狀態": "已讀取"},
    {"日期": "2026-08-31", "場域": "示範池B",
     "水溫": 28.1, "pH": 7.1, "狀態": "已讀取"},
]

path = Path("巡查資料_字典版.csv")
with path.open("x", encoding="utf-8-sig", newline="") as file:
    writer = csv.DictWriter(file, fieldnames=fieldnames)
    writer.writeheader()
    writer.writerows(records)

使用 x 模式可避免不小心覆蓋既有紀錄。若確實要產生新版,請更換檔名並保留來源與版本資訊。

五、DictReader:用欄位名稱取資料

import csv
from pathlib import Path

path = Path("巡查資料_字典版.csv")

with path.open("r", encoding="utf-8-sig", newline="") as file:
    reader = csv.DictReader(file)
    print("標題列:", reader.fieldnames)

    for record in reader:
        temperature = float(record["水溫"])
        ph = float(record["pH"])
        print(record["場域"], temperature, ph)

六、先檢查必要欄位

def validate_header(fieldnames):
    required = {"日期", "場域", "水溫", "pH", "狀態"}
    actual = set(fieldnames or [])
    missing = required - actual
    if missing:
        names = "、".join(sorted(missing))
        raise ValueError(f"CSV缺少必要欄位:{names}")


validate_header(["日期", "場域", "水溫", "pH", "狀態"])
print("標題列檢查完成")

欄位檢查應在逐筆轉換之前完成,否則資料跑到一半才發現整份格式不相容。

七、驗證每一列並隔離錯誤

from datetime import date


def parse_record(record):
    patrol_date = date.fromisoformat(record["日期"].strip())
    site = record["場域"].strip()
    if not site:
        raise ValueError("場域不可空白")

    temperature = float(record["水溫"])
    ph = float(record["pH"])
    if not 0.0 <= ph <= 14.0:
        raise ValueError("pH超出理論範圍")

    return {
        "日期": patrol_date.isoformat(),
        "場域": site,
        "水溫": temperature,
        "pH": ph,
        "狀態": record["狀態"].strip(),
    }

這裡只檢查日期、數字格式及pH理論範圍,不判定養殖安全。場域管理門檻必須由養殖戶與專業人員共同設定。

八、完整匯入:單筆錯誤不中止整批

import csv
from pathlib import Path


def load_patrol_csv(path):
    valid_records = []
    errors = []

    with Path(path).open(
        "r", encoding="utf-8-sig", newline=""
    ) as file:
        reader = csv.DictReader(file)
        validate_header(reader.fieldnames)

        for row_number, record in enumerate(reader, start=2):
            try:
                valid_records.append(parse_record(record))
            except (KeyError, TypeError, ValueError) as error:
                errors.append({
                    "列號": row_number,
                    "原因": str(error),
                    "原始資料": dict(record),
                })

    return valid_records, errors
設計重點:成功資料與錯誤資料分開,錯誤紀錄保留列號與原因;但錯誤報告若含個資或敏感營運內容,不可直接公開。

九、建立七座匿名教學場域資料

demo_records = [
    {"日期": "2026-08-31", "場域": f"示範池{letter}",
     "水溫": 26.0 + index * 0.4,
     "pH": 7.0 + index * 0.1,
     "狀態": "教學資料"}
    for index, letter in enumerate("ABCDEFG")
]

for record in demo_records:
    print(record)

十、計算摘要,不把錯誤值混進去

def summarize(records):
    if not records:
        return {"筆數": 0, "平均水溫": None, "平均pH": None}

    temperature_average = sum(
        record["水溫"] for record in records
    ) / len(records)
    ph_average = sum(
        record["pH"] for record in records
    ) / len(records)

    return {
        "筆數": len(records),
        "平均水溫": round(temperature_average, 2),
        "平均pH": round(ph_average, 2),
    }


print(summarize(demo_records))

平均值只描述已通過格式驗證的教學資料,不代表整個場域狀態。報告應同時呈現錯誤筆數、缺漏情況與資料時間範圍。

十一、匯出前防範試算表公式注入

若CSV會由試算表開啟,來自外部的文字若以 =+-@ 開頭,可能被當成公式。可在匯出前標記為文字,並限制可接受欄位。

def safe_csv_text(value):
    text = str(value)
    if text.startswith(("=", "+", "-", "@")):
        return "'" + text
    return text


examples = ["正常", '=HYPERLINK("bad")', "@command"]
for value in examples:
    print(safe_csv_text(value))
注意:這是降低常見風險的教學措施,不是完整安全保證。匯入外部CSV仍應視為不可信資料,避免啟用巨集或未知連結。

十二、安全匯出驗證後資料

import csv
from pathlib import Path


def export_records(path, records):
    fieldnames = ["日期", "場域", "水溫", "pH", "狀態"]
    path = Path(path)
    path.parent.mkdir(parents=True, exist_ok=True)

    with path.open(
        "x", encoding="utf-8-sig", newline=""
    ) as file:
        writer = csv.DictWriter(file, fieldnames=fieldnames)
        writer.writeheader()
        for record in records:
            safe_record = {
                key: safe_csv_text(record[key])
                for key in fieldnames
            }
            writer.writerow(safe_record)


# export_records("輸出/七座場域_驗證版.csv", demo_records)

十三、常見錯誤檢查表

現象原因修正
CSV多出空白列未設定newline=""開啟CSV時加上newline=""
Excel中文亂碼編碼辨識不同交換規格約定UTF-8,必要時utf-8-sig
數字無法計算CSV讀出的是字串驗證後轉成float或int
含逗號備註被拆壞自行split(",")使用csv模組
一筆壞資料整批中止未做逐列例外隔離保留列號、原因並繼續
開啟後執行公式外部文字被試算表解讀驗證、標記文字並避免不可信檔案

十四、USR實作挑戰

挑戰A|七座場域巡查表
建立七座匿名場域各三筆教學資料,混入缺欄、錯誤日期及非數字讀值,輸出成功資料與錯誤清單。
挑戰B|資料品質報告
除平均值外,列出總筆數、有效筆數、錯誤筆數、日期範圍及各場域資料量。
挑戰C|公開版CSV
建立不含養殖戶姓名、精確位置及設備識別資訊的公開版,請場域合作夥伴確認欄位與授權範圍。

十五、與生成式AI協作

請擔任Python CSV與智慧養殖資料品質助教。
請檢查我的巡查CSV程式:
1. 使用csv模組,不可自行split逗號;
2. 明確處理UTF-8與newline="";
3. 先檢查標題列,再逐列驗證;
4. 單筆錯誤不能拖垮整批;
5. 無效值不可當成0或正常;
6. 匯出時防範試算表公式注入;
7. 公開資料不得包含個資、精確位置或設備憑證。
請提供測試資料與預期結果。
本篇小結
CSV看似簡單,可靠使用卻需要處理編碼、換行、引號、資料型別與錯誤列。先驗證、再計算、最後才匯出,才能讓場域資料真正可用。

沒有留言:

張貼留言