content-hash-cache-pattern

内容ハッシュファイルキャッシュパターン（Content-Hash File Cache Pattern）

ファイルパスではなく SHA-256 内容ハッシュをキャッシュキーとして使用し、PDF 解析、テキスト抽出、画像分析などのコストが高いファイル処理結果をキャッシュします。パスベースのキャッシュとは異なり、この方法はファイルの移動や名前変更後も有効で、内容が変更されると自動的に無効になります。

使用するタイミング

• PDF、画像、テキスト抽出などのファイル処理パイプラインを構築するとき
• 同じファイルを繰り返し処理する必要があり、処理コストが高いとき
• --cache/--no-cache コマンドラインオプションを提供する必要があるとき
• 既存の純粋関数（Pure Functions）を変更せずにキャッシュ機能を追加したいとき

コアパターン

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"File not found: {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. サービス層ラッパー（単一責任の原則 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("Cache hit: %s (hash=%s)", file_path.name, file_hash[:12])
        return cached.document

    # 缓存未命中 -> 提取 -> 存储
    logger.info("Cache miss: %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) で検索できます
サービス層ラッパー（Wrapper）	単一責任の原則（SRP）に準拠します。抽出ロジックは純粋に保ち、キャッシュは独立した関心事とします
手動 JSON シリアライズ	凍結データクラスのシリアライズを完全に制御できます
破損時に None を返す	優雅にフォールバックし、次回の実行時に再処理します
cache_dir.mkdir(parents=True)	初回書き込み時にディレクトリを遅延作成します

ベストプラクティス

• パスではなく内容をハッシュ化する —— パスは変わりますが、内容の識別情報（Identity）は変わりません。
• 大きなファイルは分割してハッシュ化する —— ファイル全体をメモリに読み込むのを避けます。
• 処理関数は純粋なままにする —— キャッシュについて何も認識させないでください。
• キャッシュのヒット／ミスをログに記録する —— デバッグしやすいように、切り詰めたハッシュ値を使用します。
• キャッシュの破損を適切に処理する —— 無効なキャッシュエントリはミスとして扱い、決してクラッシュしないようにします。

避けるべきアンチパターン（Anti-Patterns）

# 错误做法：基于路径的缓存（文件移动/重命名后失效）
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）
• 複数回の実行で同じファイルが処理されるバッチタスク
• 既存の純粋関数を変更せずにキャッシュ機能を追加する場合

適さない用途

• 絶対的なリアルタイム性を保つ必要があるデータ（リアルタイムデータストリーム）
• キャッシュエントリのサイズが非常に大きい場合（ストリーミング処理を検討してください）
• 結果がファイルの内容以外のパラメーターにも依存する場合（例：抽出設定の違い）