easyTrans.py

import os
import subprocess
import sys
import time
import json  # 添加 json 模块导入
import traceback
import requests
import re  # 添加正则表达式模块导入
from docx import Document
from docx.enum.style import WD_STYLE_TYPE

# 选择文件对话框
try:
    import tkinter as tk
    from tkinter import filedialog, messagebox
except Exception:
    tk = None

# 进度条
try:
    from tqdm import tqdm
except Exception:
    tqdm = None

# 配置
SUFFIX = " 白话文"
OLLAMA_HOST = "http://localhost:11434"  # 改为实际的 ollama 服务器地址
MODEL = "deepseek-r1:8b"
#MODEL = "qwen3:8b"
# INSTRUCTION = (
# '''你是一个文言文到现代白话文的翻译器，需严格按照以下要求执行翻译任务：
# 输出要求：直接输出翻译结果，不进行任何解释、摘要或附加说明。确保译文简洁、直译为主、通俗易懂，适合通过TTS朗读。
# 内容完整性：
# 不可跳过、删减或添加原文中的任何内容。
# 确保每一句原文都有对应的翻译，不遗漏任何信息。
# 表达方式：
# 译文需贴近现代白话文的表达习惯，避免使用文言文句式和词汇。
# 保持口语化风格，确保朗读时自然流畅。
# 人名处理：
# 文言文中习惯用单字称呼人名（如释道安简称为“安”，佛图澄简称为“澄”），翻译时需补全为全名（如“释道安”“佛图澄”），以避免歧义。
# 格式要求：
# 译文后不可加括号附上原文，确保译文简洁且适合朗读。'''
# )

INSTRUCTION = (
'''你是一个文言文到现代白话文的翻译器，需严格按照以下要求执行翻译任务：
翻译要求：
1. 完全逐句对应，不得意译或总结。
2. 不得省略任何内容。
3. 用简体中文翻译。文需贴近现代白话文的表达习惯，避免使用文言文句式和词汇
4. 保留人名、典故、地名。文言文中习惯用单字称呼人名（如释道安简称为“安”，佛图澄简称为“澄”），翻译时需补全为全名（如“释道安”“佛图澄”），以避免歧义。'''
)

# helper: 解析段落样式名得到 heading level
def heading_level(paragraph):
    style = paragraph.style
    if style is None:
        return None
    name = getattr(style, "name", "")
    if not isinstance(name, str):
        return None
    name = name.lower()
    if "heading" in name:
        for i in range(1, 10):
            if f"heading {i}" in name or f"heading{i}" in name:
                return i
    return None


def call_ollama_stream(prompt):
    """通过 REST API 调用 ollama"""
    full_prompt = f"{INSTRUCTION}\n\n原文：\n{prompt}"
    
    headers = {
        "Content-Type": "application/json"
    }
    
    data = {
        "model": MODEL,
        "prompt": full_prompt,
        "stream": True,
        "options": {
            "temperature": 0.1,  # 降低随机性，使输出更确定
            "top_p": 0.9
        }
    }
    
    try:
        # 使用 stream=True 来流式获取响应
        response = requests.post(
            f"{OLLAMA_HOST}/api/generate",
            headers=headers,
            json=data,
            stream=True
        )
        
        if response.status_code != 200:
            raise RuntimeError(f"API 调用失败: {response.text}")
        
        # 收集输出
        result = []
        for line in response.iter_lines(decode_unicode=True):
            if not line:
                continue
            # 解析 JSON 响应
            try:
                resp = json.loads(line)
                if "response" in resp:
                    chunk = resp["response"]
                    # 直接添加到结果中
                    result.append(chunk)
            except json.JSONDecodeError:
                continue
        
        # 将所有结果合并为一个字符串
        full_output = "".join(result)
        # 使用正则表达式去掉 <think> 包裹的内容
        cleaned_output = re.sub(r'<think>.*?</think>', '', full_output, flags=re.DOTALL).strip()
        
        return cleaned_output
        
    except requests.exceptions.RequestException as e:
        raise RuntimeError(f"调用 Ollama API 失败: {str(e)}")

# def call_ollama(prompt, timeout=600):
#     full_prompt = f"{INSTRUCTION}\n\n原文：\n{prompt}"
#     try:
#         proc = subprocess.run(
#             [OLLAMA_CMD, "run", MODEL],
#             input=full_prompt,        # ✅ 改这里
#             capture_output=True,
#             text=True,
#             timeout=timeout,
#         )
#     except FileNotFoundError:
#         raise RuntimeError("未找到 ollama 可执行程序，请确保已安装并在 PATH 中。")
#     except subprocess.TimeoutExpired:
#         raise RuntimeError("调用 ollama 超时。")

