#!/usr/bin/env python3
"""Atelier Nymphar SCD2. Données fictives, SQLite en mémoire, aucune dépendance."""
import json
import platform
import sqlite3

db = sqlite3.connect(":memory:")
db.execute("PRAGMA foreign_keys = ON")
db.executescript("""
CREATE TABLE dim_customer (
    customer_key INTEGER PRIMARY KEY,
    customer_id TEXT NOT NULL,
    segment TEXT NOT NULL,
    valid_from TEXT NOT NULL,
    valid_to TEXT,
    UNIQUE(customer_id, valid_from),
    CHECK(valid_to IS NULL OR valid_to > valid_from)
);
CREATE UNIQUE INDEX one_open_version
ON dim_customer(customer_id) WHERE valid_to IS NULL;
CREATE TABLE staging_sales (
    sale_id TEXT PRIMARY KEY,
    customer_id TEXT NOT NULL,
    sold_on TEXT NOT NULL,
    received_on TEXT NOT NULL,
    amount_cents INTEGER NOT NULL CHECK(amount_cents > 0)
);
CREATE TABLE fact_sales (
    sale_id TEXT PRIMARY KEY,
    customer_key INTEGER NOT NULL REFERENCES dim_customer(customer_key),
    sold_on TEXT NOT NULL,
    amount_cents INTEGER NOT NULL
);
INSERT INTO dim_customer VALUES
 (101, 'C001', 'Standard', '2026-01-01', NULL),
 (201, 'C002', 'Standard', '2026-01-01', NULL);
INSERT INTO staging_sales VALUES
 ('V1', 'C001', '2026-05-20', '2026-05-20', 10000),
 ('V2', 'C001', '2026-06-01', '2026-06-01', 6000),
 ('V3', 'C001', '2026-05-28', '2026-06-03', 4000),
 ('V4', 'C002', '2026-06-02', '2026-06-02', 8000);
""")
MATCH = """
SELECT d.customer_key
FROM dim_customer d
WHERE d.customer_id = ?
  AND ? >= d.valid_from
  AND (d.valid_to IS NULL OR ? < d.valid_to)
"""
OVERLAP = """
SELECT a.customer_key, b.customer_key
FROM dim_customer a JOIN dim_customer b
  ON a.customer_id = b.customer_id
 AND a.customer_key < b.customer_key
 AND (b.valid_to IS NULL OR a.valid_from < b.valid_to)
 AND (a.valid_to IS NULL OR b.valid_from < a.valid_to)
ORDER BY a.customer_key, b.customer_key
"""

def match(customer_id, sold_on):
    return [r[0] for r in db.execute(MATCH, (customer_id, sold_on, sold_on))]

def load_sale(sale_id):
    s = db.execute("SELECT * FROM staging_sales WHERE sale_id=?", (sale_id,)).fetchone()
    keys = match(s[1], s[2])
    if len(keys) != 1:
        raise ValueError(f"{sale_id}: {len(keys)} versions, chargement refusé")
    db.execute("INSERT INTO fact_sales VALUES (?, ?, ?, ?)", (s[0], keys[0], s[2], s[4]))

def scalar(sql):
    return db.execute(sql).fetchone()[0]

def require(condition, label):
    if not condition:
        raise RuntimeError(label)

# V1 existe dans les faits avant le changement de segment.
load_sale("V1")
db.commit()
original_v1 = db.execute("SELECT * FROM fact_sales WHERE sale_id='V1'").fetchone()

# Changement ordinaire, daté explicitement, reçu dans l'ordre. Transaction atomique.
# Ceci n'est PAS un chargeur générique de CDC ni de corrections rétroactives.
with db:
    cur = db.execute("""
        UPDATE dim_customer SET valid_to='2026-06-01'
        WHERE customer_id='C001' AND valid_to IS NULL
          AND valid_from < '2026-06-01'
    """)
    require(cur.rowcount == 1, "Une seule version précédente doit être clôturée")
    db.execute("""
        INSERT INTO dim_customer VALUES
        (102, 'C001', 'Premium', '2026-06-01', NULL)
    """)
