import sys
sys.path.insert(0, "..")

import math
import zlib
from collections import Counter

import pandas as pd
import matplotlib.pyplot as plt

from common.datasets import dokumen_teks, log_transaksi

def info_hdfs(ukuran_mb, blok_mb=128, replikasi=3):
    blok = math.ceil(ukuran_mb / blok_mb)
    return {"ukuran_mb": ukuran_mb, "blok": blok,
            "salinan_blok": blok * replikasi, "disk_mb": ukuran_mb * replikasi,
            "blok_terakhir_mb": ukuran_mb - (blok - 1) * blok_mb if blok else 0}

pd.DataFrame([info_hdfs(x) for x in [100, 128, 500, 1024, 10_000]])

import random
random.seed(0)
rack = {"dn1": "R1", "dn2": "R1", "dn3": "R1", "dn4": "R2", "dn5": "R2", "dn6": "R2"}

def tempatkan(blok_id, penulis="dn1"):
    r1 = penulis
    lain_rack = [d for d in rack if rack[d] != rack[r1]]
    r2 = random.choice(lain_rack)
    r3 = random.choice([d for d in lain_rack if d != r2])
    return {"blok": blok_id, "replika": [r1, r2, r3], "rack": [rack[r1], rack[r2], rack[r3]]}

pd.DataFrame([tempatkan(f"B{i}") for i in range(4)])

def bagi_split(data, n_split):
    q, r = divmod(len(data), n_split)
    hasil, i = [], 0
    for s in range(n_split):
        uk = q + (1 if s < r else 0)
        hasil.append(data[i:i + uk]); i += uk
    return hasil

def shuffle_sort(pasangan):
    grup = {}
    for k, v in pasangan:
        grup.setdefault(k, []).append(v)
    return sorted(grup.items())

def mapreduce(data, mapper, reducer, combiner=None, n_split=4, verbose=False):
    keluaran_map = []
    for i, split in enumerate(bagi_split(data, n_split)):
        out = [kv for rec in split for kv in mapper(rec)]
        if combiner:
            out = [combiner(k, vs) for k, vs in shuffle_sort(out)]
        if verbose:
            print(f"mapper {i}: {len(split)} record -> {len(out)} pasangan  {out[:4]}...")
        keluaran_map.extend(out)
    grup = shuffle_sort(keluaran_map)
    if verbose:
        print("setelah shuffle & sort:", grup[:3], "...")
    return dict(reducer(k, vs) for k, vs in grup)

docs = dokumen_teks()
for d in docs[:3]:
    print("-", d)

def mapper_wc(doc):
    return [(w, 1) for w in doc.lower().split()]

def reducer_sum(k, vs):
    return (k, sum(vs))

wc = mapreduce(docs, mapper_wc, reducer_sum, n_split=3, verbose=True)
pd.Series(wc).sort_values(ascending=False).head(10)

assert wc == dict(Counter(w for d in docs for w in d.split()))
print("Hasil sama dengan Counter ✔")

def hitung_pasangan(data, mapper, combiner=None, n_split=4):
    total = 0
    for split in bagi_split(data, n_split):
        out = [kv for rec in split for kv in mapper(rec)]
        if combiner:
            out = [combiner(k, vs) for k, vs in shuffle_sort(out)]
        total += len(out)
    return total

log = log_transaksi()
mapper_kota = lambda b: [(b.split(",")[1], int(b.split(",")[3]) * int(b.split(",")[4]))]

for n_split in [2, 8, 32]:
    a = hitung_pasangan(log, mapper_kota, None, n_split)
    b = hitung_pasangan(log, mapper_kota, reducer_sum, n_split)
    print(f"{n_split:>3} split: tanpa combiner {a:>5} pasangan | dengan combiner {b:>4} pasangan")

print(log[:3])
pendapatan = mapreduce(log, mapper_kota, reducer_sum, combiner=reducer_sum)
s = pd.Series(pendapatan).sort_values()
s.plot.barh(title="Total pendapatan per kota (MapReduce)")
plt.show()

df = pd.DataFrame([b.split(",") for b in log], columns=["tanggal", "kota", "produk", "jumlah", "harga"])
df["pendapatan"] = df["jumlah"].astype(int) * df["harga"].astype(int)
acuan = df.groupby("kota")["pendapatan"].sum()
print((acuan == pd.Series(pendapatan)[acuan.index]).all())

mapper_qty = lambda b: [(b.split(",")[2], int(b.split(",")[3]))]
avg = lambda k, vs: (k, sum(vs) / len(vs))
salah = mapreduce(log, mapper_qty, avg, combiner=avg, n_split=7)

mapper_qty2 = lambda b: [(b.split(",")[2], (int(b.split(",")[3]), 1))]
gabung = lambda k, vs: (k, (sum(v[0] for v in vs), sum(v[1] for v in vs)))
benar = {k: t / n for k, (t, n) in mapreduce(log, mapper_qty2, gabung, combiner=gabung, n_split=7).items()}

df["jumlah"] = df["jumlah"].astype(int)
cmp = pd.DataFrame({"pandas": df.groupby("produk")["jumlah"].mean(),
                    "combiner_salah": pd.Series(salah), "combiner_benar": pd.Series(benar)})
cmp.head(6)

R = 3
part = {k: zlib.crc32(k.encode()) % R for k in pendapatan}
pd.Series(part, name="reducer").sort_values()

MAPPER = '''import sys
for baris in sys.stdin:
    tgl, kota, produk, jumlah, harga = baris.strip().split(",")
    print(f"{kota}\\t{int(jumlah) * int(harga)}")
'''
REDUCER = '''import sys
kota_lama, total = None, 0
for baris in sys.stdin:                      # input SUDAH terurut per key
    kota, nilai = baris.strip().split("\\t")
    if kota != kota_lama and kota_lama is not None:
        print(f"{kota_lama}\\t{total}")
        total = 0
    kota_lama = kota
    total += int(nilai)
if kota_lama is not None:
    print(f"{kota_lama}\\t{total}")
'''
print(MAPPER)
print(REDUCER)

# Simulasi pipeline `mapper | sort | reducer` tanpa menulis file
import io, contextlib

def jalankan_skrip(kode, teks_input):
    keluar = io.StringIO()
    with contextlib.redirect_stdout(keluar):
        lama = sys.stdin
        sys.stdin = io.StringIO(teks_input)
        try:
            exec(kode, {})
        finally:
            sys.stdin = lama
    return keluar.getvalue()

keluaran_map = jalankan_skrip(MAPPER, "\n".join(log))
terurut = "".join(sorted(keluaran_map.splitlines(keepends=True)))
print(jalankan_skrip(REDUCER, terurut))

# 1.

# 2.

# 3.

# 4.