內容雜湊檔案快取模式
使用 SHA-256 內容雜湊作為快取鍵,快取昂貴的檔案處理結果(PDF 解析、文字萃取、圖片分析)。與基於路徑的快取不同,此方法在檔案移動或重新命名時仍能命中快取,並在內容變更時自動失效。
啟動時機
- 建構檔案處理管線(PDF、圖片、文字萃取)
- 處理成本高昂,且相同檔案會重複處理
- 需要提供
--cache/--no-cacheCLI 選項 - 希望為現有的純函式加入快取功能,而不修改函式本身
核心模式
1. 基於內容雜湊的快取鍵
使用檔案內容(而非路徑)作為快取鍵:
import hashlib
from pathlib import Path
_HASH_CHUNK_SIZE = 65536 # 大檔案以 64KB 區塊讀取
def compute_file_hash(path: Path) -> str:
"""計算檔案內容的 SHA-256 雜湊(針對大檔案進行區塊讀取)。"""
if not path.is_file():
raise FileNotFoundError(f"找不到檔案:{path}")
sha256 = hashlib.sha256()
with open(path, "rb") as f:
while True:
chunk = f.read(_HASH_CHUNK_SIZE)
if not chunk:
break
sha256.update(chunk)
return sha256.hexdigest()
為何使用內容雜湊? 檔案重新命名或移動 = 快取命中。內容變更 = 自動失效。不需要索引檔案。
2. 凍結資料類別(Frozen Dataclass)儲存快取條目
from dataclasses import dataclass
@dataclass(frozen=True, slots=True)
class CacheEntry:
file_hash: str
source_path: str
document: ExtractedDocument # 快取的處理結果
3. 基於檔案的快取儲存
每個快取條目儲存為 {hash}.json — 透過雜湊值進行 O(1) 查詢,不需要索引檔案。
import json
from typing import Any
def write_cache(cache_dir: Path, entry: CacheEntry) -> None:
cache_dir.mkdir(parents=True, exist_ok=True)
cache_file = cache_dir / f"{entry.file_hash}.json"
data = serialize_entry(entry)
cache_file.write_text(json.dumps(data, ensure_ascii=False), encoding="utf-8")
def read_cache(cache_dir: Path, file_hash: str) -> CacheEntry | None:
cache_file = cache_dir / f"{file_hash}.json"
if not cache_file.is_file():
return None
try:
raw = cache_file.read_text(encoding="utf-8")
data = json.loads(raw)
return deserialize_entry(data)
except (json.JSONDecodeError, ValueError, KeyError):
return None # 將損毀視為快取未命中
4. 服務層包裝器(單一職責原則)
保持處理函式純淨。將快取作為獨立的服務層加入。
def extract_with_cache(
file_path: Path,
*,
cache_enabled: bool = True,
cache_dir: Path = Path(".cache"),
) -> ExtractedDocument:
"""服務層:檢查快取 → 萃取 → 寫入快取。"""
if not cache_enabled:
return extract_text(file_path) # 純函式,不了解快取
file_hash = compute_file_hash(file_path)
# 檢查快取
cached = read_cache(cache_dir, file_hash)
if cached is not None:
logger.info("快取命中:%s(雜湊=%s)", file_path.name, file_hash[:12])
return cached.document
# 快取未命中 → 萃取 → 儲存
logger.info("快取未命中:%s(雜湊=%s)", file_path.name, file_hash[:12])
doc = extract_text(file_path)
entry = CacheEntry(file_hash=file_hash, source_path=str(file_path), document=doc)
write_cache(cache_dir, entry)
return doc
關鍵設計決策
| 決策 | 理由 |
|---|---|
| SHA-256 內容雜湊 | 與路徑無關,內容變更時自動失效 |
{hash}.json 檔案命名 | O(1) 查詢,不需要索引檔案 |
| 服務層包裝器 | 單一職責原則:萃取保持純淨,快取是獨立關注點 |
| 手動 JSON 序列化 | 完整控制凍結資料類別的序列化過程 |
損毀時回傳 None | 優雅降級,下次執行重新處理 |
cache_dir.mkdir(parents=True) | 首次寫入時懶建立目錄 |
最佳實踐
- 雜湊內容,而非路徑 — 路徑會變,內容身份不變
- 處理大檔案時進行區塊讀取 — 避免將整個檔案載入記憶體
- 保持處理函式純淨 — 它們不應該知道快取的存在
- 記錄快取命中/未命中,並使用截斷的雜湊以利除錯
- 優雅處理損毀 — 將無效的快取條目視為未命中,絕不崩潰
應避免的反模式
# 錯誤:基於路徑的快取(檔案移動/重新命名時會失效)
cache = {"/path/to/file.pdf": result}
# 錯誤:將快取邏輯放入處理函式內部(違反單一職責原則)
def extract_text(path, *, cache_enabled=False, cache_dir=None):
if cache_enabled: # 現在這個函式有兩個職責
...
# 錯誤:對巢狀凍結資料類別使用 dataclasses.asdict()
# (複雜的巢狀型別可能導致問題)
data = dataclasses.asdict(entry) # 應使用手動序列化
使用時機
- 檔案處理管線(PDF 解析、OCR、文字萃取、圖片分析)
- 受益於
--cache/--no-cache選項的 CLI 工具 - 批次處理中,相同檔案在不同執行之間重複出現
- 為現有的純函式加入快取功能,而不修改函式本身
不應使用時機
- 資料必須永遠保持最新(即時資料流)
- 快取條目會非常龐大(應考慮串流處理)
- 結果依賴於檔案內容以外的參數(例如不同的萃取設定)