tugas-pak-nanangsesi10/main.py

# -*- coding: utf-8 -*-
"""
PRAKTIKUM RAG  |  Lab 1 (Chunking)  ->  Lab 2 (Indeks & Cari)  ->  Lab 3 (Hit@k)

Cara pakai:
  1. python -m pip install pdfplumber PyPDF2 sentence-transformers numpy pandas
  2. Buat folder `dokumen/` di samping file ini, isi 3 file PDF buku cerita.
  3. Jalankan per-cell di VS Code (tanda `# %%`) atau jalankan langsung `python main.py`.

Catatan desain:
  - Embedding : sentence-transformers (model multilingual, cocok untuk Bahasa Indonesia).
  - Vector DB : numpy + cosine similarity.
"""

# %% ============================================================
# 0. IMPORT & KONFIGURASI
# ===============================================================
import os
import re
import random
from glob import glob

import numpy as np

FOLDER_DOKUMEN = "dokumen"          # folder berisi PDF
OVERLAP_PERSEN = 0.15               # overlap 15%
MODEL_EMBEDDING = "paraphrase-multilingual-MiniLM-L12-v2"  # mendukung Bahasa Indonesia
random.seed(42)                     # supaya chunk "acak" bisa direproduksi


# %% ============================================================
# 1. FUNGSI INTI (dipakai oleh Lab 1, 2, dan 3)
# ===============================================================
def baca_pdf(filepath):
    """
    Membaca 1 file PDF dan mengembalikan seluruh teksnya (string).

    - Utama : pdfplumber (lebih baik untuk layout/tabel).
    - Cadangan: PyPDF2, dipakai jika pdfplumber tidak terpasang / gagal.
    - Teks dinormalisasi: spasi berlebih dirapikan, halaman digabung dengan newline.
    """
    halaman = []
    try:
        import pdfplumber
        with pdfplumber.open(filepath) as pdf:
            for p in pdf.pages:
                halaman.append(p.extract_text() or "")
    except Exception as e:  # fallback
        print(f"[peringatan] pdfplumber gagal ({e}); beralih ke PyPDF2")
        from PyPDF2 import PdfReader
        reader = PdfReader(filepath)
        halaman = [(pg.extract_text() or "") for pg in reader.pages]

    teks = "\n".join(halaman)
    teks = re.sub(r"[ \t]+", " ", teks)        # rapikan spasi/tab berulang
    teks = re.sub(r"\n{3,}", "\n\n", teks)     # maksimal 1 baris kosong
    return teks.strip()


def potong(teks, chunk_size, overlap):
    """
    Memotong `teks` menjadi potongan (chunk) berukuran `chunk_size` karakter.

    Parameter:
      chunk_size : panjang tiap chunk (karakter).
      overlap    : tumpang-tindih antar chunk.
                   Jika 0 < overlap < 1 -> dianggap persentase (mis. 0.15 = 15%).
                   Jika >= 1            -> dianggap jumlah karakter.
    Return: list[str]
    """
    if 0 < overlap < 1:
        overlap = int(chunk_size * overlap)
    overlap = int(overlap)
    if overlap >= chunk_size:
        raise ValueError("overlap harus lebih kecil dari chunk_size")

    langkah = chunk_size - overlap
    hasil = []
    for mulai in range(0, len(teks), langkah):
        potongan = teks[mulai:mulai + chunk_size].strip()
        if potongan:
            hasil.append(potongan)
        if mulai + chunk_size >= len(teks):   # sudah mencapai akhir teks
            break
    return hasil


class IndeksVektor:
    """Indeks vektor sederhana: simpan chunk + embedding, cari dengan cosine similarity."""

    _model = None  # model dimuat sekali saja (cache di level kelas)

    def __init__(self, nama="indeks"):
        self.nama = nama
        self.chunks = []        # list[dict]: {"teks", "sumber", "id"}
        self.vektor = None      # np.ndarray (n_chunk, dim), sudah dinormalisasi

    @classmethod
    def model(cls):
        if cls._model is None:
            from sentence_transformers import SentenceTransformer
            cls._model = SentenceTransformer(MODEL_EMBEDDING)
        return cls._model

    def tambah(self, daftar_teks, sumber):
        """Menambahkan chunk dari satu dokumen ke indeks lalu meng-embed-nya."""
        awal = len(self.chunks)
        for i, t in enumerate(daftar_teks):
            self.chunks.append({"teks": t, "sumber": sumber, "id": awal + i})
        emb = self.model().encode(daftar_teks, normalize_embeddings=True,
                                  show_progress_bar=False)
        self.vektor = emb if self.vektor is None else np.vstack([self.vektor, emb])

    def cari(self, pertanyaan, k=3):
        """Mengembalikan top-k chunk paling mirip (list[dict] dengan kunci 'skor')."""
        q = self.model().encode([pertanyaan], normalize_embeddings=True)[0]
        skor = self.vektor @ q                       # cosine (vektor sudah ternormalisasi)
        idx = np.argsort(-skor)[:k]
        return [{**self.chunks[i], "skor": float(skor[i])} for i in idx]