#     if proc.returncode != 0:
#         raise RuntimeError(f"调用 ollama 失败: {proc.stderr.strip()}")
#     return proc.stdout.strip()


def split_text_by_punctuation(text, max_length=500):
    """按标点符号拆分文本，确保句子完整性"""
    # 如果文本长度小于最大长度，直接返回
    if len(text) <= max_length:
        return [text]
    
    # 使用正则表达式找到完整的句子
    # 匹配到句子末尾的标点符号，包括句子内容
    pattern = r'[^，。！？,\.!?；;：:\n]+[，。！？,\.!?；;：:\n]'
    sentences = re.finditer(pattern, text)
    
    result = []
    current_chunk = []
    current_length = 0
    
    for sentence in sentences:
        sent = sentence.group()
        sent_length = len(sent)
        
        # 如果当前块加上新句子不超过最大长度
        if current_length + sent_length <= max_length:
            current_chunk.append(sent)
            current_length += sent_length
        else:
            # 如果当前块非空，加入结果列表
            if current_chunk:
                result.append(''.join(current_chunk))
            # 开启新的块
            current_chunk = [sent]
            current_length = sent_length
    
    # 处理最后一个块
    if current_chunk:
        result.append(''.join(current_chunk))
    
    # 处理可能的剩余文本（没有标点符号结尾的部分）
    last_end = sum(len(x) for x in result)
    remaining = text[last_end:].strip()
    if remaining:
        result.append(remaining)
    
    return result

# 把整个 h4 下的内容一次性发送给模型并返回分段结果
def translate_section(paragraph_texts, pause=0.5, max_chars=1000):
    """翻译章节内容，对超长文本进行分段处理"""
    # 移除 None 并保留原始空字符串以保序
    paras = [p if p is not None else "" for p in paragraph_texts]
    joined = "\n\n".join(paras).strip()
    if not joined:
        return [""] * len(paragraph_texts)

    # 检查总字数
    total_chars = len(joined)
    if total_chars <= max_chars:
        # 如果字数在限制范围内，直接翻译
        print(f"\n开始翻译整章（字数：{total_chars}）：")
        out = call_ollama_stream(joined)
        time.sleep(pause)
    else:
        # 超过字数限制，需要分段处理
        print(f"\n文本较长（{total_chars}字），进行分段翻译：")
        chunks = split_text_by_punctuation(joined)
        translated_chunks = []
        
        for i, chunk in enumerate(chunks, 1):
            print(f"\n翻译第 {i}/{len(chunks)} 段（{len(chunk)}字）：\n{chunk}\n")
            chunk_result = call_ollama_stream(chunk)
            translated_chunks.append(chunk_result)
            time.sleep(pause)
        
        # 合并所有翻译结果
        out = "\n\n".join(translated_chunks)

    # 将翻译结果分段
    parts = [p.strip() for p in out.split("\n\n") if p.strip()]
    if not parts:
        return [out.strip()] if out.strip() else [""]

    return parts

# def translate_paragraphs(paragraph_texts, pause=0.5):
#     results = []
#     for txt in paragraph_texts:
#         if not txt.strip():
#             results.append("")
#             continue
#         print("\n开始翻译段落：", txt)
#         out = call_ollama_stream(txt)
#         results.append(out)
#         time.sleep(pause)
#     return results

# 解析输入文档：收集文档开头（在第一个 h4 之前）和所有 h4 章节
def parse_input_h4_sections(input_path):
    doc = Document(input_path)
    paragraphs = doc.paragraphs
    prelude = []  # 段落文本，在第一个 h4 之前，直接复制（不翻译）
    sections = []  # 每个 section: {'heading': text, 'paras':[...]}
    i = 0
    # collect prelude until first h4
    while i < len(paragraphs):
        p = paragraphs[i]
        lvl = heading_level(p)
        if lvl == 4:
            break
        prelude.append({'type': 'para', 'text': p.text or ''} if lvl is None else {'type': 'heading', 'level': lvl, 'text': p.text or ''})
        i += 1

    # collect sections starting at each h4
    while i < len(paragraphs):
        p = paragraphs[i]
        lvl = heading_level(p)
        # 改为检测第 4 级标题（h4）
        if lvl == 4:
            heading_text = p.text or ""
            i += 1
            paras = []
            # collect non-heading paragraphs until next h4
            while i < len(paragraphs):
                q = paragraphs[i]
                qlvl = heading_level(q)
                if qlvl == 4:
                    break
                paras.append(q.text or "")
                i += 1
            sections.append({'heading': heading_text, 'paras': paras})
        else:
            # skip other non-h4 content between h4 blocks (treat them as prelude if before any h4)
            # but to keep simple, attach them to prelude if before any h4 (already done), otherwise ignore here
            i += 1
    return prelude, sections

