| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111 |
- 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]], n: int = 2, m: int = 2) -> List[List[str]]:
- """提取前n行m列数据用于检查"""
- try:
- sample = []
- for row in data[:n]:
- sample.append(row[:m])
- 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}")
-
- # 记录前start_row - 5行数据
- logger.info("前5行数据:")
- for i, row in enumerate(data[start_row:start_row+5]):
- logger.info(f"行 {i}: {', '.join(row)}")
-
- # 记录最后5行数据
- logger.info("最后5行数据:")
- for i, row in enumerate(data[-5:]):
- logger.info(f"行 {len(data)-5+i}: {', '.join(row)}")
-
- 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)
- logger.info(f"Sample data extracted for inspection:\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()
|