# Koleksi merchant dari MongoDB (simulator mongomock)
col = db["merchant"]

# (a) Merchant Kuliner di Bandung, punya promo "cashback", rating >= 4,0.
#     Filter pada field array ("promo": "cashback") otomatis berarti "array mengandung elemen tsb".
filter_a = {
    "kategori": "Kuliner",
    "kota": "Bandung",
    "promo": "cashback",
    "rating": {"$gte": 4.0}
}
hasil_a = list(col.find(filter_a, {"_id": 0, "nama": 1, "rating": 1, "promo": 1}))
tabel_merchant = pd.DataFrame(hasil_a).rename(columns={
    "nama": "Nama Merchant",
    "rating": "Rating",
    "promo": "Promo"
})
tampilkan_tabel(
    tabel_merchant,
    f"C1(a) Merchant yang sesuai kriteria ({len(hasil_a)} merchant)"
)

# (b) Jumlah merchant yang memiliki field jam_buka (skema fleksibel).
jml_jam_buka = col.count_documents({"jam_buka": {"$exists": True}})
print("\nC1(b) Merchant dengan field jam_buka:", jml_jam_buka, "dari", col.count_documents({}))

# (c) Aggregation pipeline: jumlah merchant & rata-rata rating per kategori, urut rating tertinggi.
pipeline = [
    {"$group": {"_id": "$kategori",
                "jumlah_merchant": {"$sum": 1},
                "rata_rating": {"$avg": "$rating"}}},
    {"$sort": {"rata_rating": -1}}
]
hasil_c = pd.DataFrame(list(col.aggregate(pipeline))).rename(columns={"_id": "kategori"})
hasil_c["rata_rating"] = hasil_c["rata_rating"].round(3)
tampilkan_tabel(hasil_c, "C1(c) Jumlah merchant dan rata-rata rating per kategori")

# C2(a) - Membaca dan parsing log transaksi menggunakan Spark RDD

# Membuat accumulator untuk menghitung baris rusak
baris_rusak = sc.accumulator(0)

# Membaca file log menggunakan Spark
log_rdd = sc.textFile(jalur_log)

# Fungsi parsing setiap baris log
def parse_log(baris):
    global baris_rusak

    try:
        kolom = baris.split("|")

        if len(kolom) != 6:
            baris_rusak.add(1)
            return None

        waktu, id_trx, id_user, id_merchant, nominal, status = kolom

        return (
            waktu,
            id_trx,
            id_user,
            id_merchant,
            int(nominal),
            status
        )

    except (ValueError, TypeError):
        baris_rusak.add(1)
        return None


# Parsing dan menyimpan hasil agar tidak dihitung berulang
parsed_rdd = log_rdd.map(parse_log).persist()

# Memicu proses Spark
total_baris = parsed_rdd.count()

# Mengambil transaksi valid
valid_rdd = parsed_rdd.filter(lambda x: x is not None)

print("C2(a) - Hasil parsing log transaksi:")

print("Total baris:", total_baris)

print("Baris valid:", valid_rdd.count())

print("Baris rusak:", baris_rusak.value)

# C2(b) - Menghitung transaksi berdasarkan status

jumlah_status = (
    valid_rdd
    .map(lambda x: (x[5], 1))
    .reduceByKey(lambda a, b: a + b)
    .sortByKey()
)

tampilkan_tabel(
    pd.DataFrame(jumlah_status.collect(), columns=["Status Transaksi", "Jumlah Transaksi"]),
    "C2(b) Jumlah transaksi berdasarkan status"
)

# C2(c) - Menghitung tingkat kegagalan per merchant

# Format: id_merchant -> (total transaksi, transaksi gagal)
transaksi_merchant = (
    valid_rdd
    .map(lambda x: (
        x[3],
        (1, 1 if x[5] == "GAGAL" else 0)
    ))
    .reduceByKey(
        lambda a, b: (a[0] + b[0], a[1] + b[1])
    )
)

# Menghitung persentase kegagalan
tingkat_gagal = (
    transaksi_merchant
    .filter(lambda x: x[1][0] >= 1500)
    .map(lambda x: (
        x[0],
        x[1][0],
        x[1][1],
        x[1][1] / x[1][0] * 100
    ))
    .sortBy(lambda x: x[3], ascending=False)
)

