API Reference¶

def configure_search(self) -> Search:
    """Configures the search object.

    Returns:
        Search: A configured search object.
    """
    search = self.search()
    search = search.source(self.search_source)
    search = search.params(scroll=self.scan_scroll, size=self.pool_size)
    return search

def create_client(self) -> Elasticsearch:
    """Create an Elasticsearch client instance.

    Returns:
        Configured Elasticsearch client.
    """
    return Elasticsearch(self.url)

def create_translated_hit(self, hit: ObjectBase) -> TranslatedHit:
    """Create a TranslatedHit wrapper for a document hit.

    Args:
        hit: Document hit object.

    Returns:
        TranslatedHit instance ready for translation.
    """
    return TranslatedHit(hit, self.source_field, self.target_field, self.force)

def create_translation_queue(self) -> JoinableQueue:
    """Creates a queue that can translate documents in parallel.

    Returns:
        JoinableQueue: A queue for parallel document translation.
    """
    return JoinableQueue(self.pool_size)

def find_document(self, params: dict[str, str]) -> Document:
    """Find a document by ID and routing.

    Args:
        params: Dictionary containing 'index', 'id', and optionally 'routing'.

    Returns:
        The found Document object.
    """
    using = self.create_client()
    routing = getattr(params, 'routing', params['id'])
    return Document.get(index=params['index'], id=params['id'], routing=routing, using=using)

def init_interpreter(self) -> Any:
    """Initializes the interpreter.

    Returns:
        Any: The initialized interpreter.
    """
    pack_dir = path.join(self.data_dir, 'packs', self.interpreter_name)
    interpreters = (
        Apertium,
        Argos,
    )
    Interpreter = next(i for i in interpreters if i.name.lower() == self.interpreter_name.lower())
    # Pass device option only to Argos (Apertium doesn't support GPU)
    if Interpreter == Argos:
        return Interpreter(
            self.source_language, self.target_language, self.intermediary_language, pack_dir, self.device
        )
    return Interpreter(self.source_language, self.target_language, self.intermediary_language, pack_dir)

def instantiate_interpreter(self) -> Any:
    """Instantiates the interpreter.

    Returns:
        Any: An instance of the interpreter.
    """
    if not hasattr(self, 'interpreter'):
        with self.print_done(f'Instantiating {self.interpreter_name} interpreter'):
            self.interpreter = self.init_interpreter()
    return self.interpreter

@contextmanager
def print_done(self, string: str) -> Generator[None, None, None]:
    """Print progress message and yield, showing done/error status.

    Args:
        string: The status message to be printed.

    Returns:
        Generator for wrapping operations with status output.
    """
    logger.info(string)
    if self.stdout_loglevel > 20:
        string = f'\r{string}...'
        self.print_flush(string)
        try:
            yield
            print(f'{string} \033[92mdone\033[0m')
        except (FatalTranslationException, ElasticsearchException, Full) as error:
            logger.error(error, exc_info=True)
            print(f'{string} \033[91merror\033[0m')
            sys.exit(1)
    else:
        yield

def print_flush(self, string: str) -> None:
    """Print and flush a string to stdout.

    Args:
        string: The string to be printed.
    """
    sys.stdout.write(f'\r{string}')
    sys.stdout.flush()

def process_document(
    self, translation_queue: JoinableQueue, hit: Any, progress: Progress, task: Any, shared_fatal_error: Manager
) -> None:
    """Processes a document.

    Args:
        translation_queue (JoinableQueue): A queue for parallel document translation.
        hit (Any): The document to be translated.
        index (int): The index of the document.
        progress (Progress): A progress object.
        task (Any): The current task.
        shared_fatal_error (Manager): A shared manager for fatal errors.
    """
    translation_queue.put((self, hit), True, self.pool_timeout)
    progress.advance(task)
    if shared_fatal_error.value:
        raise FatalTranslationException(shared_fatal_error.value)

def search(self) -> Search:
    """Executes a search query.

    Returns:
        Search: The search result.
    """
    using = self.create_client()
    search = Search(index=self.index, using=using)
    if self.query_string:
        search = search.query('query_string', query=self.query_string)
    return search

