昨天，我們打開韓國急診檢傷與急迫度分級量表（Korean Triage and Acuity Scale, KTAS）的 Kaggle 公開資料，確認了 1,267 筆紀錄、24 個欄位、兩種人工級數，以及哪些事後欄位不能交給模型。

但「我曾經成功讀到資料」和「別人能重建完全相同的資料」是兩件事。

假設三位讀者都說自己下載了同一份 data.csv：第一位使用分號切欄，得到 24 欄；第二位使用預設逗號，只得到 1 欄；第三位下載時 Kaggle 上的檔案已更新，卻沒有發現內容不同。三個檔名看起來一樣，實際上已經不是同一份實驗輸入。

所以今天不會靠一連串手動指令與記憶操作。我們要建立一條資料管線（Data Pipeline）：讓程式依固定順序取得來源、驗證內容、解壓縮、解析、正規化、依欄位角色拆檔，最後留下可核對的產物清單。

下圖先用工作台呈現這個概念。請觀察同一個來源資料包，如何依序通過多道檢查，而不是下載後直接進入模型。

上圖的重點不是把流程畫得很長，而是讓每一關都有「通過條件」。只要來源雜湊、壓縮檔內容、欄位名稱或資料筆數不符合預期，流程就應該停止，而不是勉強產生一份看似正常的結果。

今天要完成什麼

讀完並實際操作後，你應該能夠：

1. 從公開 Kaggle 資料集識別字串（Dataset Slug）取得本系列鎖定的 KTAS 資料版本。
2. 分別驗證壓縮檔與解壓後 data.csv 的內容。
3. 說明為什麼原始層、處理層與模型層不能混在同一個目錄。
4. 用一個命令完成下載、驗證、安全解壓縮與欄位角色拆分。
5. 閱讀來源與處理產物清單，知道本次執行使用哪些輸入、程式與相依套件。
6. 從空目錄重跑流程，確認 1,267 × 24 的來源資料會產生 1,267 × 15 的候選輸入檔。
7. 確認逐筆急診資料沒有被提交到公開程式碼儲存庫。

本篇會用到的名詞

詞彙表提供快速索引，下面會再說明每個概念如何落到路徑、指令與驗證結果。

先定義「可重跑」到底代表什麼

資料管線不是把多行指令包成一行而已。這條管線至少要同時回答五個問題：

1. 來源是哪一個？ 必須固定 Kaggle dataset slug，不能只記得搜尋關鍵字。
2. 內容是哪一版？ 必須驗證壓縮檔與解壓後 CSV 的 SHA-256。
3. 怎麼解讀內容？ 必須明確指定分隔符、文字編碼、缺失代碼與小數點正規化。
4. 哪些欄位能去哪裡？ 必須套用 Day 11 建立的欄位角色契約，避免標籤與事後結果混入輸入。
5. 最後產生了什麼？ 必須保存檔案尺寸、內容雜湊、程式版本與可觀察結果。

本篇把「可重跑」定義為：在相同 Kaggle 檔案、資料卡、欄位政策、處理程式與相依套件下，流程能從空白資料目錄建立相同的核心輸出。

這個定義沒有宣稱任何電腦、任何未鎖定套件版本都一定產生完全相同的位元。Day 13 會再處理環境鎖定與整個 repository 的執行契約；今天先把資料來源與資料轉換鎖住。

SHA-256 鎖定內容，不代表內容一定正確

安全雜湊演算法 256 位元版本（Secure Hash Algorithm 256-bit, SHA-256）會依檔案內容產生 64 個十六進位字元。內容只要改變，檢查碼通常也會改變。

本系列目前鎖定：

Kaggle ZIP
f78df0a06c31b6df873c410b0596ecb95ac9335485beaf890dcabf7c69560f26

data.csv
0e2c088e358fd4cdfd0dcc2fd4c2f085aa303856e3a6df8ceb44a69cef8ed2de

SHA-256 能回答「這次檔案是否和已稽核版本相同」，不能回答資料欄位是否適合臨床使用、標籤是否完美或來源是否代表整體族群。後三個問題仍要依資料卡、欄位政策與來源研究判斷。

Dataset slug、資料卡與 manifest 各有不同責任

資料集識別字串（Dataset Slug）是 Kaggle 網址中的 ilkeryildiz/emergency-service-triage-application。它告訴程式要去哪一個資料集頁面，不足以單獨鎖定某次下載內容。

本篇稍後會帶你建立完整的 Kaggle KTAS 資料卡 configs/data/day-12-kaggle-ktas-dataset-card.json。它會保存在專案設定資料夾中，內容包括：

• 來源頁面、dataset slug 與出處文章。
• ZIP 與 CSV 的檔名、SHA-256、筆數、欄數、分隔符與文字編碼。
• 適合用途、不適合用途與已知限制。
• 主要參考標籤、人類現場參考與五級數字方向。
• 逐筆資料不得重新提交到 Git 的儲存政策。

執行產物清單（Preparation Manifest）則是某一次本機處理留下的機器可讀證據。它會記錄實際讀到的來源雜湊、政策與程式雜湊、Python 與 Pandas 版本、輸出尺寸，以及每個核心產物的 SHA-256。

