#!/usr/bin/env python3 """Reproduce the published illustration from its downloadable CSV and JSON. Python 3.9+; standard library only; no network requests or account access. Run in the downloaded asset directory: python3 reproduce-benchmark.py Use --input-dir DIRECTORY to read another local copy. JSON is written to stdout. This checks arithmetic, not collection accuracy, representativeness or causality. """ import argparse import csv import hashlib import json import math import re import statistics import sys from collections import defaultdict from datetime import datetime from pathlib import Path FORMATS = {"Video/Reel", "Carousel", "Image"} COLUMNS = ["sample_id", "post_index", "format", "published_at_utc", "likes", "comments", "visible_interactions"] def utc_timestamp(value): if not isinstance(value, str) or not re.fullmatch(r"\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(?:\.\d+)?Z", value): raise ValueError("Expected an ISO 8601 UTC timestamp ending in Z") return datetime.fromisoformat(value[:-1] + "+00:00") def count(value, label): """Blank is unavailable. Zero is an observed value. Reject invalid counts.""" if value == "": return None if not re.fullmatch(r"(?:0|[1-9][0-9]*)", value): raise ValueError(f"{label}: expected a non-negative whole count or blank") result = int(value) if result > 9007199254740991: raise ValueError(f"{label}: exceeds the source metric's safe integer range") return result def read_posts(filename): with filename.open(newline="", encoding="utf-8-sig") as stream: reader = csv.DictReader(stream) if reader.fieldnames != COLUMNS: raise ValueError("Post CSV columns do not match the published dataset") posts = [] identities = set() for line, row in enumerate(reader, start=2): if None in row or any(value is None for value in row.values()): raise ValueError(f"CSV line {line}: incomplete or extra columns") index = count(row["post_index"], f"CSV line {line} post index") if not row["sample_id"] or index is None or index < 1: raise ValueError(f"CSV line {line}: missing sample or positive post index") identity = (row["sample_id"], index) if identity in identities: raise ValueError(f"CSV line {line}: duplicate sample and post index") identities.add(identity) if row["format"] not in FORMATS: raise ValueError(f"CSV line {line}: unsupported format") utc_timestamp(row["published_at_utc"]) likes = count(row["likes"], f"CSV line {line} likes") comments = count(row["comments"], f"CSV line {line} comments") recorded = count(row["visible_interactions"], f"CSV line {line} interactions") calculated = None if likes is None or comments is None else likes + comments if calculated is not None and calculated > 9007199254740991: raise ValueError(f"CSV line {line}: interaction sum exceeds the source metric's safe integer range") if calculated != recorded: raise ValueError(f"CSV line {line}: visible interactions do not equal the available likes plus comments") posts.append({**row, "post_index": index, "interactions": calculated}) if not posts: raise ValueError("Post CSV is empty") return posts def summarize(posts): values = [post["interactions"] for post in posts if post["interactions"] is not None] return { "count": len(posts), "eligible": len(values), "missing": len(posts) - len(values), "total": sum(values) if values else None, "average": statistics.mean(values) if values else None, "median": statistics.median(values) if values else None, "highest": max(values) if values else None, } def assert_same(actual, expected, label): for key, actual_value in actual.items(): if key not in expected: raise ValueError(f"{label}.{key}: missing from published summary") expected_value = expected[key] same = actual_value == expected_value if isinstance(actual_value, (int, float)) and isinstance(expected_value, (int, float)): same = math.isclose(actual_value, expected_value, rel_tol=1e-12, abs_tol=1e-9) if not same: raise ValueError(f"{label}.{key}: calculated {actual_value!r}, published {expected_value!r}") def reproduce(directory): posts_file = directory / "benchmark-posts-anonymized.csv" summary_file = directory / "benchmark-summary.json" manifest_file = directory / "source-manifest.json" posts = read_posts(posts_file) published = json.loads(summary_file.read_text(encoding="utf-8")) manifest = json.loads(manifest_file.read_text(encoding="utf-8")) sources = manifest["sources"] by_id = {source["id"]: source for source in sources} if len(by_id) != len(sources): raise ValueError("Source manifest contains duplicate sample IDs") groups = defaultdict(list) for post in posts: if post["sample_id"] not in by_id: raise ValueError("Post CSV contains a sample absent from the source manifest") if utc_timestamp(post["published_at_utc"]) > utc_timestamp(by_id[post["sample_id"]]["observedAt"]): raise ValueError("A publication timestamp follows its sample's observation timestamp") groups[post["sample_id"]].append(post) if set(groups) != set(by_id): raise ValueError("Source manifest and CSV sample IDs differ") if len(posts) != published["totalPosts"]: raise ValueError("Total post count differs from the published summary") overall = summarize(posts) assert_same(overall, published["overall"], "overall") formats = {name: summarize([post for post in posts if post["format"] == name]) for name in sorted(FORMATS)} if set(formats) != set(published["overallFormats"]): raise ValueError("Format groups differ from the published summary") for name, summary in formats.items(): assert_same(summary, published["overallFormats"][name], "overallFormats." + name) published_samples = {sample["id"]: sample for sample in published["samples"]} if len(published_samples) != len(published["samples"]) or set(published_samples) != set(groups): raise ValueError("Published summary sample IDs differ or are duplicated") paired = [] account_summaries = [] for sample_id in sorted(groups): rows = groups[sample_id] source = by_id[sample_id] sample = published_samples[sample_id] if source["brand"] != sample["brand"] or source["observedAt"] != sample["observedAt"]: raise ValueError("Source manifest and summary attribution differ") total = summarize(rows) assert_same(total, sample["total"], sample_id) by_format = {name: summarize([row for row in rows if row["format"] == name]) for name in sorted({row["format"] for row in rows})} if set(by_format) != set(sample["byFormat"]): raise ValueError("Sample format groups differ from the published summary") for name, summary in by_format.items(): assert_same(summary, sample["byFormat"][name], sample_id + "." + name) account_summaries.append({"sampleId": sample_id, "brand": source["brand"], "observedAt": source["observedAt"], "total": total, "byFormat": by_format}) reel, carousel = by_format.get("Video/Reel"), by_format.get("Carousel") if reel and carousel and reel["median"] is not None and carousel["median"] is not None: winner = "Tie" if reel["median"] == carousel["median"] else "Video/Reel" if reel["median"] > carousel["median"] else "Carousel" paired.append({"brand": source["brand"], "videoReel": reel, "carousel": carousel, "higherMedian": winner}) published_pairs = {row["brand"]: row for row in published["pairedFormats"]} if len(published_pairs) != len(published["pairedFormats"]) or set(published_pairs) != {row["brand"] for row in paired}: raise ValueError("Paired-format account set differs from the published summary") for row in paired: expected = published_pairs[row["brand"]] assert_same(row["videoReel"], expected["videoReel"], row["brand"] + ".videoReel") assert_same(row["carousel"], expected["carousel"], row["brand"] + ".carousel") if row["higherMedian"] != expected["higherMedian"]: raise ValueError("Paired-format median comparison differs from the published summary") return { "schemaVersion": 1, "verification": "Published interaction arithmetic reproduced from downloaded rows", "sourceMetadataStatus": {"summary": published.get("status"), "manifest": manifest.get("status")}, "metric": "Observed likes plus observed comments; missing inputs remain unavailable", "inputs": [{"file": item.name, "sha256": hashlib.sha256(item.read_bytes()).hexdigest()} for item in [posts_file, summary_file, manifest_file]], "overall": overall, "formats": formats, "samples": account_summaries, "pairedFormats": paired, "limitations": [ "Six selected retained account samples; not a representative Instagram benchmark.", "Accounts have different observation dates, publication windows and post ages.", "Pooled means and medians do not establish a causal effect of post format.", "The existing CSV filename says anonymized, but the public source register identifies each sample's brand; it is not anonymous.", "No private reach, saves, shares, revenue, or follower-normalized rate is reconstructed.", "Cadence is not reproduced: this public CSV does not contain the pinned-post eligibility fields needed for that calculation.", "The manifest's historical capture hashes refer to source snapshots, not hashes of these CSV downloads.", "A successful arithmetic check does not verify collection completeness, original observations, publication approval or permission to reuse third-party material." ], } def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--input-dir", type=Path, default=Path(__file__).resolve().parent, help="Directory containing the three downloaded source files") args = parser.parse_args() try: result = reproduce(args.input_dir) except (OSError, ValueError, KeyError, TypeError) as error: print("Reproduction failed: " + str(error), file=sys.stderr) return 1 print(json.dumps(result, ensure_ascii=False, indent=2, allow_nan=False)) return 0 if __name__ == "__main__": raise SystemExit(main())