# Mengambil 5 merchant dengan tingkat kegagalan tertinggi
top5_gagal = tingkat_gagal.take(5)

tabel_top5_gagal = pd.DataFrame(top5_gagal, columns=[
    "ID Merchant", "Total Transaksi", "Transaksi Gagal", "Tingkat Gagal (%)"
])
tabel_top5_gagal["Tingkat Gagal (%)"] = tabel_top5_gagal["Tingkat Gagal (%)"].round(2)
tampilkan_tabel(tabel_top5_gagal, "C2(c) Lima merchant dengan tingkat kegagalan tertinggi")

# C3(a): definisikan skema eksplisit dan ubah RDD menjadi DataFrame.
from datetime import datetime
from pyspark.sql.types import (
    StructType, StructField, StringType,
    TimestampType, LongType
)

# Menentukan skema DataFrame
schema_transaksi = StructType([
    StructField("waktu", TimestampType(), True),
    StructField("id_trx", StringType(), True),
    StructField("id_user", StringType(), True),
    StructField("id_merchant", StringType(), True),
    StructField("nominal", LongType(), True),
    StructField("status", StringType(), True)
])

# Mengubah RDD menjadi format sesuai skema
transaksi_row = valid_rdd.map(
    lambda x: (
        datetime.strptime(x[0], "%Y-%m-%d %H:%M:%S"),
        x[1], x[2], x[3], x[4], x[5]
    )
)

df_transaksi = spark.createDataFrame(
    transaksi_row,
    schema=schema_transaksi
)

print("C3(a) - Skema DataFrame transaksi:")

df_transaksi.printSchema()

print("Jumlah transaksi valid:", df_transaksi.count())

# C3(b): buat DataFrame merchant dengan tiga kolom yang diminta.
# Mengambil data merchant dari MongoDB
data_merchant = list(db["merchant"].find(
    {},
    {"_id": 1, "kategori": 1, "kota": 1}
))

# Membuat DataFrame merchant
merchant_row = [
    (m["_id"], m["kategori"], m["kota"])
    for m in data_merchant
]

schema_merchant = StructType([
    StructField("id_merchant", StringType(), True),
    StructField("kategori", StringType(), True),
    StructField("kota", StringType(), True)
])

df_merchant = spark.createDataFrame(
    merchant_row,
    schema=schema_merchant
)

# Broadcast join transaksi sukses dengan merchant
df_join = (
    df_transaksi
    .filter(F.col("status") == "SUKSES")
    .join(
        F.broadcast(df_merchant),
        on="id_merchant",
        how="inner"
    )
)

print("C3(b) - Jumlah transaksi sukses setelah join:")

print(df_join.count())

# C3(c): total nominal per kota dan kategori.
# Menghitung total nominal per kota dan kategori
df_kota_kategori = (
    df_join
    .groupBy("kota", "kategori")
    .agg(
        F.sum("nominal").alias("total_nominal")
    )
)

print("C3(c) - Total nominal transaksi per kota dan kategori:")

tabel_kota_kategori = (
    df_kota_kategori.orderBy("kota", F.desc("total_nominal"))
    .toPandas()
    .rename(columns={
        "kota": "Kota",
        "kategori": "Kategori",
        "total_nominal": "Total Nominal (Rp)"
    })
)
tabel_kota_kategori["Total Nominal (Rp)"] = (
    tabel_kota_kategori["Total Nominal (Rp)"].map(lambda x: f"Rp {x:,.0f}")
)
tampilkan_tabel(tabel_kota_kategori, "C3(c) Total nominal transaksi per kota dan kategori")

# C3(d): ranking kategori dalam setiap kota dan porsi terhadap total kota.
# Menghitung total transaksi setiap kota
window_kota = Window.partitionBy("kota")

# Mengurutkan kategori berdasarkan nominal tertinggi
window_rank = Window.partitionBy("kota").orderBy(
    F.desc("total_nominal")
)

df_top_kategori = (
    df_kota_kategori
    .withColumn(
        "total_kota",
        F.sum("total_nominal").over(window_kota)
    )
    .withColumn(
        "share_persen",
        F.round(
            F.col("total_nominal") /
            F.col("total_kota") * 100,
            2
        )
    )
    .withColumn(
        "peringkat",
        F.row_number().over(window_rank)
    )
    .filter(F.col("peringkat") == 1)
    .select(
        "kota",
        "kategori",
        "total_nominal",
        "share_persen"
    )
    .orderBy("kota")
)