一句話區分：資料卡描述這份資料應該是什麼；manifest 記錄這次程式實際做出了什麼。

一個命令要經過哪些關卡

在繼續之前，要先解釋指令中的 scripts/ 到底是什麼。

scripts/ 是這個專案存放自動化程式的資料夾

腳本（Script）是一份可以交給直譯器依序執行的程式。這個專案把資料下載、驗證與轉換等自動化程式集中放在 scripts/ 資料夾；它不是需要另外安裝的 Python 套件，也不是網路服務。

以後面會執行的指令為例：

python scripts/run_ktas_pipeline.py

這行指令可以拆成三部分理解：

1. python：啟動 Python 直譯器，也就是實際讀取並執行程式的工具。
2. scripts/：從專案根目錄往下找到存放自動化程式的資料夾。
3. run_ktas_pipeline.py：要執行的 Python 原始碼檔；.py 是 Python 程式的副檔名。

因此，這行指令的完整意思是：「請 Python 執行 scripts/ 資料夾裡的 run_ktas_pipeline.py。」後面的完整程式碼小節會先帶你建立這支程式；只要目前位於專案根目錄，Python 就能依相對路徑找到它。

本篇的四個檔案各自負責什麼

資料管線使用兩份設定檔與兩支 Python 程式。設定檔負責保存決策，Python 程式負責執行動作：

專案根目錄/
├── configs/
│   └── data/
│       ├── day-11-ktas-field-policy.json
│       └── day-12-kaggle-ktas-dataset-card.json
└── scripts/
    ├── run_ktas_pipeline.py
    └── prepare_ktas.py

configs/ 是組態設定資料夾。組態（Configuration）指的是程式執行時要遵守、但不需要寫死在程式內的決策，例如來源雜湊、欄位角色與輸出限制。這兩份設定檔使用 JavaScript 物件表示法（JavaScript Object Notation, JSON）保存，因此可以直接用文字閱讀，也能由 Python 載入。

四個檔案的責任如下：

run_ktas_pipeline.py 是最外層入口，負責安排工作順序。它本身不猜測欄位用途，而是依序完成：

1. 讀取資料卡，取得 Kaggle 下載位置與預期雜湊。
2. 讀取欄位政策，確認後續資料轉換應遵守的角色規則。
3. 下載或接收 ZIP，驗證壓縮檔內容。
4. 安全解壓出 data.csv，再驗證 CSV 雜湊。
5. 把已驗證的 CSV 路徑與欄位政策交給 prepare_ktas.py。
6. 等 prepare_ktas.py 完成拆檔後，計算輸出雜湊並建立 preparation manifest。

prepare_ktas.py 的責任比較單純：它不連上 Kaggle，也不負責下載。它只處理已經驗證過的 data.csv，把來源表格轉成候選輸入、稽核欄位、參考標籤、禁止欄位清單與摘要。這樣拆成兩支程式，是為了讓「來源取得與版本驗證」和「表格內容轉換」能分開測試；其中一邊出錯時，也能更快判斷問題發生在哪一層。

讀到這裡，應該已經知道每個檔案的角色。Day 11 已完整提供 prepare_ktas.py 與欄位政策；本篇下一個實作小節會完整提供資料卡與 run_ktas_pipeline.py，讀者不需要前往其他程式碼網站尋找內容。

下圖把一個命令背後的七道關卡展開。閱讀方向先由左到右走完上排，再從第四關往下，最後由右向左走到 manifest。

上圖顯示，ZIP 與 CSV 需要分開驗證。前者能確認整個下載包相同，後者能確認真正交給表格處理程式的內容相同。即使未來壓縮方式改變但 CSV 沒變，也不能靜默略過差異；我們要先記錄版本變化，再決定是否更新資料卡。

安全解壓縮也不是多餘的步驟。程式只接受 ZIP 中剛好有一個 data.csv，而且檔名不能帶有上層路徑。這能同時偵測來源結構改變，並避免壓縮檔把內容寫到預期目錄之外。

先在自己的專案資料夾建立本篇完整檔案

接下來不會要求你前往任何程式碼網站。請在自己的電腦開啟專案資料夾，依下列順序建立檔案；每個程式碼區塊都是該檔案的完整內容，不含省略號。

本篇沿用 Day 11 已完整建立的欄位政策與 prepare_ktas.py；現在新增資料卡與最外層管線入口。

先從專案根目錄建立需要的資料夾：

mkdir -p configs/data scripts

如果指令沒有印出訊息是正常的。可用 test -d 資料夾路徑 && echo "資料夾已建立" 驗證單一資料夾。接著使用你熟悉的文字編輯器新增各檔案，把對應區塊完整貼入後儲存。

檔案 1：建立 configs/data/day-12-kaggle-ktas-dataset-card.json

鎖定 Kaggle 來源、ZIP 與 CSV 雜湊、格式、資料規模及使用限制。

請在文字編輯器建立 configs/data/day-12-kaggle-ktas-dataset-card.json，貼入以下完整內容並儲存：

