Google Playデータ取得 データ取得

Google PlayレビューをPythonで取得する方法｜google-play-scraperのreviews()と継続取得

2026年8月14日

Google Playレビューを分析へ使う前に、取得条件、継続取得、保存形式を揃える必要があります。この記事では、Pythonライブラリgoogle-play-scraperのreviews()で少量取得し、continuation tokenで次のページへ進み、必要な場合だけreviews_all()を検討する流れを整理します。

最短手順：ファイルを用意し、実行して出力を開く

App IDを指定してレビューを少量取得し、CSVと取得条件を保存します。この手順はGoogle Playへの通信を行います。APIキーは不要ですが、通信なしの練習ではありません。

掲載Python10ファイルのZIPをダウンロードし、「すべて展開」で任意の作業フォルダーへ展開します。ZIP内にサブフォルダーはありません。必要ファイルは collect_gp_reviews.py、gp_common.py です。使わないファイルは実行しません。以下の長いコードを手で保存する場合も、同じ名前・同じフォルダーへ置きます。

Python導入済みのWindows PowerShellで実行します。最初の入力欄へ、展開後の gp_common.py が見えるフォルダーの絶対パスを貼り付けてEnterを押してください。引用符は入力不要です。以降もその作業フォルダーから実行します。インストールには通信が必要ですが、その後の保存CSV分析とは別です。

$work = Read-Host "ZIPを展開したフォルダーの絶対パス"
Set-Location -LiteralPath $work
Get-ChildItem *.py
python -m pip install pandas google-play-scraper

APP_IDはファイル先頭で対象アプリの値を確認します。配布時はPokémon GOの例です。取得条件・上限を理解してから次を実行します。

python collect_gp_reviews.py

最後に表示されたcollectionフォルダーをエクスプローラーで開き、下表の2ファイルを確認します。次の分析には、そのフォルダー内のgoogle_play_reviews.csvを直接指定できます。標準の outputs/google_play_reviews.csv にコピーする場合だけ、後述の「保存結果を確認して分析用の入力に選ぶ」を使います。そこのPython例にある collection-実際の実行名 は、今回表示されたフォルダー名へ置き換えてから実行してください。

まず開くファイル：知りたいことから選ぶ

何件保存され、なぜ終了したか
metadata.json：saved_rows・end_reason・error。フォルダー表示だけを全件取得成功とは扱いません。

保存したレビューを確認する
google_play_reviews.csv：content・score・at。取得時刻はmetadataのstarted_at_utc／finished_at_utc。

CSVは表計算ソフトでも開けますが、保存し直すと型やファイルの確認値（hash）が変わる場合があります。元ファイルは上書きせず閲覧してください。図がない場合は、同じ実行フォルダーのmetadata.jsonにある skipped や手法別状態を確認します。

コードの前に：結果から何が分かるか

終了時に実行フォルダーが表示されても、Google Playにある全件を取り切った証明ではありません。上限到達、0件、通信の途中終了をmetadataで確認し、その範囲のCSVとして使います。基本例と実用版は代替手順で、両方を続けて実行する必要はありません。

掲載コードと詳しい条件（ZIPと同じ内容）

想定読者

  • Google PlayレビューをPythonで取得したい方
  • continuation tokenで継続取得したい方
  • 取得条件を残してCSV保存したい方

google-play-scraperとレビュー取得の位置づけ

google-play-scraperはGoogle公式APIではなく、Google Playの公開情報を取得する外部Pythonライブラリです。レビュー取得には、件数を区切って取得するreviews()と、内部で継続取得を繰り返すreviews_all()があります。

推奨する進め方

  • 最初はreviews()で3〜100件程度を確認する
  • 返却フィールドとCSV形式を決める
  • 必要な範囲だけcontinuation tokenで継続取得する

App IDとライブラリを準備する

App IDはGoogle Play URLのid=以降にあるPackage nameです。例としてcom.nianticlabs.pokemongoを使います。

