kb-skills/kb_workflow/srt/cache/semantic_cache.py

"""语义分段缓存管理器

使用内容哈希（SHA256）作为缓存键，存储在本地文件系统。

功能：
- 计算内容的 SHA256 哈希作为缓存键
- 缓存 SegmentationResult 到本地文件系统
- 支持版本控制，便于未来格式变更
- 错误容错，缓存失败不影响主流程
"""

from __future__ import annotations

import hashlib
import json
import logging
from pathlib import Path
from typing import Optional

from pydantic import ValidationError

from kb_workflow.srt.semantic_engine import SegmentationResult

logger = logging.getLogger(__name__)


def compute_content_hash(content: str) -> str:
    """计算内容的 SHA256 哈希作为缓存键

    Args:
        content: 字幕内容

    Returns:
        SHA256 哈希值（64 字符的十六进制字符串）

    Examples:
        >>> hash1 = compute_content_hash("相同内容")
        >>> hash2 = compute_content_hash("相同内容")
        >>> hash1 == hash2
        True
        >>> len(hash1)
        64
    """
    return hashlib.sha256(content.encode('utf-8')).hexdigest()


class SemanticCache:
    """语义分段结果缓存管理器

    使用内容哈希（SHA256）作为缓存键，存储在本地文件系统。

    缓存格式：
    - 文件命名: {sha256_hash}.json
    - 文件内容: JSON 格式的 SegmentationResult + version 字段
    - 缓存 TTL: 永久有效，手动清理

    Attributes:
        CACHE_VERSION: 缓存格式版本，用于未来兼容性
        DEFAULT_CACHE_DIR: 默认缓存目录

    Examples:
        >>> cache = SemanticCache()
        >>> result = cache.get(subtitle_content)
        >>> if result is None:
        ...     result = llm_segment(subtitle_content)
        ...     cache.set(subtitle_content, result)
    """

    CACHE_VERSION = "1.0"
    DEFAULT_CACHE_DIR = Path(".planning/cache/semantic")

    def __init__(self, cache_dir: Optional[Path] = None):
        """初始化缓存管理器

        Args:
            cache_dir: 缓存目录路径，默认为 .planning/cache/semantic/
        """
        self.cache_dir = cache_dir or self.DEFAULT_CACHE_DIR
        self.logger = logging.getLogger(__name__)

    def get(self, content: str) -> Optional[SegmentationResult]:
        """从缓存读取分段结果

        Args:
            content: 字幕内容

        Returns:
            缓存的 SegmentationResult，如果不存在或读取失败则返回 None

        Examples:
            >>> cache = SemanticCache()
            >>> result = cache.get("字幕内容")
            >>> if result:
            ...     print(f"从缓存读取: {result.video_type}")
        """
        cache_key = compute_content_hash(content)
        cache_file = self.cache_dir / f"{cache_key}.json"

        if not cache_file.exists():
            self.logger.debug(f"Cache miss: {cache_key}")
            return None

        try:
            data = json.loads(cache_file.read_text(encoding='utf-8'))

            # 验证版本
            if data.get('version') != self.CACHE_VERSION:
                self.logger.warning(
                    f"Cache version mismatch: {data.get('version')} != {self.CACHE_VERSION}"
                )
                return None

            # 移除 version 字段，转换为 SegmentationResult
            data.pop('version', None)
            result = SegmentationResult(**data)

            self.logger.debug(f"Cache hit: {cache_key}")
            return result

        except (json.JSONDecodeError, IOError, ValidationError) as e:
            self.logger.warning(f"Failed to load cache: {e}")
            return None

    def set(self, content: str, result: SegmentationResult) -> None:
        """保存分段结果到缓存

        Args:
            content: 字幕内容
            result: 分段结果

        Examples:
            >>> cache = SemanticCache()
            >>> result = SegmentationResult(video_type="知识类", segments=[], actual_segment_count=0)
            >>> cache.set("字幕内容", result)
        """
        cache_key = compute_content_hash(content)
        cache_file = self.cache_dir / f"{cache_key}.json"

        try:
            # 确保目录存在
            self.cache_dir.mkdir(parents=True, exist_ok=True)

            # 序列化为 JSON，添加 version 字段
            data = result.model_dump()
            data['version'] = self.CACHE_VERSION

            cache_file.write_text(
                json.dumps(data, ensure_ascii=False, indent=2),
                encoding='utf-8'
            )

            self.logger.debug(f"Cache saved: {cache_key}")

        except IOError as e:
            self.logger.warning(f"Failed to save cache: {e}")