kai/epub_bilingual_translator
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Normalize Chinese Part ordinals and support resumable complete-book translation

Browse Source
This commit is contained in:
Kai Tan
2026-10-08 08:32:05 +08:00
parent 498b5d09ce
commit 644e93b874
12 changed files with 186 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/assembly/backfiller.py
+7 -1
View File
@@ -1,7 +1,7 @@
import copy
import re
from src.common.html import parse_html
from src.preprocessing.navigation import normalize_heading_html, sync_navigation
from src.preprocessing.navigation import normalize_heading_html, sync_navigation, chinese_number
from src.assembly.format_restorer import FormatRestorer
from src.common.utils import setup_logger, add_spacing
@@ -76,6 +76,12 @@ class BackfillEngine:
else: new_tag.attrs.pop('style', None)
applied += 1
fragment = parse_html(f'<div>{entry.translated_html}</div>').body.div
record = titles.get(entry.entry_id)
if record and record.get('role') == 'part' and record.get('number'):
number = chinese_number(record['number'])
for span in fragment.find_all('span'):
if span.get_text(strip=True) == number:
span['data-translation-ordinal'] = 'true'
for node in list(fragment.contents):
new_tag.append(node.extract())
if mode == 'bilingual':
src/assembly/bilingual_style.css
+10
View File
@@ -22,3 +22,13 @@ p.translation::first-letter {
.translation[data-translation-role="body"] {
text-indent: 2em !important;
}
/* A Chinese ordinal is one phrase, even when the publisher enlarged only 4.
Limit this reset to the translated number; keep the original Part/title design. */
.translation [data-translation-ordinal="true"] {
font-family: inherit !important;
font-size: inherit !important;
font-weight: inherit !important;
color: inherit !important;
line-height: inherit !important;
vertical-align: baseline !important;
}
src/common/config.py
+2
View File
@@ -55,6 +55,8 @@ class ConfigLoader:
base_url = os.getenv("OPENAI_BASE_URL")
if base_url:
self.config["llm"]["base_url"] = base_url
if 'zenmux.ai' in self.config['llm'].get('base_url', '') and os.getenv('ZENMUX_API_KEY'):
self.config['llm']['api_key'] = os.environ['ZENMUX_API_KEY']
return self.config
src/preprocessing/text_extractor.py
+8 -3
View File
@@ -29,8 +29,9 @@ class FineGrainedExtractor:
r'toc\.x?html',
]
def __init__(self, translate_toc: bool = False):
def __init__(self, translate_toc: bool = False, complete_book: bool = False):
self.translate_toc = translate_toc
self.complete_book = complete_book
self.format_extractor = FormatExtractor()
def extract(self, structure: BookStructure, profile: Optional[BookProfile] = None) -> List[ManifestEntry]:
@@ -42,7 +43,10 @@ class FineGrainedExtractor:
manifest_entries = []
# Iterate over spine to maintain order
for item_id in structure.spine:
item_ids = list(structure.spine)
if self.complete_book:
item_ids.extend(item_id for item_id in structure.resources if item_id not in item_ids)
for item_id in item_ids:
if item_id not in structure.resources:
logger.warning(f"Item ID {item_id} in spine but not in resources.")
continue
@@ -97,8 +101,9 @@ class FineGrainedExtractor:
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'
if any(re.search(p, fname) for p in self.SKIP_TRANSLATION_PATTERNS):
return 'core' if self.complete_book else 'skip'
return 'core'
def _should_translate(self, doc_type: str, is_decorative: bool, text: str = "") -> bool:
src/translation/glossary.py
+36 -13
View File
@@ -2,7 +2,11 @@
import hashlib
import json
import re
import copy
from pathlib import Path
from src.common.utils import setup_logger
logger = setup_logger('glossary')
def source_digest(entries):
@@ -101,12 +105,16 @@ async def prepare_glossary(entries, llm, path, override=None, auto=True, chunk_c
"""Scan the whole book before translation; never race glossary updates."""
path = Path(path)
digest = source_digest(entries)
terms, conflicts, start = [], [], 0
if override:
return BookGlossary.load(override)
if path.exists():
data = json.loads(path.read_text())
if data.get('source_digest') == digest:
return BookGlossary(data['terms'])
if data.get('scan_complete', True):
return BookGlossary(data['terms'])
if data.get('chunk_chars') == chunk_chars and data.get('model') == getattr(llm, 'model', ''):
terms, conflicts, start = data['terms'], data.get('conflicts', []), data.get('next_chunk', 0)
if not auto:
return BookGlossary()
chunks, current, size = [], [], 0
@@ -116,8 +124,9 @@ async def prepare_glossary(entries, llm, path, override=None, auto=True, chunk_c
current.append(entry.original_text); size += len(entry.original_text)
if current:
chunks.append('\n'.join(current))
terms, conflicts = [], []
for chunk in chunks:
path.parent.mkdir(parents=True, exist_ok=True)
for index in range(start, len(chunks)):
chunk = chunks[index]
prompt = ('Extract recurring named people, organizations and specialist terminology. '
'Resolve transliterations across Japanese/English/Chinese. Return only JSON '
'{"terms":[{"source":"English name","target":"Chinese canonical name",'
@@ -127,21 +136,35 @@ async def prepare_glossary(entries, llm, path, override=None, auto=True, chunk_c
json.dumps(terms, ensure_ascii=False) + '\nExcerpt:\n' + chunk)
response = await llm.raw_chat_completion('You are a book terminology editor.', prompt)
response = re.sub(r'^```(?:json)?\s*|\s*```$', '', response.strip())
proposals = BookGlossary(json.loads(response)['terms'])
for term in proposals.terms:
# Validate proposals individually: a contradictory model suggestion must
# not discard the already frozen terms or stop the entire book scan.
for proposal in json.loads(response)['terms']:
try:
term = BookGlossary([proposal]).terms[0]
except (ValueError, AttributeError) as exc:
conflicts.append({'proposal': proposal, 'reason': str(exc)})
continue
aliases = {a.casefold() for a in [term['source']] + term.get('aliases', [])}
existing = next((t for t in terms if aliases & {a.casefold() for a in [t['source']] + t.get('aliases', [])}), None)
if existing and existing['target'] != term['target']:
conflicts.append({'existing': existing, 'proposal': term})
continue
candidate = copy.deepcopy(terms)
if existing:
existing['aliases'] = sorted(set(existing.get('aliases', []) + [term['source']] + term.get('aliases', [])) - {existing['source']})
candidate[terms.index(existing)]['aliases'] = sorted(set(existing.get('aliases', []) + [term['source']] + term.get('aliases', [])) - {existing['source']})
else:
terms.append(term)
BookGlossary(terms) # Fail on alias collisions before saving.
path.parent.mkdir(parents=True, exist_ok=True)
temp = path.with_suffix('.tmp')
temp.write_text(json.dumps({'source_digest': digest, 'terms': terms, 'conflicts': conflicts},
ensure_ascii=False, indent=2), encoding='utf-8')
temp.replace(path)
candidate.append(term)
try:
BookGlossary(candidate)
except ValueError as exc:
conflicts.append({'proposal': term, 'reason': str(exc)})
continue
terms = candidate
temp = path.with_suffix('.tmp')
temp.write_text(json.dumps({'source_digest': digest, 'terms': terms, 'conflicts': conflicts,
'scan_complete': index + 1 == len(chunks), 'next_chunk': index + 1,
'chunk_chars': chunk_chars, 'model': getattr(llm, 'model', '')},
ensure_ascii=False, indent=2), encoding='utf-8')
temp.replace(path)
logger.info('Terminology scan %s/%s: %s terms, %s conflicts recorded', index + 1, len(chunks), len(terms), len(conflicts))
return BookGlossary(terms)