{
  "schema_version": "0.1",
  "status": "source-audited-pipeline-ready",
  "title": "Kaggle KTAS 公開急診資料卡",
  "dataset_id": "kaggle_ktas_emergency_service_triage",
  "source": {
    "display_name": "Emergency Service - Triage Application",
    "kaggle_slug": "ilkeryildiz/emergency-service-triage-application",
    "kaggle_url": "https://www.kaggle.com/datasets/ilkeryildiz/emergency-service-triage-application",
    "download_url": "https://www.kaggle.com/api/v1/datasets/download/ilkeryildiz/emergency-service-triage-application",
    "provenance_article": "https://doi.org/10.1371/journal.pone.0216972",
    "checked_on": "2026-08-02"
  },
  "version_lock": {
    "archive_filename": "ktas-kaggle.zip",
    "archive_sha256": "f78df0a06c31b6df873c410b0596ecb95ac9335485beaf890dcabf7c69560f26",
    "data_filename": "data.csv",
    "data_sha256": "0e2c088e358fd4cdfd0dcc2fd4c2f085aa303856e3a6df8ceb44a69cef8ed2de",
    "record_count": 1267,
    "field_count": 24,
    "delimiter": ";",
    "encoding": "cp1254"
  },
  "intended_use": [
    "建立五級 KTAS 離線分類與檢索增強生成實驗",
    "比較模型預測、護理師登錄級數與專家重新判定級數",
    "分析檢傷不足、檢傷過度、跨級距離與各級召回率"
  ],
  "not_intended_use": [
    "直接支援真實臨床決策或取代檢傷護理師",
    "把 KTAS 結果改稱 TTAS 或 ESI 效能",
    "使用急診診斷、去向、停留時間或衍生誤差欄位預測檢傷級數",
    "宣稱資料代表韓國或其他地區所有急診病患"
  ],
  "label_policy": {
    "primary_reference_label": "KTAS_expert",
    "human_reference": "KTAS_RN",
    "label_direction": "1 最緊急，5 最不緊急"
  },
  "known_limits": [
    "資料只有 1,267 筆紀錄，專家第一級只有 26 筆",
    "資料沒有可用的病患識別碼，無法保證病患層級切分",
    "資料來自兩所都市教學醫院與隨機選出的 20 天",
    "Kaggle CSV 不是完整 KTAS 規則手冊"
  ],
  "storage_policy": "逐筆原始與處理後資料只保存在不納入版本控制的 data/ 目錄；專案的版本控制只保存來源、雜湊、欄位政策、程式與聚合摘要。"
}

儲存後先確認檔名與相對路徑完全一致，再繼續建立下一個檔案。

檔案 2：建立 scripts/run_ktas_pipeline.py

負責下載或接收 ZIP、安全解壓、驗證雜湊、呼叫 Day 11 轉換程式並建立 manifest。

請在文字編輯器建立 scripts/run_ktas_pipeline.py，貼入以下完整內容並儲存：

#!/usr/bin/env python3
"""從 Kaggle ZIP 建立可追溯的 KTAS 本機資料產物。

預設會從公開 Kaggle 下載端點取得資料；也可用 ``--archive`` 指定已下載的
ZIP，方便離線驗證。逐筆資料只會寫入被 Git 忽略的 data/ 目錄。
"""

from __future__ import annotations

import argparse
import hashlib
from importlib.metadata import version
import json
import os
from pathlib import Path
import platform
import shutil
import subprocess
import sys
from typing import Any
from urllib.request import Request, urlopen
import zipfile


ROOT = Path(__file__).resolve().parents[1]
POLICY_PATH = ROOT / "configs" / "data" / "day-11-ktas-field-policy.json"
DATASET_CARD_PATH = (
    ROOT / "configs" / "data" / "day-12-kaggle-ktas-dataset-card.json"
)
PREPARE_SCRIPT = ROOT / "scripts" / "prepare_ktas.py"


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="下載、驗證、解壓縮並準備 Kaggle KTAS 公開資料。"
    )
    parser.add_argument(
        "--archive",
        type=Path,
        help="選用：使用既有 Kaggle ZIP；省略時從資料卡的公開端點下載。",
    )
    parser.add_argument(
        "--raw-dir",
        type=Path,
        default=ROOT / "data" / "raw" / "ktas",
        help="不可變來源層；預設為 data/raw/ktas。",
    )
    parser.add_argument(
        "--output-dir",
        type=Path,
        default=ROOT / "data" / "interim" / "ktas-v1",
        help="角色拆分與正規化輸出；預設為 data/interim/ktas-v1。",
    )
    parser.add_argument("--policy", type=Path, default=POLICY_PATH)
    parser.add_argument("--dataset-card", type=Path, default=DATASET_CARD_PATH)
    return parser.parse_args()


def read_json(path: Path) -> dict[str, Any]:
    with path.open("r", encoding="utf-8") as file:
        return json.load(file)


def write_json(path: Path, payload: dict[str, Any]) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    temporary = path.with_suffix(path.suffix + ".tmp")
    with temporary.open("w", encoding="utf-8") as file:
        json.dump(payload, file, ensure_ascii=False, indent=2)
        file.write("\n")
    os.replace(temporary, path)


