Files
epub_bilingual_translator/archive/v0.12/pipeline/01_preprocess.py
T

70 lines
2.3 KiB
Python

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
from src.common.paths import get_work_dirs
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)
work = get_work_dirs(input_path)
ensure_directory(work["root"])
structure_path = work["structure"]
manifest_path = work["manifest"]
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)