require(db.execute("SELECT * FROM fact_sales WHERE sale_id='V1'").fetchone() == original_v1,
        "Le changement ordinaire ne doit pas modifier V1")

for sale_id in ["V2", "V4", "V3"]:  # ordre de réception, V3 arrive en retard
    load_sale(sale_id)
db.commit()

historical = dict(db.execute("""
SELECT d.segment, SUM(f.amount_cents)
FROM fact_sales f JOIN dim_customer d ON d.customer_key=f.customer_key
GROUP BY d.segment ORDER BY d.segment
"""))
current = dict(db.execute("""
SELECT d.segment, SUM(s.amount_cents)
FROM staging_sales s JOIN dim_customer d
 ON d.customer_id=s.customer_id AND d.valid_to IS NULL
GROUP BY d.segment ORDER BY d.segment
"""))
natural_only = scalar("""
SELECT SUM(s.amount_cents)
FROM staging_sales s JOIN dim_customer d ON d.customer_id=s.customer_id
""")
inclusive = scalar("""
SELECT SUM(s.amount_cents)
FROM staging_sales s JOIN dim_customer d ON d.customer_id=s.customer_id
 AND s.sold_on BETWEEN d.valid_from AND COALESCE(d.valid_to, '9999-12-31')
""")
require(historical == {"Premium": 6000, "Standard": 22000}, "Historique incorrect")
require(current == {"Premium": 20000, "Standard": 8000}, "Vue courante incorrecte")
require(natural_only == 48000 and inclusive == 34000, "Contre-exemples incorrects")
require(match("C001", "2026-06-01") == [102], "Borne du 1er juin")
require(match("C001", "2026-05-28") == [101], "Vente tardive")
require(db.execute(OVERLAP).fetchall() == [], "Chevauchement inattendu")
require(scalar("SELECT COUNT(*) FROM fact_sales") == 4, "Une ligne par vente")
require(scalar("SELECT SUM(amount_cents) FROM fact_sales") == 28000, "Total perdu")

# Deux défauts injectés séparément, sans contaminer le résultat principal.
db.execute("SAVEPOINT no_match")
db.execute("INSERT INTO staging_sales VALUES ('X0','C999','2026-06-02','2026-06-02',100)")
try:
    load_sale("X0")
    raise RuntimeError("La vente sans version a été acceptée")
except ValueError as err:
    missing_error = str(err)
db.execute("ROLLBACK TO no_match")
db.execute("RELEASE no_match")

db.execute("SAVEPOINT overlap")
db.execute("""
INSERT INTO dim_customer VALUES
(103,'C001','Standard','2026-05-01','2026-06-15')
""")
overlap_pairs = db.execute(OVERLAP).fetchall()
require(overlap_pairs == [(101, 103), (102, 103)], "Détection des chevauchements")
db.execute("INSERT INTO staging_sales VALUES ('X2','C001','2026-06-02','2026-06-02',100)")
try:
    load_sale("X2")
    raise RuntimeError("La vente ambiguë a été acceptée")
except ValueError as err:
    overlap_error = str(err)
db.execute("ROLLBACK TO overlap")
db.execute("RELEASE overlap")
require(scalar("SELECT COUNT(*) FROM fact_sales") == 4, "Rejets non isolés")

result = {
    "python": platform.python_version(),
    "sqlite": sqlite3.sqlite_version,
    "data": "fictives, dates civiles ISO au jour, montants en centimes entiers",
    "sales_keys": dict(db.execute("SELECT sale_id,customer_key FROM fact_sales ORDER BY sale_id")),
    "historical_cents": historical,
    "current_segments_cents": current,
    "correct_total_cents": 28000,
    "wrong_business_key_only_total_cents": natural_only,
    "wrong_inclusive_between_total_cents": inclusive,
    "v1_unchanged_after_regular_change": True,
    "injected_overlap_pairs": overlap_pairs,
    "rejected_missing_version": missing_error,
    "rejected_ambiguous_version": overlap_error,
    "status": "PASS"
}
print(json.dumps(result, ensure_ascii=False, indent=2))
db.close()