Bagian 4: Menulis Hasil dan Testing

"""Pengolah File CSV — Bagian 4: Menulis hasil dan testing."""

import csv
import sys
from datetime import datetime

KOLOM_WAJIB = ("tanggal", "produk", "kategori", "jumlah", "harga")


def baca_data(nama_file):
    """Baca seluruh baris file CSV dan kembalikan sebagai list of dict."""
    with open(nama_file, newline="", encoding="utf-8") as berkas:
        return list(csv.DictReader(berkas))


def hitung_omzet(baris):
    """Hitung omzet satu baris: jumlah dikali harga."""
    return int(baris["jumlah"]) * int(baris["harga"])


def validasi_baris(baris):
    """Periksa satu baris data dan kembalikan daftar pesan galat."""
    galat = []

    for kolom in KOLOM_WAJIB:
        nilai = baris.get(kolom)
        if nilai is None or nilai.strip() == "":
            galat.append(f"kolom {kolom} kosong")

    if baris.get("tanggal"):
        try:
            datetime.strptime(baris["tanggal"], "%Y-%m-%d")
        except ValueError:
            galat.append("tanggal tidak berformat YYYY-MM-DD")

    for kolom in ("jumlah", "harga"):
        nilai = baris.get(kolom)
        if nilai is None or nilai.strip() == "":
            continue
        try:
            angka = int(nilai)
        except ValueError:
            galat.append(f"nilai {kolom} bukan angka bulat: {nilai!r}")
        else:
            if angka <= 0:
                galat.append(f"nilai {kolom} harus lebih besar dari 0")

    return galat


def pisahkan_valid(data):
    """Pisahkan baris valid dari baris bermasalah."""
    valid = []
    galat_semua = []
    for nomor, baris in enumerate(data, start=2):
        galat = validasi_baris(baris)
        if galat:
            for pesan in galat:
                galat_semua.append(f"baris {nomor}: {pesan}")
        else:
            valid.append(baris)
    return valid, galat_semua


def ringkas_per_kategori(barisan):
    """Agregasi transaksi, jumlah item, dan omzet per kategori."""
    ringkasan = {}
    for baris in barisan:
        kategori = baris["kategori"]
        if kategori not in ringkasan:
            ringkasan[kategori] = {"transaksi": 0, "jumlah": 0, "omzet": 0}
        ringkasan[kategori]["transaksi"] += 1
        ringkasan[kategori]["jumlah"] += int(baris["jumlah"])
        ringkasan[kategori]["omzet"] += hitung_omzet(baris)
    return ringkasan


def jumlah_per_produk(barisan):
    """Total jumlah item terjual per produk."""
    hasil = {}
    for baris in barisan:
        hasil[baris["produk"]] = hasil.get(baris["produk"], 0) + int(baris["jumlah"])
    return hasil


def tulis_ringkasan_csv(ringkasan, nama_file):
    """Tulis ringkasan per kategori ke file CSV."""
    with open(nama_file, "w", newline="", encoding="utf-8") as berkas:
        kolom = ("kategori", "transaksi", "jumlah", "omzet", "rata_rata")
        penulis = csv.DictWriter(berkas, fieldnames=kolom)
        penulis.writeheader()
        urutan = sorted(ringkasan.items(), key=lambda x: x[1]["omzet"], reverse=True)
        for kategori, nilai in urutan:
            penulis.writerow(
                {
                    "kategori": kategori,
                    "transaksi": nilai["transaksi"],
                    "jumlah": nilai["jumlah"],
                    "omzet": nilai["omzet"],
                    "rata_rata": round(nilai["omzet"] / nilai["transaksi"]),
                }
            )


def buat_laporan(data, valid, galat_semua):
    """Susun laporan analisis sebagai teks siap simpan."""
    baris_laporan = []
    total_omzet = sum(hitung_omzet(baris) for baris in valid)
    tertinggi = max(valid, key=hitung_omzet)
    peringkat = sorted(jumlah_per_produk(valid).items(), key=lambda x: (-x[1], x[0]))

    baris_laporan.append("LAPORAN PENJUALAN")
    baris_laporan.append("=================")
    baris_laporan.append(f"Total baris       : {len(data)}")
    baris_laporan.append(f"Baris valid       : {len(valid)}")
    baris_laporan.append(f"Baris tidak valid : {len(data) - len(valid)}")
    baris_laporan.append("")
    baris_laporan.append(f"Total omzet        : Rp{total_omzet:,}")
    baris_laporan.append(f"Rata-rata transaksi: Rp{total_omzet / len(valid):,.2f}")
    baris_laporan.append(
        f"Transaksi tertinggi: {tertinggi['tanggal']} {tertinggi['produk']} "
        f"(Rp{hitung_omzet(tertinggi):,})"
    )
    baris_laporan.append(f"Produk terlaris    : {peringkat[0][0]} ({peringkat[0][1]} item)")
    baris_laporan.append("")
    baris_laporan.append("OMZET PER KATEGORI")
    baris_laporan.append("------------------")
    baris_laporan.append(f"{'Kategori':<12}{'Transaksi':>10}{'Jumlah':>8}{'Omzet':>12}")
    ringkasan = ringkas_per_kategori(valid)
    urutan = sorted(ringkasan.items(), key=lambda x: x[1]["omzet"], reverse=True)
    for kategori, nilai in urutan:
        baris_laporan.append(
            f"{kategori:<12}"
            f"{nilai['transaksi']:>10}"
            f"{nilai['jumlah']:>8}"
            f"{nilai['omzet']:>12,}"
        )
    baris_laporan.append("")
    if galat_semua:
        baris_laporan.append("BARIS DILEWATI")
        baris_laporan.append("--------------")
        for pesan in galat_semua:
            baris_laporan.append(f"- {pesan}")

    return "\n".join(baris_laporan) + "\n"


