Automating App Store Reviews with Gemini API and App Store Connect API: Implementation Notes from Running 50M-Download Apps

def get_app_ids(client: AppStoreConnectClient) -> list[dict]:
    """Fetch a list of apps you manage"""
    data = client.get("/apps", params={"limit": 50})
    return [
        {
            "id": app["id"],
            "bundle_id": app["attributes"]["bundleId"],
            "name": app["attributes"]["name"],
        }
        for app in data.get("data", [])
    ]

from datetime import datetime, timedelta, timezone
from dataclasses import dataclass
 
@dataclass
class Review:
    """Structured review data"""
    id: str
    app_id: str
    app_name: str
    rating: int
    title: str
    body: str
    reviewer_nickname: str
    created_date: datetime
    territory: str  # Country/region code
    developer_response: str | None = None
 
def fetch_recent_reviews(
    client: AppStoreConnectClient,
    app_id: str,
    app_name: str,
    days: int = 7
) -> list[Review]:
    """Fetch reviews from the last N days"""
    reviews = []
    cursor = None
    since = datetime.now(timezone.utc) - timedelta(days=days)
 
    while True:
        params = {"limit": 200, "sort": "-createdDate"}
        if cursor:
            params["cursor"] = cursor
 
        data = client.get(f"/apps/{app_id}/customerReviews", params=params)
 
        for item in data.get("data", []):
            attrs = item["attributes"]
            created = datetime.fromisoformat(attrs["createdDate"].replace("Z", "+00:00"))
 
            if created < since:
                return reviews  # Older than window — stop paginating
 
            # Fetch developer response if it exists
            response_text = None
            if item.get("relationships", {}).get("response", {}).get("data"):
                response_data = client.get(f"/customerReviews/{item['id']}/response")
                if response_data.get("data"):
                    response_text = response_data["data"]["attributes"].get("responseBody")
 
            reviews.append(Review(
                id=item["id"],
                app_id=app_id,
                app_name=app_name,
                rating=attrs["rating"],
                title=attrs.get("title", ""),
                body=attrs.get("body", ""),
                reviewer_nickname=attrs.get("reviewerNickname", "Anonymous"),
                created_date=created,
                territory=attrs.get("territory", "USA"),
                developer_response=response_text,
            ))
 
        next_link = data.get("links", {}).get("next")
        if not next_link:
            break
        cursor = next_link.split("cursor=")[-1].split("&")[0]
 
    return reviews

import google.generativeai as genai
import json
 
genai.configure(api_key=os.getenv("GEMINI_API_KEY"))
model = genai.GenerativeModel("gemini-2.5-flash")
 
ANALYSIS_SYSTEM_PROMPT = """
You are an App Store review analysis expert.
Analyze each review and return results ONLY in the specified JSON format.
 
Analysis fields:
- sentiment: Overall tone (positive/negative/neutral/mixed)
- category: Issue type (bug/ui_ux/feature_request/performance/praise/support/other)
- severity: Urgency level (1=low to 5=critical, for negative reviews only)
- key_issue: Main issue or praise point (max 50 characters)
- needs_response: Whether a reply is warranted (true/false)
- response_priority: Reply urgency (1=low to 5=urgent)
- language: Detected language code (ja/en/zh/ko/fr/de/es/etc.)
 
Return a JSON array only. No explanations.
"""
 
