Automating Multilingual App Review Replies with Claude API — Living with the App Store's Undocumented Limits

from anthropic import Anthropic
from dataclasses import dataclass
from enum import Enum
 
class ReviewComplexity(Enum):
    SIMPLE = "simple"     # 1-2 sentences, clear sentiment
    MODERATE = "moderate" # 3-5 sentences, multiple signals
    COMPLEX = "complex"   # Long-form, bug reports, feature requests
 
@dataclass
class ReviewMetrics:
    word_count: int
    has_bug_report: bool
    has_feature_request: bool
    sentiment_score: float  # -1.0 (negative) to 1.0 (positive)
 
def determine_complexity(review: str, metrics: ReviewMetrics) -> ReviewComplexity:
    if metrics.has_bug_report or metrics.has_feature_request:
        return ReviewComplexity.COMPLEX
    if metrics.word_count > 50 or abs(metrics.sentiment_score) < 0.3:
        return ReviewComplexity.MODERATE
    return ReviewComplexity.SIMPLE
 
def select_model(complexity: ReviewComplexity) -> str:
    return {
        ReviewComplexity.SIMPLE: "claude-haiku-4-5-20251001",
        ReviewComplexity.MODERATE: "claude-haiku-4-5-20251001",
        ReviewComplexity.COMPLEX: "claude-sonnet-4-6",
    }[complexity]
 
def generate_reply(
    client: Anthropic,
    review_text: str,
    app_name: str,
    locale: str,
    complexity: ReviewComplexity
) -> str:
    model = select_model(complexity)
    
    system_prompt = f"""You are the support representative for {app_name}.
Write a natural reply in {locale} to the user's review.
 
Guidelines:
- Keep the reply to 3-4 sentences (longer replies don't get read)
- Match the tone appropriate for {app_name}
- Reference something specific from the review — avoid generic openers
- For bug reports and feature requests: express genuine appreciation and forward momentum
- Vary your language — don't reuse phrases from other replies"""
 
    response = client.messages.create(
        model=model,
        max_tokens=300,
        system=system_prompt,
        messages=[{"role": "user", "content": f"Reply to this review:\n\n{review_text}"}]
    )
    
    return response.content[0].text

import hashlib
from difflib import SequenceMatcher
 
class ReplyVariationManager:
    """
    Tracks generated replies to enforce diversity.
    Prevents similarity-based template spam detection.
    """
    
    def __init__(self, similarity_threshold: float = 0.6):
        self.recent_replies: list[str] = []
        self.reply_hashes: set[str] = set()
        self.similarity_threshold = similarity_threshold
        self.window_size = 50
    
    def is_too_similar(self, new_reply: str) -> bool:
        new_hash = hashlib.md5(new_reply.encode()).hexdigest()
        if new_hash in self.reply_hashes:
            return True
        
        for existing in self.recent_replies[-self.window_size:]:
            ratio = SequenceMatcher(None, new_reply, existing).ratio()
            if ratio > self.similarity_threshold:
                return True
        return False
    
    def add_reply(self, reply: str):
        self.recent_replies.append(reply)
        self.reply_hashes.add(hashlib.md5(reply.encode()).hexdigest())
    
    def get_variation_instruction(self) -> str:
        """Returns a prompt addition listing recently used openings to avoid."""
        recent = self.recent_replies[-5:]
        if not recent:
            return ""
        
        used_openings = []
        for reply in recent:
            opening = reply[:50].split(".")[0]
            used_openings.append(opening)
        
        return "\n\nAvoid starting with these recently used phrases:\n" + \
               "\n".join(f"- {o}" for o in used_openings)

APP_PROFILES = {
    "beautiful-hd-wallpapers": {
        "tone": "warm and visually enthusiastic",
        "system_extra": "Empathize with the visual experience — how the wallpaper feels each time they see their screen",
        "avoid_openers": ["Thank you for your", "We appreciate"],
    },
    "law-of-attraction": {
        "tone": "encouraging and genuinely sincere",
        "system_extra": "Convey a real wish for the user's daily life to improve — not motivational-speaker energy",
        "avoid_openers": ["We're so glad", "That's wonderful"],
    },
    "relaxing-healing": {
        "tone": "calm and understated",
        "system_extra": "Match the app's quiet atmosphere — no exclamation marks, peaceful pacing",
        "avoid_openers": ["So happy to hear", "Amazing"],
    },
    "ukiyo-e-wallpapers": {
        "tone": "culturally respectful and informative",
        "system_extra": "Show appreciation for the user's interest in Japanese traditional art",
        "avoid_openers": ["Thank you for downloading"],
    }
}

