english-to-chinese/code/translate_epub_v4(单线程版本)V3.py

import os
import re
from bs4 import BeautifulSoup
import openai
import time
from tqdm import tqdm
import sqlite3
import json
from datetime import datetime
import logging
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
import asyncio
import aiohttp
from concurrent.futures import ThreadPoolExecutor
from functools import lru_cache
import hashlib
import yaml
from pathlib import Path

# 配置管理
class Config:
    def __init__(self, config_path='config.yaml'):
        self.config_path = config_path
        self.config = self.load_config()
        
        # 设置日志
        self.setup_logging()
        
        # 初始化OpenAI客户端
        self.setup_openai()
    
    def load_config(self):
        """加载配置文件"""
        if not os.path.exists(self.config_path):
            # 创建默认配置
            default_config = {
                'logging': {
                    'level': 'INFO',
                    'format': '%(asctime)s - %(levelname)s - %(message)s',
                    'file': 'translation.log'
                },
                'openai': {
                    'base_url': 'https://api.siliconflow.cn/v1',
                    'api_key': 'sk-',
                    'model_name': 'deepseek-ai/DeepSeek-R1',
                    'max_retries': 3,
                    'retry_delay': 2,
                    'timeout': 30,
                    'max_concurrent_requests': 5
                },
                'translation': {
                    'min_line_count': 1,
                    'max_line_count': 5,
                    'initial_line_count': 2,
                    'error_threshold': 3,
                    'success_threshold': 5,
                    'error_cooldown': 60,
                    'cache_size': 1000
                },
                'database': {
                    'path': 'translation_progress.db',
                    'pool_size': 5
                },
                'paths': {
                    'input_dir': '002/Ops',
                    'output_dir': '002/Ops_translated'
                }
            }
            
            # 保存默认配置
            with open(self.config_path, 'w', encoding='utf-8') as f:
                yaml.dump(default_config, f, allow_unicode=True)
            
            return default_config
        
        # 加载现有配置
        with open(self.config_path, 'r', encoding='utf-8') as f:
            return yaml.safe_load(f)
    
    def setup_logging(self):
        """设置日志"""
        logging.basicConfig(
            level=getattr(logging, self.config['logging']['level']),
            format=self.config['logging']['format'],
            handlers=[
                logging.FileHandler(self.config['logging']['file']),
                logging.StreamHandler()
            ]
        )
    
    def setup_openai(self):
        """设置OpenAI客户端"""
        self.client = openai.OpenAI(
            base_url=self.config['openai']['base_url'],
            api_key=self.config['openai']['api_key']
        )
    
    def get(self, *keys):
        """获取配置值"""
        value = self.config
        for key in keys:
            value = value[key]
        return value
    
    def update(self, updates):
        """更新配置"""
        def deep_update(d, u):
            for k, v in u.items():
                if isinstance(v, dict):
                    d[k] = deep_update(d.get(k, {}), v)
                else:
                    d[k] = v
            return d
        
        self.config = deep_update(self.config, updates)
        
        # 保存更新后的配置
        with open(self.config_path, 'w', encoding='utf-8') as f:
            yaml.dump(self.config, f, allow_unicode=True)
        
        # 重新设置日志和OpenAI客户端
        self.setup_logging()
        self.setup_openai()

# 创建全局的配置实例
config = Config()

# 更新全局变量
MODEL_CONFIG = {
    "model_name": config.get('openai', 'model_name'),
    "max_retries": config.get('openai', 'max_retries'),
    "retry_delay": config.get('openai', 'retry_delay'),
    "timeout": config.get('openai', 'timeout'),
    "max_concurrent_requests": config.get('openai', 'max_concurrent_requests'),
    "cache_size": config.get('translation', 'cache_size')
}

MIN_LINE_COUNT = config.get('translation', 'min_line_count')
MAX_LINE_COUNT = config.get('translation', 'max_line_count')
INITIAL_LINE_COUNT = config.get('translation', 'initial_line_count')
ERROR_THRESHOLD = config.get('translation', 'error_threshold')
SUCCESS_THRESHOLD = config.get('translation', 'success_threshold')