tabel_top_kategori = df_top_kategori.toPandas().rename(columns={
    "kota": "Kota",
    "kategori": "Kategori Teratas",
    "total_nominal": "Total Nominal (Rp)",
    "share_persen": "Porsi (%)"
})
tabel_top_kategori["Total Nominal (Rp)"] = (
    tabel_top_kategori["Total Nominal (Rp)"].map(lambda x: f"Rp {x:,.0f}")
)
tampilkan_tabel(tabel_top_kategori, "C3(d) Kategori transaksi tertinggi per kota")

import shutil
from pyspark.sql.types import StructType, StructField, StringType, LongType

# Skema eksplisit untuk event JSON (tidak memakai inferSchema agar tipe pasti dan stream tidak
# perlu memindai berkas lebih dulu).
skema_event = StructType([
    StructField("id_trx", StringType(), True),
    StructField("id_merchant", StringType(), True),
    StructField("nominal", LongType(), True),
    StructField("waktu", StringType(), True),
])

# Lokasi keluaran dan checkpoint (dibersihkan agar chunk dapat dijalankan ulang dari awal).
out_valid   = os.path.join(folder, "out_valid")
out_invalid = os.path.join(folder, "out_invalid")
ckpt        = os.path.join(folder, "checkpoint")
for p in (out_valid, out_invalid, ckpt):
    shutil.rmtree(p, ignore_errors=True)

catatan_batch = []   # catatan per batch (dikumpulkan di driver)

def proses_batch(df_batch, batch_id):
    """Dipanggil Spark untuk setiap micro-batch: pisahkan valid/tidak valid lalu simpan."""
    df_batch = df_batch.cache()
    valid   = df_batch.filter(F.col("nominal").isNotNull())
    invalid = df_batch.filter(F.col("nominal").isNull())

    n_valid, n_invalid = valid.count(), invalid.count()

    # Event valid -> Parquet (mode append); event tidak valid dikarantina di folder terpisah.
    valid.write.mode("append").parquet(out_valid)
    invalid.write.mode("append").parquet(out_invalid)

    catatan_batch.append({"batch_id": batch_id, "valid": n_valid,
                          "tidak_valid": n_invalid, "total": n_valid + n_invalid})
    df_batch.unpersist()

# Kueri streaming: baca folder landing, 1 berkas per batch, jalankan semua berkas yang ada lalu berhenti.
stream_df = (spark.readStream
             .schema(skema_event)
             .option("maxFilesPerTrigger", 1)
             .json(landing))

kueri = (stream_df.writeStream
         .foreachBatch(proses_batch)
         .option("checkpointLocation", ckpt)
         .trigger(availableNow=True)
         .start())
kueri.awaitTermination()

# Catatan per batch dan total baris Parquet.
tampilkan_tabel(
    pd.DataFrame(catatan_batch).rename(columns={
        "batch_id": "Batch ke-",
        "valid": "Event Valid",
        "tidak_valid": "Event Tidak Valid",
        "total": "Total Event"
    }),
    "C4 Ringkasan pemrosesan setiap batch"
)

total_valid   = spark.read.parquet(out_valid).count()
total_invalid = spark.read.parquet(out_invalid).count()
print("\nTotal baris Parquet (event valid)        :", total_valid)

print("Total baris karantina (event tidak valid):", total_invalid)

print("Total event diproses                     :", total_valid + total_invalid)

# ============================================================
# C5 - EDA SKALA BESAR DAN INSIGHT
# ============================================================

from pyspark.sql import functions as F
import pandas as pd
import matplotlib.pyplot as plt
from matplotlib.ticker import FuncFormatter

# ============================================================
# 10.2.1 C5(1): LAPORAN KUALITAS DATA
# ============================================================

hasil = df_transaksi.select(
    F.count("*").alias("total"),
    F.sum(F.when(F.col("nominal") < 0, 1).otherwise(0)).alias("negatif"),
    F.sum(F.when(F.col("nominal") >= 50000000, 1).otherwise(0)).alias("ekstrem"),
    F.sum(F.when(F.col("status") == "PENDING", 1).otherwise(0)).alias("pending")
).first()