取得条件や返却形式が変わったときに追跡できるよう、使用バージョンも確認しておきます。

reviews()で少量取得する

最初から全件取得を狙わず、少量のレスポンスでフィールドと型を確認します。

  1. App IDと取得条件を決める
  2. reviews()を少量で実行する
  3. 返却件数・フィールド名・continuation tokenを確認する
from google_play_scraper import Sort, reviews

app_id = "com.nianticlabs.pokemongo"

result, continuation_token = reviews(
    app_id,
    lang="ja",
    country="jp",
    sort=Sort.NEWEST,
    count=100,
    filter_score_with=None,
)

print("returned:", len(result))
print("fields:", sorted(result[0].keys()) if result else [])
print("continuation:", bool(getattr(continuation_token, "token", None)))

現行ソースでは、lang、country、sort、count、filter_score_with、filter_device_with、continuation_tokenを指定できます。countは取得希望件数であり、指定件数の返却を保証するものではありません。

取得条件を使い分ける

引数用途記録する理由
lang表示言語レビューや返信の言語条件を追跡する
countryストアの国・地域取得対象の範囲を明確にする
sort新着・評価順など取得順序による偏りを確認する
count1回の取得希望件数処理単位と負荷を記録する
filter_score_with星評価で絞り込み対象スコアを明確にする
filter_device_with端末条件で絞り込み利用した場合に条件を再現する

分析用データでは、返却されたレビューだけでなく、これらの取得条件も同じファイルや実行ログへ残します。

continuation tokenで次のページを取得する

reviews()はレビュー一覧とcontinuation tokenを返します。次の呼び出しへtokenを渡すと、同じ取得条件を引き継いで続きを取得できます。

from google_play_scraper import Sort, reviews

app_id = "com.nianticlabs.pokemongo"

first_page, token = reviews(
    app_id,
    lang="ja",
    country="jp",
    sort=Sort.NEWEST,
    count=100,
)

second_page = []
next_token = None

if token is not None:
    second_page, next_token = reviews(
        app_id,
        continuation_token=token,
    )

reviews_two_pages = first_page + second_page
print("total:", len(reviews_two_pages))
print("has next:", bool(getattr(next_token, "token", None)))

continuation tokenはライブラリ内部の継続情報を持つオブジェクトで、永続的なAPI tokenや認証情報ではありません。長期保存して別の実行で再利用する前提にせず、処理中のページ継続に使います。

reviews_all()との違い

reviews_all()は、内部でcontinuation tokenを使いながら繰り返し取得する便利関数です。

from google_play_scraper import Sort, reviews_all

all_reviews = reviews_all(
    "com.example.app",
    lang="ja",
    country="jp",
    sort=Sort.NEWEST,
    sleep_milliseconds=1000,
)

print("returned:", len(all_reviews))

レビュー数が多いアプリでは、多数のHTTPリクエストと長い処理時間が発生します。標準手順として無制限取得を推奨するものではありません。まずreviews()を少量実行し、保存形式、途中保存、失敗時の再開単位を決めたうえで、必要な場合だけ使ってください。

返却される主な項目

キー内容注意点
reviewIdレビュー識別子重複確認に利用
contentレビュー本文個人情報・引用範囲に注意
score星評価1〜5の値
thumbsUpCount参考になった数人気や正しさを直接示す値ではない
at投稿日CSV読込時に日時型を確認
reviewCreatedVersion / appVersionアプリバージョン欠損する可能性がある
replyContent / repliedAt運営からの返信返信がない場合は欠損

現行の少量確認では、reviewCreatedVersionに加えてappVersionが返る場合がありました。処理では両方の存在を確認し、欠損を許容してください。

前の少量確認と以下の実用版は代替手順です。続けて両方を実行する必要はありません。実用版は最大3回のライブラリ呼び出し・保存300行を上限にします。1呼び出しの内部HTTP回数は観測しておらず、3 HTTPリクエストという意味ではありません。内部処理が戻らない場合は60秒で打ち切り、直前までの返却済み結果を保存します。

