Maas-Model-platform/backend/app/services/translation_memory_service.py

"""
翻译记忆服务

处理翻译记忆的增删改查和相似句子匹配
"""
import logging
from datetime import datetime
from typing import Optional, List
from sqlalchemy.orm import Session
from sqlalchemy import and_, func

from app.models.translation import TranslationMemory
from app.schemas.translation_schema import (
    MemoryCreate,
    MemoryResponse,
    MemoryListResponse,
    MemoryMatch,
    MemoryMatchResponse
)

logger = logging.getLogger(__name__)


class TranslationMemoryService:
    """翻译记忆服务类"""

    def __init__(self, db: Session):
        """
        初始化服务
        
        Args:
            db: 数据库会话
        """
        self.db = db

    def add_memory(
        self,
        user_id: str,
        memory: MemoryCreate,
        source_type: str = "manual",
        task_id: Optional[str] = None
    ) -> MemoryResponse:
        """
        添加翻译记忆
        
        Args:
            user_id: 用户ID
            memory: 翻译记忆创建请求
            source_type: 来源类型
            task_id: 关联的任务ID
            
        Returns:
            翻译记忆响应
        """
        try:
            # 检查是否已存在相同的记忆
            existing = self.db.query(TranslationMemory).filter(
                and_(
                    TranslationMemory.user_id == user_id,
                    TranslationMemory.source_lang == memory.source_lang,
                    TranslationMemory.target_lang == memory.target_lang,
                    TranslationMemory.source_text == memory.source_text,
                    TranslationMemory.is_deleted == False
                )
            ).first()
            
            if existing:
                # 更新已存在的记忆
                existing.target_text = memory.target_text
                existing.updated_at = datetime.now()
                self.db.commit()
                self.db.refresh(existing)
                
                logger.info(f"更新翻译记忆: user_id={user_id}, id={existing.id}")
                
                return MemoryResponse(
                    id=existing.id,
                    source_lang=existing.source_lang,
                    target_lang=existing.target_lang,
                    source_text=existing.source_text,
                    target_text=existing.target_text,
                    source_type=existing.source_type,
                    usage_count=existing.usage_count,
                    created_at=existing.created_at.strftime("%Y-%m-%d %H:%M:%S")
                )
            
            # 创建新记忆
            new_memory = TranslationMemory(
                user_id=user_id,
                source_lang=memory.source_lang,
                target_lang=memory.target_lang,
                source_text=memory.source_text,
                target_text=memory.target_text,
                source_type=source_type,
                task_id=task_id
            )
            
            self.db.add(new_memory)
            self.db.commit()
            self.db.refresh(new_memory)
            
            logger.info(f"添加翻译记忆成功: user_id={user_id}, id={new_memory.id}")
            
            return MemoryResponse(
                id=new_memory.id,
                source_lang=new_memory.source_lang,
                target_lang=new_memory.target_lang,
                source_text=new_memory.source_text,
                target_text=new_memory.target_text,
                source_type=new_memory.source_type,
                usage_count=new_memory.usage_count,
                created_at=new_memory.created_at.strftime("%Y-%m-%d %H:%M:%S")
            )
            
        except Exception as e:
            logger.error(f"添加翻译记忆失败: {str(e)}")
            self.db.rollback()
            raise Exception(f"添加翻译记忆失败: {str(e)}")

    def get_user_memories(
        self,
        user_id: str,
        source_lang: Optional[str] = None,
        target_lang: Optional[str] = None,
        keyword: Optional[str] = None
    ) -> MemoryListResponse:
        """
        查询用户翻译记忆
        
        Args:
            user_id: 用户ID
            source_lang: 源语言筛选
            target_lang: 目标语言筛选
            keyword: 关键词搜索
            
        Returns:
            翻译记忆列表响应
        """
        query = self.db.query(TranslationMemory).filter(
            TranslationMemory.user_id == user_id,
            TranslationMemory.is_deleted == False
        )
        
        # 应用筛选条件
        if source_lang:
            query = query.filter(TranslationMemory.source_lang == source_lang)
        if target_lang:
            query = query.filter(TranslationMemory.target_lang == target_lang)
        if keyword:
            query = query.filter(
                (TranslationMemory.source_text.ilike(f"%{keyword}%")) |
                (TranslationMemory.target_text.ilike(f"%{keyword}%"))
            )
        
        # 排序（按使用次数和创建时间）
        query = query.order_by(
            TranslationMemory.usage_count.desc(),
            TranslationMemory.created_at.desc()
        )
        
        items = query.all()
        
        # 构建响应
        memory_items = [
            MemoryResponse(
                id=item.id,
                source_lang=item.source_lang,
                target_lang=item.target_lang,
                source_text=item.source_text,
                target_text=item.target_text,
                source_type=item.source_type,
                usage_count=item.usage_count,
                created_at=item.created_at.strftime("%Y-%m-%d %H:%M:%S")
            )
            for item in items
        ]
        
        return MemoryListResponse(items=memory_items)

    def delete_memory(
        self,
        user_id: str,
        memory_id: int
    ) -> bool:
        """
        删除翻译记忆（软删除）
        
        Args:
            user_id: 用户ID
            memory_id: 记忆ID
            
        Returns:
            是否删除成功
        """
        try:
            memory = self.db.query(TranslationMemory).filter(
                TranslationMemory.id == memory_id,
                TranslationMemory.user_id == user_id,
                TranslationMemory.is_deleted == False
            ).first()
            
            if not memory:
                logger.warning(f"翻译记忆不存在或已删除: memory_id={memory_id}, user_id={user_id}")
                return False
            
            # 软删除
            memory.is_deleted = True
            memory.deleted_at = datetime.now()
            self.db.commit()
            
            logger.info(f"软删除翻译记忆成功: memory_id={memory_id}, user_id={user_id}")
            return True
            
        except Exception as e:
            logger.error(f"软删除翻译记忆失败: {str(e)}")
            self.db.rollback()
            return False

    def find_similar(
        self,
        user_id: str,
        source_text: str,
        source_lang: str,
        target_lang: str,
        threshold: float = 0.8,
        limit: int = 5
    ) -> MemoryMatchResponse:
        """
        查找相似翻译
        
        使用简单的文本相似度算法（基于字符串包含关系）
        
        Args:
            user_id: 用户ID
            source_text: 源文本
            source_lang: 源语言
            target_lang: 目标语言
            threshold: 相似度阈值
            limit: 返回结果数量限制
            
        Returns:
            相似句子匹配响应
        """
        try:
            # 查询相同语言对的翻译记忆
            memories = self.db.query(TranslationMemory).filter(
                TranslationMemory.user_id == user_id,
                TranslationMemory.source_lang == source_lang,
                TranslationMemory.target_lang == target_lang,
                TranslationMemory.is_deleted == False
            ).all()
            
            matches = []
            
            for memory in memories:
                # 计算简单的相似度（基于字符串包含和长度）
                similarity = self._calculate_similarity(source_text, memory.source_text)
                
                if similarity >= threshold:
                    matches.append({
                        "memory": memory,
                        "similarity": similarity
                    })
            
            # 按相似度排序
            matches.sort(key=lambda x: x["similarity"], reverse=True)
            
            # 限制返回数量
            matches = matches[:limit]
            
            # 更新使用统计
            for match in matches:
                memory = match["memory"]
                memory.usage_count += 1
                memory.last_used_at = datetime.now()
            
            self.db.commit()
            
            # 构建响应
            match_results = [
                MemoryMatch(
                    source_text=match["memory"].source_text,
                    target_text=match["memory"].target_text,
                    similarity=match["similarity"]
                )
                for match in matches
            ]
            
            return MemoryMatchResponse(matches=match_results)
            
        except Exception as e:
            logger.error(f"查找相似翻译失败: {str(e)}")
            return MemoryMatchResponse(matches=[])

    def _calculate_similarity(self, text1: str, text2: str) -> float:
        """
        计算文本相似度（简单算法）
        
        基于最长公共子序列和字符串长度
        
        Args:
            text1: 文本1
            text2: 文本2
            
        Returns:
            相似度（0-1）
        """
        if text1 == text2:
            return 1.0
        
        # 完全包含关系
        if text1 in text2 or text2 in text1:
            shorter = min(len(text1), len(text2))
            longer = max(len(text1), len(text2))
            return shorter / longer
        
        # 计算最长公共子序列长度
        lcs_length = self._lcs_length(text1, text2)
        
        # 相似度 = LCS长度 / 较长字符串长度
        max_length = max(len(text1), len(text2))
        
        if max_length == 0:
            return 0.0
        
        return lcs_length / max_length

    def _lcs_length(self, text1: str, text2: str) -> int:
        """
        计算最长公共子序列长度
        
        Args:
            text1: 文本1
            text2: 文本2
            
        Returns:
            LCS长度
        """
        m, n = len(text1), len(text2)
        
        # 创建DP表
        dp = [[0] * (n + 1) for _ in range(m + 1)]
        
        # 填充DP表
        for i in range(1, m + 1):
            for j in range(1, n + 1):
                if text1[i - 1] == text2[j - 1]:
                    dp[i][j] = dp[i - 1][j - 1] + 1
                else:
                    dp[i][j] = max(dp[i - 1][j], dp[i][j - 1])
        
        return dp[m][n]