227 lines
8.0 KiB
Python
227 lines
8.0 KiB
Python
"""
|
||
翻译缓存管理模块 - 简化版
|
||
基于全局ID和chunk的缓存系统
|
||
"""
|
||
|
||
import json
|
||
import hashlib
|
||
from pathlib import Path
|
||
from datetime import datetime, timedelta
|
||
from typing import Dict, Optional, List
|
||
from loguru import logger
|
||
|
||
|
||
class TranslationCache:
|
||
"""翻译缓存管理器 - 简化版"""
|
||
|
||
def __init__(self, config: Dict):
|
||
"""初始化缓存管理器"""
|
||
self.config = config
|
||
cache_config = config.get('cache', {})
|
||
|
||
self.enabled = cache_config.get('enabled', True)
|
||
self.cache_dir = Path(cache_config.get('directory', 'cache'))
|
||
self.max_age_days = cache_config.get('max_age_days', 30)
|
||
|
||
if self.enabled:
|
||
self.cache_dir.mkdir(parents=True, exist_ok=True)
|
||
self.translations_dir = self.cache_dir / 'translations'
|
||
self.translations_dir.mkdir(parents=True, exist_ok=True)
|
||
|
||
logger.info(f"翻译缓存已启用: {self.cache_dir}")
|
||
|
||
def get_chunk_translation(self, chunk: List[Dict], model: str) -> Optional[Dict[str, str]]:
|
||
"""
|
||
获取chunk的缓存翻译
|
||
|
||
Args:
|
||
chunk: 段落列表（带global_id）
|
||
model: 模型名称
|
||
|
||
Returns:
|
||
{global_id: translation} 映射，如果不存在返回 None
|
||
"""
|
||
if not self.enabled:
|
||
return None
|
||
|
||
try:
|
||
cache_key = self._get_chunk_cache_key(chunk, model)
|
||
cache_file = self._get_cache_file_path(cache_key)
|
||
|
||
if not cache_file.exists():
|
||
return None
|
||
|
||
# 检查是否过期
|
||
file_age = datetime.now() - datetime.fromtimestamp(cache_file.stat().st_mtime)
|
||
if file_age > timedelta(days=self.max_age_days):
|
||
logger.debug(f"缓存已过期: {cache_key[:8]}...")
|
||
cache_file.unlink()
|
||
return None
|
||
|
||
# 读取缓存
|
||
with open(cache_file, 'r', encoding='utf-8') as f:
|
||
cache_data = json.load(f)
|
||
|
||
# 验证缓存
|
||
if (cache_data.get('success') and
|
||
cache_data.get('model') == model and
|
||
self._validate_cache_data(cache_data, chunk)):
|
||
|
||
logger.debug(f"缓存命中: {cache_key[:8]}... ({len(chunk)} 段落)")
|
||
return cache_data.get('translations', {})
|
||
|
||
return None
|
||
|
||
except Exception as e:
|
||
logger.warning(f"读取缓存失败: {e}")
|
||
return None
|
||
|
||
def save_chunk_translation(self, chunk: List[Dict], translations: Dict[str, str],
|
||
model: str, success: bool = True) -> None:
|
||
"""
|
||
保存chunk翻译到缓存
|
||
|
||
Args:
|
||
chunk: 段落列表（带global_id）
|
||
translations: {global_id: translation} 映射
|
||
model: 模型名称
|
||
success: 是否翻译成功
|
||
"""
|
||
if not self.enabled:
|
||
return
|
||
|
||
try:
|
||
cache_key = self._get_chunk_cache_key(chunk, model)
|
||
cache_file = self._get_cache_file_path(cache_key)
|
||
|
||
# 构建缓存数据
|
||
cache_data = {
|
||
'global_ids': [p['global_id'] for p in chunk],
|
||
'translations': translations,
|
||
'model': model,
|
||
'timestamp': datetime.now().isoformat(),
|
||
'success': success,
|
||
'paragraph_count': len(chunk),
|
||
'cache_version': '3.0'
|
||
}
|
||
|
||
with open(cache_file, 'w', encoding='utf-8') as f:
|
||
json.dump(cache_data, f, ensure_ascii=False, indent=2)
|
||
|
||
logger.debug(f"缓存已保存: {cache_key[:8]}... ({len(chunk)} 段落)")
|
||
|
||
except Exception as e:
|
||
logger.warning(f"保存缓存失败: {e}")
|
||
|
||
def _get_chunk_cache_key(self, chunk: List[Dict], model: str) -> str:
|
||
"""
|
||
生成chunk缓存键（基于全局ID序列）
|
||
|
||
Args:
|
||
chunk: 段落列表
|
||
model: 模型名称
|
||
|
||
Returns:
|
||
缓存键
|
||
"""
|
||
# 使用全局ID序列作为缓存键的一部分
|
||
id_sequence = ",".join(p['global_id'] for p in chunk)
|
||
combined = f"{id_sequence}|{model}"
|
||
return hashlib.md5(combined.encode('utf-8')).hexdigest()
|
||
|
||
def _get_cache_file_path(self, cache_key: str) -> Path:
|
||
"""获取缓存文件路径"""
|
||
# 使用 hash 前缀分目录，避免单目录文件过多
|
||
subdir = cache_key[:2]
|
||
cache_subdir = self.translations_dir / subdir
|
||
cache_subdir.mkdir(parents=True, exist_ok=True)
|
||
return cache_subdir / f"{cache_key}.json"
|
||
|
||
def _validate_cache_data(self, cache_data: Dict, chunk: List[Dict]) -> bool:
|
||
"""验证缓存数据的有效性"""
|
||
# 检查ID序列是否匹配
|
||
cached_ids = cache_data.get('global_ids', [])
|
||
chunk_ids = [p['global_id'] for p in chunk]
|
||
|
||
if cached_ids != chunk_ids:
|
||
logger.debug("缓存ID序列不匹配")
|
||
return False
|
||
|
||
# 只要有翻译结果就认为有效，不要求数量完全匹配
|
||
translations = cache_data.get('translations', {})
|
||
if not translations:
|
||
logger.debug("缓存无翻译结果")
|
||
return False
|
||
|
||
return True
|
||
|
||
def clear_cache(self, older_than_days: Optional[int] = None) -> int:
|
||
"""清理缓存"""
|
||
if not self.enabled or not self.translations_dir.exists():
|
||
return 0
|
||
|
||
cleared_count = 0
|
||
cutoff_time = None
|
||
|
||
if older_than_days is not None:
|
||
cutoff_time = datetime.now() - timedelta(days=older_than_days)
|
||
|
||
try:
|
||
for cache_file in self.translations_dir.rglob('*.json'):
|
||
should_delete = False
|
||
|
||
if cutoff_time is None:
|
||
should_delete = True
|
||
else:
|
||
file_time = datetime.fromtimestamp(cache_file.stat().st_mtime)
|
||
should_delete = file_time < cutoff_time
|
||
|
||
if should_delete:
|
||
cache_file.unlink()
|
||
cleared_count += 1
|
||
|
||
# 清理空目录
|
||
for date_dir in self.translations_dir.iterdir():
|
||
if date_dir.is_dir() and not any(date_dir.iterdir()):
|
||
date_dir.rmdir()
|
||
|
||
logger.info(f"清理了 {cleared_count} 个缓存文件")
|
||
return cleared_count
|
||
|
||
except Exception as e:
|
||
logger.error(f"清理缓存失败: {e}")
|
||
return 0
|
||
|
||
def get_cache_stats(self) -> Dict:
|
||
"""获取缓存统计信息"""
|
||
if not self.enabled or not self.translations_dir.exists():
|
||
return {'enabled': False}
|
||
|
||
try:
|
||
cache_files = list(self.translations_dir.rglob('*.json'))
|
||
total_files = len(cache_files)
|
||
total_size = sum(f.stat().st_size for f in cache_files)
|
||
|
||
# 统计段落数
|
||
total_paragraphs = 0
|
||
for cache_file in cache_files:
|
||
try:
|
||
with open(cache_file, 'r', encoding='utf-8') as f:
|
||
data = json.load(f)
|
||
total_paragraphs += data.get('paragraph_count', 0)
|
||
except:
|
||
continue
|
||
|
||
return {
|
||
'enabled': True,
|
||
'total_files': total_files,
|
||
'total_paragraphs': total_paragraphs,
|
||
'total_size_mb': round(total_size / 1024 / 1024, 2),
|
||
'cache_directory': str(self.cache_dir),
|
||
'max_age_days': self.max_age_days
|
||
}
|
||
|
||
except Exception as e:
|
||
logger.error(f"获取缓存统计失败: {e}")
|
||
return {'enabled': True, 'error': str(e)}
|