total = hasil["total"]

baris = [
    "10.2.1 C5(1): LAPORAN KUALITAS DATA",
    "-" * 65,
    f"{'Total baris valid':<25} : {total:>12,} (100.000%)",
    f"{'Nominal negatif':<25} : {hasil['negatif']:>12,} ({hasil['negatif']/total*100:>6.3f}%)",
    f"{'Nominal >= Rp50 juta':<25} : {hasil['ekstrem']:>12,} ({hasil['ekstrem']/total*100:>6.3f}%)",
    f"{'Status PENDING':<25} : {hasil['pending']:>12,} ({hasil['pending']/total*100:>6.3f}%)"
]

print("\n".join(baris))

# ============================================================
# 10.2.2 C5(2): DISTRIBUSI NOMINAL
# ============================================================

df_bersih = df_transaksi.filter(
    (F.col("nominal") > 0) &
    (F.col("nominal") < 50000000)
)

persentil = df_bersih.approxQuantile(
    "nominal",
    [0.05, 0.50, 0.95, 0.99],
    0.0001
)

stat_raw = df_transaksi.agg(
    F.avg("nominal").alias("rata"),
    F.max("nominal").alias("maksimum")
).first()

stat_bersih = df_bersih.agg(
    F.avg("nominal").alias("rata")
).first()

total_raw = (
    df_transaksi
    .filter(F.col("status") == "SUKSES")
    .agg(F.sum("nominal").alias("total"))
    .first()["total"] or 0
)

total_bersih = (
    df_transaksi
    .filter(
        (F.col("status") == "SUKSES") &
        (F.col("nominal") > 0) &
        (F.col("nominal") < 50000000)
    )
    .agg(F.sum("nominal").alias("total"))
    .first()["total"] or 0
)

selisih = total_raw - total_bersih
persen_selisih = selisih / total_raw * 100 if total_raw else 0

sukses_ekstrem = df_transaksi.filter(
    (F.col("status") == "SUKSES") &
    (F.col("nominal") >= 50000000)
).count()

negatif_gagal = df_transaksi.filter(
    (F.col("nominal") < 0) &
    (F.col("status") == "GAGAL")
).count()

baris = [
    "",
    "10.2.2 C5(2): DISTRIBUSI NOMINAL DENGAN approxQuantile",
    "-" * 65,
    "Persentil nominal transaksi (relativeError = 0,0001):",
    f"P5  : Rp {persentil[0]:>15,.0f}",
    f"P50 : Rp {persentil[1]:>15,.0f}",
    f"P95 : Rp {persentil[2]:>15,.0f}",
    f"P99 : Rp {persentil[3]:>15,.0f}",
    "",
    f"Rata-rata nominal mentah   : Rp {stat_raw['rata']:>15,.0f}",
    f"Rata-rata nominal bersih   : Rp {stat_bersih['rata']:>15,.0f}",
    f"Nominal maksimum           : Rp {stat_raw['maksimum']:>15,.0f}",
    f"Total nominal SUKSES mentah: Rp {total_raw:>15,.0f}",
    f"Total nominal SUKSES bersih: Rp {total_bersih:>15,.0f}",
    f"Selisih akibat nilai ekstrem: Rp {selisih:,.0f} ({persen_selisih:.1f}% dari total mentah)",
    f"Transaksi SUKSES >= Rp50 juta: {sukses_ekstrem:,}",
    f"Nominal negatif berstatus GAGAL: {negatif_gagal:,}"
]

if persentil[3] != 0:
    rasio = stat_raw["maksimum"] / persentil[3]
    baris.append(f"Nominal maksimum = {rasio:,.1f} kali P99")

print("\n".join(baris))

# ============================================================
# 10.2.3 C5(3): TRANSAKSI SUKSES BERDASARKAN JAM
# ============================================================

df_jam = (
    df_transaksi
    .filter(F.col("status") == "SUKSES")
    .withColumn("jam", F.hour("waktu"))
    .groupBy("jam")
    .agg(F.count("*").alias("jumlah_sukses"))
)

df_semua_jam = spark.createDataFrame(
    [(i,) for i in range(24)],
    ["jam"]
)