def sha256(path: Path) -> str:
    digest = hashlib.sha256()
    with path.open("rb") as file:
        for chunk in iter(lambda: file.read(1024 * 1024), b""):
            digest.update(chunk)
    return digest.hexdigest()


def verify_hash(path: Path, expected: str, label: str) -> None:
    if not path.is_file():
        raise FileNotFoundError(f"找不到{label}：{path}")
    observed = sha256(path)
    if observed != expected:
        raise ValueError(
            f"{label}的 SHA-256 不符。\n預期：{expected}\n實際：{observed}\n"
            "流程已停止；請確認來源版本，不要直接改掉預期雜湊。"
        )


def download(url: str, destination: Path) -> None:
    destination.parent.mkdir(parents=True, exist_ok=True)
    temporary = destination.with_suffix(destination.suffix + ".part")
    request = Request(url, headers={"User-Agent": "ktas-sideproject30/0.1"})
    try:
        with urlopen(request, timeout=120) as response, temporary.open("wb") as file:
            shutil.copyfileobj(response, file)
        os.replace(temporary, destination)
    except Exception:
        temporary.unlink(missing_ok=True)
        raise


def materialize_archive(source: Path | None, target: Path, download_url: str) -> str:
    if source is not None:
        source = source.expanduser().resolve()
        if not source.is_file():
            raise FileNotFoundError(f"找不到 --archive 指定的檔案：{source}")
        target.parent.mkdir(parents=True, exist_ok=True)
        if source != target.resolve():
            if target.exists() and sha256(target) != sha256(source):
                raise FileExistsError(
                    f"{target} 已存在但內容不同。請改用空目錄，流程不會覆寫來源層。"
                )
            if not target.exists():
                shutil.copyfile(source, target)
        return "local-archive"

    if not target.exists():
        download(download_url, target)
        return "kaggle-download"
    return "verified-cache"


def safe_extract(zip_path: Path, data_path: Path, expected_member: str) -> None:
    temporary = data_path.with_suffix(".csv.tmp")
    with zipfile.ZipFile(zip_path) as archive:
        files = [member for member in archive.infolist() if not member.is_dir()]
        names = [member.filename for member in files]
        if names != [expected_member]:
            raise ValueError(
                f"ZIP 內容與鎖定版本不同；預期只有 {expected_member}，實際為 {names}"
            )
        member = files[0]
        if Path(member.filename).name != member.filename:
            raise ValueError("ZIP 成員包含路徑，為避免路徑穿越，流程已停止。")
        try:
            with archive.open(member) as source, temporary.open("wb") as target:
                shutil.copyfileobj(source, target)
            os.replace(temporary, data_path)
        except Exception:
            temporary.unlink(missing_ok=True)
            raise


def project_path(path: Path) -> str:
    resolved = path.resolve()
    try:
        return str(resolved.relative_to(ROOT))
    except ValueError:
        return str(resolved)


def run_prepare(data_path: Path, output_dir: Path, policy_path: Path) -> None:
    subprocess.run(
        [
            sys.executable,
            str(PREPARE_SCRIPT),
            "--input",
            str(data_path),
            "--output-dir",
            str(output_dir),
            "--policy",
            str(policy_path),
        ],
        cwd=ROOT,
        check=True,
    )