取得条件と終了理由を残す：collect_gp_reviews.py

掲載コードを保存：collect_gp_reviews.py

"""少量の基本例とは別に実行する、上限付きレビュー取得。"""
from datetime import datetime, timezone
import json
import time
import multiprocessing
import pandas as pd
from google_play_scraper import Sort, reviews
from gp_common import Run

APP_ID = "com.nianticlabs.pokemongo"
LANG, COUNTRY = "ja", "jp"
MAX_CALLS, MAX_ROWS, PAGE_SIZE = 3, 300, 100

def _worker(pipe, args, kwargs):
    try:
        pipe.send((True, reviews(*args, **kwargs)))
    except Exception as exc:
        pipe.send((False, type(exc).__name__ + ": " + str(exc)))
    finally:
        pipe.close()

def bounded_reviews(*args, **kwargs):
    # 上流内部の通信・再試行で戻らない場合も、1呼び出し60秒で打ち切る。
    context = multiprocessing.get_context("spawn")
    parent, child = context.Pipe(duplex=False)
    process = context.Process(target=_worker, args=(child, args, kwargs))
    process.start(); child.close()
    try:
        if not parent.poll(60):
            raise TimeoutError("ライブラリ呼び出しが60秒以内に完了しませんでした。内部の途中行は取得できません。")
        success, result = parent.recv()
        if not success:
            raise RuntimeError(result)
        return result
    finally:
        if process.is_alive():
            process.terminate()
        process.join(); parent.close()

def collect(app_id=APP_ID, fetch=bounded_reviews, max_calls=MAX_CALLS, max_rows=MAX_ROWS,
            page_size=PAGE_SIZE, root="outputs/gp_runs"):
    if min(max_calls, max_rows, page_size) < 1:
        raise ValueError("上限は1以上にしてください。")
    settings = dict(app_id=app_id, lang=LANG, country=COUNTRY, sort="NEWEST",
                    filter_score_with=None, filter_device_with=None,
                    max_library_calls=max_calls, requested_saved_rows=max_rows,
                    page_size=page_size, library_call_timeout_seconds=60, conflict_policy="first_seen",
                    missing_id_policy="retain_each_row")
    run = Run("collection", settings=settings, root=root)
    rows, seen_ids, seen_tokens = [], {}, set()
    calls = returned = duplicates = conflicts = omitted_by_limit = 0
    token = None
    reason, error = "call_limit", None
    try:
        for _ in range(max_calls):
            calls += 1
            # 継続時のcountはライブラリが元tokenから復元する。上限超過分を保存しない。
            batch, next_token = fetch(app_id, lang=LANG, country=COUNTRY,
                                      sort=Sort.NEWEST, count=min(page_size, max_rows),
                                      filter_score_with=None, filter_device_with=None,
                                      continuation_token=token)
            returned += len(batch)
            old_count = len(rows)
            for raw in batch:
                keep = ["reviewId", "content", "score", "thumbsUpCount", "at", "reviewCreatedVersion", "appVersion", "replyContent", "repliedAt"]
                item = {k: raw.get(k) for k in keep}
                rid = "" if item.get("reviewId") is None else str(item["reviewId"])
                item["reviewId"] = rid
                # この版はfromtimestampによる実行機のローカルnaive日時を返す。
                for field in ["at", "repliedAt"]:
                    v = item.get(field)
                    item[field + "_raw"] = v.isoformat() if isinstance(v, datetime) else v
                    if isinstance(v, datetime):
                        item[field] = v.astimezone().isoformat()
                signature = json.dumps(item, sort_keys=True, ensure_ascii=False, default=str)
                if rid and rid in seen_ids:
                    if signature == seen_ids[rid]:
                        duplicates += 1
                    else:
                        conflicts += 1
                    continue
                if len(rows) >= max_rows:
                    omitted_by_limit += 1
                    continue
                if rid:
                    seen_ids[rid] = signature
                rows.append(item)
            value = getattr(next_token, "token", None)
            if not batch:
                reason = "empty_response_end_or_upstream_failure_unknown"
                break
            if not value:
                reason = "no_continuation_end_or_upstream_failure_unknown"
                break
            if len(rows) >= max_rows:
                reason = "saved_row_limit"
                break
            if repr(value) in seen_tokens:
                reason = "repeated_continuation"
                break
            if len(rows) == old_count:
                reason = "no_new_review"
                break
            seen_tokens.add(repr(value))
            token = next_token
    except KeyboardInterrupt:
        reason = "user_interrupted_partial"
    except Exception as exc:
        reason, error = "exception_partial", type(exc).__name__ + ": " + str(exc)
    cols = ["reviewId", "content", "score", "at", "reviewCreatedVersion", "appVersion"]
    frame = pd.DataFrame(rows)
    for col in cols:
        if col not in frame:
            frame[col] = pd.Series(dtype="str")
    run.csv("google_play_reviews.csv", frame)
    return run.finish(returned_rows=returned, saved_rows=len(frame),
                      duplicate_rows=duplicates, conflict_rows=conflicts,
                      missing_id_rows=int(frame["reviewId"].eq("").sum()),
                      omitted_by_saved_limit=omitted_by_limit,
                      library_calls=calls, internal_http_requests="not_observable",
                      end_reason=reason, error=error, all_reviews_complete="not_verified",
                      result_scope="bounded_or_partial", system_timezone_names=list(time.tzname),
                      date_policy="scraper 1.2.7 local naive -> system local offset at that date; original retained in *_raw",
                      continuation_token_persisted=False)

