diff --git a/translate_content.py b/translate_content.py new file mode 100644 index 0000000..5a98ca1 --- /dev/null +++ b/translate_content.py @@ -0,0 +1,483 @@ +#!/usr/bin/env python3 +# Main script to translate .rst files into multiple languages, converting them to both .rst and .md formats. +# Supports progress tracking with tqdm and a debug mode for streaming API responses. + +import os # For file and directory operations +import hashlib # For generating content hashes +import json # For reading/writing hash files +from openai import OpenAI, OpenAIError # OpenAI API client and error type +import pandoc # For converting .rst to .md +import re # For regex pattern matching +import time # For retry delays +from tqdm import tqdm # For progress bars +import sys # For stderr output +import tiktoken # For token counting +import argparse # For command-line argument parsing +from typing import Dict, Tuple, List, Optional # Type hints for better code clarity + +# Parse command-line arguments to enable optional debug mode +parser = argparse.ArgumentParser( + description="Translate .rst files with progress and debugging." +) +parser.add_argument( + "--debug", action="store_true", help="Enable debug mode with streaming output." +) +args = parser.parse_args() + +# AI client setup using environment variables for flexibility and security +# Defaults are provided for local testing, but should be overridden in production +base_url = os.environ.get("OPENAI_BASE_URL", "https://hermes.ai.unturf.com/v1") +api_key = os.environ.get("OPENAI_API_KEY", "choose-any-value") +client = OpenAI(base_url=base_url, api_key=api_key) +MODEL = os.environ.get("OPENAI_MODEL", "adamo1139/Hermes-3-Llama-3.1-8B-FP8-Dynamic") + +# Tokenizer setup for counting tokens in .rst content (used once per file) +TOKENIZER = tiktoken.get_encoding("cl100k_base") + +# Dictionary of top languages to translate into, mapping codes to full names +# Excludes languages like Tamil and Kuku Yalanji due to known looping issues +TOP_20_LANGS = { + "en": "English", + "zh": "Chinese (Simplified)", + "hi": "Hindi", + "es": "Spanish", + "fr": "French", + "ar": "Arabic", + "bn": "Bengali", + "ru": "Russian", + "pt": "Portuguese", + "ur": "Urdu", + "id": "Indonesian", + "de": "German", + "ja": "Japanese", + "sw": "Swahili", + "mr": "Marathi", + "te": "Telugu", + "tr": "Turkish", + "zh-tw": "Chinese (Traditional)", + "ko": "Korean", +} + +# Directory and file constants +CONTENT_DIR = "content" # Root directory for .rst files +HASH_FILE = "content/translation_hashes.json" # File to store content hashes + + +def replace_special_content(content: str) -> Tuple[str, Dict[str, str]]: + """Replace special RST content (code blocks, URIs, images) with unique placeholders. + + Args: + content: The raw .rst content to process. + + Returns: + A tuple of (replaced content, dictionary mapping placeholders to original text). + """ + # Regex patterns to identify special content in RST + patterns = [ + # Code directives (single-line and multi-line) + (re.compile(r"(\.\.\s+code::\s+[^\n]*(?:\n\s+[^\n]+)*)"), "__CODE_{}__"), + (re.compile(r"(\.\.\s+code-block::\s+[^\n]*(?:\n\s+[^\n]+)*)"), "__CODE_{}__"), + (re.compile(r"(\.\.\s+highlight::\s+[^\n]*(?:\n\s+[^\n]+)*)"), "__CODE_{}__"), + # Raw content directive + (re.compile(r"(\.\.\s+raw::\s+[^\n]*(?:\n\s+[^\n]+)*)"), "__RAW_{}__"), + # Inline literals (double and single backticks) + (re.compile(r"``([^`]+?)``"), "__CODE_{}__"), + (re.compile(r"`([^`]+?)`(?![`_])"), "__CODE_{}__"), + # URI patterns (links with optional inline code) + (re.compile(r"`([^`]+?)\s+<(https?://[^\s>]+)>`_"), "__URI_{}__"), + (re.compile(r"(`[^`]+?`_\s+)?<(https?://[^\s>]+)>"), "__URI_{}__"), + (re.compile(r"(\.\.\s+_.*?:(?:\s+https?://[^\s]+)?)"), "__URI_{}__"), + # Image directive with optional attributes + (re.compile(r"(\.\.\s+image::\s+[^\s]+(?:\n\s+:.*?)*)"), "__IMG_{}__"), + ] + + placeholders: Dict[str, str] = {} # Store original content for restoration + counter = {"code": 0, "uri": 0, "img": 0, "raw": 0} # Track placeholder indices + + def replacer(match, placeholder_template: str, key: str): + """Helper to replace matched content with a placeholder.""" + nonlocal counter + full_match = match.group(0) + placeholder = placeholder_template.format(counter[key]) + placeholders[placeholder] = full_match + counter[key] += 1 + return placeholder + + replaced_content = content + for pattern, template in patterns: + # Apply replacements based on placeholder type + if "CODE" in template: + current_template = template # Create a local variable that will be properly captured + replaced_content = pattern.sub( + lambda m: replacer(m, current_template, "code"), replaced_content + ) + elif "URI" in template: + current_template = template + replaced_content = pattern.sub( + lambda m: replacer(m, current_template, "uri"), replaced_content + ) + elif "IMG" in template: + current_template = template + replaced_content = pattern.sub( + lambda m: replacer(m, current_template, "img"), replaced_content + ) + elif "RAW" in template: + replaced_content = pattern.sub( + lambda m: replacer(m, template, "raw"), replaced_content + ) + + return replaced_content, placeholders + + +def restore_special_content(content: str, placeholders: Dict[str, str]) -> str: + """Restore original special content from placeholders. + + Args: + content: The content with placeholders. + placeholders: Dictionary mapping placeholders to original text. + + Returns: + The content with placeholders replaced by original text. + """ + for placeholder, original in placeholders.items(): + content = content.replace(placeholder, original) + return content + + +def _handle_translation_error( + e: Exception, + attempt: int, + retries: int, + delay: float, + target_lang_full: str, + debug: bool = False, +) -> bool: + """Handle translation errors with retries and optional debug output. + + Args: + e: The caught exception. + attempt: Current attempt number (0-based). + retries: Total number of retries. + delay: Delay between retries in seconds. + target_lang_full: The target language name. + debug: Whether to print debug messages. + + Returns: + True if retrying should continue, False if retries are exhausted. + + Raises: + Exception: If retries are exhausted, with chained context. + """ + if debug: + error_type = "OpenAI API" if isinstance(e, OpenAIError) else "Unexpected" + print( + f"\nDEBUG: {error_type} error on attempt {attempt + 1}/{retries}: {str(e)}", + file=sys.stderr, + ) + + if attempt < retries - 1: + if debug: + print(f"DEBUG: Retrying in {delay} seconds...", file=sys.stderr) + time.sleep(delay) + return False + else: + error_type = ( + "Failed to translate to" if isinstance(e, OpenAIError) else "Translation to" + ) + raise Exception( + f"{error_type} {target_lang_full} after {retries} attempts: {str(e)}" + ) from e + + +def translate_text( + text: str, + target_lang_full: str, + retries: int = 3, + delay: float = 5, + timeout: float = 100, +) -> str: + """Translate RST content into the target language using the OpenAI API. + + Args: + text: The RST content to translate. + target_lang_full: The full name of the target language. + retries: Number of retry attempts for API calls. + delay: Delay between retries in seconds. + timeout: Timeout for blocking API calls in seconds. + + Returns: + The translated RST content. + + Raises: + Exception: If translation fails after all retries. + """ + # Construct the prompt for the OpenAI API + prompt = ( + f"Translate the following reStructuredText (.rst) content into {target_lang_full}. " + "The input is in English and formatted as valid reStructuredText, which includes metadata " + "like titles (underlined with # or =), directives, and body text. " + "Ensure the output remains valid reStructuredText, preserving the structure, syntax, and formatting " + "of the original, including proper handling of titles, directives, and text. " + "Do not translate or modify placeholders like __CODE_0__, __URI_0__, __IMG_0__, etc., as they represent " + "code blocks, URLs, or image references that should remain unchanged. Only translate the surrounding text. " + "Here is the text to translate:\n\n" + f"{text}" + ) + messages = [{"role": "user", "content": prompt}] + + if args.debug: + # Debug mode: Stream the API response for real-time debugging + for attempt in range(retries): + try: + print( + f"\nDEBUG: Streaming attempt {attempt + 1}/{retries} for {target_lang_full}", + file=sys.stderr, + ) + print( + f"DEBUG: Translation output in {target_lang_full}: ", + end="", + file=sys.stderr, + ) + stream = client.chat.completions.create( + model=MODEL, + messages=messages, + temperature=0.5, + max_tokens=16000, # Lower limit for streaming to avoid overload + stream=True, + ) + result = "" + for chunk in stream: + if chunk.choices[0].delta.content is not None: + content = chunk.choices[0].delta.content + result += content + print(content, end="", file=sys.stderr) + sys.stderr.flush() # Ensure immediate output + print( + f"\nDEBUG: Streaming completed for {target_lang_full}", + file=sys.stderr, + ) + return result + except (OpenAIError, Exception) as e: + if not _handle_translation_error( + e, attempt, retries, delay, target_lang_full, debug=True + ): + continue + else: + # Normal mode: Blocking API call with timeout + for attempt in range(retries): + try: + response = client.chat.completions.create( + model=MODEL, + messages=messages, + temperature=0.5, + max_tokens=60000, # Higher limit for full responses + timeout=timeout, + ) + return response.choices[0].message.content + except (OpenAIError, Exception) as e: + if not _handle_translation_error( + e, attempt, retries, delay, target_lang_full + ): + continue + + +def load_hashes() -> Dict[str, Dict[str, str]]: + """Load translation hashes from the hash file. + + Returns: + A dictionary of hashes, empty if the file doesn’t exist. + """ + if os.path.exists(HASH_FILE): + with open(HASH_FILE, "r", encoding="utf-8") as f: + return json.load(f) + return {} + + +def save_hashes(hashes: Dict[str, Dict[str, str]]) -> None: + """Save translation hashes to the hash file. + + Args: + hashes: The dictionary of hashes to save. + """ + with open(HASH_FILE, "w", encoding="utf-8") as f: + json.dump(hashes, f, indent=2) + + +def get_content_hash(content: str) -> str: + """Generate a SHA-256 hash of the content. + + Args: + content: The content to hash. + + Returns: + The hexadecimal hash string. + """ + return hashlib.sha256(content.encode("utf-8")).hexdigest() + + +def process_file(file_path: str, lang_pbar) -> None: + """Process a single .rst file, translating it into all target languages. + + Args: + file_path: Path to the .rst file. + lang_pbar: tqdm progress bar for language processing. + """ + # Read the .rst file and split into header and content + with open(file_path, "r", encoding="utf-8") as f: + lines = f.readlines() + header_end = next( + (i for i, line in enumerate(lines) if not line.strip() or i > 4), len(lines) + ) + header = lines[:header_end] + content = "".join(lines[header_end:]) + + # Count tokens once for the entire content + total_tokens = len(TOKENIZER.encode(content)) + base_name = os.path.splitext(os.path.basename(file_path))[0] + slug = base_name + source_hash = get_content_hash(content) + + # Load existing hashes to check for updates + hashes = load_hashes() + file_hashes = hashes.get(file_path, {}) + needs_update = file_hashes.get("source") != source_hash + + # Replace special content with placeholders + replaced_content, placeholders = replace_special_content(content) + + # Process each language + for lang_code, lang_full in TOP_20_LANGS.items(): + lang_pbar.set_description( + f"File: {slug}.rst | Lang: {lang_full} ({lang_code}) | Total Tokens: {total_tokens}" + ) + lang_dir = f"{CONTENT_DIR}/{lang_code}" + os.makedirs(lang_dir, exist_ok=True) + + rst_file = f"{lang_dir}/{slug}.rst" + md_file = f"{lang_dir}/{slug}.md" + + # Update header with language code + new_header = [] + lang_set = False + for line in header: + if line.startswith(":lang:"): + new_header.append(f":lang: {lang_code}\n") + lang_set = True + else: + new_header.append(line) + if not lang_set: + new_header.append(f":lang: {lang_code}\n") + + # Check if translation already exists and is up-to-date + lang_hashes = file_hashes.get(lang_code, {}) + rst_hash = lang_hashes.get("rst") + md_hash = lang_hashes.get("md") + + rst_exists = os.path.exists(rst_file) + md_exists = os.path.exists(md_file) + if rst_exists and md_exists and not needs_update: + with open(rst_file, "r", encoding="utf-8") as f: + current_rst_hash = get_content_hash(f.read()) + with open(md_file, "r", encoding="utf-8") as f: + current_md_hash = get_content_hash(f.read()) + if current_rst_hash == rst_hash and current_md_hash == md_hash: + lang_pbar.update(1) + continue + + # Translate content (English is copied directly) + if lang_code == "en": + translated_replaced_content = replaced_content + else: + try: + translated_replaced_content = translate_text( + replaced_content, lang_full + ) + except Exception as e: + print( + f"Error translating {slug} to {lang_full}: {str(e)}", + file=sys.stderr, + ) + lang_pbar.update(1) + continue + + # Restore special content + translated_content = restore_special_content( + translated_replaced_content, placeholders + ) + full_rst_content = "".join(new_header) + translated_content + + # Save translated .rst and convert to .md + try: + with open(rst_file, "w", encoding="utf-8") as f: + f.write(full_rst_content) + rst_hash = get_content_hash(full_rst_content) + + replaced_full_rst = "".join(new_header) + translated_replaced_content + md_content = pandoc.read(replaced_full_rst, format="rst") + md_replaced = pandoc.write(md_content, format="markdown") + md_final = restore_special_content(md_replaced, placeholders) + + with open(md_file, "w", encoding="utf-8") as f: + f.write(md_final) + md_hash = get_content_hash(md_final) + except Exception as e: + print( + f"Error converting/saving {slug} to {lang_full}: {str(e)}", + file=sys.stderr, + ) + lang_pbar.update(1) + continue + + # Update hashes + if file_path not in hashes: + hashes[file_path] = {} + hashes[file_path]["source"] = source_hash + if lang_code not in hashes[file_path]: + hashes[file_path][lang_code] = {} + hashes[file_path][lang_code]["rst"] = rst_hash + hashes[file_path][lang_code]["md"] = md_hash + + save_hashes(hashes) + lang_pbar.update(1) + + +def main() -> None: + """Main function to process all .rst files in the content directory.""" + # Collect all source .rst files (excluding language subdirectories) + rst_files = [ + os.path.join(root, file) + for root, _, files in os.walk(CONTENT_DIR) + for file in files + if file.endswith(".rst") + and not any(f"/{lang}/" in root for lang in TOP_20_LANGS) + ] + + # Use nested progress bars for files and languages + with tqdm( + total=len(rst_files), desc="Total Files", unit="file", file=sys.stderr + ) as file_pbar, tqdm( + total=len(TOP_20_LANGS), + desc="File: N/A | Lang: N/A | Total Tokens: N/A", + unit="lang", + file=sys.stderr, + ) as lang_pbar: + for file_path in rst_files: + file_pbar.set_description( + f"Total Files (Current: {os.path.basename(file_path)})" + ) + lang_pbar.reset() + process_file(file_path, lang_pbar) + file_pbar.update(1) + + +if __name__ == "__main__": + # Verify Pandoc availability before starting + try: + pandoc.read("test", format="rst") + except Exception: + print( + "Pandoc is not installed or not functioning correctly. Install Pandoc to enable .md conversion.", + file=sys.stderr, + ) + sys.exit(1) + main()