def analyze_reviews_batch(reviews: list[Review]) -> list[dict]:
    """Batch process reviews for cost efficiency"""
    review_batch = [
        {
            "id": r.id,
            "rating": r.rating,
            "title": r.title,
            "body": r.body[:500],  # Trim long reviews
            "territory": r.territory,
            "has_response": r.developer_response is not None,
        }
        for r in reviews
    ]
 
    prompt = f"""
Analyze the following {len(review_batch)} App Store reviews.
 
Review data:
{json.dumps(review_batch, ensure_ascii=False, indent=2)}
 
Return analysis results as a JSON array, one object per review id.
Format: [{{"id": "...", "sentiment": "...", "category": "...", ...}}]
"""
 
    response = model.generate_content(
        prompt,
        generation_config=genai.GenerationConfig(
            temperature=0.1,  # Low temperature for consistent analysis
            response_mime_type="application/json",
        ),
        system_instruction=ANALYSIS_SYSTEM_PROMPT,
    )
 
    try:
        return json.loads(response.text)
    except json.JSONDecodeError:
        print(f"⚠️ JSON parse error: {response.text[:200]}")
        return [
            {"id": r["id"], "sentiment": "neutral", "category": "other", "severity": 1,
             "key_issue": "Analysis failed", "needs_response": False, "response_priority": 1, "language": "en"}
            for r in review_batch
        ]
 
def analyze_all_reviews(reviews: list[Review], batch_size: int = 50) -> list[dict]:
    """Process all reviews in batches of 50 for stability"""
    all_results = []
    for i in range(0, len(reviews), batch_size):
        batch = reviews[i:i + batch_size]
        results = analyze_reviews_batch(batch)
        all_results.extend(results)
        print(f"✅ Analyzed: {min(i + batch_size, len(reviews))}/{len(reviews)} reviews")
    return all_results

REPLY_GENERATION_PROMPT = """
You are a successful indie app developer responding to App Store reviews.
Generate sincere, warm replies that feel human and personal.
 
Reply rules:
1. Match the language of the review
2. Keep replies under 200 characters (App Store limit)
3. Reference specific details from the review — never sound generic
4. For negative reviews: acknowledge → explain (if possible) → improvement plan → support contact
5. For positive reviews: genuine thanks → share enthusiasm → tease upcoming features
6. For bug reports: acknowledge → confirm steps → status update → contact info
7. Avoid corporate-sounding phrases
 
Never use:
- "Thank you for your valuable feedback" (overused)
- "We apologize for any inconvenience" (robotic)
- Excessive formality or repetition
"""
 
def generate_reply(review: Review, analysis: dict) -> str:
    """Generate a personalized reply for a specific review"""
 
    if not analysis.get("needs_response", True):
        return ""
 
    if review.developer_response:
        return ""  # Already replied
 
    # In production, load this from a config file per app
    app_context = f"""
App name: {review.app_name}
App category: Wallpaper & Lifestyle
Latest version: 3.2.1
Recent changes: iOS 18 support, performance improvements, new categories added
"""
 
    prompt = f"""
{app_context}
 
Review details:
- Rating: {review.rating}/5
- Title: {review.title or "(no title)"}
- Body: {review.body}
- Region: {review.territory}
- Analysis: {json.dumps(analysis, ensure_ascii=False)}
 
Generate a reply to this review.
Requirements: under 200 characters, in {analysis.get('language', 'en')}.
Output the reply text only — no preamble.
"""
 
    response = model.generate_content(
        prompt,
        generation_config=genai.GenerationConfig(temperature=0.7),
        system_instruction=REPLY_GENERATION_PROMPT,
    )
    return response.text.strip()
 
def submit_reply(client: AppStoreConnectClient, review_id: str, reply_text: str) -> bool:
    """Submit a reply via App Store Connect API"""
    try:
        existing = client.get(f"/customerReviews/{review_id}/response")
 
        if existing.get("data"):
            client.patch(
                f"/customerReviewResponses/{existing['data']['id']}",
                {"data": {"type": "customerReviewResponses",
                          "id": existing['data']['id'],
                          "attributes": {"responseBody": reply_text}}}
            )
        else:
            headers = {"Authorization": f"Bearer {client.token}", "Content-Type": "application/json"}
            body = {
                "data": {
                    "type": "customerReviewResponses",
                    "attributes": {"responseBody": reply_text},
                    "relationships": {"review": {"data": {"type": "customerReviews", "id": review_id}}},
                }
            }
            httpx.post(f"{AppStoreConnectClient.BASE_URL}/customerReviewResponses",
                      headers=headers, json=body).raise_for_status()
        return True
    except Exception as e:
        print(f"❌ Reply submission error (review_id: {review_id}): {e}")
        return False

