feat: Release v0.10 - Modular Architecture & External Config

- Refactor codebase into src/ (preprocessing, translation, assembly)
- Add pipeline/ scripts for individual stages
- Externalize configuration to config/config.yaml
- Fix Cover Image preservation
- Update documentation and manuals
This commit is contained in:
谭凯
2026-01-31 22:49:44 +08:00
parent 9ef82393be
commit 7a93c52b42
306 changed files with 30313 additions and 1071 deletions
+22
View File
@@ -0,0 +1,22 @@
# LLM Configuration
llm:
# Model name (e.g., gpt-4o, gpt-3.5-turbo, deepseek-chat)
model: "gemini-3-flash-preview"
# API Base URL (default is OpenAI)
base_url: "https://api.gpt.ge/v1"
# Timeout for API requests in seconds
timeout: 60
# Rate Limiting
requests_per_minute: 60
concurrent_requests: 16
# Translation Settings
translation:
# Characters per chunk (approximate)
chunk_size: 4000
# System prompt instruction file (optional, overrides default if present)
# style_guide_path: "config/style_guide.txt"
+10
View File
@@ -0,0 +1,10 @@
{
"translation": {
"system": "你是一位精通中英文的专业翻译家。你的任务是将英文书籍内容翻译成中文。\n\n【核心要求】\n1. 准确传达原文含义,语言流畅自然,符合中文阅读习惯。\n2. 保持原文的风格和语气。\n\n【格式要求 - 极其重要】\n1. 每行格式:#N: 译文(N是行号,必须原样保留)\n2. 输入多少行,输出必须是相同数量的行\n3. **占位符规则**:原文中的 φXφ 和 φ/Xφ 标记必须原样保留在译文中\n - 成对标记:\"φ1φBoldφ/1φ\" → \"φ1φ粗体φ/1φ\"\n - 单体标记:\"φ2φ\" 表示公式或符号,保持位置不变\n - 尾注锚点:文末的 \"φ3φ\" 是超链接,必须保留\n\n【禁止事项】\n- 禁止修改 #N: 行号\n- 禁止删除或修改任何 φXφ 标记\n- 禁止添加解释或注释\n- 禁止合并或拆分行",
"user_template": "请翻译以下段落(务必原样保留所有 φnφ 格式标记):\n\n{{content}}"
},
"glossary_extraction": {
"system": "你是一位资深的文学编辑和领域专家。你的任务是分析书籍样本,提取关键术语并制定统一的译名表。",
"user_template": "请阅读以下书籍片段(包含前言和正文采样)。\n\n任务:\n1. 识别文中出现的人名(如 'Masa', 'Steve Jobs')、地名、机构名。\n2. 识别特定的行业术语或关键概念。\n3. 为上述词汇提供标准的中文译名。如果像 'Masa' 这样的昵称有对应的全名(如孙正义),请务必使用全名。\n\n请以 JSON 格式输出,格式如下:\n{\n \"Masa\": \"孙正义\",\n \"Apple\": \"苹果公司\",\n ...\n}\n\n书籍片段:\n\n{{content}}"
}
}
+90
View File
@@ -0,0 +1,90 @@
# Operation Manual & Change Log
## Core Principles
1. **Modularity**: The system is divided into three distinct phases (Preprocessing, Translation, Assembly) with clear boundaries.
2. **Immutability**: `book_structure.json` is generated once during preprocessing and should not be modified by subsequent steps.
3. **Source of Truth**: `manifest.json` is the single source of truth for translations.
4. **Idempotency**: Translation steps can be retried without side effects (existing translations are preserved).
## Directory Structure
* `pipeline/`: Executable scripts for each stage.
* `01_preprocess.py`: Clean EPUB, generate structure, extract text.
* `02_translate.py`: Translate text in manifest.
* `03_assemble.py`: Apply translations and build final EPUB.
* `src/`: Core logic modules.
* `preprocessing/`: Cleaning, extraction, profiling.
* `translation/`: LLM integration, manifest management.
* `assembly/`: Backfilling, EPUB building.
* `common/`: Shared data models and utils.
* `work/`: Working directory for intermediate files (ignored by git).
## Pipeline Usage
### Step 1: Preprocessing
```bash
python pipeline/01_preprocess.py inputs/my_book.epub
```
Generates `work/my_book/book_structure.json` and `manifest.json`.
### Step 2: Translation
```bash
python pipeline/02_translate.py --input-epub inputs/my_book.epub
```
Translates entries in `manifest.json`. ensuring `.env` has `OPENAI_API_KEY`.
### Step 3: Assembly
```bash
python pipeline/03_assemble.py inputs/my_book.epub --mode bilingual
```
Generates `output/my_book_bilingual.epub`.
## Configuration
System settings are managed via `config/config.yaml` and environment variables.
### `config/config.yaml`
Control LLM parameters and translation behavior:
```yaml
llm:
model: "gpt-3.5-turbo" # LLM Model Name
base_url: "https://api.openai.com/v1"
timeout: 60
requests_per_minute: 60 # Rate limiting
concurrent_requests: 5 # Parallel chunks
translation:
chunk_size: 4000 # Characters per chunk
```
### Environment Variables (`.env`)
Security-sensitive credentials must be set here:
```bash
OPENAI_API_KEY=sk-... # Required
OPENAI_BASE_URL=... # Optional override for config
```
## Known Issues & Troubleshooting
### Missing Placeholders Warning
During assembly, you may see logs like:
`WARNING - Restoration warning: missing placeholders {'1'}`
This indicates that the LLM translation missed a placeholder tag (e.g. `φ1φ`). The system attempts to recover, but this warning is logged for review. These are usually minor and do not prevent EPUB generation.
## Change Log
### [2026-01-28] Bug Fixes
* **Fix Cover Image**: Resolved issue where book cover execution was missing in the final EPUB. Added `cover_image_id` tracking in `BookStructure` and restored proper OPF metadata in `BilingualBuilder`.
### [2026-01-27] Externalized Configuration
* **Config**: Added `config/config.yaml` for tuning parameters (LLM model, RPM, Chunk Size).
* **Logic**: `pipeline/02_translate.py` now loads settings from `config.yaml`.
* **Dependency**: Added `PyYAML` to `requirements.txt`.
### [2026-01-27] Architecture Refactoring
* **Restructured**: Moved source files into `src/preprocessing`, `src/translation`, `src/assembly`, `src/common`.
* **Pipeline**: Created individual pipeline scripts in `pipeline/`.
* **Refactor**: Renamed `fine_grained_extractor` to `text_extractor`, `translator` to `translator_engine`, etc.
* **Logic**: Enforced 100% text coverage check in `format_extractor.py` (removed 95% threshold).
* **Docs**: Created this Operation Manual.
+100
View File
@@ -0,0 +1,100 @@
# ePub Bilingual Translator - Architecture & Data Flow (Revised)
本文档详细描述了程序处理一个 EPUB 文件的完整生命周期。整个流程旨在实现**结构稳定性**(不丢段落)与**内容精细度**(不丢格式)的最佳平衡。
## High Level Data Flow
```mermaid
graph TD
Input[Input EPUB] --> Cleaner[EpubCleaner]
Cleaner --> CleanedEPUB[1. Cleaned EPUB Temp]
CleanedEPUB --> Profiler[Book Profiler]
Profiler -->|Identify| Profile[Book Profile / Style]
CleanedEPUB --> Extractor[FineGrainedExtractor]
subgraph Extraction
Extractor -->|Step 1: P/H Tags| P[Source Paragraph]
P -->|Step 2: FormatExtract| PhText[Analyzed Text]
PhText -->|Register| Manifest[Manifest DB]
end
Manifest -->|Batch| LLM[LLM Translation]
Profile -->|Prompt Context| LLM
LLM -->|Translation| Manifest
Manifest --> Restorer[FormatRestorer]
Restorer -->|Reconstruct HTML| TargetHtml[Target HTML]
CleanedEPUB --> Backfiller[Backfill Engine]
TargetHtml --> Backfiller
Backfiller -->|Bilingual/Chinese Mode| DOM[Final DOM]
DOM --> Builder[BilingualBuilder]
Builder --> Output[Output EPUB]
```
## Detailed Workflow
### 1. Preprocessing (预处理)
**模块**: `src/epub_cleaner.py`
* **Flatten Structure**: 消除嵌套 `div`,统一转为 `<p>`,消除结构性漏译风险。
* **Auto-Fix**: 修复 TOC 死链、缺失 UID、由于 `ebooklib` bug 导致的样式丢失。
* **Result**: 产生一个标准的临时文件,后续所有操作基于此文件,不再受原始糟糕格式影响。
### 2. Intelligent Extraction (智能提取)
**模块**: `src/fine_grained_extractor.py` + `src/format_extractor.py`
#### A. 结构层 (Macro)
使用 `FineGrainedExtractor` 锁定所有正文元素 (`p`, `h1`-`h6`)。
* **Filter**: 排除页码、页眉脚。
* **Optimization**: 针对目录章节,识别 **罗马数字 (I, II)**、**单独数字 (1, 2)**、**修饰符 (***)**,这些内容**不送翻译**,直接在回填时保留原文,以维持原书排版美感。
#### B. 内容层 (Micro)
对每个提取的段落调用 `FormatExtractor`
* **Inline Style**: 将 `<b>`, `<i>` 转为配对占位符 `φ1φ...φ/1φ`
* **Formula Protection**: 识别 $E=mc^2$ 等数学公式,保护为不可变占位符。
* **Drop Cap Handling**:
- 原始: `<span class="dropcap">T</span>he`
- 提取给 LLM: "The" (完整单词,无格式干扰)
- 记录: Prefix 包含 Drop Cap 样式。
### 3. Manifest Management (清单管理)
**模块**: `src/manifest_manager.py`
Manifest 是系统的**核心状态中心 (Source of Truth)**。
* **作用**: 解耦提取和翻译。提取器只管往 Manifest 填数据,翻译器只管从 Manifest 取数据。
* **Persistence**: 支持中断续传,翻译进度实时保存。
### 4. Translation with Profiling (翻译)
**模块**: `src/book_profiler.py` & `src/translator.py`
* **Profiling**: 在翻译前,抽取部分文本分析书籍的类型(技术、小说、诗歌)、核心术语和语言风格,生成 `System Prompt`
* **Translation**: 这是纯文本层面的转换,LLM 处理的是带有 `φ` 占位符的文本。
### 5. Robust Restoration (健壮还原)
**模块**: `src/format_restorer.py`
负责将 LLM 返回的文本还原为 HTML。
* **Drop Cap Logic**:
- **原文回填**: 需要 Prefix `<span class="dropcap">T</span>`
- **译文回填**: **丢弃** Drop Cap Prefix。中文不需要首字母下沉,否则会出现 "T这本书..." 的怪诞结果。
* **Error Handling**:
- **Missing Placeholders**: 如果 LLM 丢了 `φ1φ`,自动在末尾补全或报错重试。
- **Hallucinated Placeholders**: 移除 LLM 臆造的不存在 ID。
### 6. Backfill Strategy (回填策略)
**模块**: `src/fine_grained_extractor.py` (backfill method)
支持多种模式,且**严格遵循一对一 (One-to-One) 映射**,绝不依赖顺序,而是依赖元素的内存引用或唯一 ID。
* **Mode A: Bilingual (双语)**
- 保留原文 DOM。
- 在原文后 `append` 一个新元素 `<p class="translation">译文</p>`
- 样式继承:译文元素复制原文的 `margin`, `text-align` 等关键样式。
* **Mode B: Chinese Only (仅中文)**
- **Replace**: 直接用译文元素替换原文元素。
- **Drop Cap 适配**: 此时译文通常为普通段落,不再保留首字母下沉样式,以符合中文排版习惯。
### 7. Assembly (组装)
**模块**: `src/bilingual_builder.py`
将内存中的 DOM 序列化,重新打包资源,生成最终 EPUB。
+180
View File
@@ -0,0 +1,180 @@
# 预处理与回填层技术文档
本文档详细描述 EPUB 双语翻译器的预处理(Preprocessing)和回填(Backfill)层的技术方案、数据结构、问题解决方案,用于指导代码开发、调试和维护。
---
## 1. 架构概览
```
┌─────────────┐ ┌────────────────┐ ┌───────────────────┐
│ EPUB 文件 │ ──► │ EpubCleaner │ ──► │ book_structure.json│
└─────────────┘ └────────────────┘ └───────────────────┘
┌────────────────┐ ┌───────────────────┐
│FineGrainedExt. │ ──► │ manifest.json │
└────────────────┘ └───────────────────┘
┌────────────────┐ │
│ BackfillEngine │ ◄───────────┘
└────────────────┘
┌────────────────┐ ┌───────────────────┐
│BilingualBuilder│ ──► │ 输出 EPUB │
└────────────────┘ └───────────────────┘
```
---
## 2. 目录结构
```
work/
├── {book_name}/ # 每本书独立的工作目录
│ ├── book_structure.json # 书籍结构(会被复用)
│ ├── manifest.json # 翻译清单(持久化,Source of Truth
│ └── assets/ # 解压出的 EPUB 资源文件
│ └── OEBPS/images/... # 保留原始路径结构
└── translations/ # (可选) 翻译记忆或其他中间文件
```
**关键设计**`book_structure.json` **默认复用**。为避免 UUID 不一致导致回填失败,除非显式指定强制清理,否则程序优先读取现有的结构文件。
---
## 3. 数据结构定义
### 3.1 BookStructure (book_structure.json)
记录书籍的“骨架”,确保翻译后的章节能按正确顺序和层级重组。
```json
{
"metadata": {
"title": "书名",
"author": "作者",
"language": "en",
"identifier": "ISBN或UUID"
},
"spine": ["item_id_1", "item_id_2", ...], // 阅读顺序
"resources": {
"item_id_1": {
"href": "OEBPS/chapter1.xhtml",
"media_type": "application/xhtml+xml",
"content": "<html>...</html>", // 清理并注入ID后的HTML内容
"properties": "nav"
},
...
}
}
```
### 3.2 ManifestEntry (manifest.json)
翻译清单是全生命周期的核心,记录了所有待翻译段落的状态。
```json
[
{
"entry_id": "OEBPS/c3Z.xhtml#uuid-603ff3fb",
"file_path": "OEBPS/c3Z.xhtml",
"element_id": "uuid-603ff3fb",
"original_text": "φ1φHelloφ/1φ world.",
"placeholders": {
"1": "<b>",
"/1": "</b>",
"_prefix": "<p>",
"_suffix": "</p>"
},
"translated_text": "φ1φ你好φ/1φ 世界。",
"context": "body"
}
]
```
| 字段 | 类型 | 说明 |
|------|------|------|
| `entry_id` | str | **全局唯一ID** (`file_path#element_id`),用于精准追踪。 |
| `file_path` | str | 标识该段落属于哪一章,用于按章分组批处理。 |
| `element_id` | str | HTML DOM 元素的 ID,回填时的锚点。 |
| `original_text` | str | 经过智能抽提和占位符化后的文本,发送给 LLM。 |
| `placeholders` | Dict | 格式映射表,用于还原 HTML 结构。 |
| `translated_text`| str | 翻译结果(含占位符),初始为 null。 |
---
## 4. 预处理与抽提流程 (Extraction)
流程由 `FormatExtractor` 驱动,分为三个阶段:
### Step 1: 结构索引 (Structure Indexing)
* **动作**: 解析 OPF 文件,提取 Spine 和 Metadata。
* **目的**: 建立骨架,确定处理顺序。
### Step 2: 语义识别 (Semantic Detection)
* **动作**: 使用 `HeadingDetector` 分析 DOM 节点。
* **逻辑**:
* 匹配 `h1`-`h6` 正则,区分 `Chapter`(章)与 `Section`(节)。
* 识别 `blockquote` 或特定 class 判定 `Epigraph`(引言)。
* **目的**: 为 LLM 提供差异化的 Prompt(例如翻译标题时不要加句号)。
### Step 3: 智能抽提 (Smart Extraction - `_smart_extract_v3`)
这是核心算法,将 HTML 转换为“纯文本+占位符”。
1. **首尾分离 (Prefix/Suffix Separation)**:
* 将包裹文本的外层标签(如 `<p>`, `div`)剥离到 `_prefix``_suffix`
* **目的**: 极大减少 LLM 输入 Token,且防止 LLM 随意修改外层布局。
2. **首字下沉处理 (Drop Cap Handling)**:
* 检测并合并被 `<span>` 单独包裹的首字母(如 `<span class="drop">O</span>` + `nce``Once`)。
* **目的**: 修复语意割裂,让 LLM 看到完整的单词。
3. **占位符化 (Placeholder Mapping)**:
* 将内联标签(`<a>`, `<em>`)或公式替换为短码 `φIDφ`
* **目的**: 保护 HTML 属性不被“翻译”,降低噪声干扰。
4. **完整性校验 (Integrity Check)**:
* **逻辑**: 抽提后的文本(去占位符)与原始纯文本进行归一化比对,要求覆盖率 **100%**
* **兜底**: 若校验失败(如误删内容),回退到简单模式(只剥离首尾标签)。
---
## 5. 构建与回填流程 (Backfill & Build)
### 5.1 翻译回填
* 根据 `entry_id` 找到对应的 DOM 节点。
* 使用 `FormatRestorer``translated_text` 中的占位符(`φ1φ`)还原为原始 HTML 标签(`<b>`)。
* 根据模式(双语/单语)决定将新节点插入到原文后还是替换原文。
### 5.2 链接修复 (Link Repair)
`BilingualBuilder` 中执行:
1. **ID 补全**: 为缺失 ID 的 TOC 节点自动生成 UUID。
2. **死链检测**: 检查 TOC/Nav 指向的文件是否存在。
3. **模糊修复**: 尝试通过文件名后缀匹配(解决路径前缀变更问题)或特定重定向(如 `c0.xhtml` -> `cover.xhtml`)。
4. **坏死剔除**: 无法修复的死链将从目录中移除。
### 5.3 CSS 样式恢复
`EbookLib` 默认可能会重写 `<head>` 导致样式丢失。
* **逻辑**:
* 收集所有 CSS 资源。
* 在构建每个 HTML Item 时,显式计算 HTML 到 CSS 的**相对路径**。
* 强制调用 `item.add_link(..., rel='stylesheet', type='text/css')` 注入引用。
---
## 6. 常见问题排查
### 6.1 UUID 不匹配
**现象**: `WARNING - Element uuid-xxx not found`.
**原因**: 手动删除了 `book_structure.json` 但保留了 `manifest.json`,导致重新生成的 HTML ID 与清单记录不一致。
**解决**: 清空 `work/BookName` 目录重新运行,或确保两个 JSON 文件版本一致。
### 6.2 样式丢失
**现象**: 打开书面目全非,只有黑白文字。
**检查**: 解压 EPUB,查看 HTML `<head>` 是否有 `<link rel="stylesheet">`。如果没有,检查 `BilingualBuilder` 的 Step 4 逻辑。
### 6.3 翻译错位
**现象**: 译文出现在了错误的位置。
**检查**: 确认 `entry_id` 生成逻辑是否包含文件名,且文件名在处理过程中未被意外修改。
+108
View File
@@ -0,0 +1,108 @@
# Translation Layer Technical Documentation
This document details the architecture of the **Translation Layer**, which operates primarily on `manifest.json`.
> **Core Principle**: The Translation Layer is **decoupled** from the EPUB file format. It reads translatable units from `manifest.json`, processes them using an LLM, and writes translations back to `manifest.json`. It does **not** read or parse the EPUB file directly.
---
## 1. Architecture Overview
### Data Flow
```mermaid
graph LR
A[manifest.json] -->|Load| B(ManifestManager)
B -->|Entries| C{Translator Orchestrator}
C -->|Sample Text| D[BookProfiler]
D -->|Style Guide| C
C -->|Chunks| E[LLMClient]
E -->|Translation| C
C -->|Update| B
B -->|Save| A[manifest.json]
```
1. **Input**: `manifest.json` (Generated by Preprocessing Layer).
2. **Process**:
* **Profiling**: Analyze text samples to generate a `BookProfile` (style, tone, terminology).
* **Translation**: Batch entries into chunks, send to LLM, receive translations.
3. **Output**: `manifest.json` (Updated with `translated_text` fields).
---
## 2. Core Modules (`src/translation/`)
### 2.1 ManifestManager (`manifest_manager.py`)
* **Role**: The interface for the "Source of Truth".
* **Responsibility**:
* Load `manifest.json`.
* Provide list of `ManifestEntry` objects.
* Save updates back to disk.
* **Key Method**: `update_translation(entry_id, translation)`
### 2.2 Translator Engine (`translator_engine.py`)
* **Role**: Orchestrates the translation process.
* **Responsibility**:
* **Filtering**: Identify untranslated entries.
* **Grouping**: Group entries by chapter (file path) to maintain context.
* **Chunking**: Create character-based chunks (default ~5000 chars) that do not cross chapter boundaries.
* **Concurrency**: Manage async workers (default 5 concurrent tasks).
* **Input**: `List[ManifestEntry]`, `BookProfile`.
* **Output**: Updates `ManifestEntry` objects in-place.
### 2.3 LLM Client (`llm_client.py`)
* **Role**: Handles raw communication with the LLM Provider (OpenAI compatible).
* **Responsibility**:
* **Prompt Engineering**: Construct Short-ID based prompts.
* **Rate Limiting**: Control RPM (Requests Per Minute).
* **Retry Logic**: Exponential backoff for API failures.
* **Logging**: Save raw request/response pairs to `work/{book}/chunks/` for debugging.
### 2.4 Book Profiler (`../preprocessing/profiler.py`)
* **Note**: While located in preprocessing, it is often invoked at the start of the translation phase.
* **Role**: Generates a style guide.
* **Mechanism**: Extracts a sample (Intro + Random segments) from `manifest.json` entries and asks the LLM to analyze author style.
---
## 3. Short ID Strategy
To optimize token usage and ensuring mapping accuracy, we use a **Short ID** system for LLM interaction.
**Prompt Format**:
```text
#1: First paragraph text...
#2: Second paragraph text...
```
**Response Format**:
```text
#1: 第一段翻译...
#2: 第二段翻译...
```
The `LLMClient` maintains a mapping of `Short ID (#N)` <-> `Entry ID (File#UUID)` for each chunk lifecycle.
---
## 4. Pipeline Usage
The translation is executed via the standalone pipeline script:
```bash
python pipeline/02_translate.py --input-epub inputs/my_book.epub
```
* **--input-epub**: Uses the filename to locate the `work/` directory.
* **--book-name**: Alternatively, specify the book folder name directly.
### Dependencies
* Environment variables must be set in `.env`:
* `OPENAI_API_KEY`
* `OPENAI_BASE_URL` (Optional)
---
## 5. Development & Debugging
* **Chunk Logs**: Check `.work/{book}/chunks/` to see exactly what was sent to and received from the LLM.
* **Idempotency**: The translation script skips entries that already have `translated_text`. To re-translate, you must manually clear `translated_text` in `manifest.json` or delete the manifest (to restart from preprocessing).
+155
View File
@@ -0,0 +1,155 @@
import argparse
import sys
import os
import asyncio
from pathlib import Path
from dotenv import load_dotenv
from src.common.config import load_global_config
from src.preprocessing.epub_cleaner import EpubCleaner
from src.preprocessing.profiler import BookProfiler
from src.preprocessing.text_extractor import FineGrainedExtractor
from src.translation.manifest_manager import ManifestManager
from src.translation.translator_engine import Translator
from src.translation.llm_client import LLMClient
from src.assembly.backfiller import BackfillEngine
from src.assembly.builder import BilingualBuilder
from src.common.data_model import BookStructure
from src.common.utils import setup_logger, ensure_directory
from src.common.exceptions import EpubTranslatorError
logger = setup_logger("main")
# Unified work directory structure
# .work/
# ├── {book_name}/
# │ ├── book_structure.json
# │ ├── manifest.json
# │ ├── assets/
# │ └── chunks/
def get_work_dirs(input_path: Path) -> dict:
"""Get work directory paths for a specific book."""
book_name = input_path.stem
work_root = Path(".work") / book_name
return {
"root": work_root,
"structure": work_root / "book_structure.json",
"manifest": work_root / "manifest.json",
"assets": work_root / "assets",
"chunks": work_root / "chunks",
}
async def run_pipeline(args):
input_path = Path(args.input_epub)
output_dir = Path(args.output_dir)
# Get work directories for this book
work = get_work_dirs(input_path)
ensure_directory(work["root"])
ensure_directory(output_dir)
# Load Config
config = load_global_config()
llm_conf = config.get("llm", {})
trans_conf = config.get("translation", {})
api_key = llm_conf.get("api_key")
# Base URL and Model come from config if not overridden
base_url = llm_conf.get("base_url")
# CLI model arg overrides config model, which overrides default
model = args.model if args.model else llm_conf.get("model", "gpt-3.5-turbo")
if not api_key:
logger.warning("OPENAI_API_KEY not found in env or config. LLM features may fail.")
try:
# 1. Preprocessing - Reuse book_structure.json if exists
if work["structure"].exists() and not args.force_clean:
logger.info(f"Reusing existing book_structure: {work['structure']}")
structure = BookStructure.load(work["structure"])
else:
logger.info("Cleaning EPUB and generating book_structure...")
cleaner = EpubCleaner(input_path, work["root"])
book_structure_json = cleaner.clean()
structure = BookStructure.load(book_structure_json)
# 2. Extraction
extractor = FineGrainedExtractor()
manifest_entries = extractor.extract(structure)
# 3. Manifest Management
manifest_manager = ManifestManager(work["manifest"])
manifest_manager.load() # Load existing if any
manifest_manager.add_entries(manifest_entries)
manifest_manager.save()
# 4. Translation
if not args.skip_translation:
if not api_key:
logger.error("Cannot translate without API Key. Use --skip-translation to test pipeline.")
sys.exit(1)
llm_client = LLMClient(
api_key=api_key,
base_url=base_url,
model=model,
requests_per_minute=llm_conf.get("requests_per_minute", 60),
concurrent_requests=llm_conf.get("concurrent_requests", 5)
)
# Profiling
profiler = BookProfiler(llm_client)
profile = await profiler.analyze(manifest_manager.entries)
logger.info(f"Book Profile: {profile}")
# Translation
target_chunk_size = trans_conf.get("chunk_size", 5000)
translator = Translator(llm_client, chunk_size=target_chunk_size)
await translator.translate(manifest_manager.entries, profile)
manifest_manager.save()
await llm_client.close()
else:
logger.info("Skipping translation step.")
# 5. Backfill
backfiller = BackfillEngine()
updated_structure = backfiller.backfill(structure, manifest_manager.entries, mode=args.mode)
# 6. Assembly - Pass original EPUB for TOC preservation
builder = BilingualBuilder(work["root"], original_epub_path=input_path)
output_filename = f"bilingual_{input_path.name}"
output_path = output_dir / output_filename
created_epub = builder.build(updated_structure, output_path)
logger.info(f"Pipeline completed! Output: {created_epub}")
except EpubTranslatorError as e:
logger.error(f"An error occurred: {e}")
sys.exit(1)
except Exception as e:
logger.critical(f"Unexpected error: {e}")
import traceback
traceback.print_exc()
sys.exit(1)
def main():
parser = argparse.ArgumentParser(description="EPUB Bilingual Translator")
parser.add_argument("input_epub", help="Path to the input EPUB file")
parser.add_argument("--output-dir", default="output", help="Directory for output files")
parser.add_argument("--model", default=None, help="LLM Model to use (overrides config)")
parser.add_argument("--mode", default="bilingual", choices=["bilingual", "target_only"], help="Output mode")
parser.add_argument("--skip-translation", action="store_true", help="Skip LLM translation (for testing)")
parser.add_argument("--force-clean", action="store_true", help="Force re-clean EPUB even if book_structure exists")
args = parser.parse_args()
asyncio.run(run_pipeline(args))
if __name__ == "__main__":
main()
+69
View File
@@ -0,0 +1,69 @@
import argparse
import sys
import os
from pathlib import Path
# Add project root to sys.path
sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
from src.preprocessing.epub_cleaner import EpubCleaner
from src.preprocessing.text_extractor import FineGrainedExtractor
from src.translation.manifest_manager import ManifestManager
from src.common.data_model import BookStructure
from src.common.utils import setup_logger, ensure_directory
from src.common.exceptions import EpubTranslatorError
logger = setup_logger("pipeline_preprocess")
def run_preprocess(args):
input_path = Path(args.input_epub)
if not input_path.exists():
logger.error(f"Input file not found: {input_path}")
sys.exit(1)
book_name = input_path.stem
work_root = Path("work") / book_name
ensure_directory(work_root)
structure_path = work_root / "book_structure.json"
manifest_path = work_root / "manifest.json"
try:
# 1. Clean / Load Structure
if structure_path.exists() and not args.force:
logger.info(f"Reusing existing structure: {structure_path}")
structure = BookStructure.load(structure_path)
else:
logger.info("Cleaning EPUB...")
cleaner = EpubCleaner(input_path, work_root)
structure_path = cleaner.clean()
structure = BookStructure.load(structure_path)
# 2. Extract Text
logger.info("Extracting text segments...")
extractor = FineGrainedExtractor()
entries = extractor.extract(structure)
# 3. Update Manifest
logger.info(f"Updating manifest: {manifest_path}")
manager = ManifestManager(manifest_path)
manager.load()
manager.add_entries(entries)
manager.save()
logger.info("Preprocessing complete.")
except EpubTranslatorError as e:
logger.error(f"Preprocessing failed: {e}")
sys.exit(1)
except Exception as e:
logger.critical(f"Unexpected error: {e}")
sys.exit(1)
if __name__ == "__main__":
parser = argparse.ArgumentParser(description="Step 1: Preprocessing (Clean + Extract)")
parser.add_argument("input_epub", help="Path to input EPUB")
parser.add_argument("--force", action="store_true", help="Force re-clean")
args = parser.parse_args()
run_preprocess(args)
+91
View File
@@ -0,0 +1,91 @@
import argparse
import sys
import os
import asyncio
from pathlib import Path
# Add project root to sys.path
sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
from src.translation.translator_engine import Translator
from src.translation.llm_client import LLMClient
from src.translation.manifest_manager import ManifestManager
from src.preprocessing.profiler import BookProfiler
from src.common.utils import setup_logger
from src.common.exceptions import EpubTranslatorError
from src.common.config import load_global_config
logger = setup_logger("pipeline_translate")
async def run_translate(args):
# Resolve paths
if args.book_name:
book_name = args.book_name
elif args.input_epub:
book_name = Path(args.input_epub).stem
else:
logger.error("Must provide --book-name or --input-epub")
sys.exit(1)
work_root = Path("work") / book_name
manifest_path = work_root / "manifest.json"
if not manifest_path.exists():
logger.error(f"Manifest not found: {manifest_path}. Run Step 1 first.")
sys.exit(1)
# Load Config
config = load_global_config()
llm_conf = config.get("llm", {})
trans_conf = config.get("translation", {})
api_key = llm_conf.get("api_key")
if not api_key:
logger.error("OPENAI_API_KEY not found in env or config.")
sys.exit(1)
try:
# Load Manifest
manager = ManifestManager(manifest_path)
manager.load()
# Init components
# Allow CLI args to override config if needed (not implemented yet, taking config partial priority)
llm = LLMClient(
api_key=api_key,
base_url=llm_conf.get("base_url"),
model=args.model if args.model else llm_conf.get("model", "gpt-3.5-turbo"),
requests_per_minute=llm_conf.get("requests_per_minute", 60),
concurrent_requests=llm_conf.get("concurrent_requests", 5)
)
# Profile
profiler = BookProfiler(llm)
profile = await profiler.analyze(manager.entries)
logger.info(f"Book Profile: {profile.title} ({profile.genre})")
# Translate
target_chunk_size = trans_conf.get("chunk_size", 4000)
translator = Translator(llm, chunk_size=target_chunk_size)
await translator.translate(manager.entries, profile)
# Save final state
manager.save()
await llm.close()
logger.info("Translation complete.")
except EpubTranslatorError as e:
logger.error(f"Translation failed: {e}")
sys.exit(1)
except Exception as e:
logger.critical(f"Unexpected error: {e}")
sys.exit(1)
if __name__ == "__main__":
parser = argparse.ArgumentParser(description="Step 2: Translation")
parser.add_argument("--input-epub", help="Path to original EPUB (to derive book name)")
parser.add_argument("--book-name", help="Book name (folder name in work/)")
parser.add_argument("--model", default=None, help="LLM Model (overrides config)")
args = parser.parse_args()
asyncio.run(run_translate(args))
+72
View File
@@ -0,0 +1,72 @@
import argparse
import sys
import os
from pathlib import Path
# Add project root to sys.path
sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
from src.assembly.backfiller import BackfillEngine
from src.assembly.builder import BilingualBuilder
from src.translation.manifest_manager import ManifestManager
from src.common.data_model import BookStructure
from src.common.utils import setup_logger, ensure_directory
from src.common.exceptions import EpubTranslatorError
logger = setup_logger("pipeline_assemble")
def run_assemble(args):
input_path = Path(args.input_epub)
if not input_path.exists():
logger.error(f"Input file not found: {input_path}")
sys.exit(1)
book_name = input_path.stem
work_root = Path("work") / book_name
structure_path = work_root / "book_structure.json"
manifest_path = work_root / "manifest.json"
if not structure_path.exists() or not manifest_path.exists():
logger.error("Missing structure or manifest. Run Step 1.")
sys.exit(1)
output_dir = Path(args.output_dir)
ensure_directory(output_dir)
try:
# Load Data
structure = BookStructure.load(structure_path)
manager = ManifestManager(manifest_path)
manager.load()
# Backfill
logger.info(f"Backfilling translations (Mode: {args.mode})...")
backfiller = BackfillEngine()
updated_structure = backfiller.backfill(structure, manager.entries, mode=args.mode)
# Build
logger.info("Building EPUB...")
builder = BilingualBuilder(work_root, original_epub_path=input_path)
output_filename = f"{book_name}_{args.mode}.epub"
output_path = output_dir / output_filename
builder.build(updated_structure, output_path)
logger.info(f"Assembly complete. Output: {output_path}")
except EpubTranslatorError as e:
logger.error(f"Assembly failed: {e}")
sys.exit(1)
except Exception as e:
logger.critical(f"Unexpected error: {e}")
import traceback
traceback.print_exc()
sys.exit(1)
if __name__ == "__main__":
parser = argparse.ArgumentParser(description="Step 3: Assembly (Backfill + Build)")
parser.add_argument("input_epub", help="Path to original EPUB")
parser.add_argument("--output-dir", default="output", help="Output directory")
parser.add_argument("--mode", default="bilingual", choices=["bilingual", "target_only"], help="Output mode")
args = parser.parse_args()
run_assemble(args)
+12
View File
@@ -0,0 +1,12 @@
ebooklib>=0.19
beautifulsoup4>=4.12.0
lxml>=4.9.0
openai>=1.0.0
aiohttp>=3.9.0
pydantic>=2.0.0
loguru>=0.7.0
rich>=13.0.0
asyncio-throttle>=1.0.2
tenacity>=8.0.0
python-dotenv>=1.0.0
PyYAML>=6.0
@@ -0,0 +1,51 @@
import sys
from pathlib import Path
import traceback
# Add src to path
sys.path.insert(0, str(Path(__file__).parent.parent))
from src.data_model import BookStructure, ManifestEntry
from src.manifest_manager import ManifestManager
from src.backfill_engine import BackfillEngine
from src.bilingual_builder import BilingualBuilder
from src.utils import setup_logger
logger = setup_logger("build_final")
def build_final():
BOOK_NAME = "Empire of AI Dreams and Nightmares in Sam Altmans OpenAI (Karen Hao)"
WORK_DIR = Path(f".work/{BOOK_NAME}")
MANIFEST_PATH = WORK_DIR / "manifest.json"
STRUCTURE_PATH = WORK_DIR / "book_structure.json"
OUTPUT_EPUB = Path("output/final_verification.epub")
INPUT_EPUB = Path(f"input/{BOOK_NAME}.epub")
if not MANIFEST_PATH.exists() or not STRUCTURE_PATH.exists():
logger.error("Missing manifest or structure. Run translation first.")
return
# 1. Load Data
logger.info("Loading structure and manifest...")
structure = BookStructure.load(STRUCTURE_PATH)
manager = ManifestManager(MANIFEST_PATH)
manager.load()
# 2. Backfill
logger.info("Backfilling translations...")
backfiller = BackfillEngine()
structure = backfiller.backfill(structure, manager.entries, mode="bilingual")
# 3. Build
logger.info(f"Building final EPUB to {OUTPUT_EPUB}...")
builder = BilingualBuilder(WORK_DIR, original_epub_path=INPUT_EPUB)
builder.build(structure, OUTPUT_EPUB)
logger.info("Build complete.")
if __name__ == "__main__":
try:
build_final()
except Exception as e:
traceback.print_exc()
print(f"CRITICAL ERROR: {e}")
+102
View File
@@ -0,0 +1,102 @@
import os
import sys
from pathlib import Path
from dotenv import load_dotenv
import asyncio
import httpx
from openai import AsyncOpenAI
# Add project root to sys.path
sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
from src.common.config import load_global_config
async def main():
print("--- Environment Debug ---")
load_dotenv()
env_key = os.getenv("OPENAI_API_KEY")
if env_key:
print(f"OPENAI_API_KEY found in env: {env_key[:8]}...{env_key[-4:]}")
else:
print("OPENAI_API_KEY NOT found in env!")
print("\n--- Config Loader Debug ---")
try:
config = load_global_config()
llm_conf = config.get("llm", {})
conf_key = llm_conf.get("api_key")
base_url = llm_conf.get("base_url")
model = llm_conf.get("model")
print(f"Config Base URL: {base_url}")
print(f"Config Model: {model}")
if conf_key:
print(f"Config API Key: {conf_key[:8]}...{conf_key[-4:]}")
if env_key and conf_key == env_key:
print("Config Key matches Env Key.")
else:
print("Config Key DOES NOT match Env Key!")
else:
print("Config API Key NOT found!")
print("\n--- API Connectivity Test ---")
if not conf_key or not base_url:
print("Missing params for test.")
return
headers = {"Authorization": f"Bearer {conf_key}"}
url = f"{base_url}/models"
print(f"Requesting: {url}")
async with httpx.AsyncClient() as client:
try:
resp = await client.get(url, headers=headers, timeout=10)
print(f"Status Code: {resp.status_code}")
if resp.status_code == 200:
print("Success! Models listed.")
else:
print(f"Failed. Response: {resp.text}")
except Exception as e:
print(f"Exception during request: {e}")
print("\n--- Chat Completion Test (Mimicking LLMClient) ---")
proxy_url = os.environ.get("http_proxy") or os.environ.get("https_proxy")
print(f"Proxy detected: {proxy_url}")
http_client = httpx.AsyncClient(
proxy=proxy_url,
timeout=60.0,
follow_redirects=True
) if proxy_url else None
aclient = AsyncOpenAI(
api_key=conf_key,
base_url=base_url,
http_client=http_client
)
print(f"Model: {model}")
system_prompt = "You are a senior publishing editor."
user_prompt = "Analyze this text."
try:
print("Sending request with System Prompt...")
resp = await aclient.chat.completions.create(
model=model,
messages=[
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_prompt}
],
temperature=0.3,
)
print("Success!")
print(f"Response: {resp.choices[0].message.content}")
except Exception as e:
print(f"Chat Completion failed: {type(e).__name__}: {e}")
except Exception as e:
print(f"Config loading failed: {e}")
if __name__ == "__main__":
asyncio.run(main())
+212
View File
@@ -0,0 +1,212 @@
#!/usr/bin/env python3
"""
Placeholder Backfill Test Script
Tests placeholder restoration on translated entries.
Usage:
python scripts/test_backfill.py --chapter 38
python scripts/test_backfill.py --chapter 38 --limit 10
"""
import argparse
import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).parent.parent))
from src.manifest_manager import ManifestManager
from src.format_restorer import FormatRestorer
# Configuration - Use unified .work directory
BOOK_NAME = "Empire of AI Dreams and Nightmares in Sam Altmans OpenAI (Karen Hao)"
WORK_DIR = Path(f".work/{BOOK_NAME}")
MANIFEST_PATH = WORK_DIR / "manifest.json"
def build_toc_from_manifest(manager: ManifestManager) -> list:
"""Build TOC from manifest entries."""
files = {}
for entry in manager.entries:
fp = entry.file_path
if fp not in files:
files[fp] = {'count': 0, 'first_text': ''}
files[fp]['count'] += 1
if not files[fp]['first_text'] and entry.original_text:
files[fp]['first_text'] = entry.original_text[:40].replace('\n', ' ')
toc = []
for i, (fp, info) in enumerate(sorted(files.items()), 1):
toc.append({
'index': i,
'href': fp,
'title': info['first_text'] or f"File {i}",
'paragraphs': info['count']
})
return toc
def get_chapter_entries(manager: ManifestManager, chapter_index: int) -> tuple:
"""Get entries for a specific chapter by index."""
toc = build_toc_from_manifest(manager)
if chapter_index < 1 or chapter_index > len(toc):
print(f"错误: 章节编号 {chapter_index} 无效 (范围: 1-{len(toc)})")
return None, None
chapter = toc[chapter_index - 1]
href = chapter['href']
entries = [e for e in manager.entries if e.file_path == href]
return chapter, entries
def test_backfill(chapter: dict, entries: list, limit: int = None):
"""Test placeholder restoration for a chapter."""
print(f"\n" + "=" * 70)
print(f"占位符回填测试 - 章节 #{chapter['index']}: {chapter['title'][:40]}...")
print("=" * 70)
# Filter entries with translation and placeholders
translated = [e for e in entries if e.translated_text]
with_placeholders = [e for e in translated if e.placeholders and len(e.placeholders) > 0]
print(f"\n统计:")
print(f" 总段落: {len(entries)}")
print(f" 已翻译: {len(translated)}")
print(f" 有占位符: {len(with_placeholders)}")
if not translated:
print("\n⚠️ 该章节没有已翻译的内容!")
return
# Test restoration
restorer = FormatRestorer()
success_count = 0
fail_count = 0
results = []
test_entries = with_placeholders[:limit] if limit else with_placeholders
print(f"\n测试 {len(test_entries)} 个带占位符的段落:")
print("-" * 70)
for i, entry in enumerate(test_entries, 1):
original = entry.original_text
translated = entry.translated_text
placeholders = entry.placeholders
# Get non-internal placeholders
visible_ph = {k: v for k, v in placeholders.items() if not k.startswith('_')}
# Perform restoration
restored, success = restorer.restore(translated, placeholders)
if success:
success_count += 1
status = ""
else:
fail_count += 1
status = ""
results.append({
'index': i,
'entry_id': entry.entry_id,
'original': original,
'translated': translated,
'restored': restored,
'placeholders': visible_ph,
'success': success
})
# Print summary
print(f"\n[{i}] {status} {entry.entry_id[-40:]}")
print(f" 占位符: {list(visible_ph.keys())}")
print(f" 原文: {original[:50]}...")
print(f" 译文: {translated[:50]}...")
if not success:
print(f" 还原: {restored[:50]}...")
# Show what placeholders are missing
missing = []
for k in visible_ph.keys():
if k.isdigit():
if f"φ{k}φ" not in translated and f"φ/{k}φ" not in translated:
missing.append(k)
if missing:
print(f" 缺失: {missing}")
# Summary
print("\n" + "=" * 70)
print(f"测试结果汇总")
print("=" * 70)
print(f" 成功: {success_count}/{len(test_entries)}")
print(f" 失败: {fail_count}/{len(test_entries)}")
if fail_count > 0:
print(f"\n失败案例详情:")
for r in results:
if not r['success']:
print(f"\n [{r['index']}] {r['entry_id'][-50:]}")
print(f" 原文: {r['original'][:60]}...")
print(f" 译文: {r['translated'][:60]}...")
print(f" 还原: {r['restored'][:60]}...")
print(f" 占位符: {r['placeholders']}")
# Also test entries without visible placeholders (only _prefix/_suffix)
prefix_suffix_only = [e for e in translated
if e.placeholders
and all(k.startswith('_') for k in e.placeholders.keys())]
if prefix_suffix_only:
print(f"\n\n额外测试: 只有 _prefix/_suffix 的段落 ({len(prefix_suffix_only)} 个)")
print("-" * 70)
ps_success = 0
ps_fail = 0
for entry in prefix_suffix_only[:5]: # Test first 5
restored, success = restorer.restore(entry.translated_text, entry.placeholders)
if success:
ps_success += 1
status = ""
else:
ps_fail += 1
status = ""
print(f" {status} {entry.entry_id[-40:]}")
if '_prefix' in entry.placeholders:
print(f" _prefix: {entry.placeholders['_prefix'][:30]}...")
if '_suffix' in entry.placeholders:
print(f" _suffix: {entry.placeholders['_suffix'][:30]}...")
print(f"\n 结果: {ps_success}/{min(5, len(prefix_suffix_only))} 成功")
def main():
parser = argparse.ArgumentParser(description="测试占位符回填")
parser.add_argument("--chapter", "-c", type=int, required=True, help="章节编号")
parser.add_argument("--limit", "-l", type=int, default=20, help="测试数量限制 (默认20)")
args = parser.parse_args()
if not MANIFEST_PATH.exists():
print(f"错误: Manifest 不存在: {MANIFEST_PATH}")
return
# Load manifest
manager = ManifestManager(MANIFEST_PATH)
manager.load()
print(f"已加载 manifest: {len(manager.entries)} 条目")
# Get chapter
chapter, entries = get_chapter_entries(manager, args.chapter)
if not chapter:
return
# Test backfill
test_backfill(chapter, entries, args.limit)
if __name__ == "__main__":
main()
+25
View File
@@ -0,0 +1,25 @@
import asyncio
import os
import httpx
async def test_conn():
print(f"HTTP_PROXY: {os.environ.get('http_proxy')}")
print(f"HTTPS_PROXY: {os.environ.get('https_proxy')}")
print(f"ALL_PROXY: {os.environ.get('all_proxy')}")
url = "https://api.gpt.ge/v1/models"
headers = {"Authorization": f"Bearer {os.environ.get('V3_API_KEY')}"}
print(f"Connecting to {url}...")
try:
async with httpx.AsyncClient(timeout=10, follow_redirects=True) as client:
resp = await client.get(url, headers=headers)
print(f"Status: {resp.status_code}")
print(f"Headers: {resp.headers}")
except Exception as e:
print(f"Error: {e}")
if __name__ == "__main__":
from dotenv import load_dotenv
load_dotenv()
asyncio.run(test_conn())
+40
View File
@@ -0,0 +1,40 @@
import asyncio
import os
import httpx
from openai import AsyncOpenAI
async def test_openai():
proxy_url = os.environ.get("http_proxy")
print(f"Using proxy: {proxy_url}")
http_client = httpx.AsyncClient(
proxy=proxy_url,
timeout=30.0,
follow_redirects=True
)
client = AsyncOpenAI(
base_url="https://api.gpt.ge/v1",
api_key=os.environ.get("V3_API_KEY"),
http_client=http_client
)
print("Sending request...")
try:
response = await client.chat.completions.create(
model="gpt-3.5-turbo",
messages=[{"role": "user", "content": "Hello"}],
max_tokens=5
)
print(f"Response: {response.choices[0].message.content}")
except Exception as e:
import traceback
traceback.print_exc()
print(f"Error: {e}")
finally:
await http_client.aclose()
if __name__ == "__main__":
from dotenv import load_dotenv
load_dotenv()
asyncio.run(test_openai())
@@ -0,0 +1,211 @@
#!/usr/bin/env python3
"""
Debug script to show chunk content and test short ID strategy.
"""
import asyncio
import json
import os
import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).parent.parent))
from dotenv import load_dotenv
from src.manifest_manager import ManifestManager
from src.llm_client import LLMClient
from src.data_model import ManifestEntry
load_dotenv()
# Configuration
MANIFEST_PATH = Path("cache/Empire of AI Dreams and Nightmares in Sam Altmans OpenAI (Karen Hao)_manifest.json")
API_KEY = os.getenv("V3_API_KEY")
BASE_URL = "https://api.gpt.ge/v1"
MODEL = "gemini-3-flash-preview"
EXTRA_HEADERS = {"x-foo": "true"}
def show_current_chunk_format():
"""Show current chunk format (problematic)."""
print("\n" + "="*60)
print("当前 Chunk 格式 (有问题)")
print("="*60)
# Load manifest
manager = ManifestManager(MANIFEST_PATH)
manager.load()
# Get sample entries with placeholders
samples = [e for e in manager.entries if e.placeholders][:3]
print("\n发送给 LLM 的格式 (当前):")
print("-"*60)
for item in samples:
context = item.context or "BODY"
print(f"{item.entry_id} [{context}] {item.original_text[:50]}...")
print("\n问题分析:")
print(" 1. entry_id 太长 (包含 UUID): 容易被 LLM 截断或修改")
print(" 2. [BODY] context 没必要发送")
print(" 3. 依赖 LLM 精确复制长 ID,不可靠")
def show_proposed_chunk_format():
"""Show proposed short ID chunk format."""
print("\n" + "="*60)
print("建议的 Chunk 格式 (短 ID)")
print("="*60)
manager = ManifestManager(MANIFEST_PATH)
manager.load()
samples = [e for e in manager.entries if e.placeholders][:5]
print("\n发送给 LLM 的格式 (建议):")
print("-"*60)
# Build with short IDs
id_map = {} # short_id -> entry_id
for i, item in enumerate(samples, 1):
short_id = f"#{i}"
id_map[short_id] = item.entry_id
text = item.original_text[:60]
print(f"{short_id}: {text}...")
print("\n期望 LLM 返回的格式:")
print("-"*60)
print("#1: φ1φpenguinrandomhouse.com(保持不翻译)")
print("#2: 该产品在欧盟的产品安全授权代表为 φ1φPenguin Random House Irelandφ/1φ...")
print("#3: φ1φ献辞")
print("#4: φ1φ题记")
print("#5: φ1φ作者说明")
print("\n优势:")
print(" 1. 短 ID (#1, #2...) 不会被 LLM 弄乱")
print(" 2. 去掉了无用的 context 标签")
print(" 3. 解析更可靠:用正则 ^#(\\d+): 匹配")
print(" 4. ID 映射表保留在代码中,用于还原")
print("\n映射表 (代码内保留):")
for short_id, full_id in id_map.items():
print(f" {short_id} -> {full_id[:50]}...")
async def test_short_id_translation():
"""Test translation with short ID format."""
print("\n" + "="*60)
print("测试短 ID 翻译")
print("="*60)
manager = ManifestManager(MANIFEST_PATH)
manager.load()
# Get 5 entries with varied content
samples = [e for e in manager.entries if len(e.original_text) > 20][:5]
# Build prompt with short IDs
id_map = {}
lines = []
for i, item in enumerate(samples, 1):
short_id = f"#{i}"
id_map[short_id] = item.entry_id
text = item.original_text.replace('\n', ' ').strip()
lines.append(f"{short_id}: {text}")
user_prompt = "\n".join(lines)
print("\n发送给 LLM 的 Prompt:")
print("-"*60)
print(user_prompt)
# Create client
from openai import AsyncOpenAI
client = AsyncOpenAI(
base_url=BASE_URL,
api_key=API_KEY,
default_headers=EXTRA_HEADERS
)
system_prompt = """You are a professional English to Chinese translator.
Translate each line to Chinese. Keep the format:
- Each line starts with #N: (keep this ID exactly)
- Preserve any φXφ placeholders exactly as-is
- Only output translations, no explanations
Example input:
#1: Hello world
#2: φ1φClick hereφ/1φ to continue
Example output:
#1: 你好世界
#2: φ1φ点击这里φ/1φ 继续"""
print("\n发送请求...")
try:
resp = await client.chat.completions.create(
model=MODEL,
messages=[
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_prompt}
],
temperature=0.3,
)
raw_response = resp.choices[0].message.content.strip()
print("\nLLM 返回:")
print("-"*60)
print(raw_response)
# Parse with short ID
print("\n解析结果:")
print("-"*60)
import re
results = {}
for line in raw_response.split("\n"):
line = line.strip()
match = re.match(r'^#(\d+):\s*(.+)$', line)
if match:
short_id = f"#{match.group(1)}"
translation = match.group(2)
if short_id in id_map:
full_id = id_map[short_id]
results[full_id] = translation
print(f" {short_id} -> {translation[:40]}...")
print(f"\n成功解析: {len(results)}/{len(samples)}")
finally:
await client.close()
async def main():
print("="*60)
print("Chunk ID 策略分析与测试")
print("="*60)
if not MANIFEST_PATH.exists():
print(f"ERROR: Manifest not found at {MANIFEST_PATH}")
return
# 1. Show current format (problems)
show_current_chunk_format()
# 2. Show proposed format
show_proposed_chunk_format()
# 3. Test short ID translation
if API_KEY:
await test_short_id_translation()
else:
print("\n跳过测试 (V3_API_KEY 未设置)")
print("\n" + "="*60)
print("分析完成")
print("="*60)
if __name__ == "__main__":
asyncio.run(main())
@@ -0,0 +1,355 @@
#!/usr/bin/env python3
"""
Translation Pipeline Debug Script
Shows visible results at each step of the translation process.
"""
import asyncio
import json
import os
import sys
from pathlib import Path
# Add project root to path
sys.path.insert(0, str(Path(__file__).parent.parent))
from dotenv import load_dotenv
from src.manifest_manager import ManifestManager
from src.llm_client import LLMClient
from src.book_profiler import BookProfiler
from src.translator import Translator
from src.format_restorer import FormatRestorer
from src.data_model import ManifestEntry, BookProfile
load_dotenv()
# Configuration
MANIFEST_PATH = Path("cache/Empire of AI Dreams and Nightmares in Sam Altmans OpenAI (Karen Hao)_manifest.json")
API_KEY = os.getenv("V3_API_KEY")
BASE_URL = "https://api.gpt.ge/v1"
MODEL = "gemini-3-flash-preview"
EXTRA_HEADERS = {"x-foo": "true"}
# Chunk size configuration
MAX_CHUNK_SIZE = 15 # Maximum entries per chunk
def group_entries_by_file(entries: list) -> dict:
"""Group manifest entries by their source file (chapter)."""
grouped = {}
for entry in entries:
file_path = entry.file_path
if file_path not in grouped:
grouped[file_path] = []
grouped[file_path].append(entry)
return grouped
def create_chapter_aware_chunks(entries: list, max_size: int = MAX_CHUNK_SIZE) -> list:
"""
Create chunks that respect chapter boundaries.
Returns list of (file_path, chunk_entries) tuples.
"""
grouped = group_entries_by_file(entries)
chunks = []
for file_path, file_entries in grouped.items():
# Split this file's entries into chunks of max_size
for i in range(0, len(file_entries), max_size):
chunk = file_entries[i:i + max_size]
chunks.append((file_path, chunk))
return chunks
def get_file_type(file_path: str) -> str:
"""Determine the type of content based on file name."""
fname = file_path.lower()
if any(k in fname for k in ['toc', 'contents', 'nav']):
return 'toc'
elif any(k in fname for k in ['title', 'cover']):
return 'cover'
elif any(k in fname for k in ['copyright', 'colophon']):
return 'legal'
elif any(k in fname for k in ['author', 'about']):
return 'author_bio'
elif any(k in fname for k in ['index', 'bibliography', 'endnote', 'footnote']):
return 'reference'
else:
return 'body'
async def step1_load_manifest():
"""Step 1: Load manifest and show statistics."""
print("\n" + "="*60)
print("STEP 1: Loading Manifest")
print("="*60)
manager = ManifestManager(MANIFEST_PATH)
manager.load()
entries = manager.entries
untranslated = [e for e in entries if not e.translated_text]
print(f" Total entries: {len(entries)}")
print(f" Untranslated: {len(untranslated)}")
# Show grouping by file
grouped = group_entries_by_file(entries)
print(f" Unique files: {len(grouped)}")
# Show sample entry
if entries:
sample = entries[0]
print(f"\n Sample entry:")
print(f" ID: {sample.entry_id}")
print(f" File: {sample.file_path}")
print(f" Original: {sample.original_text[:80]}...")
print(f" Placeholders: {sample.placeholders}")
return manager
async def step2_profile_book(manager: ManifestManager, llm_client: LLMClient):
"""Step 2: Generate book profile."""
print("\n" + "="*60)
print("STEP 2: Generating Book Profile")
print("="*60)
profiler = BookProfiler(llm_client)
profile = await profiler.analyze(manager.entries)
print(f" Title: {profile.title}")
print(f" Author: {profile.author}")
print(f" Genre: {profile.genre}")
print(f" Keywords: {profile.keywords}")
print(f" Style Guide: {profile.style_guide[:200]}..." if profile.style_guide else " Style Guide: (none)")
return profile
async def step3_create_chunks(manager: ManifestManager):
"""Step 3: Create chapter-aware chunks."""
print("\n" + "="*60)
print("STEP 3: Creating Chapter-Aware Chunks")
print("="*60)
untranslated = [e for e in manager.entries if not e.translated_text]
chunks = create_chapter_aware_chunks(untranslated)
print(f" Total chunks: {len(chunks)}")
# Show chunk distribution
print(f"\n Chunk distribution by file type:")
type_counts = {}
for file_path, chunk_entries in chunks:
ftype = get_file_type(file_path)
type_counts[ftype] = type_counts.get(ftype, 0) + 1
for ftype, count in sorted(type_counts.items()):
print(f" {ftype}: {count} chunks")
# Show first 3 chunks
print(f"\n First 3 chunks:")
for i, (file_path, chunk_entries) in enumerate(chunks[:3]):
ftype = get_file_type(file_path)
print(f" [{i}] {file_path} ({ftype}): {len(chunk_entries)} entries")
if chunk_entries:
print(f" First: {chunk_entries[0].original_text[:50]}...")
return chunks
async def step4_translate_sample(chunks: list, llm_client: LLMClient, profile: BookProfile):
"""Step 4: Translate a sample chunk and show results."""
print("\n" + "="*60)
print("STEP 4: Translating Sample Chunk")
print("="*60)
if not chunks:
print(" No chunks to translate!")
return
# Pick a proper body chapter (skip first few files which are usually cover/copyright/toc)
sample_chunk = None
skip_prefixes = ['cM', 'c9', 'c18'] # Cover, title, contents pages
for file_path, chunk_entries in chunks:
# Skip non-body files and known cover/toc files
ftype = get_file_type(file_path)
fname = Path(file_path).stem
if ftype == 'body' and fname not in skip_prefixes and len(chunk_entries) > 3:
sample_chunk = (file_path, chunk_entries[:5]) # Limit to 5 entries for demo
break
if not sample_chunk:
# Fallback to any body chunk
for file_path, chunk_entries in chunks:
if get_file_type(file_path) == 'body':
sample_chunk = (file_path, chunk_entries[:5])
break
if not sample_chunk:
sample_chunk = chunks[0]
sample_chunk = (sample_chunk[0], sample_chunk[1][:5])
file_path, entries = sample_chunk
ftype = get_file_type(file_path)
print(f" Selected chunk: {file_path} ({ftype})")
print(f" Entries: {len(entries)}")
# Show entries before translation
print(f"\n === Before Translation ===")
for i, entry in enumerate(entries):
print(f" [{i}] {entry.entry_id}")
print(f" Original: {entry.original_text[:60]}...")
if entry.placeholders:
print(f" Placeholders: {list(entry.placeholders.keys())}")
# Translate
print(f"\n Translating...")
results = await llm_client.translate_chunk(
entries,
instruction=profile.style_guide,
mode="bilingual"
)
# Apply results and show
print(f"\n === After Translation ===")
restorer = FormatRestorer()
for i, entry in enumerate(entries):
if entry.entry_id in results:
translated = results[entry.entry_id]
entry.translated_text = translated
print(f" [{i}] {entry.entry_id}")
print(f" Original: {entry.original_text[:50]}...")
print(f" Translated: {translated[:50]}...")
# Restore format
if entry.placeholders:
restored, success = restorer.restore(translated, entry.placeholders)
print(f" Restored OK: {success}")
if not success:
print(f" Restored: {restored[:50]}...")
else:
print(f" [{i}] MISSING: {entry.entry_id}")
return entries
async def step5_test_placeholders(manager: ManifestManager, llm_client: LLMClient, profile: BookProfile):
"""Step 5: Test placeholder handling with entries that have placeholders."""
print("\n" + "="*60)
print("STEP 5: Testing Placeholder Handling")
print("="*60)
# Find entries with placeholders
entries_with_ph = [e for e in manager.entries if e.placeholders and len(e.placeholders) > 1]
print(f" Entries with placeholders: {len(entries_with_ph)}")
if not entries_with_ph:
print(" No entries with placeholders found!")
return
# Pick 5 diverse samples
samples = entries_with_ph[:5]
print(f"\n === Selected Samples ({len(samples)}) ===")
for i, entry in enumerate(samples):
ph_keys = [k for k in entry.placeholders.keys() if not k.startswith('_')]
print(f" [{i}] {entry.entry_id}")
print(f" Original: {entry.original_text[:60]}...")
print(f" Placeholders: {ph_keys}")
# Translate
print(f"\n Translating {len(samples)} entries with placeholders...")
results = await llm_client.translate_chunk(
samples,
instruction=profile.style_guide,
mode="bilingual"
)
# Show results with restoration
print(f"\n === Translation Results ===")
restorer = FormatRestorer()
success_count = 0
for i, entry in enumerate(samples):
print(f"\n [{i}] {entry.entry_id}")
print(f" Original: {entry.original_text[:50]}...")
if entry.entry_id in results:
translated = results[entry.entry_id]
print(f" Translated: {translated[:50]}...")
# Check if placeholders are preserved
ph_keys = [k for k in entry.placeholders.keys() if not k.startswith('_')]
preserved = all(f"φ{k}φ" in translated or f"φ/{k}φ" in translated for k in ph_keys if k.isdigit())
print(f" PH Preserved: {preserved}")
# Restore format
restored, success = restorer.restore(translated, entry.placeholders)
print(f" Restore OK: {success}")
if success:
success_count += 1
else:
print(f" Restored: {restored[:50]}...")
else:
print(f" MISSING from results!")
print(f"\n Summary: {success_count}/{len(samples)} restored successfully")
async def main():
"""Run all steps."""
print("="*60)
print("TRANSLATION PIPELINE DEBUG")
print("="*60)
if not API_KEY:
print("ERROR: V3_API_KEY not found in .env")
return
if not MANIFEST_PATH.exists():
print(f"ERROR: Manifest not found at {MANIFEST_PATH}")
print("Run the main pipeline first to generate the manifest.")
return
# Initialize LLM client
llm_client = LLMClient(
api_key=API_KEY,
base_url=BASE_URL,
model=MODEL,
extra_headers=EXTRA_HEADERS
)
try:
# Step 1: Load manifest
manager = await step1_load_manifest()
# Step 2: Profile book
profile = await step2_profile_book(manager, llm_client)
# Step 3: Create chunks
chunks = await step3_create_chunks(manager)
# Step 4: Translate sample (simple text)
await step4_translate_sample(chunks, llm_client, profile)
# Step 5: Test placeholders
await step5_test_placeholders(manager, llm_client, profile)
print("\n" + "="*60)
print("DEBUG COMPLETE")
print("="*60)
finally:
await llm_client.close()
if __name__ == "__main__":
asyncio.run(main())
+353
View File
@@ -0,0 +1,353 @@
#!/usr/bin/env python3
"""
Chapter Translation Test Script
Translate a complete chapter to test the full pipeline.
Usage:
python scripts/translate_chapter.py --show-toc # 显示章节目录
python scripts/translate_chapter.py --chapter 5 # 翻译第5章
python scripts/translate_chapter.py --chapter 5 --test # 测试模式,只翻译前2个chunk
"""
import argparse
import asyncio
import json
import os
import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).parent.parent))
from dotenv import load_dotenv
from src.manifest_manager import ManifestManager
from src.llm_client import LLMClient
from src.book_profiler import BookProfiler
from src.format_restorer import FormatRestorer
from src.data_model import ManifestEntry, BookStructure
load_dotenv()
# Configuration - Use unified .work directory
BOOK_NAME = "Empire of AI Dreams and Nightmares in Sam Altmans OpenAI (Karen Hao)"
WORK_DIR = Path(f".work/{BOOK_NAME}")
MANIFEST_PATH = WORK_DIR / "manifest.json"
STRUCTURE_PATH = WORK_DIR / "book_structure.json"
CHUNK_DIR = WORK_DIR / "chunks"
API_KEY = os.getenv("V3_API_KEY")
BASE_URL = "https://api.gpt.ge/v1"
MODEL = "gpt-4o-mini" # "gemini-3-flash-preview"
EXTRA_HEADERS = {"x-foo": "true"}
# Chunk config - around 5000 chars per chunk
CHUNK_SIZE_CHARS = 5000
def load_toc_from_structure() -> list:
"""Load TOC from book structure for readable chapter names."""
if not STRUCTURE_PATH.exists():
return []
try:
structure = BookStructure.load(STRUCTURE_PATH)
# Build TOC from spine order with chapter titles
toc = []
for i, item_id in enumerate(structure.spine, 1):
if item_id in structure.resources:
resource = structure.resources[item_id]
href = resource.href
# Try to extract title from content
title = extract_title_from_html(resource.content) if resource.content else None
toc.append({
'index': i,
'item_id': item_id,
'href': href,
'title': title or f"Chapter {i}"
})
return toc
except Exception as e:
print(f"警告: 无法加载书籍结构: {e}")
return []
def extract_title_from_html(html: str) -> str:
"""Extract title from HTML content."""
from bs4 import BeautifulSoup
soup = BeautifulSoup(html, 'html.parser')
# Try h1, h2, h3 in order
for tag in ['h1', 'h2', 'h3']:
elem = soup.find(tag)
if elem:
return elem.get_text().strip()[:50]
# Try first paragraph
p = soup.find('p')
if p:
text = p.get_text().strip()[:50]
if text:
return text + "..."
return None
def build_toc_from_manifest(manager: ManifestManager) -> list:
"""Build TOC from manifest entries."""
files = {}
for entry in manager.entries:
fp = entry.file_path
if fp not in files:
files[fp] = {
'count': 0,
'first_text': '',
'total_chars': 0
}
files[fp]['count'] += 1
files[fp]['total_chars'] += len(entry.original_text)
if not files[fp]['first_text'] and entry.original_text:
files[fp]['first_text'] = entry.original_text[:40].replace('\n', ' ')
toc = []
for i, (fp, info) in enumerate(sorted(files.items()), 1):
toc.append({
'index': i,
'href': fp,
'title': info['first_text'] or f"File {i}",
'paragraphs': info['count'],
'chars': info['total_chars']
})
return toc
def show_toc(manager: ManifestManager):
"""Display TOC with chapter numbers."""
toc = build_toc_from_manifest(manager)
print("\n" + "=" * 70)
print("章节目录 (Table of Contents)")
print("=" * 70)
print(f"{'#':>3} | {'段落':>5} | {'字符':>6} | 章节标题")
print("-" * 70)
for item in toc:
title = item['title'][:45] if len(item['title']) > 45 else item['title']
print(f"{item['index']:3d} | {item['paragraphs']:5d} | {item['chars']:6d} | {title}")
print("-" * 70)
print(f"{len(toc)} 个章节")
print("\n用法: python scripts/translate_chapter.py --chapter <编号>")
print("示例: python scripts/translate_chapter.py --chapter 5")
def get_chapter_entries(manager: ManifestManager, chapter_index: int) -> tuple:
"""Get entries for a specific chapter by index."""
toc = build_toc_from_manifest(manager)
if chapter_index < 1 or chapter_index > len(toc):
print(f"错误: 章节编号 {chapter_index} 无效 (范围: 1-{len(toc)})")
return None, None
chapter = toc[chapter_index - 1]
href = chapter['href']
entries = [e for e in manager.entries if e.file_path == href]
return chapter, entries
def create_char_based_chunks(entries: list, chunk_size: int = CHUNK_SIZE_CHARS) -> list:
"""
Create chunks based on character count (~5000 chars each).
Returns list of entry lists.
"""
chunks = []
current_chunk = []
current_size = 0
for entry in entries:
text_len = len(entry.original_text)
# If adding this entry exceeds limit and we have content, start new chunk
if current_size + text_len > chunk_size and current_chunk:
chunks.append(current_chunk)
current_chunk = []
current_size = 0
current_chunk.append(entry)
current_size += text_len
if current_chunk:
chunks.append(current_chunk)
return chunks
async def translate_chapter(chapter: dict, entries: list, manager: ManifestManager,
llm_client: LLMClient, profile, test_mode: bool = False):
"""Translate a complete chapter."""
print(f"\n开始翻译章节 #{chapter['index']}: {chapter['title'][:40]}...")
print(f" 文件: {chapter['href']}")
print(f" 总段落: {len(entries)}")
# Filter untranslated
untranslated = [e for e in entries if not e.translated_text]
print(f" 待翻译: {len(untranslated)}")
if not untranslated:
print(" ✅ 该章节已全部翻译!")
return
# Create character-based chunks
chunks = create_char_based_chunks(untranslated)
print(f" 分块: {len(chunks)} 个 Chunk (约{CHUNK_SIZE_CHARS}字符/块)")
if test_mode:
print(" [测试模式] 只翻译前2个 Chunk")
chunks = chunks[:2]
# Show chunk stats
for i, chunk in enumerate(chunks, 1):
total_chars = sum(len(e.original_text) for e in chunk)
print(f" Chunk {i}: {len(chunk)} 段落, {total_chars} 字符")
# Translate
restorer = FormatRestorer()
total_success = 0
total_failed = 0
for i, chunk in enumerate(chunks, 1):
chunk_chars = sum(len(e.original_text) for e in chunk)
print(f"\n 翻译 Chunk {i}/{len(chunks)} ({len(chunk)} 段, {chunk_chars} 字符)...")
try:
results = await llm_client.translate_chunk(
chunk,
instruction=profile.style_guide if hasattr(profile, 'style_guide') else None,
mode="bilingual"
)
# Apply results
chunk_success = 0
chunk_failed = 0
for entry in chunk:
if entry.entry_id in results:
translated = results[entry.entry_id]
entry.translated_text = translated
# Verify placeholder restoration
if entry.placeholders:
_, success = restorer.restore(translated, entry.placeholders)
if success:
chunk_success += 1
else:
chunk_failed += 1
print(f" ⚠️ 占位符还原警告: {entry.entry_id[-30:]}")
else:
chunk_success += 1
else:
chunk_failed += 1
print(f" ❌ 缺失: {entry.entry_id[-30:]}")
total_success += chunk_success
total_failed += chunk_failed
print(f" ✓ 成功: {chunk_success}, 失败: {chunk_failed}")
# Save after each chunk
manager.save()
except Exception as e:
print(f" ❌ Chunk {i} 翻译失败: {e}")
total_failed += len(chunk)
print(f"\n翻译完成:")
print(f" ✅ 成功: {total_success}")
print(f" ❌ 失败: {total_failed}")
# Show sample results
print(f"\n翻译样例 (前3段):")
print("-" * 60)
translated_entries = [e for e in entries if e.translated_text][:3]
for entry in translated_entries:
orig = entry.original_text[:40].replace('\n', ' ')
trans = entry.translated_text[:40].replace('\n', ' ') if entry.translated_text else "(无)"
print(f" 原: {orig}...")
print(f" 译: {trans}...")
print()
async def main():
parser = argparse.ArgumentParser(description="翻译指定章节")
parser.add_argument("--show-toc", action="store_true", help="显示章节目录")
parser.add_argument("--chapter", "-c", type=int, help="章节编号 (从1开始)")
parser.add_argument("--test", "-t", action="store_true", help="测试模式 (只翻译前2个chunk)")
args = parser.parse_args()
if not MANIFEST_PATH.exists():
print(f"错误: Manifest 不存在: {MANIFEST_PATH}")
print("请先运行主管道生成 manifest。")
return
# Load manifest
manager = ManifestManager(MANIFEST_PATH)
manager.load()
print(f"已加载 manifest: {len(manager.entries)} 条目")
# Show TOC
if args.show_toc or not args.chapter:
show_toc(manager)
return
if not API_KEY:
print("错误: V3_API_KEY 未设置")
return
# Get chapter entries
chapter, entries = get_chapter_entries(manager, args.chapter)
if not chapter:
return
# Initialize LLM client
from src.utils import ensure_directory
ensure_directory(CHUNK_DIR)
llm_client = LLMClient(
api_key=API_KEY,
base_url=BASE_URL,
model=MODEL,
extra_headers=EXTRA_HEADERS,
chunk_dir=CHUNK_DIR
)
try:
# Generate profile
print("\n生成书籍 Profile... (Skipping for debug)")
# profiler = BookProfiler(llm_client)
# profile = await profiler.analyze(manager.entries)
# print(f" 风格: {profile.style_guide[:80] if profile.style_guide else '(无)'}...")
class DummyProfile:
style_guide = "Keep technical terms. Translate accurately."
profile = DummyProfile()
# Translate chapter
await translate_chapter(chapter, entries, manager, llm_client, profile, args.test)
print(f"\n✅ Manifest 已保存: {MANIFEST_PATH}")
print(f"✅ Chunk 文件保存在: {CHUNK_DIR}")
finally:
await llm_client.close()
if __name__ == "__main__":
print("DEBUG: Script started execution")
try:
asyncio.run(main())
print("DEBUG: Script finished execution")
except Exception as e:
import traceback
traceback.print_exc()
print(f"CRITICAL ERROR: {e}")
+47
View File
@@ -0,0 +1,47 @@
import sys
from pathlib import Path
import traceback
# Add src to path
sys.path.insert(0, str(Path(__file__).parent.parent))
from src.epub_cleaner import EpubCleaner
from src.bilingual_builder import BilingualBuilder
from src.utils import setup_logger
from src.data_model import BookStructure
logger = setup_logger("verify_toc")
def verify_fix():
INPUT_EPUB = Path("input/Empire of AI Dreams and Nightmares in Sam Altmans OpenAI (Karen Hao).epub")
OUTPUT_EPUB = Path("output/verify_toc.epub")
WORK_DIR = Path("tmp/verify_toc")
if not INPUT_EPUB.exists():
logger.error(f"Input not found: {INPUT_EPUB}")
return
# Clean
logger.info("Step 1: Cleaning EPUB...")
cleaner = EpubCleaner(INPUT_EPUB, WORK_DIR)
json_path = cleaner.clean()
# Load structure
with open(json_path, 'r') as f:
structure = BookStructure.model_validate_json(f.read())
# Build
logger.info("Step 2: Building EPUB with TOC preservation...")
builder = BilingualBuilder(WORK_DIR, original_epub_path=INPUT_EPUB)
builder.build(structure, OUTPUT_EPUB)
logger.info(f"Step 3: EPUB generated at {OUTPUT_EPUB}")
if __name__ == "__main__":
print("Starting verification script...")
try:
verify_fix()
print("Verification script finished.")
except Exception as e:
traceback.print_exc()
print(f"CRITICAL ERROR: {e}")
View File
+96
View File
@@ -0,0 +1,96 @@
from pathlib import Path
from typing import List, Dict, Optional
from bs4 import BeautifulSoup
from src.common.data_model import BookStructure, ManifestEntry
from src.assembly.format_restorer import FormatRestorer
from src.common.utils import setup_logger
logger = setup_logger("backfill_engine")
class BackfillEngine:
"""
Applies translations back to the BookStructure.
"""
def __init__(self):
self.restorer = FormatRestorer()
def backfill(self, structure: BookStructure, manifest_entries: List[ManifestEntry], mode: str = "bilingual") -> BookStructure:
"""
Modifies the BookStructure in-place with translations.
Args:
structure: The BookStructure (from book_structure.json).
manifest_entries: List of translations.
mode: 'bilingual' or 'target_only'.
"""
logger.info(f"Backfilling with mode: {mode}")
# Index manifest by file and element ID for faster lookup
# Map: file_path -> element_id -> ManifestEntry
manifest_map: Dict[str, Dict[str, ManifestEntry]] = {}
for entry in manifest_entries:
if not entry.translated_text:
continue # Skip untranslated entries
if entry.file_path not in manifest_map:
manifest_map[entry.file_path] = {}
manifest_map[entry.file_path][entry.element_id] = entry
# Iterate resources in structure
for item_id, resource in structure.resources.items():
if resource.media_type != "application/xhtml+xml" or resource.href not in manifest_map:
continue
file_entries = manifest_map[resource.href]
if not file_entries:
continue
logger.debug(f"Processing {resource.href} with {len(file_entries)} translations")
soup = BeautifulSoup(resource.content, 'html.parser')
modified = False
for element_id, entry in file_entries.items():
element = soup.find(id=element_id)
if not element:
logger.warning(f"Element {element_id} not found in {resource.href}")
continue
# Restore formatting
restored_html, _ = self.restorer.restore(entry.translated_text, entry.placeholders)
# Create translated tag
new_tag = soup.new_tag(element.name)
# Parse restored HTML to get content nodes
inner_soup = BeautifulSoup(restored_html, 'html.parser')
if inner_soup.body:
for child in list(inner_soup.body.children):
new_tag.append(child)
else:
for child in list(inner_soup.children):
new_tag.append(child)
# Copy classes and add 'translation'
classes = element.get('class', [])
if isinstance(classes, str):
classes = classes.split()
new_tag['class'] = classes + ['translation']
# Copy style
style = element.get('style')
if style:
new_tag['style'] = style
if mode == "bilingual":
element.insert_after(new_tag)
else:
element.replace_with(new_tag)
modified = True
if modified:
resource.content = str(soup)
return structure
+300
View File
@@ -0,0 +1,300 @@
import shutil
import uuid
from pathlib import Path
from ebooklib import epub
from src.common.data_model import BookStructure
from src.common.utils import setup_logger, ensure_directory
logger = setup_logger("bilingual_builder")
class BilingualBuilder:
"""
Assembles the final EPUB from BookStructure.
Preserves the original TOC structure by reading it from the original EPUB.
"""
def __init__(self, work_dir: Path, original_epub_path: Path = None):
self.work_dir = work_dir
self.assets_dir = work_dir / "assets"
self.original_epub_path = original_epub_path
self._original_book = None
def _load_original_book(self):
"""Lazy load original book for TOC extraction."""
if self._original_book is None and self.original_epub_path and self.original_epub_path.exists():
try:
self._original_book = epub.read_epub(str(self.original_epub_path))
logger.debug(f"Loaded original EPUB for TOC: {self.original_epub_path}")
except Exception as e:
logger.warning(f"Failed to load original EPUB: {e}")
return self._original_book
def _sanitize_toc(self, toc):
"""
Ensure all TOC nodes have IDs (for ebooklib compatibility).
From v0.08 bilingual_builder.py
"""
result = []
for item in toc:
if isinstance(item, (epub.Link, epub.Section)):
if not getattr(item, 'uid', None):
item.uid = f"navPoint-{uuid.uuid4().hex[:8]}"
result.append(item)
elif isinstance(item, tuple) and len(item) == 2:
# Handle (Section, [children]) structure
section, children = item
if isinstance(section, (epub.Link, epub.Section)):
if not getattr(section, 'uid', None):
section.uid = f"navPoint-{uuid.uuid4().hex[:8]}"
sanitized_children = self._sanitize_toc(children)
result.append((section, sanitized_children))
else:
result.append(item)
return result
def _validate_and_fix_toc(self, toc, book_items):
"""
Recursively validate and fix TOC links.
Removes nodes with broken links that cannot be fixed.
"""
fixed_toc = []
for item in toc:
if isinstance(item, (epub.Link, epub.Section)):
# Check href
href = getattr(item, 'href', '')
if href:
# Remove anchor for check
clean_href = href.split('#')[0]
# Check if item exists in book (by file_name)
found = False
for existing_item in book_items.values():
if existing_item.file_name == clean_href:
found = True
break
if clean_href.endswith(existing_item.file_name) or existing_item.file_name.endswith(clean_href):
item.href = existing_item.file_name
found = True
break
if not found:
if 'c0.xhtml' in clean_href:
for existing_item in book_items.values():
if 'titlepage' in existing_item.file_name or 'cover' in existing_item.file_name.lower():
if existing_item.media_type == "application/xhtml+xml":
logger.info(f"Fixed TOC link: {href} -> {existing_item.file_name}")
item.href = existing_item.file_name
found = True
break
if not found:
logger.warning(f"Removing broken TOC link: {href}")
continue
if isinstance(item, tuple) and len(item) == 2:
section, children = item
fixed_children = self._validate_and_fix_toc(children, book_items)
fixed_toc.append((section, fixed_children))
else:
fixed_toc.append(item)
return fixed_toc
def build(self, structure: BookStructure, output_path: Path) -> Path:
"""
Builds the EPUB file.
Returns the path to the generated EPUB.
"""
logger.info(f"Building final EPUB: {output_path}")
output_path.parent.mkdir(parents=True, exist_ok=True)
book = epub.EpubBook()
# 1. Metadata
book.set_identifier(structure.metadata.identifier or f"uuid-{uuid.uuid4().hex[:12]}")
book.set_title(structure.metadata.title)
book.set_language(structure.metadata.language)
book.add_author(structure.metadata.author)
# 2. Copy TOC from original EPUB if available
original_book = self._load_original_book()
if original_book and hasattr(original_book, 'toc') and original_book.toc:
book.toc = self._sanitize_toc(original_book.toc)
logger.info("Copied TOC structure from original EPUB")
# 3. Add Resources - First pass: Collect CSS items
items_map = {} # id -> epub_item
css_items = [] # List of CSS EpubItem for linking
html_items = [] # List of (item_id, EpubHtml) tuples
for item_id, resource in structure.resources.items():
# Skip NCX - we'll handle it separately
if resource.media_type == "application/x-dtbncx+xml":
continue
if resource.media_type == "application/xhtml+xml":
# HTML Item - create but don't add yet (need to add CSS links)
item = epub.EpubHtml(
uid=item_id,
file_name=resource.href,
media_type=resource.media_type,
content=resource.content.encode('utf-8')
)
html_items.append((item_id, item))
elif resource.file_path:
# Binary/Asset Item
asset_full_path = self.work_dir / resource.file_path
if not asset_full_path.exists():
logger.warning(f"Asset missing: {asset_full_path}")
continue
with open(asset_full_path, 'rb') as f:
content = f.read()
# Check if this is the cover image
if structure.metadata.cover_image_id == item_id:
logger.info(f"Setting cover image: {item_id}")
# set_cover automatically creates the item and sets metadata
book.set_cover(resource.href, content)
# We still need to track it in items_map for spine/TOC references if needed?
# ebooklib set_cover creates an item withuid='cover-img' (default) or similar?
# Actually set_cover logic:
# def set_cover(self, file_name, content, create_page=True):
# c = EpubCover(file_name=file_name)
# c.content = content
# self.add_item(c)
# self.add_metadata(None, 'meta', '', {'name': 'cover', 'content': 'cover-img'})
# Be careful: ebooklib might change the ID.
# If we use set_cover, we should verify how it affects references.
# However, for the cover image specifically, usually it's referenced by the cover page
# (which set_cover creates if create_page=True).
# If create_page=False, we just get metadata.
# Let's use set_cover with create_page=False (since we likely preserved the cover page HTML)
# and rely on existing HTML to point to it?
# Or let ebooklib handle it.
# Most translator users want the cover to just work.
book.set_cover(resource.href, content, create_page=False)
# We also need to add it to items_map so we don't try to add it again
# But set_cover adds it to the book.
# We need to find the item added by set_cover to put in items_map
# standard ebooklib set_cover adds item with id derived or fixed?
# Actually, if we use set_cover, we might introduce a duplicate if we're not careful about IDs.
# Simplified approach:
# 1. Add as normal EpubImage
# 2. Add metadata manually pointing to it
item = epub.EpubImage(
uid=item_id,
file_name=resource.href,
media_type=resource.media_type,
content=content
)
book.add_item(item)
book.add_metadata(None, 'meta', item_id, {'name': 'cover', 'content': item_id})
items_map[item_id] = item
continue
if "image" in resource.media_type:
item = epub.EpubImage(
uid=item_id,
file_name=resource.href,
media_type=resource.media_type,
content=content
)
else:
item = epub.EpubItem(
uid=item_id,
file_name=resource.href,
media_type=resource.media_type,
content=content
)
# Track CSS items
if resource.media_type == "text/css":
css_items.append(item)
book.add_item(item)
items_map[item_id] = item
else:
logger.warning(f"Skipping resource {item_id}: No content or file path.")
continue
# 4. Add HTML items with CSS links
for item_id, item in html_items:
html_dir = Path(item.file_name).parent
for css_item in css_items:
css_path = Path(css_item.file_name)
# Calculate relative path from HTML directory to CSS file
try:
relative_css_path = Path(css_path).relative_to(html_dir)
except ValueError:
# Not a subpath, calculate full relative
# Go up from html_dir, then down to css_path
up_count = len(html_dir.parts)
relative_css_path = Path("/".join([".."] * up_count)) / css_path
item.add_link(href=str(relative_css_path), rel='stylesheet', type='text/css')
book.add_item(item)
items_map[item_id] = item
# 5. Copy missing items from original EPUB (cover, etc.)
# This ensures TOC links don't break
if original_book:
added_hrefs = {item.file_name for item in items_map.values() if hasattr(item, 'file_name')}
for orig_item in original_book.get_items():
orig_name = orig_item.get_name()
if orig_name not in added_hrefs:
# Skip NCX and NAV - we generate these
if 'ncx' in orig_name.lower() or orig_name.endswith('nav.xhtml'):
continue
# Copy the item directly
try:
book.add_item(orig_item)
items_map[orig_item.id] = orig_item
logger.debug(f"Copied missing item from original: {orig_name}")
except Exception as e:
logger.warning(f"Failed to copy item {orig_name}: {e}")
# 6. Spine
book.spine = []
for item_id in structure.spine:
if item_id in items_map:
book.spine.append(items_map[item_id])
else:
logger.warning(f"Spine item {item_id} not found in resources.")
# Add missing spine items from original
if original_book:
for spine_id, _ in original_book.spine:
if spine_id not in [i.id for i in book.spine]:
orig_item = original_book.get_item_with_id(spine_id)
if orig_item and orig_item.id in items_map:
book.spine.append(items_map[orig_item.id])
# Validate and fix TOC
if original_book and hasattr(book, 'toc') and book.toc:
try:
book.toc = self._validate_and_fix_toc(book.toc, items_map)
logger.info("Validated and fixed TOC links")
except Exception as e:
logger.error(f"Failed to validate TOC: {e}")
# 7. Navigation - NCX and Nav
book.add_item(epub.EpubNcx())
book.add_item(epub.EpubNav())
# 7. Write
epub.write_epub(str(output_path), book)
logger.info(f"EPUB created successfully at {output_path}")
return output_path
@@ -0,0 +1,70 @@
import re
from typing import Dict, Tuple, List, Optional
from loguru import logger
from src.common.utils import setup_logger
logger = setup_logger("format_restorer")
class FormatRestorer:
"""
Restores HTML formatting from placeholders.
"""
PLACEHOLDER_REGEX = re.compile(r'φ(/?\d+)φ')
def restore(self, text_with_placeholders: str, placeholder_map: Dict[str, str]) -> Tuple[str, bool]:
"""
Restores HTML from text with placeholders.
Returns (restored_html, success).
"""
if not placeholder_map:
return text_with_placeholders or "", True
if not text_with_placeholders:
prefix = placeholder_map.get("_prefix", "")
suffix = placeholder_map.get("_suffix", "")
return prefix + suffix, True
prefix = placeholder_map.get("_prefix", "")
suffix = placeholder_map.get("_suffix", "")
inner_map = {k: v for k, v in placeholder_map.items() if not k.startswith("_")}
found_ids = set(self.PLACEHOLDER_REGEX.findall(text_with_placeholders))
expected_ids = set(inner_map.keys())
success = True
missing_ids = expected_ids - found_ids
if missing_ids:
logger.warning(f"Restoration warning: missing placeholders {missing_ids}")
success = False
unknown_ids = found_ids - expected_ids
if unknown_ids:
real_unknowns = set()
for pid in unknown_ids:
if pid.startswith('/') and pid[1:] in expected_ids:
continue
real_unknowns.add(pid)
if real_unknowns:
logger.warning(f"Restoration warning: unknown placeholders {real_unknowns}")
success = False
def replace_match(match):
pid = match.group(1)
if pid in inner_map:
return inner_map[pid]
else:
return ""
try:
restored_inner = self.PLACEHOLDER_REGEX.sub(replace_match, text_with_placeholders)
restored_html = prefix + restored_inner + suffix
return restored_html, success
except Exception as e:
logger.error(f"Restoration failed: {e}")
return prefix + self._strip_placeholders(text_with_placeholders) + suffix, False
def _strip_placeholders(self, text: str) -> str:
return self.PLACEHOLDER_REGEX.sub("", text)
+71
View File
@@ -0,0 +1,71 @@
import os
import yaml
from pathlib import Path
from typing import Dict, Any
from dotenv import load_dotenv
from src.common.utils import setup_logger
logger = setup_logger("config_loader")
class ConfigLoader:
def __init__(self, config_path: str = "config/config.yaml"):
# Resolve absolute path relative to project root if needed,
# but usually running from root so relative is fine.
self.config_path = Path(config_path)
self.config = self._load_defaults()
def _load_defaults(self) -> Dict[str, Any]:
return {
"llm": {
"model": "gpt-3.5-turbo",
"base_url": "https://api.openai.com/v1",
"timeout": 60,
"requests_per_minute": 60,
"concurrent_requests": 5
},
"translation": {
"chunk_size": 4000
}
}
def load_config(self) -> Dict[str, Any]:
# 1. Load YAML
if self.config_path.exists():
try:
with open(self.config_path, 'r', encoding='utf-8') as f:
file_config = yaml.safe_load(f)
if file_config:
self._deep_update(self.config, file_config)
logger.info(f"Loaded config from {self.config_path}")
except Exception as e:
logger.error(f"Failed to load config file: {e}")
else:
logger.warning(f"Config file not found at {self.config_path}, using defaults")
# 2. Env overrides (Priority: Env > Config File > Defaults)
load_dotenv()
# API Key is mandatory from Env (security best practice)
api_key = os.getenv("OPENAI_API_KEY")
if api_key:
self.config["llm"]["api_key"] = api_key
# Base URL override
base_url = os.getenv("OPENAI_BASE_URL")
if base_url:
self.config["llm"]["base_url"] = base_url
return self.config
def _deep_update(self, d, u):
for k, v in u.items():
if isinstance(v, dict):
d[k] = self._deep_update(d.get(k, {}), v)
else:
d[k] = v
return d
def load_global_config(path: str = "config/config.yaml") -> Dict[str, Any]:
loader = ConfigLoader(path)
return loader.load_config()
+52
View File
@@ -0,0 +1,52 @@
from pathlib import Path
from typing import List, Optional, Dict, Any
from pydantic import BaseModel, Field
class ManifestEntry(BaseModel):
"""Represents a single translatable unit."""
entry_id: str = Field(..., description="Global unique ID (e.g. file.html#paragraph_id)")
file_path: str = Field(..., description="Internal path in EPUB")
element_id: str = Field(..., description="HTML ID (e.g. uuid-1234)")
original_text: str
placeholders: Dict[str, str] = Field(default_factory=dict)
translated_text: Optional[str] = None
context: Optional[str] = None
class BookMetaData(BaseModel):
title: str = "Unknown Title"
author: str = "Unknown Author"
language: str = "en"
identifier: str = ""
cover_image_id: Optional[str] = None
class ResourceItem(BaseModel):
href: str
media_type: str
content: Optional[str] = None # For text/html
file_path: Optional[str] = None # For binary/assets (relative to assets dir)
properties: Optional[str] = None
class BookStructure(BaseModel):
metadata: BookMetaData
spine: List[str] = Field(default_factory=list, description="Ordered list of item IDs in spine")
resources: Dict[str, ResourceItem] = Field(default_factory=dict, description="Map of item_id to ResourceItem")
@classmethod
def load(cls, path: Path) -> "BookStructure":
"""Load BookStructure from JSON file."""
with open(path, 'r', encoding='utf-8') as f:
return cls.model_validate_json(f.read())
def save(self, path: Path):
"""Save BookStructure to JSON file."""
with open(path, 'w', encoding='utf-8') as f:
f.write(self.model_dump_json(indent=2))
class BookProfile(BaseModel):
"""Represents the profile of the book."""
title: str
author: str
genre: str = "General"
keywords: List[str] = Field(default_factory=list)
style_guide: str = ""
+18
View File
@@ -0,0 +1,18 @@
class EpubTranslatorError(Exception):
"""Base exception for Epub Translator."""
pass
class CleaningError(EpubTranslatorError):
pass
class ExtractionError(EpubTranslatorError):
pass
class TranslationError(EpubTranslatorError):
pass
class RestorationError(EpubTranslatorError):
pass
class BuildError(EpubTranslatorError):
pass
+25
View File
@@ -0,0 +1,25 @@
import logging
from pathlib import Path
def setup_logger(name: str, log_file: Path = None, level=logging.INFO):
"""Sets up a logger with the given name."""
logger = logging.getLogger(name)
logger.setLevel(level)
formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
if log_file:
file_handler = logging.FileHandler(log_file)
file_handler.setFormatter(formatter)
logger.addHandler(file_handler)
stream_handler = logging.StreamHandler()
stream_handler.setFormatter(formatter)
logger.addHandler(stream_handler)
return logger
def ensure_directory(path: Path):
"""Ensures a directory exists."""
if not path.exists():
path.mkdir(parents=True, exist_ok=True)
@@ -0,0 +1,184 @@
import shutil
import uuid
import zipfile
from pathlib import Path
from typing import Dict, Set
import ebooklib
from bs4 import BeautifulSoup, Tag
from ebooklib import epub
from src.common.data_model import BookStructure, BookMetaData, ResourceItem
from src.common.exceptions import CleaningError
from src.common.utils import setup_logger, ensure_directory
logger = setup_logger("epub_cleaner")
class EpubCleaner:
def __init__(self, input_path: Path, work_dir: Path):
self.input_path = input_path
self.work_dir = work_dir
self.assets_dir = work_dir / "assets"
self.json_path = work_dir / "book_structure.json"
def clean(self) -> Path:
"""
Cleans the input EPUB and generates book_structure.json.
Returns the path to the JSON file.
"""
logger.info(f"Starting cleanup for {self.input_path}")
ensure_directory(self.work_dir)
ensure_directory(self.assets_dir)
try:
book = epub.read_epub(self.input_path)
# 1. Extract Metadata
metadata = self._extract_metadata(book)
# 2. Process Resources
resources = {}
# Use zipfile for binary extraction to avoid ebooklib's memory overhead/decoding issues
with zipfile.ZipFile(self.input_path, 'r') as zf:
# Map ebooklib items to zip entries isn't straightforward directly via name
# So we iterate ebooklib items and assume standard structure or handle content bytes
for item in book.get_items():
item_id = item.get_id()
file_name = item.get_name()
media_type = item.get_type() # ebooklib constant
if media_type == ebooklib.ITEM_DOCUMENT:
# Clean HTML
content_str = item.get_content().decode('utf-8')
cleaned_content = self._clean_html(content_str, file_name)
resources[item_id] = ResourceItem(
href=file_name,
media_type="application/xhtml+xml",
content=cleaned_content
)
elif media_type in (ebooklib.ITEM_IMAGE, ebooklib.ITEM_STYLE, ebooklib.ITEM_FONT, ebooklib.ITEM_COVER):
# Save asset
# Check if it is a cover
if media_type == ebooklib.ITEM_COVER:
metadata.cover_image_id = item_id
logger.info(f"Found cover image: {item_id} ({file_name})")
# Preserve directory structure to avoid collisions
asset_path = self.assets_dir / file_name
ensure_directory(asset_path.parent)
# Ebooklib might change filenames, safer to use item.get_content()
with open(asset_path, "wb") as f:
f.write(item.get_content())
resources[item_id] = ResourceItem(
href=file_name,
media_type=self._get_media_type_str(item),
file_path=str(asset_path.relative_to(self.work_dir))
)
elif media_type == ebooklib.ITEM_NAVIGATION:
# NCX or NAV document - preserve for TOC
content_bytes = item.get_content()
asset_path = self.assets_dir / file_name
ensure_directory(asset_path.parent)
with open(asset_path, "wb") as f:
f.write(content_bytes)
# Determine media type
if file_name.endswith('.ncx'):
mt = "application/x-dtbncx+xml"
else:
mt = "application/xhtml+xml"
resources[item_id] = ResourceItem(
href=file_name,
media_type=mt,
file_path=str(asset_path.relative_to(self.work_dir))
)
logger.debug(f"Preserved navigation: {file_name}")
else:
# Skip other items (scripts, etc.)
pass
# 3. Extract Spine
spine_ids = [item[0] for item in book.spine]
# 4. Construct Structure
structure = BookStructure(
metadata=metadata,
spine=spine_ids,
resources=resources
)
# 5. Serialize
with open(self.json_path, "w", encoding="utf-8") as f:
f.write(structure.model_dump_json(indent=2))
logger.info(f"Cleanup finished. Structure saved to {self.json_path}")
return self.json_path
except Exception as e:
logger.error(f"Cleaning failed: {e}")
raise CleaningError(f"Failed to clean EPUB: {e}") from e
def _extract_metadata(self, book: epub.EpubBook) -> BookMetaData:
title = book.get_metadata('DC', 'title')[0][0] if book.get_metadata('DC', 'title') else "Unknown"
author = book.get_metadata('DC', 'creator')[0][0] if book.get_metadata('DC', 'creator') else "Unknown"
lang = book.get_metadata('DC', 'language')[0][0] if book.get_metadata('DC', 'language') else "en"
ident = book.get_metadata('DC', 'identifier')[0][0] if book.get_metadata('DC', 'identifier') else ""
return BookMetaData(
title=str(title),
author=str(author),
language=str(lang),
identifier=str(ident)
)
def _get_media_type_str(self, item) -> str:
# Helper to map ebooklib type to mime string if needed
# ebooklib doesn't expose easy MIME string for all types directly on item object sometimes
if hasattr(item, 'media_type'):
return item.media_type
return "application/octet-stream"
def _clean_html(self, content: str, filename: str) -> str:
soup = BeautifulSoup(content, 'html.parser')
# 1. Provide IDs for structural/translatable elements
self._ensure_element_ids(soup)
# 2. Flatten divs (Disable to prevent style loss)
# self._flatten_divs(soup)
return str(soup)
def _ensure_element_ids(self, soup: BeautifulSoup):
"""
Injects UUIDs into p, h1-h6, li tags if they don't have an ID.
This provides the anchor for translation backfilling.
"""
targets = soup.find_all(['p', 'h1', 'h2', 'h3', 'h4', 'h5', 'h6', 'li'])
for tag in targets:
if not tag.has_attr('id'):
tag['id'] = f"uuid-{uuid.uuid4()}"
def _flatten_divs(self, soup: BeautifulSoup):
"""
Converts generic divs containing only inline text/styles to p tags.
Recursive strategies can be complex, sticking to simple heuristic from archive.
"""
inline_tags = {'span', 'em', 'strong', 'i', 'b', 'u', 'a', 'br', 'sub', 'sup', 'small'}
for div in list(soup.find_all('div')):
# If div has no block children, convert to p
has_block = any(
isinstance(c, Tag) and c.name not in inline_tags
for c in div.children
)
if not has_block:
div.name = 'p'
@@ -0,0 +1,372 @@
import re
from bs4 import BeautifulSoup, Tag
from typing import Tuple, Dict, List
class HeadingDetector:
"""Detects heading types and paragraph roles."""
CHAPTER_PATTERNS = [
r'^(chapter|chap\.?|part)\s+([0-9]+|[ivxlc]+|[a-z])',
r'^(第\s*[0-9一二三四五六七八九十百]+\s*[章节部篇])',
r'^(\d+|[IVXLC]+|[A-Z])\.$'
]
EPIGRAPH_CLASSES = {
'epigraph', 'quote', 'blockquote', 'motto',
'dedication', 'verse', 'poetry', 'poem'
}
def detect(self, element: Tag, text: str) -> str:
if self._is_epigraph(element):
return "epigraph"
tag_name = element.name.lower()
if tag_name in ['h1', 'h2']:
return "chapter" if self._matches_chapter_pattern(text) else "section"
if tag_name == 'h3':
return "section"
if tag_name in ['h4', 'h5', 'h6']:
return "subsection"
if self._is_pseudo_heading(element, text):
return "subsection"
return "body"
def _is_epigraph(self, element: Tag) -> bool:
if element.name == 'blockquote':
return True
current = element
for _ in range(3):
if not current: break
classes = current.get('class', [])
if isinstance(classes, list):
classes = ' '.join(classes)
if any(k in classes.lower() for k in self.EPIGRAPH_CLASSES):
return True
current = current.parent
return False
def _matches_chapter_pattern(self, text: str) -> bool:
text = text.strip().lower()
for pattern in self.CHAPTER_PATTERNS:
if re.match(pattern, text, re.IGNORECASE):
return True
return False
def _is_pseudo_heading(self, element: Tag, text: str) -> bool:
if element.name != 'p':
return False
text = text.strip()
if not text or len(text) > 80:
return False
children = list(element.children)
if len(children) == 1 and isinstance(children[0], Tag):
if children[0].name in ['strong', 'b']:
return True
return False
class FormatExtractor:
"""
HTML Format Extractor (Ported from v0.09 v3)
Handles inline styles, formulas, and drop caps.
"""
FORMULA_CHARS = re.compile(
r'^[\d\s\+\-\×\÷\=\(\)\[\]\{\}\<\>\^\*\/\.\,\;\:\'\"\`\~\@\#\$\%\&\|\\'
r'αβγδεζηθικλμνξοπρστυφχψωΑΒΓΔΕΖΗΘΙΚΛΜΝΞΟΠΡΣΤΥΦΧΨΩ'
r'a-zA-Z]+$'
)
def __init__(self):
self.detector = HeadingDetector()
def extract(self, element_html: str) -> Tuple[str, str, Dict[str, str], str, List[str]]:
"""
Extracts format information.
Returns:
clean_text: Pure text
text_with_placeholders: Text with inline placeholders
placeholder_map: Map of placeholders
paragraph_type: Detected type
endnote_anchors: List of detected endnote IDs
"""
soup = BeautifulSoup(element_html, 'html.parser')
root = list(soup.children)[0] if list(soup.children) else soup
clean_text = root.get_text().strip()
clean_text = re.sub(r'\s+', ' ', clean_text)
p_type = self.detector.detect(root, clean_text) if isinstance(root, Tag) else "body"
inner_html = root.decode_contents() if isinstance(root, Tag) else str(root)
text_with_ph, local_map = self._smart_extract_v3(inner_html)
if text_with_ph:
text_with_ph = re.sub(r'\s+', ' ', text_with_ph).strip()
# Verify integrity
stripped_text = self._strip_placeholders(text_with_ph)
stripped_text = re.sub(r'\s+', ' ', stripped_text).strip()
if not self._verify_content_integrity(clean_text, stripped_text):
# Fallback
text_with_ph, local_map = self._fallback_extract(inner_html, clean_text)
endnote_anchors = []
for pid, html in local_map.items():
if pid.startswith("_"):
continue
if re.match(r'<(span|a)\s+id="[a-zA-Z][a-zA-Z0-9]{2,5}"\s*>\s*</\1>', html):
endnote_anchors.append(pid)
return clean_text, text_with_ph, local_map, p_type, endnote_anchors
def _strip_placeholders(self, text: str) -> str:
return re.sub(r'φ/?[0-9]+φ', '', text)
def _verify_content_integrity(self, clean_text: str, stripped_text: str) -> bool:
def normalize(s):
s = re.sub(r'\s+', '', s)
s = s.lower()
return s
norm_clean = normalize(clean_text)
norm_stripped = normalize(stripped_text)
if norm_clean == norm_stripped:
return True
return False
def _fallback_extract(self, inner_html: str, clean_text: str) -> Tuple[str, Dict[str, str]]:
return clean_text, {"_prefix": "", "_suffix": ""}
def _smart_extract_v3(self, inner_html: str) -> Tuple[str, Dict[str, str]]:
parts = re.split(r'(<[^>]+>)', inner_html)
parts = [p for p in parts if p]
if not parts:
return "", {"_prefix": "", "_suffix": ""}
part_types = []
for part in parts:
if part.startswith('<'):
part_types.append('tag')
elif not part.strip():
part_types.append('whitespace')
elif self._is_translatable_text(part):
part_types.append('translatable')
else:
part_types.append('formula')
first_trans_idx = None
last_trans_idx = None
for i, t in enumerate(part_types):
if t == 'translatable':
if first_trans_idx is None:
first_trans_idx = i
last_trans_idx = i
if first_trans_idx is None:
return "", {"_prefix": inner_html, "_suffix": ""}
# Prefix Separation
safe_prefix_end = 0
for i in range(first_trans_idx):
if part_types[i] == 'tag':
tag = parts[i]
is_self_closing = tag.endswith('/>') or re.match(r'<(br|hr|img|meta|link)\b', tag, re.I)
is_closing = tag.startswith('</')
if is_self_closing or is_closing:
safe_prefix_end = i + 1
else:
break
elif part_types[i] == 'whitespace':
safe_prefix_end = i + 1
else:
break
# Suffix Separation
safe_suffix_start = len(parts)
for i in range(len(parts) - 1, last_trans_idx, -1):
if part_types[i] == 'tag':
tag = parts[i]
is_closing = tag.startswith('</')
is_self_closing = tag.endswith('/>') or re.match(r'<(br|hr|img|meta|link)\b', tag, re.I)
if is_closing or is_self_closing:
safe_suffix_start = i
else:
break
elif part_types[i] == 'whitespace':
safe_suffix_start = i
else:
break
prefix_parts = parts[:safe_prefix_end]
middle_parts = parts[safe_prefix_end:safe_suffix_start]
middle_types = part_types[safe_prefix_end:safe_suffix_start]
suffix_parts = parts[safe_suffix_start:]
# Drop Cap Check
if prefix_parts and middle_parts:
prefix_parts, middle_parts, middle_types = self._handle_drop_cap(
prefix_parts, middle_parts, middle_types
)
local_map = {}
if prefix_parts:
local_map["_prefix"] = "".join(prefix_parts)
if suffix_parts:
local_map["_suffix"] = "".join(suffix_parts)
# Middle processing
placeholder_counter = 1
result_parts = []
tag_stack = []
i = 0
while i < len(middle_parts):
part = middle_parts[i]
ptype = middle_types[i]
if ptype == 'translatable':
result_parts.append(part)
i += 1
elif ptype == 'tag':
is_closing = part.startswith('</')
if is_closing:
if tag_stack:
open_id, open_tag = tag_stack.pop()
local_map[f"/{open_id}"] = part
result_parts.append(f"φ/{open_id}φ")
else:
pid = str(placeholder_counter)
placeholder_counter += 1
local_map[pid] = part
result_parts.append(f"φ{pid}φ")
i += 1
else:
has_translatable_after = False
for j in range(i + 1, len(middle_parts)):
if middle_types[j] == 'translatable':
has_translatable_after = True
break
elif middle_types[j] == 'tag' and middle_parts[j].startswith('</'):
break
if has_translatable_after:
pid = str(placeholder_counter)
placeholder_counter += 1
local_map[pid] = part
result_parts.append(f"φ{pid}φ")
tag_stack.append((pid, part))
i += 1
else:
block_parts = []
while i < len(middle_parts) and middle_types[i] != 'translatable':
block_parts.append(middle_parts[i])
i += 1
if block_parts:
block_html = "".join(block_parts)
pid = str(placeholder_counter)
placeholder_counter += 1
local_map[pid] = block_html
result_parts.append(f"φ{pid}φ")
else:
result_parts.append(part)
i += 1
text_with_ph = "".join(result_parts)
text_with_ph = re.sub(r'\s+', ' ', text_with_ph).strip()
# Merge consecutive placeholders
def merge_match(m):
full_match = m.group(0)
pids = re.findall(r'φ(/?\d+)φ', full_match)
if len(pids) <= 1:
return full_match
merged_html = ""
for pid in pids:
if pid in local_map:
merged_html += local_map[pid]
del local_map[pid]
new_pid = pids[0] if pids[0].isdigit() else pids[0][1:]
local_map[new_pid] = merged_html
return f"φ{new_pid}φ"
text_with_ph = re.sub(r'(φ/?\d+φ)(φ/?\d+φ)+', merge_match, text_with_ph)
return text_with_ph, local_map
def _handle_drop_cap(self, prefix_parts: List[str], middle_parts: List[str], middle_types: List[str]):
if not prefix_parts or not middle_parts:
return prefix_parts, middle_parts, middle_types
prefix_text = ""
for part in prefix_parts:
if not part.startswith('<'):
prefix_text = part.strip()
if not prefix_text or len(prefix_text) != 1 or not prefix_text.isupper():
return prefix_parts, middle_parts, middle_types
first_middle_text = ""
first_middle_idx = -1
for i, (part, ptype) in enumerate(zip(middle_parts, middle_types)):
if ptype == 'translatable':
first_middle_text = part.strip()
first_middle_idx = i
break
if not first_middle_text:
return prefix_parts, middle_parts, middle_types
is_drop_cap = False
first_char = first_middle_text[0] if first_middle_text else ''
if first_char.islower() or first_char.isupper():
is_drop_cap = True
combined = prefix_text + first_middle_text.split()[0] if first_middle_text else ""
if not (len(combined) >= 2 and combined.isalpha()):
return prefix_parts, middle_parts, middle_types
new_prefix = []
skip_until_close = False
found_letter = False
for part in prefix_parts:
if part.startswith('<') and not part.startswith('</'):
skip_until_close = True
elif part.startswith('</'):
if skip_until_close:
skip_until_close = False
continue
new_prefix.append(part)
elif part.strip() == prefix_text:
found_letter = True
continue
else:
if not skip_until_close:
new_prefix.append(part)
if found_letter:
middle_parts = middle_parts.copy()
middle_parts[first_middle_idx] = prefix_text + middle_parts[first_middle_idx]
prefix_parts = new_prefix
return prefix_parts, middle_parts, middle_types
def _is_translatable_text(self, text: str) -> bool:
text = text.strip()
if not text:
return False
if re.search(r'[a-zA-Z]{3,}', text):
return True
if ' ' in text and re.search(r'[a-zA-Z]', text):
return True
if re.search(r'\d', text):
return True
return False
@@ -0,0 +1,76 @@
import json
import random
from typing import Dict, List
from loguru import logger
from src.translation.llm_client import LLMClient
from src.translation.manifest_manager import ManifestManager
from src.common.data_model import BookProfile
class BookProfiler:
def __init__(self, llm_client: LLMClient):
self.llm_client = llm_client
def extract_sample_text(self, entries: List[object], char_limit: int = 3000) -> str:
"""Extract sample text from manifest entries."""
if not entries: return ""
# Simple sampling strategy: First few + random middle
intro_text = []
for entry in entries[:50]:
if len(entry.original_text) > 50:
intro_text.append(entry.original_text)
body_text = []
candidates = [e for e in entries[50:] if len(e.original_text) > 80]
if candidates:
samples = random.sample(candidates, min(5, len(candidates)))
body_text = [e.original_text for e in samples]
full_text = "\n\n".join(intro_text[:5] + body_text)
return full_text[:char_limit]
async def analyze(self, entries: List[object]) -> BookProfile:
"""Generate Book Profile."""
sample = self.extract_sample_text(entries)
if not sample:
return BookProfile(title="Unknown", author="Unknown")
logger.info("Generating Book Profile from sample text...")
system_prompt = "You are a senior publishing editor. Analyze the text and output JSON."
user_prompt = f"""
Please analyze the following book excerpt.
Output JSON format:
{{
"title": "Book Title",
"author": "Author Name",
"genre": "Genre",
"style": "Style description",
"keywords": ["keyword1", "keyword2"],
"style_guide": "Specific instruction for translator"
}}
Excerpt:
{sample}
"""
try:
response = await self.llm_client.raw_chat_completion(system_prompt, user_prompt)
json_str = response.strip()
# Basic cleanup
if "```json" in json_str:
json_str = json_str.split("```json")[1].split("```")[0].strip()
elif "```" in json_str:
json_str = json_str.split("```")[1].split("```")[0].strip()
data = json.loads(json_str)
return BookProfile(
title=data.get("title", "Unknown"),
author=data.get("author", "Unknown"),
genre=data.get("genre", "General"),
keywords=data.get("keywords", []),
style_guide=data.get("style_guide", "")
)
except Exception as e:
logger.error(f"Profile generation failed: {e}")
return BookProfile(title="Unknown", author="Unknown")
@@ -0,0 +1,132 @@
import re
from typing import List, Dict, Any, Optional
from bs4 import BeautifulSoup
from loguru import logger
from src.common.data_model import BookStructure, ManifestEntry, BookProfile
from src.preprocessing.format_extractor import FormatExtractor
from src.common.utils import setup_logger
logger = setup_logger("fine_grained_extractor")
class FineGrainedExtractor:
"""
Extracts translatable text segments from BookStructure.
Uses FormatExtractor for detailed content analysis.
"""
SKIP_TRANSLATION_PATTERNS = [
r'index\.x?html',
r'bibliography\.x?html',
r'endnotes?\.x?html',
r'footnotes?\.x?html',
r'copyright\.x?html',
]
TOC_PATTERNS = [
r'nav\.x?html',
r'toc\.x?html',
]
def __init__(self, translate_toc: bool = False):
self.translate_toc = translate_toc
self.format_extractor = FormatExtractor()
def extract(self, structure: BookStructure, profile: Optional[BookProfile] = None) -> List[ManifestEntry]:
"""
Extracts translatable segments from the BookStructure.
Iteration follows the spine order.
"""
logger.info("Starting extraction from BookStructure...")
manifest_entries = []
# Iterate over spine to maintain order
for item_id in structure.spine:
if item_id not in structure.resources:
logger.warning(f"Item ID {item_id} in spine but not in resources.")
continue
resource = structure.resources[item_id]
# Only process HTML/XHTML
if resource.media_type != "application/xhtml+xml" or not resource.content:
continue
file_path = resource.href
doc_type = self._classify_document(file_path)
# Parse HTML
soup = BeautifulSoup(resource.content, 'html.parser')
target_tags = ['p', 'h1', 'h2', 'h3', 'h4', 'h5', 'h6', 'li']
for element in soup.find_all(target_tags):
# Ensure element has ID (should have been done by Cleaner)
element_id = element.get('id')
if not element_id:
logger.warning(f"Element in {file_path} missing ID, skipping: {element.name}")
continue
# Check translation eligibility
raw_text = element.get_text(separator=' ', strip=True)
if not raw_text.strip():
continue
is_decorative = self._is_decorative(raw_text)
should_translate = self._should_translate(doc_type, is_decorative, raw_text)
if should_translate:
# Extract detailed format
outer_html = str(element)
clean_text, text_with_ph, ph_map, p_type, _ = self.format_extractor.extract(outer_html)
if clean_text.strip() and text_with_ph.strip():
entry = ManifestEntry(
entry_id=f"{file_path}#{element_id}",
file_path=file_path,
element_id=element_id,
original_text=text_with_ph,
placeholders=ph_map,
context=p_type
)
manifest_entries.append(entry)
logger.info(f"Extracted {len(manifest_entries)} entries in total.")
return manifest_entries
def _classify_document(self, file_name: str) -> str:
if not file_name: return 'core'
fname = file_name.lower()
if any(re.search(p, fname) for p in self.SKIP_TRANSLATION_PATTERNS): return 'skip'
if any(re.search(p, fname) for p in self.TOC_PATTERNS): return 'toc'
return 'core'
def _should_translate(self, doc_type: str, is_decorative: bool, text: str = "") -> bool:
if is_decorative: return False
if self._is_roman_numeral(text): return False
if doc_type == 'core': return True
if doc_type == 'toc': return self.translate_toc
return False
def _is_roman_numeral(self, text: str) -> bool:
text = text.strip().upper()
if not text: return False
pattern = re.compile(r"^M{0,4}(CM|CD|D?C{0,3})(XC|XL|L?X{0,3})(IX|IV|V?I{0,3})$")
return bool(pattern.match(text))
def _is_decorative(self, text: str) -> bool:
s = text.strip()
if not s: return False
if not any(c.isalnum() for c in s): return True
if len(s) > 20: return False
patterns = [
r'^[*\-_~•◆◇■□▪▫●○◉◎⦿⦾]+$',
r'^[*\-_~•◆◇■□▪▫●○◉◎⦿⦾\s]+$',
]
for p in patterns:
if re.match(p, s): return True
unique = set(s.replace(' ', ''))
if len(unique) <= 3 and (unique & set('*-_~•◆◇■□▪▫●○◉◎⦿⦾')):
return True
return False
+220
View File
@@ -0,0 +1,220 @@
import asyncio
import json
import re
import time
from datetime import datetime
from pathlib import Path
from typing import List, Dict, Optional
from openai import AsyncOpenAI
from tenacity import retry, stop_after_attempt, wait_exponential
from loguru import logger
from src.common.data_model import ManifestEntry
from src.common.utils import setup_logger, ensure_directory
logger = setup_logger("llm_client")
# Default chunk save directory (can be overridden)
DEFAULT_CHUNK_DIR = Path("tmp/chunks")
class RateLimiter:
"""Rate limiter for concurrency and RPM."""
def __init__(self, requests_per_minute: int, concurrent_requests: int):
self.semaphore = asyncio.Semaphore(concurrent_requests)
self.min_interval = 60.0 / requests_per_minute if requests_per_minute > 0 else 0
self.last_request_time = 0
self._lock = asyncio.Lock()
async def acquire(self):
await self.semaphore.acquire()
async with self._lock:
current_time = time.time()
wait_time = self.min_interval - (current_time - self.last_request_time)
if wait_time > 0:
await asyncio.sleep(wait_time)
self.last_request_time = time.time()
def release(self):
self.semaphore.release()
class LLMClient:
"""Generic OpenAI-compatible API Client with short ID strategy."""
def __init__(self, api_key: str, base_url: str, model: str = "gpt-3.5-turbo",
requests_per_minute: int = 60, concurrent_requests: int = 5,
extra_headers: Dict = None, chunk_dir: Path = None):
# Configure proxy client to avoid SOCKS issues and ensure connectivity
import httpx
import os
# Prefer HTTP proxy if available to avoid missing socksio support
proxy_url = os.environ.get("http_proxy") or os.environ.get("https_proxy")
http_client = httpx.AsyncClient(
proxy=proxy_url,
timeout=60.0,
follow_redirects=True
) if proxy_url else None
self.client = AsyncOpenAI(
base_url=base_url,
api_key=api_key,
default_headers=extra_headers,
http_client=http_client
)
self.model = model
self.rate_limiter = RateLimiter(requests_per_minute, concurrent_requests)
self.prompts = self._load_prompts()
self._chunk_counter = 0
# Chunk directory for debug output
self.chunk_dir = chunk_dir or DEFAULT_CHUNK_DIR
ensure_directory(self.chunk_dir)
def _load_prompts(self) -> Dict:
try:
with open("config/prompts.json", "r", encoding="utf-8") as f:
return json.load(f)
except Exception as e:
logger.error(f"Failed to load config/prompts.json: {e}")
return {}
async def translate_chunk(self, items: List[ManifestEntry], glossary: Dict = None,
instruction: str = None, mode: str = "bilingual") -> Dict[str, str]:
"""
Translate a chunk of items using short ID strategy.
Returns: Dict[entry_id, translated_text]
"""
if not items: return {}
# Build prompt with short IDs
id_map, prompt = self._build_prompt_with_short_ids(items)
try:
# Build System Prompt
base_sys_prompt = self.prompts.get("translation", {}).get("system",
"You are a professional English to Chinese translator.")
if instruction:
base_sys_prompt += f"\n\nBook Style Guide:\n{instruction}"
if glossary:
glossary_text = "\n".join([f"{k} -> {v}" for k, v in glossary.items()])
base_sys_prompt += f"\n\nTerminology:\n{glossary_text}"
# Short ID format instructions
base_sys_prompt += """
Output Format:
- Each line MUST start with #N: (keep this ID exactly as given)
- Preserve any φXφ or φ/Xφ placeholders EXACTLY as-is
- Only output translations, no explanations
- Match the number of output lines to input lines"""
# Save chunk before translation
chunk_id = self._save_chunk("before", prompt, base_sys_prompt)
logger.debug(f"Sending request to LLM (Chunk: {chunk_id}, Items: {len(items)})")
raw_response = await self._make_request(base_sys_prompt, prompt)
# Save chunk after translation
self._save_chunk("after", raw_response, base_sys_prompt, chunk_id)
# Parse with short ID mapping
results = self._parse_short_id_response(raw_response, id_map)
return results
except Exception as e:
logger.error(f"Translation failed: {e}")
return {item.entry_id: f"[Error - {str(e)}]" for item in items}
def _build_prompt_with_short_ids(self, items: List[ManifestEntry]) -> tuple:
"""
Build prompt with short IDs (#1, #2, ...).
Returns: (id_map, prompt_text)
"""
id_map = {} # short_id -> entry_id
lines = []
for i, item in enumerate(items, 1):
short_id = f"#{i}"
id_map[short_id] = item.entry_id
# Clean text (remove extra whitespace)
text = re.sub(r'\s+', ' ', item.original_text).strip()
lines.append(f"{short_id}: {text}")
return id_map, "\n".join(lines)
def _parse_short_id_response(self, response: str, id_map: Dict[str, str]) -> Dict[str, str]:
"""
Parse response with short ID format.
Returns: Dict[entry_id, translated_text]
"""
results = {}
for line in response.split("\n"):
line = line.strip()
if not line:
continue
# Match #N: pattern
match = re.match(r'^#(\d+):\s*(.+)$', line)
if match:
short_id = f"#{match.group(1)}"
translation = match.group(2).strip()
if short_id in id_map:
full_id = id_map[short_id]
results[full_id] = translation
else:
logger.warning(f"Unknown short ID in response: {short_id}")
return results
def _save_chunk(self, stage: str, content: str, system_prompt: str = None,
chunk_id: str = None) -> str:
"""Save chunk to tmp directory for debugging."""
if chunk_id is None:
self._chunk_counter += 1
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
chunk_id = f"{timestamp}_{self._chunk_counter:04d}"
filename = self.chunk_dir / f"chunk_{chunk_id}_{stage}.txt"
with open(filename, "w", encoding="utf-8") as f:
if system_prompt and stage == "before":
f.write("=== SYSTEM PROMPT ===\n")
f.write(system_prompt)
f.write("\n\n=== USER PROMPT ===\n")
f.write(content)
logger.debug(f"Saved chunk: {filename}")
return chunk_id
async def raw_chat_completion(self, system_prompt: str, user_prompt: str) -> str:
"""Generic chat completion."""
return await self._make_request(system_prompt, user_prompt)
@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10))
async def _make_request(self, system_prompt: str, user_prompt: str) -> str:
await self.rate_limiter.acquire()
try:
resp = await self.client.chat.completions.create(
model=self.model,
messages=[
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_prompt}
],
temperature=0.3,
)
return resp.choices[0].message.content.strip()
finally:
self.rate_limiter.release()
async def close(self):
await self.client.close()
@@ -0,0 +1,74 @@
from pathlib import Path
from typing import List, Dict, Optional
import json
from src.common.data_model import ManifestEntry
from src.common.utils import setup_logger
logger = setup_logger("manifest_manager")
class ManifestManager:
"""
Manages the translation manifest (Source of Truth).
Handles persistence and state updates.
"""
def __init__(self, manifest_path: Path):
self.manifest_path = manifest_path
self.entries: List[ManifestEntry] = []
self._entries_map: Dict[str, ManifestEntry] = {}
def load(self):
"""Loads manifest from disk if it exists."""
if not self.manifest_path.exists():
logger.info(f"Manifest not found at {self.manifest_path}, starting empty.")
return
try:
with open(self.manifest_path, 'r', encoding='utf-8') as f:
data = json.load(f)
self.entries = [ManifestEntry.model_validate(item) for item in data]
self._rebuild_map()
logger.info(f"Loaded {len(self.entries)} entries from manifest.")
except Exception as e:
logger.error(f"Failed to load manifest: {e}")
raise
def save(self):
"""Saves current state to disk."""
try:
# Pydantic v2: model_dump(mode='json') or just list dump
data = [entry.model_dump(mode='json') for entry in self.entries]
with open(self.manifest_path, 'w', encoding='utf-8') as f:
json.dump(data, f, indent=2, ensure_ascii=False)
logger.info(f"Saved {len(self.entries)} entries to manifest.")
except Exception as e:
logger.error(f"Failed to save manifest: {e}")
raise
def add_entries(self, new_entries: List[ManifestEntry]):
"""
Adds new entries to the manifest.
If an entry with the same ID exists, it keeps the EXISTING one (to preserve translations).
"""
count = 0
for entry in new_entries:
if entry.entry_id not in self._entries_map:
self.entries.append(entry)
self._entries_map[entry.entry_id] = entry
count += 1
if count > 0:
logger.info(f"Added {count} new entries to manifest.")
def update_translation(self, entry_id: str, translation: str):
"""Updates translation for a specific entry."""
if entry_id in self._entries_map:
self._entries_map[entry_id].translated_text = translation
else:
logger.warning(f"Attempted to update translation for unknown ID: {entry_id}")
def get_entry(self, entry_id: str) -> Optional[ManifestEntry]:
return self._entries_map.get(entry_id)
def _rebuild_map(self):
self._entries_map = {e.entry_id: e for e in self.entries}
@@ -0,0 +1,285 @@
"""
Translator Module - Handles translation of ManifestEntry items.
Key features:
- Character-based chunking (~5000 chars per chunk)
- Chapter-aware grouping (chunks don't cross file boundaries)
- Concurrent translation with asyncio.gather
- Progress tracking and error handling
"""
import asyncio
from typing import List, Dict
from collections import defaultdict
from loguru import logger
from src.common.data_model import ManifestEntry, BookProfile
from src.translation.llm_client import LLMClient
from src.assembly.format_restorer import FormatRestorer
from src.common.utils import setup_logger
logger = setup_logger("translator")
# Default chunk size in characters
DEFAULT_CHUNK_SIZE = 5000
# Maximum concurrent translations
MAX_CONCURRENT = 5
class Translator:
"""
Translates ManifestEntry items using LLM with chapter-aware chunking.
Supports both sequential and concurrent translation modes.
"""
def __init__(self, llm_client: LLMClient, chunk_size: int = DEFAULT_CHUNK_SIZE,
max_concurrent: int = MAX_CONCURRENT):
self.llm_client = llm_client
self.chunk_size = chunk_size
self.max_concurrent = max_concurrent
self.restorer = FormatRestorer()
async def translate(self, entries: List[ManifestEntry], profile: BookProfile,
concurrent: bool = True) -> List[ManifestEntry]:
"""
Translates all untranslated entries.
Args:
entries: All manifest entries
profile: Book profile with style guide
concurrent: Use concurrent translation (default True)
Returns:
The same entries list with translated_text populated
"""
untranslated = [e for e in entries if not e.translated_text]
if not untranslated:
logger.info("No new entries to translate.")
return entries
logger.info(f"Found {len(untranslated)} entries to translate")
# Group by chapter (file_path)
chapters = self._group_by_chapter(untranslated)
logger.info(f"Grouped into {len(chapters)} chapters")
# Create all chunks
all_chunks = []
for file_path, chapter_entries in chapters.items():
chapter_chunks = self._create_char_based_chunks(chapter_entries)
for chunk in chapter_chunks:
all_chunks.append((file_path, chunk))
total_chunks = len(all_chunks)
logger.info(f"Created {total_chunks} chunks (avg ~{self.chunk_size} chars each)")
if concurrent:
await self._translate_concurrent(all_chunks, profile)
else:
await self._translate_sequential(all_chunks, profile)
translated_count = sum(1 for e in entries if e.translated_text)
logger.info(f"Translation complete: {translated_count}/{len(entries)} entries translated")
return entries
async def _translate_concurrent(self, all_chunks: List, profile: BookProfile):
"""Translate chunks concurrently with semaphore control."""
semaphore = asyncio.Semaphore(self.max_concurrent)
completed = [0] # Use list for mutable counter in closure
total = len(all_chunks)
success = [0]
failed = [0]
async def translate_chunk_task(file_path: str, chunk: List[ManifestEntry], idx: int):
async with semaphore:
try:
results = await self.llm_client.translate_chunk(
chunk,
instruction=profile.style_guide if profile else None,
mode="bilingual"
)
for entry in chunk:
if entry.entry_id in results:
entry.translated_text = results[entry.entry_id]
success[0] += 1
else:
failed[0] += 1
except Exception as e:
logger.error(f"Chunk {idx} failed: {e}")
failed[0] += len(chunk)
finally:
completed[0] += 1
if completed[0] % 5 == 0 or completed[0] == total:
logger.info(f"Progress: {completed[0]}/{total} chunks ({success[0]} translated)")
tasks = [
translate_chunk_task(file_path, chunk, i)
for i, (file_path, chunk) in enumerate(all_chunks)
]
await asyncio.gather(*tasks)
logger.info(f"Concurrent translation: {success[0]} success, {failed[0]} failed")
async def _translate_sequential(self, all_chunks: List, profile: BookProfile):
"""Translate chunks sequentially."""
total_chunks = len(all_chunks)
success_count = 0
fail_count = 0
for idx, (file_path, chunk) in enumerate(all_chunks):
try:
results = await self.llm_client.translate_chunk(
chunk,
instruction=profile.style_guide if profile else None,
mode="bilingual"
)
for entry in chunk:
if entry.entry_id in results:
entry.translated_text = results[entry.entry_id]
success_count += 1
else:
fail_count += 1
logger.warning(f"Missing translation for: {entry.entry_id[-40:]}")
except Exception as e:
logger.error(f"Chunk {idx} translation failed: {e}")
fail_count += len(chunk)
if (idx + 1) % 10 == 0 or idx + 1 == total_chunks:
logger.info(f"Progress: {idx + 1}/{total_chunks} chunks ({success_count} entries translated)")
logger.info(f"Sequential translation: {success_count} success, {fail_count} failed")
async def translate_chapter(self, entries: List[ManifestEntry], file_path: str,
profile: BookProfile) -> Dict[str, int]:
"""
Translate a single chapter.
Args:
entries: All entries (will filter by file_path)
file_path: Chapter file path to translate
profile: Book profile
Returns:
Dict with 'success' and 'failed' counts
"""
chapter_entries = [e for e in entries if e.file_path == file_path and not e.translated_text]
if not chapter_entries:
logger.info(f"Chapter {file_path} has no untranslated entries")
return {"success": 0, "failed": 0}
logger.info(f"Translating chapter: {file_path} ({len(chapter_entries)} entries)")
chunks = self._create_char_based_chunks(chapter_entries)
logger.info(f"Created {len(chunks)} chunks")
success = 0
failed = 0
for i, chunk in enumerate(chunks, 1):
chunk_chars = sum(len(e.original_text) for e in chunk)
logger.debug(f"Chunk {i}/{len(chunks)}: {len(chunk)} entries, {chunk_chars} chars")
try:
results = await self.llm_client.translate_chunk(
chunk,
instruction=profile.style_guide if profile else None,
mode="bilingual"
)
for entry in chunk:
if entry.entry_id in results:
entry.translated_text = results[entry.entry_id]
# Verify placeholder preservation
if entry.placeholders:
_, restored_ok = self.restorer.restore(
entry.translated_text,
entry.placeholders
)
if not restored_ok:
logger.warning(f"Placeholder issue: {entry.entry_id[-40:]}")
success += 1
else:
failed += 1
except Exception as e:
logger.error(f"Chunk {i} failed: {e}")
failed += len(chunk)
logger.info(f"Chapter done: {success} success, {failed} failed")
return {"success": success, "failed": failed}
def _group_by_chapter(self, entries: List[ManifestEntry]) -> Dict[str, List[ManifestEntry]]:
"""Group entries by file_path (chapter)."""
chapters = defaultdict(list)
for entry in entries:
chapters[entry.file_path].append(entry)
return dict(chapters)
def _create_char_based_chunks(self, entries: List[ManifestEntry]) -> List[List[ManifestEntry]]:
"""
Create chunks based on character count.
Each chunk contains approximately self.chunk_size characters.
Chunks never cross chapter boundaries (entries from same file only).
"""
chunks = []
current_chunk = []
current_size = 0
for entry in entries:
text_len = len(entry.original_text)
# If adding this entry exceeds limit and we have content, start new chunk
if current_size + text_len > self.chunk_size and current_chunk:
chunks.append(current_chunk)
current_chunk = []
current_size = 0
current_chunk.append(entry)
current_size += text_len
if current_chunk:
chunks.append(current_chunk)
return chunks
def get_chapter_stats(self, entries: List[ManifestEntry]) -> List[Dict]:
"""
Get statistics for each chapter.
Returns list of dicts with: file_path, total, translated, pending, chars
"""
chapters = self._group_by_chapter(entries)
stats = []
for file_path, chapter_entries in sorted(chapters.items()):
total = len(chapter_entries)
translated = sum(1 for e in chapter_entries if e.translated_text)
total_chars = sum(len(e.original_text) for e in chapter_entries)
# Get first text as title preview
first_text = ""
for e in chapter_entries:
if e.original_text:
first_text = e.original_text[:40].replace('\n', ' ')
break
stats.append({
"file_path": file_path,
"title": first_text,
"total": total,
"translated": translated,
"pending": total - translated,
"chars": total_chars
})
return stats