"""Run with Python 3. Synthetic campaign reporting and replay on local SQLite."""

import hashlib
import json
import sqlite3
from contextlib import closing


EXPECTED_ACCOUNTS = {("google_ads", "g-1"), ("meta_ads", "m-1")}


def publish(db, day, revision, completed_accounts, detail, approved_report):
    if completed_accounts != EXPECTED_ACCOUNTS:
        raise ValueError("Account coverage incomplete")
    if any(row[:2] not in completed_accounts for row in detail):
        raise ValueError("Unexpected source account")

    payload = [sorted(completed_accounts), sorted(detail), sorted(approved_report)]
    digest = hashlib.sha256(json.dumps(payload).encode()).hexdigest()
    with db:
        db.execute("BEGIN IMMEDIATE")
        current = db.execute(
            "SELECT revision, digest FROM release WHERE day = ?", (day,)
        ).fetchone()
        if current:
            if revision < current[0]:
                raise ValueError("Stale revision")
            if revision == current[0]:
                if digest != current[1]:
                    raise ValueError("Revision contents changed")
                return "receipt reused"

        db.execute("DELETE FROM detail")
        try:
            db.executemany("INSERT INTO detail VALUES (?, ?, ?, ?, ?, ?, ?, ?)", detail)
        except sqlite3.IntegrityError as error:
            raise ValueError("Invalid or duplicate detail key") from error
        report = db.execute(
            """
            SELECT source, account, campaign, currency,
                   SUM(spend), SUM(impressions), SUM(completions)
            FROM detail
            GROUP BY source, account, campaign, currency
            ORDER BY source, account, campaign, currency
            """
        ).fetchall()
        if report != sorted(approved_report):
            raise ValueError("Report differs from approved controls")

        db.execute("DELETE FROM report WHERE day = ?", (day,))
        db.executemany(
            "INSERT INTO report VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
            [(day, *row) for row in report],
        )
        db.execute(
            "INSERT INTO release VALUES (?, ?, ?) "
            "ON CONFLICT(day) DO UPDATE SET revision = excluded.revision, digest = excluded.digest",
            (day, revision, digest),
        )
    return "published"


def expect_hold(reason, *args):
    db = args[0]
    before = list(db.iterdump())
    try:
        publish(*args)
    except ValueError as error:
        assert str(error) == reason, error
    else:
        raise AssertionError(f"Expected hold: {reason}")
    assert list(db.iterdump()) == before, "A held batch changed the approved state"
    print(f"Held: {reason}")


def expect_result(expected, *args):
    result = publish(*args)
    assert result == expected, result


def main():
    day = "2026-01-12"
    next_day = "2026-01-13"
    detail = [
        ("google_ads", "g-1", "g-c1", "USD", "desktop", 4_000, 100, 50),
        ("google_ads", "g-1", "g-c1", "USD", "mobile", 6_000, 900, 90),
        ("meta_ads", "m-1", "m-c1", "USD", "mobile", 20_000, 500, 100),
    ]
    approved = [
        ("google_ads", "g-1", "g-c1", "USD", 10_000, 1_000, 140),
        ("meta_ads", "m-1", "m-c1", "USD", 20_000, 500, 100),
    ]
    corrected_detail = [(*detail[0][:5], 6_000, *detail[0][6:]), *detail[1:]]
    corrected_report = [(*approved[0][:4], 12_000, *approved[0][5:]), approved[1]]
    next_detail = [("google_ads", "g-1", "g-c1", "USD", "desktop", 9_000, 50, 5)]
    next_report = [("google_ads", "g-1", "g-c1", "USD", 9_000, 50, 5)]

    with closing(sqlite3.connect(":memory:")) as db:
        db.executescript(
            """
            CREATE TEMP TABLE detail (
              source TEXT NOT NULL, account TEXT NOT NULL, campaign TEXT NOT NULL,
              currency TEXT NOT NULL, device TEXT NOT NULL,
              spend INTEGER NOT NULL CHECK (spend >= 0),
              impressions INTEGER NOT NULL CHECK (impressions >= 0),
              completions INTEGER NOT NULL CHECK (completions BETWEEN 0 AND impressions),
              PRIMARY KEY (source, account, campaign, currency, device)
            );
            CREATE TABLE report (
              day TEXT, source TEXT, account TEXT, campaign TEXT, currency TEXT,
              spend INTEGER, impressions INTEGER, completions INTEGER,
              PRIMARY KEY (day, source, account, campaign, currency)
            );
            CREATE TABLE release (day TEXT PRIMARY KEY, revision INTEGER, digest TEXT);
            """
        )
        expect_result("published", db, day, 1, EXPECTED_ACCOUNTS, detail, approved)
        expect_result("published", db, next_day, 1, EXPECTED_ACCOUNTS, next_detail, next_report)
        count, spend = db.execute(
            "SELECT COUNT(*), SUM(spend) FROM report WHERE day = ?", (day,)
        ).fetchone()
        completions, impressions = db.execute(
            "SELECT completions, impressions FROM report WHERE day = ? AND source = 'google_ads'",
            (day,),
        ).fetchone()
        print(f"Baseline: {len(detail)} detail rows matched {count} report rows and {spend:,} USD cents.")
        print(f"Google campaign completion rate: {completions:,} / {impressions:,} = {completions / impressions:.0%}.")

        expect_hold("Account coverage incomplete", db, day, 2, {("google_ads", "g-1")}, detail[:2], approved)
        expect_hold("Invalid or duplicate detail key", db, day, 2, EXPECTED_ACCOUNTS, detail + detail[:1], approved)
        expect_hold("Report differs from approved controls", db, day, 2, EXPECTED_ACCOUNTS, corrected_detail, approved)
        expect_result("published", db, day, 2, EXPECTED_ACCOUNTS, corrected_detail, corrected_report)
        expect_result("receipt reused", db, day, 2, EXPECTED_ACCOUNTS, corrected_detail, corrected_report)
        expect_hold("Revision contents changed", db, day, 2, EXPECTED_ACCOUNTS, corrected_detail, approved)
        expect_hold("Revision contents changed", db, day, 2, EXPECTED_ACCOUNTS, detail, approved)
        expect_hold("Stale revision", db, day, 1, EXPECTED_ACCOUNTS, detail, approved)

        assert db.execute("SELECT * FROM report ORDER BY day, source").fetchall() == [
            (day, *corrected_report[0]), (day, *corrected_report[1]), (next_day, *next_report[0]),
        ]
        assert db.execute("SELECT day, revision FROM release ORDER BY day").fetchall() == [(day, 2), (next_day, 1)]
        print("Correction: 32,000 USD cents. Exact replay reused the receipt.")
        print("Adjacent date: 9,000 USD cents. Meta completed with zero rows.")


if __name__ == "__main__":
    main()