import sqlite3
from enum import Enum
 
class ReplyStatus(Enum):
    PENDING = "pending"    # Generated, awaiting human review
    APPROVED = "approved"  # Approved, queued for submission
    SENT = "sent"          # Successfully submitted
    REJECTED = "rejected"  # Rejected, switching to manual reply
    SKIPPED = "skipped"    # No reply needed
 
def setup_database(db_path: str = "reviews.db"):
    """Initialize SQLite database for review tracking"""
    conn = sqlite3.connect(db_path)
    conn.execute("""
        CREATE TABLE IF NOT EXISTS reviews (
            id TEXT PRIMARY KEY,
            app_id TEXT,
            app_name TEXT,
            rating INTEGER,
            title TEXT,
            body TEXT,
            territory TEXT,
            created_date TEXT,
            sentiment TEXT,
            category TEXT,
            severity INTEGER,
            key_issue TEXT,
            generated_reply TEXT,
            reply_status TEXT DEFAULT 'pending',
            sent_at TEXT,
            created_at TEXT DEFAULT CURRENT_TIMESTAMP
        )
    """)
    conn.commit()
    return conn

import httpx
 
ITUNES_SEARCH_API = "https://itunes.apple.com"
 
class CompetitorMonitor:
    """Monitor competitor apps using iTunes Search API"""
 
    def __init__(self, competitor_app_ids: list[str], country: str = "us"):
        self.competitor_ids = competitor_app_ids
        self.country = country
 
    def fetch_competitor_info(self, app_id: str) -> dict:
        """Fetch competitor data from iTunes Search API"""
        response = httpx.get(
            f"{ITUNES_SEARCH_API}/lookup",
            params={"id": app_id, "country": self.country},
            timeout=10
        )
        data = response.json()
        if data["resultCount"] == 0:
            return {}
 
        result = data["results"][0]
        return {
            "app_id": app_id,
            "name": result.get("trackName"),
            "seller": result.get("sellerName"),
            "current_version": result.get("version"),
            "user_rating": result.get("averageUserRating", 0),
            "rating_count": result.get("userRatingCount", 0),
            "current_version_rating": result.get("averageUserRatingForCurrentVersion", 0),
            "release_notes": result.get("releaseNotes", ""),
            "release_date": result.get("currentVersionReleaseDate"),
            "price": result.get("price", 0),
            "description": result.get("description", "")[:500],
            "fetched_at": datetime.utcnow().isoformat(),
        }
 
    def analyze_competitor_update(self, current: dict, previous: dict | None) -> str:
        """Use Gemini to analyze what changed in a competitor update"""
        if not previous:
            return "New competitor tracking initiated"
 
        if current.get("current_version") == previous.get("current_version"):
            return "No version change"
 
        prompt = f"""
Analyze this competitor app update briefly (2-3 sentences max).
 
App: {current['name']}
Version: {previous.get('current_version', 'unknown')} → {current['current_version']}
 
Release notes:
{current.get('release_notes', 'Not provided')}
 
Rating change:
- Overall: {previous.get('user_rating', 0):.1f} → {current['user_rating']:.1f}
- Review count: {previous.get('rating_count', 0):,} → {current['rating_count']:,}
 
Cover: 1) Key changes (features/fixes/UI), 2) Strategic implications for us.
"""
        response = model.generate_content(prompt, generation_config=genai.GenerationConfig(temperature=0.3))
        return response.text.strip()
 
    def run_monitoring(self) -> list[dict]:
        """Monitor all tracked competitors"""
        results = []
        for app_id in self.competitor_ids:
            try:
                info = self.fetch_competitor_info(app_id)
                if info:
                    results.append(info)
                    print(f"✅ Fetched data for: {info['name']}")
            except Exception as e:
                print(f"❌ Failed for App ID {app_id}: {e}")
        return results