# 更新其他类的初始化参数
class TranslationStats:
    def __init__(self):
        self.start_time = time.time()
        self.total_chars = 0
        self.translated_chars = 0
        self.total_requests = 0
        self.successful_requests = 0
        self.failed_requests = 0
    
    def update_stats(self, original_text, translated_text, success=True):
        self.total_chars += len(original_text)
        self.translated_chars += len(translated_text)
        self.total_requests += 1
        if success:
            self.successful_requests += 1
        else:
            self.failed_requests += 1
    
    def get_stats(self):
        elapsed_time = time.time() - self.start_time
        chars_per_second = self.translated_chars / elapsed_time if elapsed_time > 0 else 0
        success_rate = (self.successful_requests / self.total_requests * 100) if self.total_requests > 0 else 0
        
        return {
            "总字符数": self.total_chars,
            "已翻译字符数": self.translated_chars,
            "翻译速度": f"{chars_per_second:.2f} 字符/秒",
            "成功率": f"{success_rate:.1f}%",
            "总请求数": self.total_requests,
            "成功请求": self.successful_requests,
            "失败请求": self.failed_requests,
            "运行时间": f"{elapsed_time:.1f} 秒"
        }

# 创建全局的统计对象
translation_stats = TranslationStats()

class DatabaseManager:
    def __init__(self):
        self.db_path = config.get('database', 'path')
        self.conn = None
        self.init_db()
    
    def get_connection(self):
        """获取数据库连接"""
        if self.conn is None:
            self.conn = sqlite3.connect(self.db_path)
            self.conn.row_factory = sqlite3.Row
        return self.conn
    
    def close(self):
        """关闭数据库连接"""
        if self.conn:
            self.conn.close()
            self.conn = None
    
    def init_db(self):
        """初始化数据库"""
        conn = self.get_connection()
        c = conn.cursor()
        
        # 创建文件进度表
        c.execute('''
            CREATE TABLE IF NOT EXISTS file_progress (
                file_path TEXT PRIMARY KEY,
                total_lines INTEGER,
                processed_lines INTEGER,
                status TEXT,
                last_updated TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                error_count INTEGER DEFAULT 0,
                retry_count INTEGER DEFAULT 0
            )
        ''')
        
        # 创建行进度表
        c.execute('''
            CREATE TABLE IF NOT EXISTS line_progress (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                file_path TEXT,
                line_index INTEGER,
                original_text TEXT,
                translated_text TEXT,
                status TEXT,
                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                error_count INTEGER DEFAULT 0,
                retry_count INTEGER DEFAULT 0,
                UNIQUE(file_path, line_index)
            )
        ''')
        
        # 创建错误日志表
        c.execute('''
            CREATE TABLE IF NOT EXISTS error_log (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                file_path TEXT,
                line_index INTEGER,
                error_type TEXT,
                error_message TEXT,
                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                resolved_at TIMESTAMP,
                resolution TEXT
            )
        ''')
        
        conn.commit()
    
    def begin_transaction(self):
        """开始事务"""
        self.get_connection().execute('BEGIN TRANSACTION')
    
    def commit_transaction(self):
        """提交事务"""
        self.get_connection().commit()
    
    def rollback_transaction(self):
        """回滚事务"""
        self.get_connection().rollback()
    
    def get_file_progress(self, file_path):
        """获取文件翻译进度"""
        c = self.get_connection().cursor()
        c.execute('SELECT * FROM file_progress WHERE file_path = ?', (file_path,))
        return c.fetchone()
    
    def update_file_progress(self, file_path, total_lines, processed_lines, status):
        """更新文件翻译进度"""
        c = self.get_connection().cursor()
        c.execute('''
            INSERT OR REPLACE INTO file_progress 
            (file_path, total_lines, processed_lines, status, last_updated)
            VALUES (?, ?, ?, ?, CURRENT_TIMESTAMP)
        ''', (file_path, total_lines, processed_lines, status))
        self.get_connection().commit()
    
    def get_line_progress(self, file_path, line_index):
        """获取行翻译进度"""
        c = self.get_connection().cursor()
        c.execute('''
            SELECT * FROM line_progress 
            WHERE file_path = ? AND line_index = ?
        ''', (file_path, line_index))
        return c.fetchone()
    
    def update_line_progress(self, file_path, line_index, original_text, translated_text, status):
        """更新行翻译进度"""
        c = self.get_connection().cursor()
        c.execute('''
            INSERT OR REPLACE INTO line_progress 
            (file_path, line_index, original_text, translated_text, status, updated_at)
            VALUES (?, ?, ?, ?, ?, CURRENT_TIMESTAMP)
        ''', (file_path, line_index, original_text, translated_text, status))
        self.get_connection().commit()
    
    def log_error(self, file_path, line_index, error_type, error_message):
        """记录错误"""
        c = self.get_connection().cursor()
        c.execute('''
            INSERT INTO error_log 
            (file_path, line_index, error_type, error_message)
            VALUES (?, ?, ?, ?)
        ''', (file_path, line_index, error_type, error_message))
        self.get_connection().commit()
    
    def get_error_stats(self):
        """获取错误统计信息"""
        c = self.get_connection().cursor()
        c.execute('''
            SELECT 
                COUNT(*) as total_errors,
                COUNT(CASE WHEN resolved_at IS NULL THEN 1 END) as unresolved_errors,
                COUNT(CASE WHEN created_at > datetime('now', '-1 hour') THEN 1 END) as recent_errors
            FROM error_log
        ''')
        return c.fetchone()