def bangun_indeks(folder, chunk_size, overlap=OVERLAP_PERSEN, nama="indeks"):
    """
    Membaca semua PDF di `folder`, memotong, meng-embed, lalu mengembalikan
    (indeks, laporan) dimana laporan = {nama_file: jumlah_chunk}.
    """
    indeks = IndeksVektor(nama)
    laporan = {}
    file_pdf = sorted(glob(os.path.join(folder, "*.pdf")))
    if not file_pdf:
        raise FileNotFoundError(f"Tidak ada PDF di folder '{folder}/'")
    for path in file_pdf:
        teks = baca_pdf(path)
        chunks = potong(teks, chunk_size, overlap)
        indeks.tambah(chunks, sumber=os.path.basename(path))
        laporan[os.path.basename(path)] = len(chunks)
    return indeks, laporan


INDEKS_AKTIF = None  # diisi di Lab 2; dipakai default oleh query()


def query(pertanyaan, k=3, index=None, tampil=True):
    """
    Mencari chunk paling relevan untuk `pertanyaan`.

    - index : objek IndeksVektor; jika None memakai INDEKS_AKTIF (hasil Lab 2).
    - k     : jumlah hasil teratas.
    - tampil: True -> cetak hasil dengan rapi.
    Return  : list[dict] berisi teks, sumber, id, skor.
    """
    index = index or INDEKS_AKTIF
    hasil = index.cari(pertanyaan, k)
    if tampil:
        print(f"\n❓ {pertanyaan}")
        for r, h in enumerate(hasil, 1):
            cuplikan = h["teks"][:300].replace("\n", " ")
            print(f"  [{r}] skor={h['skor']:.3f} | {h['sumber']} #chunk{h['id']}\n      {cuplikan}...")
    return hasil


# %% ============================================================
# LAB 1 - PREPROCESSING & CHUNKING
# ===============================================================
# --- 1.1 Baca PDF + cetak 500 karakter pertama ---
file_pdf = sorted(glob(os.path.join(FOLDER_DOKUMEN, "*.pdf")))
assert file_pdf, f"Letakkan PDF di folder '{FOLDER_DOKUMEN}/' dulu."
FILE_LAB1 = file_pdf[0]

teks_lab1 = baca_pdf(FILE_LAB1)
print(f"File          : {FILE_LAB1}")
print(f"Total karakter: {len(teks_lab1):,}")
print("=" * 60)
print(teks_lab1[:500])
print("=" * 60)


# %%
# --- 1.2 Eksperimen chunking: 500 / 1000 / 1500 dengan overlap 15% ---
UKURAN_CHUNK = [500, 1000, 1500]
hasil_chunk = {}
for ukuran in UKURAN_CHUNK:
    hasil_chunk[ukuran] = potong(teks_lab1, ukuran, OVERLAP_PERSEN)

print(f"{'Chunk size':>12} | {'Overlap':>8} | {'Jumlah chunk':>12}")
print("-" * 40)
for ukuran, daftar in hasil_chunk.items():
    print(f"{ukuran:>12} | {int(ukuran*OVERLAP_PERSEN):>8} | {len(daftar):>12}")


# %%
# --- 1.3 Evaluasi kualitas: 3 chunk acak per ukuran + deteksi otomatis ---
def periksa_potongan(chunk):
    """Heuristik sederhana untuk membantu analisis manual."""
    c = chunk.strip()
    catatan = []
    if c and c[0].islower():
        catatan.append("diawali huruf kecil -> kalimat terpotong di awal")
    if c and c[-1] not in ".!?:;)\"'":
        catatan.append("tidak diakhiri tanda baca -> kalimat terpotong di akhir")
    baris = c.split("\n")
    tabel = [b for b in baris if b.count("  ") >= 2 or b.count("|") >= 2
             or len(re.findall(r"\d+[.,]?\d*", b)) >= 4]
    if tabel:
        catatan.append(f"terindikasi tabel/angka berderet ({len(tabel)} baris) -> cek apakah tabel terbelah")
    return catatan or ["tidak ada indikasi pemotongan"]

