excel_tool/mylib/translate_utils.py

import os
import logging
from pathlib import Path
from typing import List, Tuple
from mylib.pdfzh_translator import OpenAITranslator
from mylib.read_encoding_cvs import read_csv
from logging_config import setup_logging
from mylib.logging_config import setup_logging

# Setup custom logging
setup_logging()
logger = logging.getLogger(__name__)

def insert_empty_columns(data: List[List[str]], column_indices: List[int]) -> List[List[str]]:
    """在指定列之后插入空列"""
    try:
        # 按从大到小排序，防止插入影响后续索引
        column_indices.sort(reverse=True)
        for row in data:
            for index in column_indices:
                row.insert(index + 1, '')
        return data
    except Exception as e:
        logger.error(f"Error inserting empty columns: {e}")
        raise

def extract_sample_data(data: List[List[str]], start_row: int = 0, column_index: int = 0, n: int = 3, m: int = 2) -> List[List[str]]:
    """提取指定行和列开始的样本数据"""
    try:
        sample = []
        # 确保不超过数据范围
        end_row = min(start_row + n, len(data))
        end_col = min(column_index + m, len(data[0]) if data else 0)
        
        for row in data[start_row:end_row]:
            sample.append(row[column_index:end_col])
        return sample
    except Exception as e:
        logger.error(f"Error extracting sample data: {e}")
        raise

def log_data_details(data: List[List[str]], search_term_index: int, start_row: int = 3):
    """记录数据详细信息"""
    try:
        # 记录行号和列号
        logger.info(f"行号范围: {start_row}-{len(data)-1}")
        logger.info(f"翻译列号: {search_term_index}")
        
        # 提取并记录被翻译列的内容
        translated_column = [row[search_term_index] for row in data[start_row:]]
        logger.info(f"被翻译列内容: {translated_column}")
            
    except Exception as e:
        logger.error(f"记录数据详细信息时出错: {e}")
        raise

def process_batch_translations(data: List[List[str]], 
                             search_term_index: int,
                             start_row: int = 3) -> Tuple[List[List[str]], List[List[str]]]:
    """批量处理搜索词翻译"""
    try:
        # 首先提取样本数据用于检查
        sample_data = extract_sample_data(data, start_row, search_term_index)
        logger.info(f"从第{start_row}行第{search_term_index}列开始的样本数据:\n{sample_data}")
        
        # 记录数据详细信息
        log_data_details(data, search_term_index, start_row)
        
        # 初始化翻译器
        translator = OpenAITranslator()
        
        # 直接提取需要翻译的搜索词
        search_terms = [row[search_term_index] for row in data[start_row-1:]]
        
        # 批量翻译
        logger.info("Starting search term translations...")
        
        if os.getenv('DEBUG', '').lower() in ('true', '1', 'True'):
            # DEBUG模式：使用模拟翻译
            search_translations = [f"{text} 翻译测试" for text in search_terms]
        else:
            # 正常模式：调用真实翻译
            search_translations = translator.translate(search_terms)
        
        logger.info("Search term translations completed")
        
        # 更新数据
        for i, row in enumerate(data[start_row-1:], start=start_row-1):
            try:
                # 更新搜索词翻译列
                row[search_term_index + 1] = search_translations[i-(start_row-1)]
                
            except Exception as e:
                logger.error(f"Error processing row {i}: {e}")
                raise
                
        return data, sample_data
    except Exception as e:
        logger.error(f"Error in batch translation: {e}")
        raise

def main():
    output_dir = Path('temp')
    input_file = output_dir/"测试.csv"
    output_file = output_dir/"processed_测试.csv"
    data = read_csv(input_file)
    process_batch_translations(data, 2)

if __name__ == "__main__":
    main()