#!/usr/bin/env python3 """Validate an exported Actor dataset before it enters a report. Usage: python tools/validate_output.py output.json The input may be a JSON array, a single object, or an Apify ``items`` wrapper. """ from __future__ import annotations import argparse import json import sys from collections import Counter from math import isclose from datetime import datetime from pathlib import Path SUMMARY_REQUIRED = {"type", "appId", "country", "reviewsFetched", "fetchedAverageRating", "ratingDistribution", "complaintThemes"} REVIEW_REQUIRED = {"type", "appId", "country", "reviewId", "rating", "text", "date"} def load_items(path: Path) -> list[dict]: data = json.loads(path.read_text(encoding="utf-8")) if isinstance(data, dict) and "items" in data: if not isinstance(data["items"], list): raise ValueError("items must be a JSON array") data = data["items"] if isinstance(data, dict): data = [data] if not isinstance(data, list) or not all(isinstance(x, dict) for x in data): raise ValueError("expected a JSON object, array, or {\"items\": [...]} wrapper") return data def _count(value: object) -> bool: return isinstance(value, int) and not isinstance(value, bool) and value >= 0 def _identifier(value: object) -> bool: return (isinstance(value, str) and bool(value.strip())) or ( isinstance(value, int) and not isinstance(value, bool) and value > 0) def _scope(row: dict) -> tuple[str, str] | None: app_id, country = row.get("appId"), row.get("country") if _identifier(app_id) and isinstance(country, str) and country.strip(): return str(app_id), country return None def validate(items: list[dict], *, require_reviews: bool = False) -> list[str]: errors: list[str] = [] summaries = [x for x in items if x.get("type") == "summary"] reviews = [x for x in items if x.get("type") == "review"] unknown = [x.get("type") for x in items if x.get("type") not in ("summary", "review")] if unknown: errors.append(f"unknown type(s): {sorted(set(map(str, unknown)))}") if not summaries: errors.append("no summary row found") for kind, rows, required in (("summary", summaries, SUMMARY_REQUIRED), ("review", reviews, REVIEW_REQUIRED)): for i, row in enumerate(rows, 1): missing = sorted(required - row.keys()) if missing: errors.append(f"{kind} #{i}: missing {', '.join(missing)}") if _scope(row) is None: errors.append(f"{kind} #{i}: appId must be a non-empty ID and country a non-empty string") for i, row in enumerate(summaries, 1): dist = row.get("ratingDistribution") valid_dist = isinstance(dist, dict) and set(dist) == {str(k) for k in range(1, 6)} if not valid_dist: errors.append(f"summary #{i}: ratingDistribution must be an object with exactly keys 1..5") elif not all(_count(v) for v in dist.values()): errors.append(f"summary #{i}: ratingDistribution values must be non-negative integers") valid_dist = False fetched = row.get("reviewsFetched") if not _count(fetched): errors.append(f"summary #{i}: reviewsFetched must be a non-negative integer") elif fetched == 0: errors.append(f"summary #{i}: zero reviews; check app ID and country feed before interpreting") if valid_dist and _count(fetched) and sum(dist.values()) != fetched: errors.append(f"summary #{i}: ratingDistribution total {sum(dist.values())} != reviewsFetched {fetched}") average = row.get("fetchedAverageRating") if not isinstance(average, (int, float)) or isinstance(average, bool) or not 1 <= average <= 5: # Empty Actor summaries legitimately use null; the zero-review diagnostic above # already explains why such a report must not be interpreted. if not (_count(fetched) and fetched == 0 and average is None): errors.append(f"summary #{i}: fetchedAverageRating must be between 1 and 5") if (valid_dist and _count(fetched) and fetched > 0 and sum(dist.values()) == fetched and isinstance(average, (int, float)) and not isinstance(average, bool) and 1 <= average <= 5): expected_average = round(sum(int(star) * count for star, count in dist.items()) / fetched, 3) if not isclose(average, expected_average, rel_tol=0, abs_tol=0.0005): errors.append(f"summary #{i}: fetchedAverageRating {average} != ratingDistribution weighted average {expected_average} (3 decimals)") themes = row.get("complaintThemes") if not isinstance(themes, list): errors.append(f"summary #{i}: complaintThemes must be an array") continue for j, theme in enumerate(themes, 1): if not isinstance(theme, dict): errors.append(f"summary #{i}: theme #{j} must be an object") continue if not isinstance(theme.get("theme"), str) or not theme["theme"].strip(): errors.append(f"summary #{i}: theme #{j} must have a non-empty theme name") if not isinstance(theme.get("example"), str) or not theme["example"].strip(): errors.append(f"summary #{i}: theme #{j} has no evidence example") for i, row in enumerate(reviews, 1): rating = row.get("rating") if not _count(rating) or not 1 <= rating <= 5: errors.append(f"review #{i}: rating must be an integer 1..5") if not _identifier(row.get("reviewId")): errors.append(f"review #{i}: reviewId must be a non-empty ID") if not isinstance(row.get("text"), str): errors.append(f"review #{i}: text must be a string") date_value = row.get("date") if not isinstance(date_value, str) or not date_value.strip(): errors.append(f"review #{i}: date must be an ISO-8601 string") else: try: datetime.fromisoformat(date_value.replace("Z", "+00:00")) except ValueError: errors.append(f"review #{i}: date must be ISO-8601") # Ignore invalid scopes here: field diagnostics above already cover them. # Normalize numeric IDs to strings so concatenating API exports cannot bypass # duplicate checks merely by mixing numeric and string ID representations. summary_scopes = [_scope(x) for x in summaries if _scope(x) is not None] review_scopes = {_scope(x) for x in reviews if _scope(x) is not None} missing_summary_scopes = sorted(review_scopes - set(summary_scopes)) if missing_summary_scopes: errors.append(f"review scope without summary: {missing_summary_scopes}") if len(set(summary_scopes)) != len(summary_scopes): errors.append("duplicate summary scope: appId+country must be unique") seen_review_ids: set[tuple[str, str, str]] = set() for i, row in enumerate(reviews, 1): scope = _scope(row) review_id = row.get("reviewId") if scope is None or not _identifier(review_id): continue key = (*scope, str(review_id)) if key in seen_review_ids: errors.append(f"review #{i}: duplicate reviewId within appId+country") seen_review_ids.add(key) if require_reviews: # Opt-in: a filtered/summary-only export need not include all review rows. # Compare each scope independently so matching global totals cannot hide # missing rows in one storefront and excess rows in another. review_counts: Counter = Counter() review_ratings: dict[tuple[str, str], Counter] = {} for row in reviews: scope = _scope(row) if scope is None: continue review_counts[scope] += 1 rating = row.get("rating") if _count(rating) and 1 <= rating <= 5: review_ratings.setdefault(scope, Counter())[str(rating)] += 1 for i, row in enumerate(summaries, 1): scope = _scope(row) fetched = row.get("reviewsFetched") if scope is None or not _count(fetched): continue if review_counts[scope] != fetched: errors.append(f"summary #{i}: complete export requires {fetched} review rows for {scope}, found {review_counts[scope]}") dist = row.get("ratingDistribution") if (isinstance(dist, dict) and set(dist) == {str(k) for k in range(1, 6)} and all(_count(v) for v in dist.values())): observed = {str(k): review_ratings.get(scope, Counter())[str(k)] for k in range(1, 6)} if observed != dist: errors.append(f"summary #{i}: review-row rating distribution does not match summary for {scope}") return errors def main() -> int: ap = argparse.ArgumentParser(description="Check App Store Review Insights JSON before reporting") ap.add_argument("json_file", type=Path) ap.add_argument("--json", dest="json_output", action="store_true", help="emit one machine-readable result object") ap.add_argument("--require-reviews", action="store_true", help="require all review rows per app/country and matching star counts (not for filtered or summary-only exports)") args = ap.parse_args() try: items = load_items(args.json_file) except (OSError, ValueError, json.JSONDecodeError) as exc: if args.json_output: print(json.dumps({"ok": False, "errors": [str(exc)]}, ensure_ascii=False)) else: print(f"ERROR: {exc}") return 2 errors = validate(items, require_reviews=args.require_reviews) summary_count = len([x for x in items if x.get("type") == "summary"]) review_count = len([x for x in items if x.get("type") == "review"]) if args.json_output: print(json.dumps({"ok": not errors, "summaryRows": summary_count, "reviewRows": review_count, "errors": errors}, ensure_ascii=False, separators=(",", ":"))) return 1 if errors else 0 if errors: print("FAIL") for error in errors: print(f"- {error}") return 1 print(f"PASS: {summary_count} summary row(s), {review_count} review row(s); evidence fields present") return 0 if __name__ == "__main__": raise SystemExit(main())