def main() -> None:
    args = parse_args()
    policy = read_json(args.policy)
    card = read_json(args.dataset_card)
    source = policy["public_source"]
    locked = card["version_lock"]

    if source["archive_sha256"] != locked["archive_sha256"]:
        raise ValueError("Day 11 欄位政策與 Day 12 資料卡的 ZIP 雜湊不一致。")
    if source["data_sha256"] != locked["data_sha256"]:
        raise ValueError("Day 11 欄位政策與 Day 12 資料卡的 CSV 雜湊不一致。")

    raw_dir = args.raw_dir.expanduser().resolve()
    output_dir = args.output_dir.expanduser().resolve()
    raw_dir.mkdir(parents=True, exist_ok=True)
    archive_path = raw_dir / locked["archive_filename"]
    data_path = raw_dir / locked["data_filename"]

    acquisition = materialize_archive(
        args.archive,
        archive_path,
        card["source"]["download_url"],
    )
    verify_hash(archive_path, locked["archive_sha256"], "Kaggle ZIP")

    if data_path.exists():
        verify_hash(data_path, locked["data_sha256"], "Kaggle data.csv")
    else:
        safe_extract(archive_path, data_path, locked["data_filename"])
        verify_hash(data_path, locked["data_sha256"], "Kaggle data.csv")

    source_manifest = {
        "status": "source-verified",
        "dataset_id": card["dataset_id"],
        "kaggle_slug": card["source"]["kaggle_slug"],
        "acquisition_method": acquisition,
        "archive": {
            "path": project_path(archive_path),
            "bytes": archive_path.stat().st_size,
            "sha256": sha256(archive_path),
        },
        "data_file": {
            "path": project_path(data_path),
            "bytes": data_path.stat().st_size,
            "sha256": sha256(data_path),
        },
        "note": "只有雜湊完全相同才視為本系列鎖定的來源版本。",
    }
    write_json(raw_dir / "source-manifest.json", source_manifest)

    run_prepare(data_path, output_dir, args.policy)
    summary = read_json(output_dir / "summary.json")
    input_field_count = sum(
        field["role"] == "model_input_candidate" for field in policy["fields"]
    )
    artifact_names = [
        "model-input-candidates.csv",
        "audit-strata.csv",
        "reference-labels.csv",
        "excluded-fields.json",
        "summary.json",
    ]
    artifacts = {
        name: {
            "bytes": (output_dir / name).stat().st_size,
            "sha256": sha256(output_dir / name),
        }
        for name in artifact_names
    }
    preparation_manifest = {
        "status": "preparation-complete",
        "dataset_id": card["dataset_id"],
        "source_manifest": project_path(raw_dir / "source-manifest.json"),
        "inputs": {
            "data_sha256": sha256(data_path),
            "policy_path": project_path(args.policy),
            "policy_sha256": sha256(args.policy),
            "dataset_card_path": project_path(args.dataset_card),
            "dataset_card_sha256": sha256(args.dataset_card),
        },
        "implementation": {
            "runner_path": project_path(Path(__file__)),
            "runner_sha256": sha256(Path(__file__)),
            "preparer_path": project_path(PREPARE_SCRIPT),
            "preparer_sha256": sha256(PREPARE_SCRIPT),
            "python": platform.python_version(),
            "pandas": version("pandas"),
        },
        "observed": {
            "record_count": summary["record_count"],
            "source_field_count": policy["observed_structure"]["field_count"],
            "model_input_shape": [summary["record_count"], input_field_count + 1],
            "expert_label_counts": summary["expert_label_counts"],
        },
        "artifacts": artifacts,
        "reproducibility_note": "時間戳不寫入 manifest；相同來源、政策與程式應得到相同產物雜湊。",
    }
    write_json(output_dir / "preparation-manifest.json", preparation_manifest)

    print("資料管線完成")
    print(f"來源層：{raw_dir}")
    print(f"處理層：{output_dir}")
    print(
        f"驗證：{summary['record_count']:,} 筆、"
        f"{policy['observed_structure']['field_count']} 個來源欄位、"
        f"{input_field_count + 1} 個候選輸入檔欄位"
    )


if __name__ == "__main__":
    main()

儲存後先確認檔名與相對路徑完全一致，再繼續建立下一個檔案。

完成後產生必要檔案並做靜態檢查

以下命令會在需要時產生套件鎖定檔，接著檢查設定格式與 Python 語法；它們不會啟動模型，也不會把病患資料送到網路：

python3 -m json.tool configs/data/day-12-kaggle-ktas-dataset-card.json
python3 -m py_compile scripts/run_ktas_pipeline.py scripts/prepare_ktas.py

每個命令都應正常結束。若 JSON 顯示行號，先檢查貼上時是否遺漏逗號、引號或括號；若 py_compile 報錯，先依行號修正縮排或漏貼內容。靜態檢查通過後，再執行本文後面的正式步驟。

從已完成 Day 11 的專案資料夾執行完整管線

以下每個步驟都在專案根目錄執行。指令不會把逐筆資料加入 Git；但完成後仍要用 git status 驗證 .gitignore 是否生效。

步驟 1：準備隔離的 Python 環境

• 目的：讓資料處理套件不與系統 Python 混用。
• 前置條件：電腦已安裝 Python 3，終端機位於自己的專案根目錄，而且已完成前一節的完整檔案建立。
• 輸入：本步驟沒有病患資料輸入。
• 操作：建立虛擬環境、啟用環境並安裝 Pandas。Pandas 是 Python 的表格處理套件，本篇用它讀取與輸出 CSV。

python3 -m venv .venv
source .venv/bin/activate
python -m pip install 'pandas>=2.0,<3'

• 預期輸出：安裝過程沒有錯誤，命令提示字元通常會出現 (.venv)。
• 驗證方式：執行下列指令，應印出 Python 與 Pandas 的實際版本。

python -c "import platform, pandas; print(platform.python_version(), pandas.__version__)"

這裡先限制 Pandas 的主要版本範圍，實際版本會寫入 preparation manifest。Day 13 會再把相依套件鎖定方式納入整個 repository 的可重現設計。

步驟 2：執行一鍵資料管線

• 目的：從 Kaggle 公開來源建立經過版本驗證的原始層與處理層。
• 前置條件：虛擬環境已啟用，電腦可連到 Kaggle；data/raw/ktas/ 尚未放入內容不同的同名檔案。
• 輸入：Day 12 資料卡、Day 11 欄位政策與 Kaggle 公開下載端點。
• 操作：執行資料管線入口。

python scripts/run_ktas_pipeline.py

• 預期輸出：終端機最後應看到：

完成：1267 筆、24 欄
模型輸入不含護理師級數、專家級數、事後結果或衍生誤差欄位。
資料管線完成
來源層：.../data/raw/ktas
處理層：.../data/interim/ktas-v1
驗證：1,267 筆、24 個來源欄位、15 個候選輸入檔欄位