def start(self) -> None:
    """Starts or plans the translation process."""
    if self.plan:
        self.start_later()
    else:
        self.start_now()

def start_later(self) -> None:
    """Queue translation tasks for later execution via Celery."""
    self.instantiate_interpreter()
    total = self.search().count()
    plural = 's' if total != 1 else ''
    desc = f'Planning translation for {total} document{plural}'
    with self.print_done(desc):
        search = self.configure_search()
        for hit in search.scan():
            logger.info(f'Planned translation for doc {hit.meta.id}')
            translate_document_task.delay(self.options, hit.meta.to_dict())

def start_now(self) -> None:
    """Start the translation process immediately."""
    self.instantiate_interpreter()
    total = self.search().count()
    desc = f'Translating {total} document(s)'
    with self.print_done(desc):
        search = self.configure_search()
        if self.pool_size == 1:
            # Direct translation without multiprocessing (better for GPU)
            self.translate_documents_direct(search, total)
        else:
            translation_queue = self.create_translation_queue()
            with self.with_shared_fatal_error() as shared_fatal_error:
                self.translate_documents_in_pool(search, translation_queue, shared_fatal_error, total)

def translate_document(self, hit: ObjectBase) -> None:
    """Translate a single document.

    Args:
        hit: Document hit object to translate.
    """
    self.instantiate_interpreter()
    # Translate the document
    logger.info(f'Translating doc {hit.meta.id}')
    translated_hit = self.create_translated_hit(hit)
    translated_hit.add_translation(self.interpreter, max_content_length=self.max_content_length)
    logger.info(f'Translated doc {hit.meta.id}')
    # Save the translated document if not in dry run mode
    if not self.dry_run:
        translated_hit.save(self.create_client())
        logger.info(f'Saved translation for doc {hit.meta.id}')

def translate_documents_direct(self, search: Search, total: int) -> None:
    """Translate documents directly without multiprocessing.

    Used when pool_size=1 for simpler execution and better GPU compatibility.

    Args:
        search: A search object.
        total: The total number of documents.
    """
    from time import sleep

    with Progress(disable=self.no_progressbar, transient=True) as progress:
        plural = 's' if total != 1 else ''
        task = progress.add_task(f'Translating {total} document{plural}', total=total)
        for hit in search.scan():
            try:
                self.translate_document(hit)
                sleep(self.throttle / 1000)
            except ElasticsearchException as error:
                logger.error(f'An error occurred when saving doc {hit.meta.id}')
                logger.error(error)
                raise FatalTranslationException(error)
            except Exception as error:
                logger.warning(f'Unable to translate doc {hit.meta.id}')
                logger.warning(error)
            finally:
                progress.advance(task)

def translate_documents_in_pool(
    self, search: Search, translation_queue: JoinableQueue, shared_fatal_error: Manager, total: int
) -> None:
    """Translates documents using multiprocessing pool.

    Args:
        search (Search): A search object.
        translation_queue (JoinableQueue): A queue for parallel document translation.
        shared_fatal_error (Manager): A shared manager for fatal errors.
        total (int): The total number of documents.
    """
    with (
        Pool(self.pool_size, translation_worker, (translation_queue, shared_fatal_error)),
        Progress(disable=self.no_progressbar, transient=True) as progress,
    ):
        plural = 's' if total != 1 else ''
        task = progress.add_task(f'Translating {total} document{plural}', total=total)
        for hit in search.scan():
            self.process_document(translation_queue, hit, progress, task, shared_fatal_error)
        translation_queue.join()

@contextmanager
def with_shared_fatal_error(self) -> Generator[Any, None, None]:
    """Creates a context manager for managing shared fatal errors.

    Returns:
        Generator yielding a shared manager value.
    """
    with Manager() as manager:
        yield manager.Value('b', None)