# 判断输出文档中某个 h4 章节是否已被翻译（判断方法：找到同名 h4，且之后有非空段落）
def is_section_done_in_output(output_path, section_heading):
    if not os.path.isfile(output_path):
        return False
    try:
        outdoc = Document(output_path)
    except Exception:
        return False
    paras = outdoc.paragraphs
    i = 0
    while i < len(paras):
        p = paras[i]
        # 改为检测第 4 级标题（h4）
        if heading_level(p) == 4 and (p.text or "").strip() == (section_heading or "").strip():
            # look ahead for next non-heading paragraph
            j = i + 1
            while j < len(paras):
                if heading_level(paras[j]) is not None:
                    break
                if (paras[j].text or "").strip():
                    return True
                j += 1
            return False
        i += 1
    return False

# 写入 prelude（仅在没有输出文档时写入）
def write_prelude_to_out(out_doc, prelude):
    for item in prelude:
        if item.get('type') == 'heading':
            lvl = item.get('level') or 1
            out_doc.add_heading(item.get('text') or "", level=min(max(int(lvl), 1), 9))
        else:
            out_doc.add_paragraph(item.get('text') or "")

def pick_input_file():
    if tk is None:
        print("tkinter 不可用，请在命令行中直接指定文件路径。")
        return None
    root = tk.Tk()
    root.withdraw()
    path = filedialog.askopenfilename(title="选择要处理的 Word 文档", filetypes=[("Word 文档", "*.docx")])
    root.update()
    root.destroy()
    return path

def main():
    # 选择输入文档
    input_path = pick_input_file()
    if not input_path:
        print("未选择文件，退出。")
        return

    base, ext = os.path.splitext(input_path)
    output_path = f"{base}{SUFFIX}{ext}"

    try:
        prelude, sections = parse_input_h4_sections(input_path)
    except Exception as e:
        print("解析输入文档失败:", str(e))
        traceback.print_exc()
        sys.exit(1)

    # 计算哪些章节已完成
    done_flags = []
    for s in sections:
        done = is_section_done_in_output(output_path, s['heading'])
        done_flags.append(done)

    # remaining count
    remaining = sum(1 for d in done_flags if not d)
    if remaining == 0:
        print("没有未完成的四级章节要翻译。输出文档:", output_path)
        return

    print(f"输入: {input_path}")
    print(f"输出: {output_path}")
    print(f"总章节数: {len(sections)}, 未完成: {remaining}")

    # 加载（或新建）输出文档
    if os.path.isfile(output_path):
        out_doc = Document(output_path)
        print("检测到已有输出文档，将在现有文档末尾追加未完成章节。")
    else:
        out_doc = Document()
        # write prelude (only when creating new output)
        write_prelude_to_out(out_doc, prelude)
        out_doc.save(output_path)
        print("创建新输出文档并写入开头内容。")

    # progress bar
    if tqdm:
        pbar = tqdm(total=remaining, desc="翻译章节")
    else:
        pbar = None
        print("开始翻译（Ctrl+C 可中断）...")

    try:
        for idx, section in enumerate(sections):
            if done_flags[idx]:
                # 已完成，跳过
                continue
            # 在输出中添加此章节标题（避免重复：如果输出中已存在同名 h4 且无内容，仍添加一个新的标题以写入翻译）
            out_doc.add_heading(section['heading'] or "", level=4)
            # 一次性翻译该章节下所有段落
            try:
                translated = translate_section(section['paras'])
            except KeyboardInterrupt:
                print("\n用户中断（在调用模型时）。保存已完成部分并退出。")
                out_doc.save(output_path)
                if pbar:
                    pbar.close()
                return
            except Exception as e:
                print("翻译章节失败:", str(e))
                # 写入原始段落以占位，或选择继续（这里写入空段以便标记）
                for _ in section['paras']:
                    out_doc.add_paragraph("")  # 占位空段
                out_doc.save(output_path)  # 保存失败情况
                if pbar:
                    pbar.update(1)
                continue

            # 写入翻译后的段落，完成后保存（每处理完一个 h4 即保存）
            if translated:
                for t in translated:
                    out_doc.add_paragraph(t)
            else:
                # 保持占位，避免丢失顺序
                for _ in section['paras']:
                    out_doc.add_paragraph("")
            out_doc.save(output_path)

            if pbar:
                pbar.update(1)
            else:
                done_so_far = sum(1 for d in done_flags[:idx+1] if d) + 1  # approximate
                print(f"已完成 {done_so_far}/{remaining}")

    except KeyboardInterrupt:
        print("\n用户中断（Ctrl+C）。已保存当前进度:", output_path)
    finally:
        if pbar:
            pbar.close()
        print("已保存:", output_path)

if __name__ == "__main__":
    main()