#!/usr/bin/env python3
"""Reproduce The PenTest Index's calculations from the adjacent final CSVs.

Run: python3 reproduce.py
Uses only the Python standard library. It writes calculation-results.json.
Source-reported survey results and forecasts remain attributed to their sources;
this script verifies this compilation's arithmetic, not proprietary source models.
"""
from pathlib import Path
from decimal import Decimal, ROUND_HALF_UP
from collections import Counter
import csv
import json
import math
import hashlib
import datetime

HERE = Path(__file__).resolve().parent
DATE = "2026-10-08"


def read(name):
    with (HERE / name).open(encoding="utf-8-sig", newline="") as f:
        return list(csv.DictReader(f))


def values(rows, key):
    return [Decimal(r[key]) for r in rows if r[key] != ""]


def percentile(data, p):
    ordered = sorted(data)
    rank = (len(ordered) - 1) * Decimal(str(p))
    lower = int(rank)
    if lower == len(ordered) - 1:
        return ordered[lower]
    return ordered[lower] + (ordered[lower + 1] - ordered[lower]) * (rank - lower)


def display(number, places=2):
    return str(number.quantize(Decimal(10) ** -places, rounding=ROUND_HALF_UP))


def summary(data):
    return {
        "records": len(data),
        "minimum": str(min(data)),
        "q1_raw": str(percentile(data, .25)),
        "median_raw": str(percentile(data, .5)),
        "q3_raw": str(percentile(data, .75)),
        "maximum": str(max(data)),
        "mean_raw": str(sum(data) / len(data)),
        "display_q1": display(percentile(data, .25)),
        "display_median": display(percentile(data, .5)),
        "display_q3": display(percentile(data, .75)),
        "display_mean": display(sum(data) / len(data)),
    }


stats = read("penetration-testing-statistics-2026.csv")
market = read("pentest-market-size-estimates-2026.csv")
rules = read("pentest-rule-tracker-2026.csv")
countries = read("eurostat-ict-security-tests-by-country-2024.csv")
providers = read("pentest-published-price-check-2026-10-08.csv")
offers = read("pentest-published-offers-2026-10-08.csv")
rates = read("gsa-penetration-tester-ceiling-rates-2026-10-08.csv")
directory = read("gsa-pentest-contractor-listings-2026-10-08.csv")
popular = read("pentest-popular-numbers-checked-2026.csv")
benchmarks = read("pentest-fix-time-and-closure-benchmarks-2026.csv")

assert len(stats) == 100
assert [int(r["id"]) for r in stats] == list(range(1, 101))
assert len({r["anchor"] for r in stats}) == 100
assert len(market) == 35 and len(rules) == 22 and len(countries) == 35
assert len(providers) == 43 and len(offers) == 54 and len(benchmarks) == 73
assert len(rates) == 417 and len(directory) == 609

included25 = [r for r in market if r["counted_in_2025_middle_estimate"] == "Yes"]
included26 = [r for r in market if r["counted_in_2026_middle_estimate"] == "Yes"]
m25 = values(included25, "value_2025_usd_billion")
m26 = values(included26, "value_2026_usd_billion")
cagrs = values(included25, "yearly_growth_cagr_percent")
ptaas = values([r for r in market if r["market"] == "PTaaS"], "value_2026_usd_billion")
cognitive = next(r for r in market if r["firm"] == "Cognitive Market Research")
rate = percentile(cagrs, .5) / 100
assert percentile(m25, .5) == Decimal("2.4")
assert percentile(m26, .5) == Decimal("2.8036")
assert percentile(cagrs, .5) == Decimal("15.29")

rate_values = values(rates, "ceiling_rate_usd_per_hour")
assert all("penetration" in r["labor_category"].lower() and "tester" in r["labor_category"].lower() for r in rates)
assert len({r["gsa_record_id"] for r in rates}) == len(rates)
bands = {}
for label, low, high in [("0–2", 0, 2), ("3–5", 3, 5), ("6–9", 6, 9), ("10+", 10, 1000)]:
    selected = [r for r in rates if low <= int(r["min_years_experience"]) <= high]
    bands[label] = summary(values(selected, "ceiling_rate_usd_per_hour"))