• 驗證方式：data/raw/ktas/ 應有 ZIP、CSV 與來源 manifest；data/interim/ktas-v1/ 應有五個資料／摘要檔與一份 preparation manifest。

公開下載端點在本系列 2026 年 8 月 2 日稽核時可直接取得。若 Kaggle 後續調整下載流程，可以從公開資料頁手動取得同一個 ZIP，再改用：

python scripts/run_ktas_pipeline.py \
  --archive /你的下載目錄/ktas-kaggle.zip

--archive 只改變取得方式，不會降低驗證標準。指定檔案仍要通過完全相同的 ZIP 與 CSV SHA-256。這份 Kaggle 公開資料不要求簽署醫療資料使用協議；Kaggle 平台本身是否調整帳號或下載介面，則要以資料頁當下規則為準。

步驟 3：讀取來源 manifest

• 目的：確認原始層實際保存的是哪一個來源版本。
• 前置條件：步驟 2 已完成。
• 輸入：data/raw/ktas/source-manifest.json。
• 操作：用 Python 內建的 JSON 格式化工具閱讀檔案。

python -m json.tool data/raw/ktas/source-manifest.json

• 預期輸出：內容至少包含資料集識別碼、Kaggle slug、取得方式、ZIP 與 CSV 的路徑、位元組數及 SHA-256。
• 驗證方式：status 必須是 source-verified，兩個雜湊必須和資料卡一致。

source-manifest.json 不會把「下載完成」當成「驗證完成」。網路下載即使成功，也可能拿到更新版資料、錯誤頁面或不完整檔案；只有內容雜湊相同，狀態才會進入 source-verified。

三個資料層要解決不同問題

原始資料層（Raw Data Layer）保留 Kaggle 來源位元；中介資料層（Interim Data Layer）保存缺失代碼正規化與欄位角色拆分；模型資料層（Processed Data Layer）才會保存切分、補值與衍生特徵後的版本。

下圖請先看每層下方的責任，再看檔名。橘色的模型層目前刻意標成「尚未建立」。

上圖顯示，Day 12 的輸出仍叫「候選輸入」，不是「最終模型資料」。我們尚未決定缺失處理、交叉驗證折數、隨機種子與主訴文字正規化方式。若現在就建立訓練與測試檔，很容易在規則尚未鎖定前反覆調整測試集合。

原始層：只保存與驗證，不直接修改

data/raw/ktas/ 會包含：

data/raw/ktas/
├── ktas-kaggle.zip
├── data.csv
└── source-manifest.json

原始層的原則是不可變（Immutable）：data.csv 若需要修正空白、缺失代碼或小數點，應把結果寫到另一層，不能直接存回原檔。否則來源 SHA-256 會改變，也無法判斷差異來自 Kaggle 還是我們的手動編輯。

處理層：依角色拆開，不保留一張方便但危險的大表

data/interim/ktas-v1/ 會包含六個檔案：

處理層不另外輸出急診診斷、去向、急診停留時間與檢傷作業時間。它們仍存在原始 CSV，未來若要做事後結果分析，必須建立另一條明確用途的流程；現在先避免「因為方便」而讓事後資訊靠近模型輸入。

模型層：等待切分契約，不提前建立

模型資料層要等 Day 15 才建立，因為資料只有 1,267 筆，專家第一級只有 26 筆，而且沒有可用的病患識別碼。切分方式會直接影響每一級的樣本量與結論不確定性。

下一階段至少要先固定：

1. 訓練、驗證與測試索引如何產生。
2. 分層時使用哪一個標籤，以及少數級別如何處理。
3. 缺失值只用訓練資料估計，不能偷看驗證或測試分布。
4. 所有模型是否共用同一組切分。
5. 沒有病患識別碼造成的相依性限制如何揭露。

所以「尚未建立」不是缺少進度，而是防止資料洩漏的主動邊界。

解析與正規化規則不能交給套件猜

逗號分隔值（Comma-Separated Values, CSV）是以純文字保存表格的格式，但副檔名並不保證分隔符真的為逗號。這份 data.csv 實際使用分號 ; 切欄，文字編碼是 Windows-1254，也就是 Python 的 cp1254。

scripts/prepare_ktas.py 明確指定：

pd.read_csv(
    data_path,
    sep=";",
    encoding="cp1254",
    dtype=object,
    keep_default_na=False,
)

先以文字讀取，是為了在數值轉型前看見來源真正使用的缺失代碼。接著程式依下列順序處理：

1. 去除每個文字欄位前後的多餘空白。
2. 只在應為數值或序位標籤的欄位中，辨識空字串、?? 與 #BOŞ!。
3. 把已知缺失代碼轉成真正的缺失值，不補成 0。
4. 把數值中的逗號小數點轉成句點，例如 5,00 轉成 5.00。
5. 嘗試轉成數值；若出現政策未登錄的非數值字串，立刻停止。
6. 驗證 KTAS_RN 與 KTAS_expert 都沒有缺失，且只落在第一到第五級。
7. 依欄位角色輸出候選輸入、稽核與標籤檔。