def generate_weekly_aso_report(
    reviews: list[Review],
    analyses: list[dict],
    competitor_data: list[dict],
    my_apps: list[dict],
) -> str:
    """Generate a Markdown-formatted weekly ASO intelligence report"""
 
    total = len(reviews)
    if total == 0:
        return "No reviews this week."
 
    rating_dist = {1: 0, 2: 0, 3: 0, 4: 0, 5: 0}
    sentiment_dist = {"positive": 0, "negative": 0, "neutral": 0, "mixed": 0}
    category_dist = {}
 
    for review, analysis in zip(reviews, analyses):
        rating_dist[review.rating] = rating_dist.get(review.rating, 0) + 1
        sentiment = analysis.get("sentiment", "neutral")
        sentiment_dist[sentiment] = sentiment_dist.get(sentiment, 0) + 1
        cat = analysis.get("category", "other")
        category_dist[cat] = category_dist.get(cat, 0) + 1
 
    avg_rating = sum(r.rating for r in reviews) / total
 
    # Critical issues requiring urgent attention
    urgent = [
        (r, a) for r, a in zip(reviews, analyses)
        if a.get("sentiment") == "negative" and a.get("severity", 0) >= 4
    ]
 
    stats_json = json.dumps({
        "total_reviews": total,
        "avg_rating": round(avg_rating, 2),
        "rating_distribution": rating_dist,
        "sentiment_distribution": sentiment_dist,
        "category_distribution": category_dist,
        "urgent_issues_count": len(urgent),
        "urgent_issues": [
            {"title": r.title, "body": r.body[:100], "key_issue": a.get("key_issue")}
            for r, a in urgent[:5]
        ],
        "competitor_updates": [
            {"name": c["name"], "version": c.get("current_version"),
             "rating": c.get("user_rating"), "notes_snippet": c.get("release_notes", "")[:200]}
            for c in competitor_data if c.get("release_notes")
        ],
    }, ensure_ascii=False)
 
    prompt = f"""
Generate a weekly ASO Intelligence Report in Markdown based on this data:
 
{stats_json}
 
Structure:
1. ## Weekly Summary (3 lines max — key takeaways)
2. ## Review Analysis (rating trends, sentiment breakdown, category insights)
3. ## Urgent Action Items (severity 4-5 negative reviews requiring immediate attention)
4. ## Competitor Intelligence (update summaries and strategic implications)
5. ## Action Plan for Next Week (3 specific, executable tasks)
 
Write in English. Focus on actionable insights over raw data recitation.
"""
 
    response = model.generate_content(
        prompt,
        generation_config=genai.GenerationConfig(temperature=0.4, max_output_tokens=2000),
    )
    return response.text
 
def send_slack_report(report_text: str, urgent_count: int):
    """Send report to Slack with color-coded urgency"""
    color = "#ff0000" if urgent_count > 0 else "#36a64f"
    emoji = "🚨" if urgent_count > 0 else "📊"
 
    payload = {
        "attachments": [{
            "color": color,
            "title": f"{emoji} Weekly ASO Intelligence Report",
            "text": report_text[:2900],
            "footer": f"Gemini Lab ASO System | {datetime.now().strftime('%Y-%m-%d %H:%M')} UTC",
            "mrkdwn_in": ["text"],
        }]
    }
    httpx.post(os.getenv("SLACK_WEBHOOK_URL"), json=payload, timeout=10)
    print("✅ Slack notification sent")

# .github/workflows/aso-monitor.yml
name: ASO Intelligence Monitor
 
on:
  schedule:
    - cron: '0 0 * * *'    # Daily at UTC 00:00 (runs daily review analysis)
    - cron: '0 22 * * 0'   # Weekly Sunday UTC 22:00 (weekly report)
  workflow_dispatch:         # On-demand manual trigger
 