def main():
    if len(sys.argv) != 2:
        print("Pemakaian: python3 main.py <file.csv>")
        return 1

    nama_file = sys.argv[1]
    try:
        data = baca_data(nama_file)
    except FileNotFoundError:
        print(f"File tidak ditemukan: {nama_file}")
        return 1

    valid, galat_semua = pisahkan_valid(data)
    if not valid:
        print("Tidak ada baris valid untuk dianalisis.")
        return 1

    tulis_ringkasan_csv(ringkas_per_kategori(valid), "ringkasan.csv")

    with open("laporan.txt", "w", encoding="utf-8") as berkas:
        berkas.write(buat_laporan(data, valid, galat_semua))

    print(f"Baris valid       : {len(valid)}")
    print(f"Baris tidak valid : {len(data) - len(valid)}")
    print("Hasil ditulis ke  : ringkasan.csv, laporan.txt")

    return 0


if __name__ == "__main__":
    sys.exit(main())

"""Test untuk fungsi inti pengolah CSV."""

import csv

import pytest

from main import (
    buat_laporan,
    hitung_omzet,
    ringkas_per_kategori,
    tulis_ringkasan_csv,
    validasi_baris,
)


@pytest.fixture
def baris_valid():
    return {
        "tanggal": "2025-06-01",
        "produk": "Kopi Susu",
        "kategori": "Minuman",
        "jumlah": "3",
        "harga": "18000",
    }


@pytest.fixture
def data_penjualan():
    return [
        {
            "tanggal": "2025-06-01",
            "produk": "Kopi Susu",
            "kategori": "Minuman",
            "jumlah": "3",
            "harga": "18000",
        },
        {
            "tanggal": "2025-06-01",
            "produk": "Roti Cokelat",
            "kategori": "Makanan",
            "jumlah": "2",
            "harga": "12000",
        },
        {
            "tanggal": "2025-06-02",
            "produk": "Kopi Susu",
            "kategori": "Minuman",
            "jumlah": "1",
            "harga": "18000",
        },
    ]


def test_hitung_omzet(baris_valid):
    assert hitung_omzet(baris_valid) == 54000


def test_validasi_baris_valid(baris_valid):
    assert validasi_baris(baris_valid) == []


@pytest.mark.parametrize(
    ("perubahan", "pesan"),
    [
        ({"tanggal": "01/06/2025"}, "tanggal tidak berformat YYYY-MM-DD"),
        ({"jumlah": "-1"}, "nilai jumlah harus lebih besar dari 0"),
        ({"jumlah": "0"}, "nilai jumlah harus lebih besar dari 0"),
        ({"harga": "1x000"}, "nilai harga bukan angka bulat: '1x000'"),
        ({"produk": ""}, "kolom produk kosong"),
    ],
)
def test_validasi_baris_tidak_valid(baris_valid, perubahan, pesan):
    baris_valid.update(perubahan)
    assert pesan in validasi_baris(baris_valid)


def test_validasi_kolom_hilang(baris_valid):
    del baris_valid["harga"]
    assert "kolom harga kosong" in validasi_baris(baris_valid)


def test_ringkas_per_kategori(data_penjualan):
    ringkasan = ringkas_per_kategori(data_penjualan)

    assert ringkasan["Minuman"] == {"transaksi": 2, "jumlah": 4, "omzet": 72000}
    assert ringkasan["Makanan"] == {"transaksi": 1, "jumlah": 2, "omzet": 24000}


def test_tulis_ringkasan_csv(data_penjualan, tmp_path):
    ringkasan = ringkas_per_kategori(data_penjualan)
    berkas_csv = tmp_path / "ringkasan.csv"

    tulis_ringkasan_csv(ringkasan, str(berkas_csv))

    with berkas_csv.open(newline="", encoding="utf-8") as berkas:
        hasil = list(csv.DictReader(berkas))

    assert hasil == [
        {
            "kategori": "Minuman",
            "transaksi": "2",
            "jumlah": "4",
            "omzet": "72000",
            "rata_rata": "36000",
        },
        {
            "kategori": "Makanan",
            "transaksi": "1",
            "jumlah": "2",
            "omzet": "24000",
            "rata_rata": "24000",
        },
    ]


def test_buat_laporan_memuat_ringkasan(data_penjualan):
    laporan = buat_laporan(data_penjualan, data_penjualan, [])

    assert "Total baris       : 3" in laporan
    assert "Total omzet        : Rp96,000" in laporan
    assert "Produk terlaris    : Kopi Susu (4 item)" in laporan