上一步的輸出會自然成為下一步的輸入：來源文字先完成缺失辨識，才能安全做數值轉型；數值與標籤通過驗證後，才能依角色拆檔；拆檔完成後，資料管線入口程式才會計算產物雜湊並建立 manifest。

用 manifest 核對這次執行

JavaScript 物件表示法（JavaScript Object Notation, JSON）是一種能同時讓人與程式閱讀的結構化文字格式。本篇用 JSON 保存資料卡、欄位政策與兩份 manifest，避免把關鍵版本只寫在終端機畫面或作者記憶裡。

步驟 4：閱讀 preparation manifest

• 目的：追溯處理後檔案使用的來源、規則、程式與執行環境。
• 前置條件：完整資料管線已成功執行。
• 輸入：data/interim/ktas-v1/preparation-manifest.json。
• 操作：格式化並閱讀 manifest。

python -m json.tool data/interim/ktas-v1/preparation-manifest.json

• 預期輸出：頂層至少有 status、inputs、implementation、observed 與 artifacts。
• 驗證方式：逐項確認：

manifest 刻意不寫入現在時間。這讓相同輸入、程式與環境重跑時，manifest 本身也能保持相同內容。若要另外記錄下載或執行時間，可以放在未參與核心產物比較的操作日誌中；不要讓每次都變動的時間戳破壞冪等性檢查。

步驟 5：自動檢查尺寸與禁止欄位

• 目的：不用人工閱讀 1,267 列，也能驗證候選輸入沒有答案或事後結果。
• 前置條件：Pandas 已安裝，處理層六個檔案都存在。
• 輸入：候選輸入、排除清單與 preparation manifest。
• 操作：執行下列唯讀檢查。

python - <<'PY'
import json
from pathlib import Path

import pandas as pd

root = Path("data/interim/ktas-v1")
inputs = pd.read_csv(root / "model-input-candidates.csv")
excluded = json.loads((root / "excluded-fields.json").read_text(encoding="utf-8"))
manifest = json.loads((root / "preparation-manifest.json").read_text(encoding="utf-8"))

forbidden = {
    field
    for role, fields in excluded.items()
    if role != "note"
    for field in fields
}
leaked = forbidden.intersection(inputs.columns)

if inputs.shape != (1267, 15):
    raise SystemExit(f"候選輸入尺寸錯誤：{inputs.shape}")
if manifest["status"] != "preparation-complete":
    raise SystemExit(f"manifest 狀態錯誤：{manifest['status']}")
if leaked:
    raise SystemExit(f"發現禁止欄位：{sorted(leaked)}")

print("通過：1,267 × 15，manifest 完整，未發現禁止欄位")
PY

• 預期輸出：通過：1,267 × 15，manifest 完整，未發現禁止欄位。
• 驗證方式：若看到尺寸或洩漏錯誤，先停止後續建模，檢查 Day 11 欄位政策與 scripts/prepare_ktas.py；不能只在模型程式中暫時忽略多出的欄位。

下圖把一次成功執行的最低驗證項目放在同一個畫面。請注意，這些是來源與管線結果，不是模型準確率。

上圖顯示「沒有報錯」只是第一層條件。來源內容、資料尺寸、缺失規則、輸出角色與 manifest 都一致，才能說本次資料準備通過契約。

用第二次執行檢查冪等性

冪等性（Idempotency）在本篇代表：相同的來源、資料卡、欄位政策、程式與環境重跑後，五個核心資料／摘要檔與 preparation manifest 的內容雜湊不變。

步驟 6：記錄雜湊、重跑，再比較

• 目的：驗證流程不是依賴 Notebook 儲存格順序、人工編輯或上一次記憶狀態。
• 前置條件：步驟 2 至 5 已通過，原始檔沒有被修改。
• 輸入：目前的處理層產物與同一份來源。
• 操作：先記錄雜湊，再執行相同命令，最後比較。

