內容雜湊檔案快取模式
使用 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. 凍結資料類別作為快取條目
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. 服務層包裝器(SRP)
保持處理函式純淨。將快取作為單獨的服務層加入。
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 (hash=%s)", file_path.name, file_hash[:12])
return cached.document
# 快取未命中 -> 執行萃取 -> 儲存
logger.info("快取未命中:%s (hash=%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) 查詢,無需索引檔案 |
| 服務層包裝器 | SRP:萃取保持純淨,快取是獨立關注點 |
| 手動 JSON 序列化 | 對凍結資料類別的序列化有完整控制 |
損毀時回傳 None | 優雅降級,下次執行時重新處理 |
cache_dir.mkdir(parents=True) | 第一次寫入時才延遲建立目錄 |
最佳實踐
- 對內容進行雜湊,而不是路徑 — 路徑會變,內容身分不會
- 對大型檔案進行區塊讀取 — 避免將整個檔案載入記憶體
- 保持處理函式純淨 — 它們不應該知道快取的存在
- 記錄快取命中/未命中 — 使用截斷的雜湊值以便除錯
- 優雅處理損毀 — 將無效的快取條目視為未命中,絕不崩潰
應避免的反模式
# 錯誤:基於路徑的快取(檔案移動/重新命名時失效)
cache = {"/path/to/file.pdf": result}
# 錯誤:將快取邏輯放在處理函式內部(違反 SRP)
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 工具 - 跨執行批次處理中會出現相同檔案的情況
- 為現有的純函式加入快取而不修改它們
何時不應使用
- 資料必須永遠保持最新(即時饋送)
- 快取條目極大(應考慮串流處理)
- 結果取決於檔案內容以外的參數(例如不同的萃取設定)