for ukuran, daftar in hasil_chunk.items():
    print("\n" + "#" * 70)
    print(f"CHUNK SIZE {ukuran}  (total {len(daftar)} chunk)")
    print("#" * 70)
    for no in random.sample(range(len(daftar)), k=min(3, len(daftar))):
        print(f"\n--- chunk #{no} ({len(daftar[no])} karakter) ---")
        print(daftar[no])
        print("  >> Deteksi otomatis:", "; ".join(periksa_potongan(daftar[no])))


# %% ============================================================
# LAB 2 - BANGUN INDEKS & CARI
# ===============================================================
# --- 2.1 Indexing multi-dokumen (2-3 PDF dalam folder dokumen/) ---
INDEKS_AKTIF, laporan = bangun_indeks(FOLDER_DOKUMEN, chunk_size=1000,
                                      overlap=OVERLAP_PERSEN, nama="Indeks-1000")
print("LAPORAN INDEXING")
print("-" * 50)
for nama_file, jml in laporan.items():
    print(f"{nama_file:<35} : {jml:>4} chunk")
print("-" * 50)
print(f"{'TOTAL':<35} : {sum(laporan.values()):>4} chunk")


# %%
# --- 2.2 Direct query: 5 pertanyaan spesifik (Cerita Rakyat) ---
PERTANYAAN_LANGSUNG = [
    "Siapa nama gadis baik hati dalam cerita Bawang Merah Bawang Putih?",
    "Benda apa yang diberikan nenek di dalam gua kepada Bawang Putih?",
    "Apa janji yang dilanggar oleh Toba kepada istrinya?",
    "Siapa nama anak laki-laki Toba yang sangat lapar dan serakah?",
    "Dikena hukuman atau dikutuk menjadi apa Malin Kundang oleh ibunya?"
]
for p in PERTANYAAN_LANGSUNG:
    query(p, k=3)


# %%
# --- 2.3 Uji semantik / parafrase: 2 pertanyaan (Cerita Rakyat) ---
PERTANYAAN_PARAFRASE = [
    "Buah apa yang dibawa pulang oleh Bawang Merah yang berisi perhiasan atau hewan buas?",
    "Bencana alam apa yang tercipta akibat luapan air mata dan amarah istri Toba?"
]
for p in PERTANYAAN_PARAFRASE:
    query(p, k=3)


# %% ============================================================
# LAB 3 - UKUR HIT@K
# ===============================================================
# --- 3.1 Ground truth: 15 pertanyaan + kata kunci jawaban cerita rakyat ---
DATASET_UJI = [
    ("Siapa nama gadis manis yang baik hati dan rajin?", ["Bawang Putih"]),
    ("Apa yang diberikan nenek tua di dalam gua kepada Bawang Putih?", ["labu"]),
    ("Benda apa yang hilang terbawa arus sungai saat Bawang Putih mencuci?", ["selendang", "baju"]),
    ("Siapa nama saudara tiri Bawang Putih yang malas dan serakah?", ["Bawang Merah"]),
    ("Di mana Bawang Putih bertemu dengan nenek tua?", ["gua"]),
    ("Siapa nama pemuda nelayan yang menangkap ikan mas ajaib?", ["Toba"]),
    ("Ikan mas ajaib berubah menjadi sosok apa?", ["wanita", "perempuan"]),
    ("Apa syarat yang diajukan jelita ikan mas sebelum menikah dengan Toba?", ["rahasia", "janji"]),
    ("Siapa nama anak laki-laki Toba yang sangat pemakan dan lapar?", ["Samosir"]),
    ("Danau apa yang terbentuk dari luapan air mata dan banjir besar?", ["Danau Toba"]),
    ("Siapa nama pemuda miskin yang pergi merantau naik kapal pedagang?", ["Malin Kundang"]),
    ("Di mana Malin Kundang tinggal bersama ibunya saat masih kecil?", ["desa", "pantai"]),
    ("Bagaimana sikap Malin Kundang saat bertemu kembali dengan ibu kandungnya?", ["durhaka", "menolak"]),
    ("Malin Kundang dikutuk oleh ibunya menjadi apa?", ["batu"]),
    ("Apa yang hancur dihantam badai dahsyat di tengah laut?", ["kapal"])
]

pertanyaan_list = [p for p, _ in DATASET_UJI]
ground_truth_list = [g for _, g in DATASET_UJI]
assert len(DATASET_UJI) == 15, "Dataset uji harus 15 pertanyaan"


