Google Play検索で見つかった範囲のアプリを、運営会社別に整理します。市場全体のシェア・売上順位ではなく、検索語・言語・国・取得時点に依存する構成です。
以下のファイルは同じ作業フォルダーへ保存してください。共通入出力の gp_common.py全文と、各節に記載した依存ファイルが必要です。掲載Python全ファイルのZIPも利用できます。ZIPを展開しても取得やモデル学習は自動実行しません。
検索条件と対応表を保存して、別実行で比較する
python -m pip install google-play-scraper pandas plotlygp_common.pyと次のファイルを保存します。searchは候補、appは詳細の取得です。検索語ごとの上限は30件。返却が少ないことだけで全候補を取得したとは判断しません。検索語をむやみに増やさず、目的に必要な少量から確認してください。
gp_publishers.py
import argparse
import time
import pandas as pd
import plotly.express as px
from google_play_scraper import search, app
from gp_common import Run, checked_run
def collect(queries, fetch_search=search, fetch_app=app, root="outputs/gp_runs"):
run = Run("publisher-source", settings={"queries": queries, "lang": "ja", "country": "jp", "n_hits": 30}, root=root)
pairs, states, details = [], [], []
for q in queries:
try:
items = fetch_search(q, lang="ja", country="jp", n_hits=30)
pairs.extend((q, str(i["appId"])) for i in items if i.get("appId"))
states.append((q, "response_nonempty" if items else "empty_response", len(items), ""))
except Exception as exc:
states.append((q, "failed", 0, type(exc).__name__ + ": " + str(exc)))
time.sleep(0.5)
for app_id in sorted({pair[1] for pair in pairs}):
try:
d = fetch_app(app_id, lang="ja", country="jp")
details.append((app_id, "success", d.get("developerId"), d.get("developer"), d.get("score"), d.get("reviews"), d.get("minInstalls"), d.get("genreId"), d.get("genre"), ""))
except Exception as exc:
details.append((app_id, "failed", None, None, None, None, None, None, None, type(exc).__name__ + ": " + str(exc)))
time.sleep(0.5)
run.csv("query_apps.csv", pd.DataFrame(pairs, columns=["query", "appId"]).drop_duplicates())
run.csv("search_status.csv", pd.DataFrame(states, columns=["query", "status", "returned_rows", "error"]))
run.csv("app_details.csv", pd.DataFrame(details, columns=["appId", "status", "developerId", "developer", "score", "reviews", "minInstalls", "genreId", "genre", "error"]))
return run.finish(unique_target_apps=len(details), search_scope="responses only; exhaustive coverage not verified", partial_failure=any(s[1] == "failed" for s in states) or any(d[1] == "failed" for d in details))
def aggregate(folder, root="outputs/gp_runs"):
folder, meta = checked_run(folder)
pairs = pd.read_csv(folder / "query_apps.csv", dtype=str, keep_default_na=False)
d = pd.read_csv(folder / "app_details.csv", dtype=str, keep_default_na=False)
if d.appId.duplicated().any() or set(pairs.appId) != set(d.appId):
raise ValueError("App IDの対応・重複を確認してください。")
good = d[d.status.eq("success")].copy()
for col in ["score", "reviews", "minInstalls"]:
good[col] = pd.to_numeric(good[col], errors="coerce")
good["publisher"] = ["id:" + i if i else "name:" + n if n else "unknown_app:" + a for i, n, a in zip(good.developerId, good.developer, good.appId)]
groups = good.groupby("publisher").agg(app_count=("appId", "nunique"), score_valid=("score", "count"),
score_mean=("score", "mean"), reviews_valid=("reviews", "count"), reviews_sum=("reviews", lambda s: s.sum(min_count=1)),
installs_valid=("minInstalls", "count"), minimum_installs_sum=("minInstalls", lambda s: s.sum(min_count=1))).reset_index()
run = Run("publisher-aggregates", folder / "metadata.json", {"source_run": meta["run_id"], "publisher_identity": "developerId else exact name; unknown kept per app"}, root)
run.csv("publishers.csv", groups)
good["genreId"] = good["genreId"].replace("", "unknown")
good["genre"] = good["genre"].replace("", "unknown")
genres = good.groupby(["publisher", "genreId", "genre"]).agg(app_count=("appId", "nunique")).reset_index()
run.csv("publisher_genres.csv", genres)
if not groups.empty:
run.figure("publishers.html", px.bar(groups, x="app_count", y="publisher", orientation="h", title="検索で取得できた範囲のアプリ数", hover_data=list(groups.columns)))
run.figure("publisher_genres.html", px.bar(genres, x="publisher", y="app_count", color="genre", title="取得できたアプリの運営会社・ストアジャンル別件数", hover_data=["genreId"]))
else:
run.meta["skipped"]["publishers.html"] = "詳細取得成功0件"
return run.finish(query_count=len(meta["settings"]["queries"]), unique_target_apps=len(d), details_success=len(good),
aggregation_apps=int(groups.app_count.sum()), not_market_share=True)
if __name__ == "__main__":
p = argparse.ArgumentParser(); p.add_argument("mode", choices=["collect", "aggregate"]); p.add_argument("values", nargs="+")
a = p.parse_args()
collect(a.values) if a.mode == "collect" else aggregate(a.values[0])python gp_publishers.py collect RPG パズル
# 取得が終わった後、新しいPython実行で保存済みCSVだけを集計
python gp_publishers.py aggregate "outputs/gp_runs/publisher-source-実際の実行名"query_apps.csvは検索語×App IDの多対多表です。同じアプリが複数検索で見つかっても、詳細とアプリ数は一度だけ数えます。search_status.csvで空応答と失敗、app_details.csvで詳細成功と失敗を区別します。0件でも列を保存し、図を省略します。失敗アプリを0点・レビュー0件・インストール0件とは扱わず、全欠損の合計も欠損を維持します。
運営会社はdeveloperId、なければ完全一致のdeveloper名で整理します。表記違い・同名の別会社までは解決していません。両方不明ならunknown_app:App IDとし、別会社を一つにまとめません。metadataには検索語数・対象アプリ数・詳細成功数・集計対象数を保存します。各指標の有効件数もpublishers.csvで確認できます。インストールは表示帯下限の合計で、売上や期間内新規インストールではありません。
従来の分析グラフ(保存済みの例)
以下は以前の取得条件・コードで作成した保存済みの実分析例です。今回の改訂コードで再取得・再集計した結果ではありません。読者のローカルCSVがこの見本へ送信・反映される仕組みではありません。旧見本には外部Plotly CDNへ接続するものがあり、オフライン用の新しい出力HTMLとは異なります。値を比較する際は元の取得範囲を確認してください。
publisher_genres.csvとpublisher_genres.htmlには、詳細情報のストアジャンル別件数を保存します。検索語をジャンルそのものと見なさず、genreId・genreの不明値もunknownとして残します。ジャンル別の母集団も詳細取得成功アプリに限られます。