class AsyncTranslationManager:
    def __init__(self):
        self.semaphore = asyncio.Semaphore(config.get('openai', 'max_concurrent_requests'))
        self.session = None

class TranslationCache:
    def __init__(self):
        self.cache = {}
        self.max_size = config.get('translation', 'cache_size')
        self.hits = 0
        self.misses = 0

# 创建全局实例
line_count_manager = TranslationStats()
db_manager = DatabaseManager()
async_translation_manager = AsyncTranslationManager()
translation_cache = TranslationCache()

# 添加版本控制
VERSION = "1.0.1" # 版本号，用于区分不同版本的翻译
line_count = 2 # 每组行数，越大越快，但越容易出错

def get_completed_groups(conn, file_path):
    """获取已完成的翻译行"""
    c = conn.cursor()
    c.execute('''
        SELECT line_index, translated_text 
        FROM line_progress 
        WHERE file_path = ? AND status = 'completed'
        ORDER BY line_index
    ''', (file_path,))
    return c.fetchall()

# """ - 输出内容要求用代码块包裹起来
# ，只在必要时提供相应的语言注释
#  """
@retry(
    stop=stop_after_attempt(MODEL_CONFIG['max_retries']),
    wait=wait_exponential(multiplier=1, min=4, max=10),
    retry=retry_if_exception_type((openai.APIError, openai.APITimeoutError)),
    before_sleep=lambda retry_state: logging.warning(f"重试第 {retry_state.attempt_number} 次...")
)
def translate_text(text):
    """翻译文本，使用流式输出"""
    try:
        messages = [
            {
                "role": "system",
                "content": "- 你名为epub翻译大师，专注于将任意语言的文本翻译成中文。- 你在翻译过程中，力求保留原文语意，确保翻译的准确性和完整性。- 你特别注重翻译结果要贴合现代人的阅读习惯，使译文更加流畅易懂。- 在处理包含代码结构的文本时，你会特别注意保持代码的原样。- 你的服务旨在为用户提供高效、便捷的翻译体验，帮助用户跨越语言障碍。- 在回答问题的时候，尽可能保留原来的代码结构。- 在回答问题的时候，尽可能只返回翻译后的内容和代码结构，不要返回任何其他内容。"
                # zhipu
#                  "content": """
# # 角色
# 你是epub翻译大师，专注于精准翻译任意语言文本至中文。
# # 工作任务
# 翻译文本，确保语义准确、完整，并贴合现代阅读习惯。
# # 输出示例
# ```
# 你好，世界！

# print("你好，世界！")
# ```
# # 注意事项
# 在翻译代码结构时，保持原样，只在必要时添加注释。
#                 """

            },
            {
                "role": "user",
                "content": text
            }
        ]
        
        # 使用流式输出
        stream = config.client.chat.completions.create(
            model=MODEL_CONFIG['model_name'],
            messages=messages,
            timeout=MODEL_CONFIG['timeout'],
            stream=True  # 启用流式输出
        )
        
        # 收集流式输出的内容
        translated_text = ""
        for chunk in stream:
            if chunk.choices[0].delta.content is not None:
                content = chunk.choices[0].delta.content
                translated_text += content
                # 实时打印翻译内容
                print(content, end='', flush=True)
        
        print()  # 换行
        
        # 更新统计信息
        translation_stats.update_stats(text, translated_text, True)
        
        return translated_text
        
    except Exception as e:
        logging.error(f"翻译出错: {str(e)}")
        
        # 更新统计信息
        translation_stats.update_stats(text, "", False)
        
        raise

