Files
Document_Translator/app/models/cache.py
2025-09-02 10:31:35 +08:00

138 lines
4.7 KiB
Python

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
翻譯快取資料模型
Author: PANJIT IT Team
Created: 2024-01-28
Modified: 2024-01-28
"""
import hashlib
from sqlalchemy.sql import func
from app import db
class TranslationCache(db.Model):
"""翻譯快取表 (dt_translation_cache)"""
__tablename__ = 'dt_translation_cache'
id = db.Column(db.Integer, primary_key=True, autoincrement=True)
source_text_hash = db.Column(db.String(64), nullable=False, comment='來源文字hash')
source_language = db.Column(db.String(50), nullable=False, comment='來源語言')
target_language = db.Column(db.String(50), nullable=False, comment='目標語言')
source_text = db.Column(db.Text, nullable=False, comment='來源文字')
translated_text = db.Column(db.Text, nullable=False, comment='翻譯文字')
created_at = db.Column(db.DateTime, default=func.now(), comment='建立時間')
# 唯一約束
__table_args__ = (
db.UniqueConstraint('source_text_hash', 'source_language', 'target_language', name='uk_cache'),
db.Index('idx_languages', 'source_language', 'target_language'),
)
def __repr__(self):
return f'<TranslationCache {self.source_text_hash[:8]}...>'
def to_dict(self):
"""轉換為字典格式"""
return {
'id': self.id,
'source_text_hash': self.source_text_hash,
'source_language': self.source_language,
'target_language': self.target_language,
'source_text': self.source_text,
'translated_text': self.translated_text,
'created_at': self.created_at.isoformat() if self.created_at else None
}
@staticmethod
def generate_hash(text):
"""生成文字的 SHA256 hash"""
return hashlib.sha256(text.encode('utf-8')).hexdigest()
@classmethod
def get_translation(cls, source_text, source_language, target_language):
"""取得快取的翻譯"""
text_hash = cls.generate_hash(source_text)
cache_entry = cls.query.filter_by(
source_text_hash=text_hash,
source_language=source_language,
target_language=target_language
).first()
return cache_entry.translated_text if cache_entry else None
@classmethod
def save_translation(cls, source_text, source_language, target_language, translated_text):
"""儲存翻譯到快取"""
text_hash = cls.generate_hash(source_text)
# 檢查是否已存在
existing = cls.query.filter_by(
source_text_hash=text_hash,
source_language=source_language,
target_language=target_language
).first()
if existing:
# 更新現有記錄
existing.translated_text = translated_text
else:
# 建立新記錄
cache_entry = cls(
source_text_hash=text_hash,
source_language=source_language,
target_language=target_language,
source_text=source_text,
translated_text=translated_text
)
db.session.add(cache_entry)
db.session.commit()
return True
@classmethod
def get_cache_statistics(cls):
"""取得快取統計資料"""
total_entries = cls.query.count()
# 按語言對統計
language_pairs = db.session.query(
cls.source_language,
cls.target_language,
func.count(cls.id).label('count')
).group_by(cls.source_language, cls.target_language).all()
# 最近一週的快取命中
from datetime import datetime, timedelta
week_ago = datetime.utcnow() - timedelta(days=7)
recent_entries = cls.query.filter(cls.created_at >= week_ago).count()
return {
'total_entries': total_entries,
'language_pairs': [
{
'source_language': pair.source_language,
'target_language': pair.target_language,
'count': pair.count
}
for pair in language_pairs
],
'recent_entries': recent_entries
}
@classmethod
def clear_old_cache(cls, days_to_keep=90):
"""清理舊快取記錄"""
from datetime import datetime, timedelta
cutoff_date = datetime.utcnow() - timedelta(days=days_to_keep)
deleted_count = cls.query.filter(
cls.created_at < cutoff_date
).delete(synchronize_session=False)
db.session.commit()
return deleted_count