russellballestrini/russell.ballestrini.net
1
0
Fork 0
Code Issues Pull requests 1 Projects Releases Packages Wiki Activity Actions

modified: translate_content.py

Browse source
This commit is contained in:
Russell Ballestrini 2025-05-01 21:46:02 -04:00
parent 17ec4e853c
commit 8e974f30c1
1 changed files with 270 additions and 103 deletions
Download patch file Download diff file Expand all files Collapse all files

373
translate_content.py
View file

@ -1,28 +1,41 @@
#!/usr/bin/env python3
import os
import hashlib
import json
from openai import OpenAI, OpenAIError
import pandoc
import re
import time
from tqdm import tqdm
import sys
import tiktoken
import argparse
# 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.
# Parse command-line arguments
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.")
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
client = OpenAI(base_url="https://hermes.ai.unturf.com/v1", api_key="choose-any-value")
MODEL = "adamo1139/Hermes-3-Llama-3.1-8B-FP8-Dynamic"
# 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 (used only once per file)
# 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)",
@ -41,30 +54,48 @@ TOP_20_LANGS = {
"mr": "Marathi",
"te": "Telugu",
"tr": "Turkish",
# "ta": "Tamil",
"zh-tw": "Chinese (Traditional)",
"ko": "Korean",
}
CONTENT_DIR = "content"
HASH_FILE = "content/translation_hashes.json"
# 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):
"""Replace code blocks, URIs, and images with unique placeholders."""
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_{}__"),
(re.compile(r"(\.\.\s+image::\s+[^\s]+(?:\s*\n\s+:.*?)*)"), "__IMG_{}__"),
# Image directive with optional attributes
(re.compile(r"(\.\.\s+image::\s+[^\s]+(?:\n\s+:.*?)*)"), "__IMG_{}__"),
]
placeholders = {}
counter = {"code": 0, "uri": 0, "img": 0}
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, key):
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])
@ -74,8 +105,7 @@ def replace_special_content(content):
replaced_content = content
for pattern, template in patterns:
for pattern, template in patterns:
if "CODE" in template:
# 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(
@ -91,19 +121,97 @@ def replace_special_content(content):
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, placeholders):
"""Restore original code, URIs, and images from 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 translate_text(text, target_lang_full, retries=3, delay=5, timeout=100):
"""Translate text with optional streaming debug mode."""
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 "
@ -118,16 +226,23 @@ def translate_text(text, target_lang_full, retries=3, delay=5, timeout=100):
messages = [{"role": "user", "content": prompt}]
if args.debug:
# Streaming mode with no newline between chunks
# 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)
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=60000,
max_tokens=16000, # Lower limit for streaming to avoid overload
stream=True,
)
result = ""
@ -135,65 +250,79 @@ def translate_text(text, target_lang_full, retries=3, delay=5, timeout=100):
if chunk.choices[0].delta.content is not None:
content = chunk.choices[0].delta.content
result += content
print(content, end="", file=sys.stderr) # No newline, continuous stream
print(content, end="", file=sys.stderr)
sys.stderr.flush() # Ensure immediate output
print(f"\nDEBUG: Streaming completed for {target_lang_full}", file=sys.stderr)
print(
f"\nDEBUG: Streaming completed for {target_lang_full}",
file=sys.stderr,
)
return result
except OpenAIError as e:
print(f"\nDEBUG: OpenAI API error on attempt {attempt + 1}/{retries}: {str(e)}", file=sys.stderr)
if attempt < retries - 1:
print(f"DEBUG: Retrying in {delay} seconds...", file=sys.stderr)
time.sleep(delay)
else:
raise Exception(f"Failed to translate to {target_lang_full} after {retries} attempts: {str(e)}")
except Exception as e:
print(f"\nDEBUG: Unexpected error on attempt {attempt + 1}/{retries}: {str(e)}", file=sys.stderr)
if attempt < retries - 1:
print(f"DEBUG: Retrying in {delay} seconds...", file=sys.stderr)
time.sleep(delay)
else:
raise Exception(f"Translation to {target_lang_full} failed after {retries} attempts: {str(e)}")
except (OpenAIError, Exception) as e:
if not _handle_translation_error(
e, attempt, retries, delay, target_lang_full, debug=True
):
continue
else:
# Non-debug, blocking mode
# 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,
max_tokens=60000, # Higher limit for full responses
timeout=timeout,
)
return response.choices[0].message.content
except OpenAIError as e:
if attempt < retries - 1:
time.sleep(delay)
else:
raise Exception(f"Failed to translate to {target_lang_full} after {retries} attempts: {str(e)}")
except Exception as e:
if attempt < retries - 1:
time.sleep(delay)
else:
raise Exception(f"Translation to {target_lang_full} failed after {retries} attempts: {str(e)}")
except (OpenAIError, Exception) as e:
if not _handle_translation_error(
e, attempt, retries, delay, target_lang_full
):
continue
def load_hashes():
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") as f:
with open(HASH_FILE, "r", encoding="utf-8") as f:
return json.load(f)
return {}
def save_hashes(hashes):
with open(HASH_FILE, "w") as f:
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):
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, lang_pbar):
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(
@ -202,29 +331,32 @@ def process_file(file_path, lang_pbar):
header = lines[:header_end]
content = "".join(lines[header_end:])
# Count tokens for the entire .rst content once
# 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 code, URIs, and images with placeholders
# 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_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 lang_code
# Update header with language code
new_header = []
lang_set = False
for line in header:
@ -236,12 +368,11 @@ def process_file(file_path, lang_pbar):
if not lang_set:
new_header.append(f":lang: {lang_code}\n")
# Check existing hashes for this language
# 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")
# Check if files exist and match hashes
rst_exists = os.path.exists(rst_file)
md_exists = os.path.exists(md_file)
if rst_exists and md_exists and not needs_update:
@ -253,11 +384,21 @@ def process_file(file_path, lang_pbar):
lang_pbar.update(1)
continue
# Translate or use original content
# Translate content (English is copied directly)
if lang_code == "en":
translated_replaced_content = replaced_content
else:
translated_replaced_content = translate_text(replaced_content, lang_full)
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(
@ -265,22 +406,29 @@ def process_file(file_path, lang_pbar):
)
full_rst_content = "".join(new_header) + translated_content
# Save .rst and calculate hash
with open(rst_file, "w", encoding="utf-8") as f:
f.write(full_rst_content)
rst_hash = get_content_hash(full_rst_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)
# Convert to .md and calculate hash
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)
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)
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 for this language
# Update hashes
if file_path not in hashes:
hashes[file_path] = {}
hashes[file_path]["source"] = source_hash
@ -289,28 +437,47 @@ def process_file(file_path, lang_pbar):
hashes[file_path][lang_code]["rst"] = rst_hash
hashes[file_path][lang_code]["md"] = md_hash
# Save hashes after each language
save_hashes(hashes)
lang_pbar.update(1)
def main():
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.keys())
and not any(f"/{lang}/" in root for lang in TOP_20_LANGS)
]
with tqdm(total=len(rst_files), desc="Total Files", unit="file", file=sys.stderr) as file_pbar:
with 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)
# 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()