def process_html_file(file_path, conn):
    """处理HTML文件"""
    # 检查文件进度
    progress = db_manager.get_file_progress(file_path)
    
    try:
        # 尝试不同的编码方式读取文件
        encodings = ['utf-8', 'gbk', 'gb2312', 'latin1']
        content = None
        
        for encoding in encodings:
            try:
                with open(file_path, 'r', encoding=encoding) as f:
                    content = f.read()
                break
            except UnicodeDecodeError:
                continue
        
        if content is None:
            raise Exception(f"无法使用支持的编码读取文件: {file_path}")
    
        # 使用正则表达式提取body标签内的内容和title标签
        body_pattern = re.compile(r'<body[^>]*>(.*?)</body>', re.DOTALL)
        title_pattern = re.compile(r'<title>(.*?)</title>', re.DOTALL)
        
        body_match = body_pattern.search(content)
        title_match = title_pattern.search(content)
        
        if not body_match:
            print(f"警告: {file_path} 中没有找到body标签")
            return
        
        body_content = body_match.group(1)
        
        # 处理title标签
        if title_match:
            title_content = title_match.group(1).strip()
            if title_content:  # 只有当标题内容不为空时才处理
                print(f"\n翻译标题: {title_content}")
                translated_title = translate_text(title_content)
                # 替换原始title内容
                content = content.replace(f"<title>{title_content}</title>", f"<title>{translated_title}</title>")
            else:
                print("\n跳过空标题")
        
        # 按行分割body内容
        lines = [line.strip() for line in body_content.split('\n') if line.strip()]
        total_lines = len(lines)
        
        # 获取已完成的翻译
        completed_lines = get_completed_groups(conn, file_path)
        completed_indices = {line[0] for line in completed_lines}
        
        # 计算已处理的进度
        if progress:
            print(f"文件 {file_path} 已处理进度: {progress['processed_lines']}/{progress['total_lines']} 行 ({round(progress['processed_lines']*100/progress['total_lines'], 2)}%)")
        
        # 按4行一组处理内容
        translated_lines = []
        try:
            with tqdm(range(0, len(lines), 4), desc=f"处理文件 {os.path.basename(file_path)}", unit="组") as pbar:
                for i in pbar:
                    # 获取当前组的行
                    group_lines = lines[i:i+4]
                    
                    # 检查是否已完成
                    if all(i+j in completed_indices for j in range(len(group_lines))):
                        # 使用已完成的翻译
                        for j in range(len(group_lines)):
                            for line in completed_lines:
                                if line[0] == i+j:
                                    translated_lines.append(line[1])
                                    break
                        continue
                    
                    # 将当前组的行合并成一个字符串
                    group_text = '\n'.join(group_lines)
                    print(f"\n翻译第 {i+1}-{min(i+4, len(lines))}/{len(lines)} 行:")
                    
                    # 翻译当前组
                    translated_group = translate_text(group_text)
                    
                    # 将翻译结果按行分割并添加到结果列表
                    translated_lines.extend(translated_group.split('\n'))
                    
                    # 更新文件进度
                    db_manager.update_file_progress(file_path, total_lines, min(i+4, len(lines)), 'in_progress')
                    
                    # 显示当前统计信息
                    stats = translation_stats.get_stats()
                    pbar.set_postfix(stats)
                    
                    # 添加较小的延迟以避免API限制
                    time.sleep(0.1)
            
            # 替换原始内容
            if translated_lines:
                # 构建新的body内容
                new_body_content = '\n'.join(translated_lines)
                
                # 替换原始内容中的body部分
                new_content = content.replace(body_content, new_body_content)
                
                # 保存修改后的文件
                output_dir = config.get('paths', 'output_dir')
                os.makedirs(output_dir, exist_ok=True)
                output_path = os.path.join(output_dir, os.path.basename(file_path))
                
                with open(output_path, 'w', encoding='utf-8') as f:
                    f.write(new_content)
                
                # 更新完成状态
                db_manager.update_file_progress(file_path, total_lines, total_lines, 'completed')
                print(f"文件 {file_path} 翻译完成，已保存到 {output_path}")
                
                # 显示最终统计信息
                print("\n翻译统计信息:")
                for key, value in translation_stats.get_stats().items():
                    print(f"{key}: {value}")
                
        except KeyboardInterrupt:
            print("\n检测到中断，保存当前进度...")
            if 'processed_lines' in locals():
                db_manager.update_file_progress(file_path, total_lines, processed_lines, 'interrupted')
            # 显示中断时的统计信息
            print("\n中断时的统计信息:")
            for key, value in translation_stats.get_stats().items():
                print(f"{key}: {value}")
            raise
        except Exception as e:
            print(f"处理文件时出错: {str(e)}")
            if 'processed_lines' in locals():
                db_manager.update_file_progress(file_path, total_lines, processed_lines, 'error')
            raise
            
    except Exception as e:
        print(f"读取文件时出错: {str(e)}")
        return

