483 lines
18 KiB
Python
483 lines
18 KiB
Python
#!/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()
|