assert sum(b["records"] for b in bands.values()) == 417
assert percentile(rate_values, .5) == Decimal("157.526")

usd_non_ai = [r for r in offers if r["counted_in_usd_comparison"].startswith("Yes") and r["sold_as_ai_run"] == "No"]
usd_ai = [r for r in offers if r["counted_in_usd_comparison"].startswith("Yes") and r["sold_as_ai_run"] == "Yes"]
assert len(usd_non_ai) == 10 and len({r["provider"] for r in usd_non_ai}) == 2
assert len(usd_ai) == 2 and len({r["provider"] for r in usd_ai}) == 2
assert all(r["currency"] == "USD" for r in usd_non_ai + usd_ai)
printed_non_ai = [r for r in offers if r["counted_in_range"] == "Not AI-run range"]
printed_ai = [r for r in offers if r["counted_in_range"] == "AI-run range"]
assert len(printed_non_ai) == 30 and len(printed_ai) == 4

fix_times = values([r for r in benchmarks if r["table"] == "fix_time"], "value")
test_rates = {r["segment"]: Decimal(r["value"]) if r["value"] else None for r in benchmarks if r["table"] == "resolved_by_test_type"}
half_lives = {r["segment"]: Decimal(r["value"]) for r in benchmarks if r["table"] == "half_life_by_industry"}
assert len(fix_times) == 6 and len(test_rates) == 8 and len(half_lives) == 10
assert test_rates["External network"] is None
assert len([v for v in test_rates.values() if v is not None]) == 7
rule_counts = Counter(r["counted_as"] for r in rules if r["counted_as"] != "proposed")
assert rule_counts == {"requires": 8, "example": 8, "not_required": 4, "suspended": 1}
assert sum(rule_counts.values()) == 21

numeric_country_fields = [k for k in countries[0] if k.startswith("pct_")]
country_numeric_cells = sum(r[k] != "" for r in countries for k in numeric_country_fields)
assert country_numeric_cells == 206
assert len(countries) * len(numeric_country_fields) - country_numeric_cells == 4
eu = next(r for r in countries if r["geo_code"] == "EU27_2020")
coverage = [int(x.strip()) for x in stats[9]["value"].split(";")]
assert coverage == [10, 21, 38, 60]

def selected_offer(row):
    return {k: row[k] for k in ["provider", "offer", "amount", "currency", "scope_or_unit", "url"]}