df_jam_lengkap = (
    df_semua_jam
    .join(df_jam, "jam", "left")
    .fillna(0, subset=["jumlah_sukses"])
    .orderBy("jam")
)

data_jam = df_jam_lengkap.toPandas()
data_jam["jumlah_sukses"] = data_jam["jumlah_sukses"].astype(int)

total_sukses = data_jam["jumlah_sukses"].sum()

data_jam["Jam"] = data_jam["jam"].apply(lambda x: f"{x:02d}:00")
data_jam["Persentase (%)"] = (
    data_jam["jumlah_sukses"] / total_sukses * 100
).round(2)

# Tabel vertikal agar lebih mudah dibaca
tabel_jam = data_jam[
    ["Jam", "jumlah_sukses", "Persentase (%)"]
].rename(columns={
    "jumlah_sukses": "Jumlah SUKSES"
})

print("\n10.2.3 C5(3): TRANSAKSI SUKSES BERDASARKAN JAM")

print("-" * 55)

print(tabel_jam.to_string(index=False))

# Jam puncak dan tersepi
puncak = data_jam.loc[data_jam["jumlah_sukses"].idxmax()]
tersepi = data_jam.loc[data_jam["jumlah_sukses"].idxmin()]

rasio_jam = (
    puncak["jumlah_sukses"] / tersepi["jumlah_sukses"]
    if tersepi["jumlah_sukses"] > 0 else 0
)

tiga_teratas = data_jam.nlargest(3, "jumlah_sukses")

persen_tiga = (
    tiga_teratas["jumlah_sukses"].sum() / total_sukses * 100
    if total_sukses else 0
)

daftar_jam = ", ".join(
    tiga_teratas["Jam"].tolist()
)

print(
    f"\nJam puncak: {puncak['Jam']} "
    f"({int(puncak['jumlah_sukses']):,} transaksi)"
    f" | Jam tersepi: {tersepi['Jam']} "
    f"({int(tersepi['jumlah_sukses']):,} transaksi)"
)

print(f"Rasio puncak/tersepi: {rasio_jam:.2f}x")

print(
    f"Tiga jam teratas: {daftar_jam} "
    f"-> {persen_tiga:.1f}% dari seluruh transaksi sukses"
)

# ============================================================
# GRAFIK BATANG TRANSAKSI SUKSES BERDASARKAN JAM
# ============================================================

fig, ax = plt.subplots(figsize=(18, 9), dpi=180)

jam = data_jam["Jam"].tolist()
jumlah = data_jam["jumlah_sukses"].tolist()

batang = ax.bar(
    jam,
    jumlah,
    width=0.72
)

# Judul dan label
ax.set_title(
    "Distribusi Transaksi SUKSES Berdasarkan Jam",
    fontsize=22,
    fontweight="bold",
    pad=25
)

ax.set_xlabel("Jam Transaksi", fontsize=16, labelpad=15)
ax.set_ylabel("Jumlah Transaksi SUKSES", fontsize=16, labelpad=15)

# Memperbesar label sumbu
ax.tick_params(axis="x", labelsize=12, rotation=45)
ax.tick_params(axis="y", labelsize=12)

# Format angka sumbu Y
ax.yaxis.set_major_formatter(
    FuncFormatter(lambda x, pos: f"{x:,.0f}")
)

# Zoom sumbu Y supaya perbedaan tinggi batang terlihat
nilai_min = min(jumlah)
nilai_max = max(jumlah)

margin = (nilai_max - nilai_min) * 0.35

ax.set_ylim(
    max(0, nilai_min - margin),
    nilai_max + margin
)

# Grid horizontal
ax.grid(
    axis="y",
    linestyle="--",
    alpha=0.4
)

ax.set_axisbelow(True)

# Label pada setiap batang
for b, nilai in zip(batang, jumlah):
    ax.annotate(
        f"{nilai:,}",
        xy=(b.get_x() + b.get_width() / 2, b.get_height()),
        xytext=(0, 7),
        textcoords="offset points",
        ha="center",
        va="bottom",
        fontsize=9
    )

# Keterangan rentang sumbu
ax.text(
    0.99,
    0.97,
    "Catatan: sumbu Y dimulai dari nilai mendekati minimum",
    transform=ax.transAxes,
    ha="right",
    va="top",
    fontsize=10
)

fig.tight_layout()

plt.show()

plt.close(fig)