aso_monitor.py

"""
App Store competitor monitor - a real, runnable use case on the Chocodata App Store Scraper API.

Polls one App Store search, stores every observation in a local SQLite file, and prints
what moved since the previous run: rating changes, review-volume growth, version releases,
price changes, and any app entering or leaving the ranking. Tracking a competitive set is
the single most common reason people scrape the App Store (ASO / competitor intelligence),
so it is here end to end rather than as a snippet.

    pip install requests
    export CHOCODATA_API_KEY="your_key"     # free key (1,000 requests, one-time): https://chocodata.com
    python app_store_scraper_api_codes/aso_monitor.py "music streaming"
    # ... run it again later to see the diffs

One search call returns rating, review count, version and price for the whole set, so this
costs 1 request (5 credits) per run per term, whatever the size of the set.

Docs: https://chocodata.com/docs
"""
import json
import os
import sqlite3
import sys
import time

import requests

API = "https://api.chocodata.com/api/v1/appstore/search"
KEY = os.environ.get("CHOCODATA_API_KEY")
DB = "app_store_rankings.db"

if not KEY:
    sys.exit("Set CHOCODATA_API_KEY first. Free key: https://chocodata.com")


def _check(r) -> None:
    """Map the API's documented errors onto actionable messages instead of a traceback."""
    if r.status_code == 400:
        sys.exit(f"400 invalid_params: {json.dumps(r.json().get('issues', r.text))[:200]}")
    if r.status_code == 401:
        sys.exit("401 INVALID_API_KEY: key missing or not recognised. Get one: https://chocodata.com")
    if r.status_code == 402:
        sys.exit("402 INSUFFICIENT_CREDITS: balance exhausted. Top up or upgrade: https://chocodata.com/pricing")
    if r.status_code == 429:
        sys.exit("429 RATE_LIMITED: over your plan's concurrency. Back off and retry.")
    if r.status_code == 502:
        sys.exit("502 extraction_failed: Apple returned no usable data for this request. "
                 "Retryable. You were not charged.")
    r.raise_for_status()


def fetch(term: str, country: str, limit: int) -> list[dict]:
    """One API call -> the current ranked competitive set for this term."""
    r = requests.get(API, params={"api_key": KEY, "term": term,
                                  "country": country, "limit": limit}, timeout=90)
    _check(r)
    return r.json().get("results", [])


def setup(conn: sqlite3.Connection) -> None:
    conn.execute(
        """CREATE TABLE IF NOT EXISTS observations (
               id TEXT, term TEXT, country TEXT, position INTEGER, title TEXT,
               rating REAL, reviews_count INTEGER, version TEXT, price REAL,
               formatted_price TEXT, ts INTEGER,
               PRIMARY KEY (id, term, country, ts)
           )"""
    )


def previous(conn: sqlite3.Connection, app_id: str, term: str, country: str) -> tuple | None:
    return conn.execute(
        """SELECT position, rating, reviews_count, version, formatted_price
             FROM observations WHERE id = ? AND term = ? AND country = ?
            ORDER BY ts DESC LIMIT 1""",
        (app_id, term, country),
    ).fetchone()


def main(term: str, country: str = "us", limit: int = 20) -> None:
    conn = sqlite3.connect(DB)
    setup(conn)
    now = int(time.time())
    results = fetch(term, country, limit)

    changes = 0
    for app in results:
        app_id, title = app.get("id"), (app.get("title") or "")[:44]
        if not app_id:
            continue
        before = previous(conn, app_id, term, country)

        if before:
            pos0, rating0, count0, ver0, price0 = before
            pos, rating = app.get("position"), app.get("rating")
            count, ver = app.get("reviews_count"), app.get("version")
            price = app.get("formatted_price")

            if rating0 is not None and rating is not None and abs(rating - rating0) >= 0.005:
                d = rating - rating0
                print(f"RATING  {title:44} {rating0:.2f} -> {rating:.2f} ({d:+.2f})")
                changes += 1
            if count0 is not None and count is not None and count != count0:
                print(f"REVIEWS {title:44} {count0:,} -> {count:,} ({count - count0:+,})")
                changes += 1
            if ver0 and ver and ver != ver0:
                print(f"VERSION {title:44} {ver0} -> {ver}   (competitor shipped an update)")
                changes += 1
            if price0 and price and price != price0:
                print(f"PRICE   {title:44} {price0} -> {price}")
                changes += 1
            if pos0 is not None and pos is not None and pos != pos0:
                print(f"RANK    {title:44} #{pos0} -> #{pos}")
                changes += 1
        conn.execute(
            "INSERT OR REPLACE INTO observations VALUES (?,?,?,?,?,?,?,?,?,?,?)",
            (app_id, term, country, app.get("position"), app.get("title"), app.get("rating"),
             app.get("reviews_count"), app.get("version"), app.get("price"),
             app.get("formatted_price"), now),
        )

    conn.commit()
    tracked = conn.execute("SELECT COUNT(DISTINCT id) FROM observations").fetchone()[0]
    conn.close()

    print(f"\n{len(results)} apps this run | {changes} change(s) | {tracked} apps tracked in {DB}")
    if changes == 0:
        print("No changes yet. Run it again in an hour, or schedule it (cron / GitHub Actions).")


if __name__ == "__main__":
    main(sys.argv[1] if len(sys.argv) > 1 else "music streaming")