if __name__ == "__main__":
    collect()

保存先は表示された outputs/gp_runs/collection-…/ です。CSVとmetadata.jsonを一組で保管します。0件でも列と条件を保存します。同じID・同じ内容は重複、同じID・異なる内容は競合として別に数え、最初の内容を採用します。ID欠損行は各行を残します。ID欠損の同一性までは判定できません。

google-play-scraper 1.2.7では継続オブジェクトが返っても内部tokenが空の場合があります。また通信例外を内部で処理して途中結果を返すため、no_continuationやempty_responseだけでは末尾到達と通信失敗を区別できません。all_reviews_completeはnot_verifiedのままです。上限・重複・中断・例外等の終了理由を確認し、全件取得と断定しないでください。継続tokenを長期保存して別実行から再開する例ではありません。

この版のat・repliedAtはdatetime.fromtimestampによる実行機のローカルnaive日時です。実用版ではその環境のオフセットを付け、元の値も*_rawに残します。取得時刻はmetadataのstarted_at_utc／finished_at_utcで、レビュー投稿日とは別です。ライブラリのレビュー取得実装と日時・バージョン抽出実装を確認しています。

保存結果を確認して分析用の入力に選ぶ

まずmetadataのsaved_rows・終了理由・CSV hashを確認します。分析へ進める範囲と判断したCSVだけを標準の入力へコピーします。以下は取得処理ではありません。既存の標準CSVがある場合は停止するので、必要なら先に別名へ退避してください。元の実行フォルダーは保持します。

from pathlib import Path
import shutil
from gp_common import checked_run
folder, metadata = checked_run("outputs/gp_runs/collection-実際の実行名")
print(metadata["saved_rows"], metadata["end_reason"])
target = Path("outputs/google_play_reviews.csv")
if target.exists():
    raise SystemExit("既存CSVがあります。確認して別名へ退避してから実行してください。")
target.parent.mkdir(parents=True, exist_ok=True)
shutil.copy2(folder / "google_play_reviews.csv", target)
shutil.copy2(folder / "metadata.json", target.with_suffix(".metadata.json"))
print(target.resolve())

保存CSVを件数・評価・バージョン別に集計する。本文の公開・引用では個人情報と引用範囲に注意し、不要なユーザー情報を掲載しないでください。

-Google Playデータ取得, データ取得