LOCALE_GUIDELINES = {
    "ar": "Reply in Modern Standard Arabic. Use formal register. Follow RTL punctuation conventions.",
    "id": "Reply in Indonesian. Always use 'Anda' (formal second person). Formal register throughout.",
    "ko": "Reply in Korean. Use 합쇼체 (formal polite). Warm but professional.",
    "zh-Hans": "Reply in Simplified Chinese. Polite and approachable — not stiff.",
    "zh-Hant": "Reply in Traditional Chinese (Taiwan/Hong Kong conventions).",
    "tr": "Reply in Turkish. Formal but warm. Use 'siz' (formal you).",
    "pt-BR": "Reply in Brazilian Portuguese. Friendly and natural, not stiff.",
    "pt-PT": "Reply in European Portuguese. Avoid Brazilian expressions.",
}

async def post_and_verify_reply(
    client: AppStoreClient,
    review_id: str,
    reply_text: str,
    throttler: AppStoreReplyThrottler,
    max_verify_attempts: int = 5
) -> bool:
    await throttler.wait_before_reply()
    success = await client.post_review_reply(review_id, reply_text)
    throttler.record_reply()
    
    if not success:
        return False
    
    for _ in range(max_verify_attempts):
        await asyncio.sleep(30)
        existing = await client.get_reply(review_id)
        if existing and existing.strip() == reply_text.strip():
            return True
    
    return False  # Route to manual verification queue

DAILY_COST_LIMIT_USD = 5.0
 
async def check_and_enforce_cost_limit(tracker: UsageTracker):
    daily_cost = tracker.get_daily_cost_usd()
    if daily_cost > DAILY_COST_LIMIT_USD:
        await send_alert(f"Claude API cost exceeded limit: ${daily_cost:.2f}")
        tracker.pause_processing()

import asyncio
import heapq
from dataclasses import dataclass, field
from datetime import datetime
from typing import List
 
@dataclass(order=True)
class ReplyJob:
    priority: float
    scheduled_at: float = field(compare=False)
    app_id: str = field(compare=False)
    review_id: str = field(compare=False)
    review_text: str = field(compare=False)
    locale: str = field(compare=False)
    complexity: str = field(compare=False)
 
class MultiAppReplyOrchestrator:
    """
    Orchestrates reply generation across multiple apps
    with shared budget and API rate limit awareness.
    """
    
    APP_PRIORITY_WEIGHTS = {
        "beautiful-hd-wallpapers": 1.0,   # Highest volume
        "law-of-attraction": 0.9,
        "relaxing-healing": 0.8,
        "ukiyo-e-wallpapers": 0.7,
    }
    
    GLOBAL_SESSION_LIMIT = 80      # Across all apps per session
    INTER_APP_GAP_SECONDS = 12.0   # Slightly larger gap when switching apps
    INTRA_APP_GAP_SECONDS = 9.0    # Standard gap within same app
    
    def __init__(self, claude_client, store_client):
        self.queue: List[ReplyJob] = []
        self.claude_client = claude_client
        self.store_client = store_client
        self.session_count = 0
        self.last_app_id: str | None = None
        self.last_reply_time: float | None = None
    
    def enqueue(self, app_id: str, review_id: str, review_text: str,
                locale: str, complexity: str, age_hours: float):
        """
        Priority = weight * recency factor.
        Older reviews get a recency penalty to ensure freshness.
        """
        weight = self.APP_PRIORITY_WEIGHTS.get(app_id, 0.5)
        recency_factor = max(0.3, 1.0 - (age_hours / 168.0))  # Decays over 1 week
        priority = -(weight * recency_factor)  # Negative for min-heap
        
        job = ReplyJob(
            priority=priority,
            scheduled_at=datetime.now().timestamp(),
            app_id=app_id,
            review_id=review_id,
            review_text=review_text,
            locale=locale,
            complexity=complexity
        )
        heapq.heappush(self.queue, job)
    
    async def process_next(self) -> bool:
        """Process the highest-priority job in the queue."""
        if not self.queue or self.session_count >= self.GLOBAL_SESSION_LIMIT:
            return False
        
        job = heapq.heappop(self.queue)
        
        # Determine gap based on whether we're switching apps
        if self.last_app_id and self.last_app_id != job.app_id:
            gap = self.INTER_APP_GAP_SECONDS
        else:
            gap = self.INTRA_APP_GAP_SECONDS
        
        if self.last_reply_time:
            elapsed = asyncio.get_event_loop().time() - self.last_reply_time
            if elapsed < gap:
                await asyncio.sleep(gap - elapsed)
        
        # Generate and post
        reply = generate_reply(
            self.claude_client,
            job.review_text,
            job.app_id,
            job.locale,
            job.complexity
        )
        
        success = await self.store_client.post_reply(job.review_id, reply)
        
        self.session_count += 1
        self.last_reply_time = asyncio.get_event_loop().time()
        self.last_app_id = job.app_id
        
        return success