def download_and_install_package(self, package: Any) -> Optional[Any]:
    """Download and install a language package.

    Uses file locking to prevent concurrent downloads of the same package.
    Skips installation if the package is already installed.

    Args:
        package: The package to download and install.

    Returns:
        Installation result or None if package was already installed.

    Raises:
        ArgosPackageDownloadLockTimeout: If lock cannot be acquired within timeout.
    """
    argospackage = _get_argos_package()
    try:
        temp_dir = Path(tempfile.gettempdir())
        lock_path = temp_dir / f'{package.from_code}_{package.to_code}.lock'

        with FileLock(lock_path, timeout=600).acquire(timeout=600):
            if self.is_package_installed(package):
                return None
            download_path = package.download()
            logger.info(f'Installing Argos package {package}')
            return argospackage.install_from_path(download_path)
    except Timeout as exc:
        raise ArgosPackageDownloadLockTimeout(
            f'Another instance of the program is downloading the package {package}. Please try again later.'
        ) from exc

def download_necessary_languages(self) -> None:
    """Download necessary language packages if not installed.

    Steps:
    1. Updates the package index.
    2. Finds the necessary package.
    3. Downloads and installs the package.

    Raises:
        ArgosPairNotAvailable: If the necessary language package could not be found.
        ArgosPackageDownloadLockTimeout: If lock cannot be acquired within timeout.
    """
    self.update_package_index()
    necessary_package = self.find_necessary_package()
    self.download_and_install_package(necessary_package)

def find_necessary_package(self) -> Any:
    """Find the necessary language package.

    Searches available packages for one matching the source and target languages.

    Returns:
        The necessary language package object.

    Raises:
        ArgosPairNotAvailable: If the necessary language package could not be found.
    """
    argospackage = _get_argos_package()
    for package in argospackage.get_available_packages():
        if package.from_code == self.source_alpha_2 and package.to_code == self.target_alpha_2:
            return package
    raise ArgosPairNotAvailable

def is_package_installed(self, package: Any) -> bool:
    """Check if a package is installed.

    Args:
        package: The package to check.

    Returns:
        True if the package is installed, False otherwise.
    """
    argospackage = _get_argos_package()
    return package in argospackage.get_installed_packages()

def translate(self, text_input: str) -> str:
    """Translate input text from source language to target language.

    Args:
        text_input: The input text in the source language.

    Returns:
        The translated text in the target language.
    """
    # Always configure device before translation (needed for multiprocessing workers)
    self._ensure_device_configured()
    return self.translation.translate(text_input)

def update_package_index(self) -> None:
    """Update the Argos package index to fetch latest available packages."""
    argospackage = _get_argos_package()
    argospackage.update_package_index()

def download_intermediary_pairs(self) -> None:
    """Download both intermediary language pairs for indirect translation."""
    for pair in self.intermediary_pairs:
        self.download_pair(pair)

def download_necessary_pairs(self) -> None:
    """Download required language pair packages.

    Downloads either a direct pair or intermediary pairs depending on
    availability in the repository.
    """
    logger.info(f'Downloading necessary package(s) for {self.pair}')
    if self.any_pair_variant_in_packages:
        self.download_pair()
    else:
        self.download_intermediary_pairs()

def download_pair(self, pair: Optional[str] = None) -> str:
    """Download and install a specific language pair package.

    Args:
        pair: Language pair to download. If None, uses current pair.

    Returns:
        Path to the installed package directory.
    """
    pair = self.pair_alpha_3 if pair is None else to_alpha_3_pair(pair)
    # All commands must be run from the pack dir
    return self.repository.install_pair_package(pair)

def first_pairs_path(self, leaf: dict, lang: str) -> list[str]:
    """Find the first path from a tree leaf to a target language.

    Args:
        leaf: Tree node dictionary with 'lang' and 'children' keys.
        lang: Target language to find path to.

    Returns:
        List of language codes forming the path.
    """
    path = []
    for child_leaf in leaf['children'].values():
        if self.leaf_has_lang(child_leaf, lang):
            path.append(child_leaf['lang'])
            path = path + self.first_pairs_path(child_leaf, lang)
            break
    return path

def lang_tree(self, lang: str, pairs: list[list[str]], depth: int = 2) -> dict:
    """Build a tree of language connections from available pairs.

    Args:
        lang: Root language for the tree.
        pairs: List of language pair lists.
        depth: Maximum depth to traverse (default: 2).

    Returns:
        Dictionary tree structure with 'lang' and 'children' keys.
    """
    tree = {'lang': lang, 'children': {}}
    for pair in pairs:
        if lang in pair and depth > 0:
            child_lang = next(item for item in pair if item != lang)
            tree['children'][child_lang] = self.lang_tree(child_lang, pairs, depth - 1)
    return tree