results = {
    "checked_date": DATE,
    "scope": "Reproducible arithmetic for this selected compilation; not a validation of proprietary forecasts, a representative buyer-price survey, or a new experiment.",
    "files": {p.name: {"rows": len(read(p.name)), "sha256": hashlib.sha256(p.read_bytes()).hexdigest()} for p in sorted(HERE.glob("*.csv"))},
    "statistics": {"count": len(stats), "topics": dict(Counter(r["topic"] for r in stats)), "unique_anchors": len({r["anchor"] for r in stats})},
    "market": {
        "2025": summary(m25), "2026": summary(m26),
        "2025_high_low_ratio": str(max(m25) / min(m25)),
        "same_eight_firms_2025_median": str(percentile(values(included26, "value_2025_usd_billion"), .5)),
        "median_cagr_percent": str(percentile(cagrs, .5)),
        "doubling_years_at_median_cagr": math.log(2) / math.log(1 + float(rate)),
        "illustrative_2026_at_median_growth": str(percentile(m25, .5) * (1 + rate)),
        "2025_median_with_cognitive": str(percentile(m25 + [Decimal(cognitive["value_2025_usd_billion"])], .5)),
        "ptaas_2026": summary(ptaas), "ptaas_high_low_ratio": str(max(ptaas) / min(ptaas)),
        "360iresearch_endpoint_implied_cagr_percent": ((476.35 / 165.46) ** (1 / 6) - 1) * 100,
    },
    "gsa": {"all": summary(rate_values), "experience_bands": bands,
        "vendor_name_count": len({r["vendor_name"] for r in rates}),
        "contract_count": len({r["contract_number"] for r in rates}),
        "distinct_title_count": len({r["labor_category"] for r in rates}),
        "data_timestamps": sorted({r["gsa_data_timestamp"] for r in rates}),
        "forty_hour_illustration_raw_usd": str(percentile(rate_values, .5) * 40),
        "directory_contract_rows": len(directory),
        "directory_distinct_contractor_names": len({r["contractor_name"] for r in directory})},
    "prices": {
        "disclosure": dict(Counter(r["price_shown"] for r in providers)),
        "current_page_price_yes_no": dict(Counter(r["price_on_pricing_or_service_page"] for r in providers)),
        "no_current_price_percent": 28 / 43 * 100,
        "current_price_percent": 15 / 43 * 100,
        "price_only_elsewhere_percent": 7 / 43 * 100,
        "no_price_in_checked_pages_percent": 21 / 43 * 100,
        "usd_non_ai": {"records": len(usd_non_ai), "providers": len({r["provider"] for r in usd_non_ai}),
            "minimum_starting_price": str(min(values(usd_non_ai, "amount"))),
            "maximum_starting_price": str(max(values(usd_non_ai, "amount"))),
            "rows": [selected_offer(r) for r in usd_non_ai]},
        "usd_ai": {"records": len(usd_ai), "rows": [selected_offer(r) for r in usd_ai]},
        "printed_dollar_non_ai": {"records": len(printed_non_ai), "providers": len({r["provider"] for r in printed_non_ai}), "warning": "Printed numerals are not a single-currency comparison."},
        "printed_dollar_ai": {"records": len(printed_ai), "providers": len({r["provider"] for r in printed_ai}), "warning": "Excludes Aikido's broad scope-based range and Synack's required-platform fee."}},
    "timing": {"human_time_or_effort_records": sum(r["states_how_long_a_test_by_people_takes"].startswith("Yes") for r in providers),
        "ai_numeric_records": sum(r["states_how_fast_an_ai_run_test_returns_results"] == "Yes, with a number" for r in providers),
        "ai_groups": dict(Counter(r["ai_time_group"] for r in providers)),
        "milestone_limit": "First reports, initial results, completed tests, elapsed time, effort days and coverage windows must not be treated as one identical duration."},
    "rules": {"active_tracker_groups": dict(rule_counts), "proposal_rows": 1, "requires_default_intervals": {"annual": 5, "three_year": 2, "organization_defined": 1}, "limit": "Intervals are coded from source clauses, with exceptions and the suspended rollout separately retained."},
    "official_surveys": {"uk_size_percentages": coverage, "large_micro_ratio": coverage[-1] / coverage[0],
        "eurostat_numeric_cells": country_numeric_cells, "eurostat_missing_cells": 4, "eurostat_cells_checked": 210,
        "eu_aggregate": {k: eu[k] for k in numeric_country_fields}},
    "benchmarks": {"fix_values_days": [str(x) for x in fix_times],
        "minimum_days_divided_by_14": str(min(fix_times) / 14), "maximum_days_divided_by_14": str(max(fix_times) / 14),
        "march_1_plus_38_days": str(datetime.date(2026, 3, 1) + datetime.timedelta(days=38)),
        "march_1_plus_58_days": str(datetime.date(2026, 3, 1) + datetime.timedelta(days=58)),
        "high_risk_resolution_rates_percent": {k: str(v) if v is not None else None for k, v in test_rates.items()},
        "unresolved_complements_percent": {k: str(100 - v) if v is not None else None for k, v in test_rates.items()},
        "api_ai_gap_percentage_points": str(test_rates["API"] - test_rates["AI and LLM application"]),
        "industry_half_lives_days": {k: str(v) for k, v in half_lives.items()},
        "industry_half_life_spread_days": str(max(half_lives.values()) - min(half_lives.values())),
        "slow_fast_half_life_ratio": 249 / 10, "slow_fast_half_life_difference_days": 249 - 10},
    "source_checks": {"popular_claims": len(popular), "verdicts": dict(Counter(r["verdict"] for r in popular)), "traced_origins": 10, "untraced_claims": 2},
    "context_math": {"cve_2025_growth_percent": (48244 / 40077 - 1) * 100, "ibm_days_183_plus_64": 183 + 64, "fortra_at_least_yearly_percent": 100 - 17},
}
target = HERE / "calculation-results.json"
target.write_text(json.dumps(results, indent=2, ensure_ascii=False) + "\n", encoding="utf-8")
print(f"Reproduced all compilation checks. Wrote {target.name}.")