def build_positive_review_prompt(review: str, app_profile: dict) -> str:
    """
    For highly positive reviews: reference something specific.
    The goal is to sound like a person who actually read it.
    """
    return f"""This user left a genuinely positive, emotional review:
 
"{review}"
 
Write a reply that:
1. References one specific thing they mentioned (don't be generic)
2. Matches the {app_profile['tone']} tone of {app_profile['name']}
3. Is 2-3 sentences maximum
4. Does NOT use: "We're so glad", "Thank you for sharing", "That means so much to us"
5. Sounds like it was written by someone who read the review, not a support template
 
Write only the reply text, nothing else."""

def build_negative_review_prompt(review: str, app_profile: dict, locale: str) -> str:
    return f"""This user left a critical review (1-2 stars):
 
"{review}"
 
Write a reply in {locale} that:
1. Acknowledges their specific complaint — don't be vague
2. Provides one concrete next step (e.g., update to latest version, contact support@dolice.net)
3. Does NOT apologize excessively or become defensive
4. Is 2-4 sentences, professional and calm
5. Avoids sounding like a PR response or an automated message
 
The tone should be: a real person who is taking the feedback seriously.
Write only the reply text."""

from anthropic import Anthropic
 
def generate_reply_with_cache(
    client: Anthropic,
    review_text: str,
    cached_system: str,  # Already-formatted system prompt > 1024 tokens
    model: str
) -> str:
    """
    Use cache_control to cache the system prompt.
    Cost reduction: ~90% on cached tokens.
    """
    response = client.messages.create(
        model=model,
        max_tokens=300,
        system=[
            {
                "type": "text",
                "text": cached_system,
                "cache_control": {"type": "ephemeral"}
            }
        ],
        messages=[
            {
                "role": "user",
                "content": f"Reply to this review:\n\n{review_text}"
            }
        ]
    )
    
    # Track cache performance
    usage = response.usage
    cache_hit = hasattr(usage, 'cache_read_input_tokens') and usage.cache_read_input_tokens > 0
    
    return response.content[0].text, cache_hit

@dataclass
class ReplySessionMetrics:
    date: str
    app_id: str
    total_reviewed: int          # Total reviews fetched
    auto_replied: int            # Successfully auto-replied
    manual_queued: int           # Sent to human review
    silently_failed: int         # Posted but didn't appear (verified)
    haiku_calls: int
    sonnet_calls: int
    cache_hits: int
    estimated_cost_usd: float
    average_similarity_score: float  # Health indicator — should stay low
    
    @property
    def automation_rate(self) -> float:
        if self.total_reviewed == 0:
            return 0.0
        return self.auto_replied / self.total_reviewed
    
    @property
    def cost_per_reply(self) -> float:
        if self.auto_replied == 0:
            return 0.0
        return self.estimated_cost_usd / self.auto_replied

from langdetect import detect, LangDetectException
import anthropic
 
def detect_review_language(review_text: str, client: anthropic.Anthropic) -> str:
    """
    Primary: langdetect library (fast, free).
    Fallback: Claude for short or ambiguous text.
    """
    if len(review_text.split()) < 5:
        # Too short for langdetect to be reliable
        return detect_language_with_claude(review_text, client)
    
    try:
        detected = detect(review_text)
        # langdetect uses ISO 639-1; map to App Store locale codes as needed
        locale_map = {"zh-cn": "zh-Hans", "zh-tw": "zh-Hant", "pt": "pt-BR"}
        return locale_map.get(detected, detected)
    except LangDetectException:
        return detect_language_with_claude(review_text, client)
 
def detect_language_with_claude(text: str, client: anthropic.Anthropic) -> str:
    """Use Claude for ambiguous language detection."""
    response = client.messages.create(
        model="claude-haiku-4-5-20251001",
        max_tokens=10,
        messages=[{
            "role": "user",
            "content": f"What language is this text written in? Reply with only the ISO 639-1 code.\n\n{text}"
        }]
    )
    return response.content[0].text.strip().lower()