def leaf_has_lang(self, leaf: dict, lang: str) -> bool:
    """Check if a tree leaf contains or leads to a target language.

    Args:
        leaf: Tree node dictionary with 'lang' and 'children' keys.
        lang: Target language to search for.

    Returns:
        True if the language is found in the leaf or its descendants.
    """
    children = leaf['children'].values()
    return lang in leaf['children'] or any(self.leaf_has_lang(child_leaf, lang) for child_leaf in children)

def pair_to_pair_package(self, pair: str) -> Optional[str]:
    """Convert language pair to package name.

    Checks both the pair and its reverse for availability in remote packages.

    Args:
        pair: Language pair string (e.g., 'en-es').

    Returns:
        Package name if found, None otherwise.
    """
    pair_inversed = '-'.join(pair.split('-')[::-1])
    combinations = [to_alpha_3_pair(pair), to_alpha_3_pair(pair_inversed)]
    try:
        return next(p for p in self.remote_pairs if p in combinations)
    except StopIteration:
        return None

def translate(self, input: str) -> str:
    """Translate text through the translation pipeline.

    If using an intermediary language, translates through multiple pairs.

    Args:
        input: Text to translate.

    Returns:
        Translated text string.
    """
    for pair in self.pairs_pipeline:
        # Create a sub-process which can receive an input
        input = self.translate_with_apertium(input, pair)
    return input

def translate_with_apertium(self, input: str, pair: str) -> str:
    """Translate text using Apertium for a specific language pair.

    Args:
        input: Text to translate.
        pair: Language pair code (e.g., 'eng-spa').

    Returns:
        Translated text string.

    Raises:
        Exception: If translation fails.
    """
    apertium = _get_apertium()
    try:
        # Works with a temporary file as buffer (opened in text mode)
        with NamedTemporaryFile(mode='w+t') as temp_input_file:
            temp_input_file.writelines(input)
            temp_input_file.seek(0)
            input_translated = apertium('-ud', self.pack_dir, pair, temp_input_file.name)
    except ErrorReturnCode:
        raise Exception('Unable to translate this string.')
    return str(input_translated)

def clear_modes(self) -> None:
    """Remove all mode files from the cache directory."""
    with pushd(self.cache_dir):
        rm('-Rf', 'modes')

def create_pair_package_alias(self, package_dir: str) -> str:
    """Create symbolic links for alternative language code formats.

    Creates aliases between ISO 639-1 (2-letter) and ISO 639-3 (3-letter) codes.

    Args:
        package_dir: Directory containing the extracted package.

    Returns:
        Path to the created alias directory.
    """
    extraction_dir = dirname(package_dir) + '/'
    source, target = basename(package_dir).split('apertium-')[-1].split('-')

    # Determine alias codes based on current format
    if len(source) == 2:
        aliases = (to_alpha_3(source), to_alpha_3(target))
    else:
        aliases = (to_alpha_2(source), to_alpha_2(target))

    # Build the alias dir using the alias codes
    alias_dir = join(extraction_dir, f'apertium-{aliases[0]}-{aliases[1]}')
    mode_file = join(extraction_dir, 'modes', f'{source}-{target}.mode')
    mode_alias_file = join(extraction_dir, 'modes', f'{aliases[0]}-{aliases[1]}.mode')

    # Use symbolic links for aliases
    create_symlink(package_dir, alias_dir)
    create_symlink(mode_file, mode_alias_file)

    return alias_dir

def download_package(self, name: str, force: bool = False) -> str:
    """Download a package from the repository.

    Args:
        name: Package name to download.
        force: If True, re-download even if package already exists.

    Returns:
        Path to the downloaded .deb file.

    Raises:
        PackageNotFoundError: If package cannot be found.
    """
    package = self.find_package(name)
    if package is None:
        raise PackageNotFoundError(name)

    package_dir = join(self.cache_dir, name)
    package_file = join(package_dir, 'package.deb')
    mkdir('-p', package_dir)

    # Don't download the file twice
    if force or not isfile(package_file):
        logger.info(f'Downloading package {name}')

        # Try the URL from Packages file first
        package_url = f'{REPOSITORY_URL}/{package["Filename"]}'
        try:
            request.urlretrieve(package_url, package_file)
        except (URLError, HTTPError, OSError) as e:
            # If that fails, try to find the latest version in the pool directory
            logger.warning(f'Failed to download from Packages file URL: {e}')
            package_url = self.find_latest_package_in_pool(name, package['Filename'])
            request.urlretrieve(package_url, package_file)

    return package_file

