lsa.py

import numpy as np
import re
import math
import os
import svd


K_LATENT = 100       
MAX_WORDS = 1000     
DATASET_FOLDER = r"dataset"


# =========================================================
# UTIL
# =========================================================
def load_txt_file(path, max_words=500):
    with open(path, "r", encoding="utf-8", errors="ignore") as f:
        text = f.read()
    words = text.split()
    return " ".join(words[:max_words])


def load_documents_from_folder(folder_path, max_words=500):
    docs, names = [], []
    files = sorted([f for f in os.listdir(folder_path) if f.endswith(".txt")])


    for i, fname in enumerate(files, 1):
        path = os.path.join(folder_path, fname)
        try:
            content = load_txt_file(path, max_words)
            wc = len(content.split())

            if content.strip() and wc > 0:
                docs.append(content)
                names.append(fname)
            else:
                print(f"   SKIP: file terbaca tapi kosong setelah diproses")
        except Exception as e:
            print(f"[{i}/{len(files)}] {fname} -> ERROR: {e}")

    return docs, names


class LSASearchEngine:
    def __init__(self, documents, filenames, stopwords=None):
        self.documents = documents
        self.filenames = filenames
        self.stopwords = stopwords if stopwords is not None else set()

    def preprocess(self, text):
        text = text.lower()
        tokens = re.findall(r"\b[a-z]+\b", text)
        return tokens

    def build_term_document_matrix(self):
        all_tokens = []
        doc_tokens = []

        for doc in self.documents:
            tokens = self.preprocess(doc)
            doc_tokens.append(tokens)
            all_tokens.extend(tokens)

        self.vocab = sorted(set(all_tokens))
        self.term2idx = {t: i for i, t in enumerate(self.vocab)}

        m, n = len(self.vocab), len(self.documents)
        A = np.zeros((m, n))

        for j, tokens in enumerate(doc_tokens):
            for t in tokens:
                A[self.term2idx[t], j] += 1

        self.A = A
        return A

    def compute_tf_idf(self):
        m, n = self.A.shape
        TF = np.zeros_like(self.A)

        for j in range(n):
            s = np.sum(self.A[:, j])
            if s > 0:
                TF[:, j] = self.A[:, j] / s

        IDF = np.zeros(m)
        for i in range(m):
            df = np.count_nonzero(self.A[i, :])
            IDF[i] = math.log10(1 + n / df)

        self.IDF = IDF
        self.TFIDF = IDF[:, None] * TF
        return self.TFIDF

    def perform_lsa(self, k):
        U, s, V = svd.Svd(self.TFIDF, k)
        self.U = U[:, :k]
        self.s = s[:k]
        self.V = V[:, :k]

        self.doc_embed = self.V * self.s

    def fold_in_query(self, query_text):
        tokens = self.preprocess(query_text)
        q = np.zeros(len(self.vocab))

        for t in tokens:
            if t in self.term2idx:
                q[self.term2idx[t]] += 1

        if np.sum(q) > 0:
            q = q / np.sum(q)

        q_tfidf = q * self.IDF

        q_latent = q_tfidf @ self.U
        q_latent = np.where(self.s > 1e-12, q_latent / self.s, 0)
        return q_latent

    def cosine_similarity(self, a, b):
        na = svd.l2Norm(a)
        nb = svd.l2Norm(b)
        if na == 0 or nb == 0:
            return 0.0
        return float(np.dot(a, b) / (na * nb))

    def search(self, query_latent):
        results = []
        for i, dvec in enumerate(self.doc_embed):
            sim = self.cosine_similarity(query_latent, dvec)
            results.append((self.filenames[i], sim))
        results.sort(key=lambda x: x[1], reverse=True)
        return results


# =========================================================
# MAIN
# =========================================================
if __name__ == "__main__":
    # folder dataset sesuai struktur proyek Anda
    DATASET_FOLDER = "dataset"

    # validasi folder dataset
    if not os.path.isdir(DATASET_FOLDER):
        print("Folder dataset tidak ditemukan:", DATASET_FOLDER)
        exit()

    # 1. Load dataset
    docs, names = load_documents_from_folder(DATASET_FOLDER, MAX_WORDS)

    if len(docs) == 0:
        print("Tidak ada file .txt di folder dataset.")
        exit()

    # 2. Init engine
    engine = LSASearchEngine(docs, names)
    engine.build_term_document_matrix()
    engine.compute_tf_idf()
    engine.perform_lsa(K_LATENT)

    # 3. Input query file (default query.txt)
    query_path = input("Masukkan path file QUERY (.txt) [default: query.txt]: ").strip()
    if query_path == "":
        query_path = "query.txt"

    if not os.path.exists(query_path):
        print("File query tidak ditemukan:", query_path)
        exit()

    query_text = load_txt_file(query_path, MAX_WORDS)

    # 4. Folding-in + similarity
    q_latent = engine.fold_in_query(query_text)
    results = engine.search(q_latent)

    # 5. Output
    print("\n=== HASIL KEMIRIPAN ===")
    print("File Dataset | Similarity | Persentase")
    print("-" * 50)

    for fname, score in results:
        print(f"{fname:12s} | {score:.4f}     | {score*100:6.2f}%")