def main():
    ops_dir = config.get('paths', 'input_dir')
    html_files = [f for f in os.listdir(ops_dir) if f.endswith('.htm') or f.endswith('.html')]
    
    # 按文件名排序
    html_files.sort()
    
    total_files = len(html_files)
    print(f"找到 {total_files} 个HTML文件需要处理")
    print(f"开始时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
    
    # 初始化数据库连接
    conn = db_manager.get_connection()
    
    try:
        for file_index, filename in enumerate(html_files, 1):
            file_path = os.path.join(ops_dir, filename)
            print(f"\n开始处理第 {file_index}/{total_files} 个文件: {filename}")
            print("-" * 50)
            
            # 检查文件是否已完成
            progress = db_manager.get_file_progress(file_path)
            if progress and progress['status'] == 'completed':
                print(f"文件 {filename} 已经完成翻译，跳过")
                continue
            
            try:
                process_html_file(file_path, conn)
                print(f"\n完成第 {file_index}/{total_files} 个文件: {filename}")
                print("-" * 50)
            except Exception as e:
                print(f"\n处理文件 {filename} 时出错: {str(e)}")
                print("继续处理下一个文件...")
                continue
            
            # 显示当前总体进度
            completed_files = sum(1 for f in html_files[:file_index] 
                                if db_manager.get_file_progress(os.path.join(ops_dir, f)) and 
                                db_manager.get_file_progress(os.path.join(ops_dir, f))['status'] == 'completed')
            print(f"\n总体进度: {completed_files}/{total_files} 个文件完成 "
                  f"({round(completed_files*100/total_files, 2)}%)")
            
            # 显示统计信息
            print("\n当前统计信息:")
            for key, value in translation_stats.get_stats().items():
                print(f"{key}: {value}")
            
            # 在文件之间添加短暂延迟
            if file_index < total_files:
                print("\n等待 5 秒后处理下一个文件...")
                time.sleep(5)
    
    except KeyboardInterrupt:
        print("\n程序被用户中断")
    finally:
        db_manager.close()
        print(f"\n结束时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
        print("\n最终统计信息:")
        for key, value in translation_stats.get_stats().items():
            print(f"{key}: {value}")

if __name__ == "__main__":
    main()