def download_pair_package(self, pair: str) -> str:
    """Download a translation pair package.

    Args:
        pair: Language pair in format 'source-target'.

    Returns:
        Path to the downloaded .deb file.

    Raises:
        PairPackageNotFoundError: If no pair package is available for the given languages.
    """
    pair_package = self.find_pair_package(pair)
    if pair_package is not None:
        return self.download_package(pair_package.get('Package'))
    else:
        raise PairPackageNotFoundError(pair)

def extract_pair_package(self, file: str, extraction_dir: str = '.') -> str:
    """Extract a translation pair .deb package.

    Args:
        file: Path to the .deb file to extract.
        extraction_dir: Directory to extract files into. Defaults to '.'.

    Returns:
        Path to the working directory containing extracted files.
    """
    workdir = dirname(file)
    with pushd(workdir):
        # Extract the file from the .deb
        dpkg_deb('-x', file, extraction_dir)
        # Copy the files we need
        cp('-rlf', glob('usr/share/apertium/*'), extraction_dir)
        # Remove everything else
        rm('-Rf', 'usr')
        # Rewrite paths in modes files to point to the working directory
        for mode in glob('modes/*.mode'):
            self.replace_in_file(mode, '/usr/share/apertium', workdir)
    return workdir

def find_latest_package_in_pool(self, package_name: str, filename: str) -> str:
    """Find latest package version from pool directory.

    Used as fallback when the Packages file lists an outdated version.

    Args:
        package_name: Name of the package to find.
        filename: Original filename from Packages file (used to determine pool directory).

    Returns:
        URL of the latest package version found.

    Raises:
        Exception: If no matching package is found in the pool directory.
        URLError: If the pool directory cannot be accessed.
    """
    logger.info('Attempting to find latest version from pool directory')

    # Extract the pool directory path
    filename_parts = filename.split('/')
    pool_dir_url = f'{REPOSITORY_URL}/{"/".join(filename_parts[:-1])}/'

    try:
        # Fetch the directory listing
        response = request.urlopen(pool_dir_url)
        html_content = response.read().decode('utf-8')
    except (URLError, HTTPError) as e:
        logger.error(f'Failed to access pool directory {pool_dir_url}: {e}')
        raise

    # Find all .deb files for this package
    pattern = rf'href="({re.escape(package_name)}_[^"]+\.deb)"'
    matches = re.findall(pattern, html_content)

    if matches:
        # Use the last one (likely the newest based on alphabetical sorting)
        latest_file = matches[-1]
        package_url = pool_dir_url + latest_file
        logger.info(f'Found latest version: {latest_file}')
        return package_url
    else:
        raise PackageNotFoundError(package_name)

def find_package(self, package: str) -> Optional[dict]:
    """Find a package by name or provided name.

    Args:
        package: Package name to search for.

    Returns:
        Package metadata dictionary if found, None otherwise.
    """

    def is_package(c: dict) -> bool:
        return c.get('Package') == package or c.get('Provides') == package

    try:
        return next(filter(is_package, self.packages))
    except StopIteration:
        logger.warning(f'Unable to find package {package}')
        return None

def find_pair_package(self, pair: str) -> Optional[dict]:
    """Find a translation pair package.

    Searches for both forward (source-target) and reverse (target-source) pairs.

    Args:
        pair: Language pair in format 'source-target'.

    Returns:
        Package metadata dictionary if found, None otherwise.
    """
    pair = to_alpha_3_pair(pair)
    pair_inversed = '-'.join(pair.split('-')[::-1])

    def is_pair(c: dict) -> bool:
        package_name = c.get('Package', '')
        return package_name.endswith(pair) or package_name.endswith(pair_inversed)

    try:
        return next(filter(is_pair, self.pair_packages))
    except StopIteration:
        return None