Automating Screenshot Localization with the Gemini API

import anthropic
import base64
import json
from pathlib import Path
 
def extract_and_translate_screenshot(
    image_path: str,
    target_languages: list = ["Japanese", "German", "French"],
    source_language: str = "English"
) -> dict:
    """
    Extract text from a screenshot and translate to multiple languages in one pass
    """
    client = anthropic.Anthropic(api_key="YOUR_GEMINI_API_KEY")
 
    # Load and encode the image
    image_data = Path(image_path).read_bytes()
    base64_image = base64.standard_b64encode(image_data).decode("utf-8")
 
    # Build language instructions dynamically
    language_instructions = "\n".join([
        f"- {lang}" for lang in target_languages
    ])
 
    # Send to Gemini API with detailed extraction instructions
    message = client.messages.create(
        model="claude-3-5-sonnet-20241022",
        max_tokens=4096,
        messages=[
            {
                "role": "user",
                "content": [
                    {
                        "type": "image",
                        "source": {
                            "type": "base64",
                            "media_type": "image/png",
                            "data": base64_image,
                        },
                    },
                    {
                        "type": "text",
                        "text": f"""Analyze this app screenshot and extract the following:
 
1. All visible text (buttons, labels, headers, descriptions)
2. The purpose of each text element (CTA, informational, navigation, etc.)
3. Approximate position on screen (top/middle/bottom, left/right)
4. The overall purpose of the screen (onboarding, home, settings, etc.)
 
Then translate all extracted text to these languages:
{language_instructions}
 
Ensure translations are natural, culturally appropriate, and fit the visual context of an app screen.
 
Return the results as valid JSON with this exact structure:
 
{{
    "source_language": "{source_language}",
    "screen_type": "screen purpose here",
    "original_text_elements": [
        {{
            "id": "element_1",
            "original_text": "Original text",
            "element_type": "button|label|heading|description",
            "position": "top-center|middle-left|bottom-right",
            "context": "brief context"
        }}
    ],
    "translations": {{
        "Japanese": [
            {{
                "element_id": "element_1",
                "translated_text": "翻訳されたテキスト",
                "notes": "any localization notes"
            }}
        ],
        "German": [
            {{
                "element_id": "element_1",
                "translated_text": "Translated text",
                "notes": "any localization notes"
            }}
        ],
        "French": [
            {{
                "element_id": "element_1",
                "translated_text": "Translated text",
                "notes": "any localization notes"
            }}
        ]
    }}
}}"""
                    }
                ],
            }
        ],
    )
 
    # Parse the JSON response
    response_text = message.content[0].text
    json_start = response_text.find('{')
    json_end = response_text.rfind('}') + 1
    json_str = response_text[json_start:json_end]
 
    return json.loads(json_str)
 
# Usage example
result = extract_and_translate_screenshot(
    image_path="app_screenshot.png",
    target_languages=["Japanese", "German", "French", "Spanish"],
    source_language="English"
)
 
# Pretty print the results
print(json.dumps(result, indent=2, ensure_ascii=False))

import os
from pathlib import Path
from typing import List, Dict
from concurrent.futures import ThreadPoolExecutor, as_completed
import time
 
def process_screenshots_batch(
    screenshot_dir: str,
    output_dir: str,
    target_languages: List[str] = ["Japanese", "German", "French", "Spanish", "Spanish (Latin America)"],
    max_workers: int = 3
) -> Dict[str, any]:
    """
    Process all screenshots in a directory with concurrent API calls.
    Respects rate limits while maximizing throughput.
    """
    client = anthropic.Anthropic(api_key="YOUR_GEMINI_API_KEY")
 
    screenshot_files = sorted(Path(screenshot_dir).glob("*.png"))
    Path(output_dir).mkdir(parents=True, exist_ok=True)
 
    results = {
        "total_screenshots": len(screenshot_files),
        "languages_processed": target_languages,
        "outputs": [],
        "errors": []
    }
 
    def process_single_screenshot(screenshot_file: Path) -> Dict:
        """Process one screenshot and return structured results"""
        try:
            print(f"Processing: {screenshot_file.name}")
 
            result = extract_and_translate_screenshot(
                str(screenshot_file),
                target_languages=target_languages,
                source_language="English"
            )
 
            # Save individual language outputs
            for language in target_languages:
                lang_code = language.lower().replace(" ", "_").split("(")[0].strip()
                output_file = Path(output_dir) / f"{screenshot_file.stem}_{lang_code}.json"
 
                # Extract language-specific data
                lang_data = {
                    "screenshot": screenshot_file.name,
                    "language": language,
                    "original_elements": result.get("original_text_elements", []),
                    "translations": result.get("translations", {}).get(language, []),
                    "screen_type": result.get("screen_type", "unknown")
                }
 
                with open(output_file, 'w', encoding='utf-8') as f:
                    json.dump(lang_data, f, indent=2, ensure_ascii=False)
 
            return {
                "screenshot": screenshot_file.name,
                "status": "success",
                "languages": len(target_languages)
            }
 
        except Exception as e:
            return {
                "screenshot": screenshot_file.name,
                "status": "error",
                "error": str(e)
            }
 
    # Process with controlled concurrency
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        futures = {
            executor.submit(process_single_screenshot, f): f
            for f in screenshot_files
        }
 
        for future in as_completed(futures):
            result = future.result()
            results["outputs"].append(result)
 
            if result["status"] == "error":
                results["errors"].append(result)
 
            # Rate limiting: small delay between requests
            time.sleep(1)
 
    # Save comprehensive report
    report_file = Path(output_dir) / "batch_report.json"
    with open(report_file, 'w', encoding='utf-8') as f:
        json.dump(results, f, indent=2, ensure_ascii=False)
 
    print(f"\nBatch processing complete!")
    print(f"Successfully processed: {len(screenshot_files)} screenshots")
    print(f"Results saved to: {output_dir}")
 
    return results
 
# Usage
process_screenshots_batch(
    screenshot_dir="./app_screenshots",
    output_dir="./localized_data",
    target_languages=["Japanese", "German", "French", "Spanish"],
    max_workers=3
)