# %%
# --- 3.2 Fungsi hit_at_k ---
def _cocok(teks_chunk, kata_kunci, min_cocok=1):
    """True jika minimal `min_cocok` kata kunci ditemukan di chunk (case-insensitive)."""
    t = teks_chunk.lower()
    return sum(kk.lower() in t for kk in kata_kunci) >= min_cocok


def hit_at_k(pertanyaan_list, ground_truth_list, index, k, min_cocok=1):
    """
    Menghitung Hit@k = (jumlah pertanyaan yang 'hit') / (total pertanyaan).

    Sebuah pertanyaan dianggap HIT bila minimal satu dari top-k chunk hasil
    pencarian memuat >= `min_cocok` kata kunci ground truth.

    Return: (skor_hit, daftar_indeks_miss)
    """
    hit, miss = 0, []
    for i, (p, gt) in enumerate(zip(pertanyaan_list, ground_truth_list)):
        top_k = index.cari(p, k)
        if any(_cocok(h["teks"], gt, min_cocok) for h in top_k):
            hit += 1
        else:
            miss.append(i)
    return hit / len(pertanyaan_list), miss


# %%
# --- 3.3 Jalankan pada 2 indeks: A (chunk 500) vs B (chunk 1000) ---
indeks_A, _ = bangun_indeks(FOLDER_DOKUMEN, 500,  OVERLAP_PERSEN, nama="Indeks A (500)")
indeks_B, _ = bangun_indeks(FOLDER_DOKUMEN, 1000, OVERLAP_PERSEN, nama="Indeks B (1000)")

K_LIST = [1, 3, 5]
tabel, miss_detail = {}, {}
for idx in (indeks_A, indeks_B):
    tabel[idx.nama] = {}
    for k in K_LIST:
        skor, miss = hit_at_k(pertanyaan_list, ground_truth_list, idx, k)
        tabel[idx.nama][f"Hit@{k}"] = skor
        miss_detail[(idx.nama, k)] = miss

print("\nPERBANDINGAN HIT@K (15 pertanyaan uji)")
print("-" * 50)
print(f"{'Indeks':<20} | {'Hit@1':<8} | {'Hit@3':<8} | {'Hit@5':<8}")
print("-" * 50)
for nama_indeks, skor_dict in tabel.items():
    h1 = f"{skor_dict.get('Hit@1', 0):.2%}"
    h3 = f"{skor_dict.get('Hit@3', 0):.2%}"
    h5 = f"{skor_dict.get('Hit@5', 0):.2%}"
    print(f"{nama_indeks:<20} | {h1:<8} | {h3:<8} | {h5:<8}")
print("-" * 50)


# %%
# --- 3.4 Analisis failure case (miss pada k=5) + draf kesimpulan ---
for idx in (indeks_A, indeks_B):
    miss = miss_detail[(idx.nama, 5)]
    print(f"\n{'='*70}\n{idx.nama}: {len(miss)} pertanyaan MISS pada k=5\n{'='*70}")
    for i in miss:
        top1 = idx.cari(pertanyaan_list[i], 1)[0]
        print(f"\n[Q{i+1}] {pertanyaan_list[i]}")
        print(f"   Kata kunci   : {ground_truth_list[i]}")
        print(f"   Top-1 diambil: ({top1['sumber']}, skor={top1['skor']:.3f}) "
              f"{top1['teks'][:200].replace(chr(10), ' ')}...")


def draf_kesimpulan(tabel_data, miss_detail, nama_a, nama_b):
    """Membuat draf kesimpulan otomatis."""
    rata_a = sum(tabel_data[nama_a].values()) / len(tabel_data[nama_a])
    rata_b = sum(tabel_data[nama_b].values()) / len(tabel_data[nama_b])
    terbaik = nama_a if rata_a >= rata_b else nama_b
    miss_b = sorted({f"Q{i+1}" for (n, k), m in miss_detail.items() if k == 5 for i in m})
    return (
        f"Berdasarkan rata-rata Hit@1/3/5, {terbaik} memberikan performa terbaik "
        f"({nama_a}: {rata_a:.1%} vs {nama_b}: {rata_b:.1%}). "
        f"Kegagalan (miss) pada k=5 terjadi di pertanyaan: {', '.join(miss_b) or 'tidak ada'}. "
        "Dugaan penyebab: jawaban tersebar/terpotong di batas chunk, istilah teknis atau kata kunci "
        "terlalu spesifik/berbeda format, atau ukuran chunk yang kurang fleksibel menangkap konteks kalimat."
    )

print("\nDRAF KESIMPULAN:\n")
print(draf_kesimpulan(tabel, miss_detail, indeks_A.nama, indeks_B.nama))
# %%