jobs:
  monitor:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: '3.12'
      - run: pip install -r requirements.txt
      - run: python aso_main.py --mode=daily
        env:
          APP_STORE_KEY_ID: ${{ secrets.APP_STORE_KEY_ID }}
          APP_STORE_ISSUER_ID: ${{ secrets.APP_STORE_ISSUER_ID }}
          APP_STORE_PRIVATE_KEY: ${{ secrets.APP_STORE_PRIVATE_KEY }}
          GEMINI_API_KEY: ${{ secrets.GEMINI_API_KEY }}
          SLACK_WEBHOOK_URL: ${{ secrets.SLACK_WEBHOOK_URL }}

import time
from functools import wraps
 
def with_retry(max_retries: int = 3, backoff_factor: float = 2.0):
    """Decorator for automatic retry with exponential backoff"""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(max_retries):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    if attempt == max_retries - 1:
                        raise
                    wait = backoff_factor ** attempt
                    print(f"⚠️ Attempt {attempt + 1}/{max_retries} failed: {e}. Retrying in {wait}s...")
                    time.sleep(wait)
        return wrapper
    return decorator
 
@with_retry(max_retries=3)
def safe_analyze_reviews(reviews):
    return analyze_reviews_batch(reviews)

#!/usr/bin/env python3
"""
ASO Intelligence System — Main entry point
Usage: python aso_main.py [--mode=daily|weekly|reply|report] [--dry-run]
"""
 
import argparse
from datetime import date
 
def main():
    parser = argparse.ArgumentParser(description="ASO Intelligence System")
    parser.add_argument("--mode", choices=["daily", "weekly", "reply", "report"], default="daily")
    parser.add_argument("--dry-run", action="store_true", help="Generate replies without submitting")
    args = parser.parse_args()
 
    client = AppStoreConnectClient()
    conn = setup_database()
 
    print(f"🚀 ASO Intelligence System started (mode: {args.mode})")
 
    apps = get_app_ids(client)
    print(f"📱 Managing {len(apps)} apps")
 
    # Load competitor IDs from config in production
    competitor_ids = ["COMPETITOR_APP_ID_1", "COMPETITOR_APP_ID_2"]
    monitor = CompetitorMonitor(competitor_ids)
 
    all_reviews, all_analyses = [], []
    days = 1 if args.mode == "daily" else 7
 
    for app in apps:
        reviews = fetch_recent_reviews(client, app["id"], app["name"], days=days)
        print(f"📝 {app['name']}: {len(reviews)} reviews fetched")
 
        if not reviews:
            continue
 
        analyses = analyze_all_reviews(reviews)
        all_reviews.extend(reviews)
        all_analyses.extend(analyses)
 
        if args.mode in ["daily", "reply"]:
            for review, analysis in zip(reviews, analyses):
                if analysis.get("response_priority", 0) >= 4:
                    reply = generate_reply(review, analysis)
                    if reply and not args.dry_run:
                        success = submit_reply(client, review.id, reply)
                        status = "✅ Sent" if success else "❌ Failed"
                        print(f"  {status}: [{review.rating}★] {review.title[:30]}...")
 
    if args.mode in ["weekly", "report"]:
        competitor_data = monitor.run_monitoring()
        report = generate_weekly_aso_report(all_reviews, all_analyses, competitor_data, apps)
        urgent_count = sum(
            1 for a in all_analyses
            if a.get("sentiment") == "negative" and a.get("severity", 0) >= 4
        )
        send_slack_report(report, urgent_count)
        report_path = f"reports/aso_report_{date.today()}.md"
        with open(report_path, "w") as f:
            f.write(report)
        print(f"📄 Report saved: {report_path}")
 
    print("✅ Processing complete")
 
if __name__ == "__main__":
    main()

def get_token(self) -> str:
    cached = redis.get("ascc:jwt")
    if cached:
        return cached.decode()
    token = self._generate_token()  # exp = now + 1200
    redis.set("ascc:jwt", token, ex=1080)  # 18 minutes
    return token