shasum -a 256 data/interim/ktas-v1/*

python scripts/run_ktas_pipeline.py

shasum -a 256 data/interim/ktas-v1/*

• 預期輸出：兩次列出的六個檔案雜湊應完全相同。
• 驗證方式：若任何核心檔案不同，先比較 manifest 中的來源、政策、兩支程式、Python 與 Pandas 版本；不要只看檔名相同。

原子寫入（Atomic Write）則處理另一種失敗：如果下載或寫 manifest 的過程中斷，程式先清除 .part 或 .tmp 暫存檔，不會把半份內容冒充正式產物。這不能消除所有硬體與作業系統風險，但能避免最常見的中途失敗狀態。

若第一次執行是網路下載，source-manifest.json 的 acquisition_method 會記為 kaggle-download；使用已下載檔案時可能是 verified-cache 或 local-archive。取得方式可以不同，真正進入資料處理的 ZIP 與 CSV 雜湊仍必須相同。

步驟 7：確認逐筆資料沒有進入 Git

• 目的：確保專案的版本控制只保存重建方法，不重新散布逐筆急診資料。
• 前置條件：管線已在專案預設路徑執行。
• 輸入：Git 工作目錄狀態。
• 操作：執行：

git status --short

• 預期輸出：不應出現 data/raw/、data/interim/ 或 data/processed/ 內的檔案。
• 驗證方式：執行 git check-ignore data/raw/ktas/data.csv，應印出該路徑，表示 .gitignore 規則生效。

Kaggle 上可以公開取得，不代表我們就應該把逐筆資料重新鏡像到自己的專案。版本控制保存的是重建契約：來源頁、雜湊、資料卡、欄位政策、處理程式與聚合統計。

常見失敗與排查方式

ZIP 的 SHA-256 不符

可能原因包括下載不完整、Kaggle 更新資料、拿到其他檔案，或原始層曾被修改。管線會停止，不提供「忽略檢查」選項。先重新下載並核對 dataset slug；若 Kaggle 確實發布新版，應另存新版本、比較內容差異、重新做 Day 11 欄位稽核，再明確更新資料卡。

ZIP 通過，但 CSV 的 SHA-256 不符

這代表解壓後真正要讀取的內容不是鎖定版本。不要只因為筆數看起來一樣就繼續；相同列數仍可能有欄位值或文字不同。

讀到 1 欄而不是 24 欄

通常是把檔案交給使用逗號作預設分隔符的工具。修正方式不是手動另存成 Excel，而是明確使用 sep=";"，並重新驗證來源仍有 1,267 列、24 欄。

出現 Unicode 解碼錯誤

通常是用萬國碼轉換格式 8 位元（Unicode Transformation Format 8-bit, UTF-8）直接讀取 #BOŞ!。這份來源要使用 encoding="cp1254"。不要先用文字編輯器另存檔案，否則原始雜湊會改變。

缺失欄位轉型後出現奇怪數字

空字串、?? 與 #BOŞ! 都應先轉成缺失，再做數值解析。它們不等於 0，也不能直接以整體平均值取代。Day 15 才會在固定切分內比較缺失處理策略。

重跑後產物雜湊改變

依序比較來源雜湊、資料卡雜湊、欄位政策雜湊、管線入口與資料準備程式雜湊、Python 版本及 Pandas 版本。這些項目正是 manifest 存在的理由；少記任何一項，都可能只剩下猜測。

把 interim 當成最終模型輸入

model-input-candidates.csv 已排除答案與事後結果，但尚未完成切分、缺失處理與特徵正規化。它只能成為下一階段輸入，不能直接拿整份資料調參後再宣稱是測試結果。

因為管線通過，就宣稱資料適合臨床部署

管線通過只代表固定檔案與固定轉換符合工程契約。它不會增加資料代表性，也不會消除小樣本、病患識別碼缺失、回溯病歷與場域限制。系統仍只定位為離線決策支援實驗，不取代急診專業判斷。

本日小結

今天把 Day 11 的來源與欄位契約變成一條真正能執行的資料管線：

1. Kaggle dataset slug 負責定位來源，ZIP 與 CSV 的 SHA-256 負責鎖定實際內容。
2. Day 12 資料卡長期記錄資料用途、版本、限制與儲存政策；manifest 記錄本次執行實際使用的輸入、程式、環境與產物。
3. scripts/run_ktas_pipeline.py 依序完成下載、ZIP 驗證、安全解壓、CSV 驗證、格式解析、欄位契約與產物登錄。
4. 原始層保留來源位元，中介層保存正規化與角色拆分，模型層要等切分策略固定後才建立。
5. data.csv 必須用分號與 cp1254 解析；空字串、??、#BOŞ! 與逗號小數點都要用明確規則處理。
6. 管線會產生 1,267 × 15 的候選輸入，以及獨立的稽核、標籤、排除清單、摘要與 preparation manifest。
7. 相同輸入與環境重跑後，核心產物雜湊應保持相同；這才是本篇可觀察的冪等性。
8. 逐筆資料不進版本控制，專案只保存重建方法與聚合資訊。
9. 管線成功是工程驗證，不是模型結果，更不是臨床部署證據。

下一篇預告

Day 13 會把視角從資料管線拉到整個專案結構：

• data/、configs/、src/、tests/ 與 results/ 各自負責什麼。
• 為什麼一支 Notebook 無法獨自承擔重現責任。
• 如何讓每次模型執行追溯到資料版本、設定、提示詞、模型與隨機種子。
• 如何使用不含真實病歷的合成測試資料，驗證程式而不暴露逐筆紀錄。

Day 12 產生的資料卡、欄位政策、來源 manifest 與 preparation manifest，會成為 Day 13 設計專案追溯鏈的第一組實際案例。

參考資料

1. Kaggle. Emergency Service - Triage Application.
2. Moon, S.-H., Shim, J. L., Park, K.-S., & Park, C.-S. (2019). Triage accuracy and causes of mistriage using the Korean Triage and Acuity Scale. Public Library of Science (PLOS) ONE, 14(9), e0216972.
3. Python Software Foundation. zipfile — Work with ZIP archives.
4. Python Software Foundation. hashlib — Secure hashes and message digests.
5. Pandas documentation. pandas.read_csv.