"""少量の基本例とは別に実行する、上限付きレビュー取得。"""
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()