TIF_E41253368/api/app/data/repository.py

from abc import ABC, abstractmethod
from datetime import date
from pathlib import Path
from typing import Dict

import numpy as np
import pandas as pd
import pymysql
from pymysql import MySQLError


class DataUnavailableError(Exception):
    pass


class PriceRepository(ABC):
    @abstractmethod
    def get_series(self, commodity: str, market: str) -> pd.Series:
        raise NotImplementedError

    @abstractmethod
    def last_data_date(self, commodity: str, market: str) -> date:
        raise NotImplementedError


class CsvPriceRepository(PriceRepository):
    FILE_PATTERNS = {
        "rawit": ("*rawit*.csv",),
        "merah": ("*cabe_merah_besar*.csv", "*merah_besar*.csv"),
        "keriting": ("*merah_keriting*.csv",),
    }

    def __init__(self, data_dir: Path, unit_scale: float = 1.0) -> None:
        self.data_dir = data_dir
        self.unit_scale = unit_scale
        self._frames: Dict[str, pd.DataFrame] = {}

    def _find_file(self, commodity: str) -> Path:
        candidates = [self.data_dir / f"{commodity}.csv", self.data_dir / f"data_{commodity}.csv"]
        for pattern in self.FILE_PATTERNS.get(commodity, (f"*{commodity}*.csv",)):
            candidates.extend(sorted(self.data_dir.rglob(pattern))) if self.data_dir.exists() else None
        for candidate in candidates:
            if candidate.is_file():
                return candidate
        raise DataUnavailableError(
            f"File CSV untuk komoditas {commodity} tidak ditemukan di {self.data_dir}."
        )

    def _load(self, commodity: str) -> pd.DataFrame:
        if commodity in self._frames:
            return self._frames[commodity]
        path = self._find_file(commodity)
        frame = pd.read_csv(path)
        date_column = next(
            (column for column in frame.columns if str(column).strip().lower() in {"date", "tanggal", "datetime"}),
            None,
        )
        if date_column is None:
            raise DataUnavailableError(
                f"CSV {path.name} harus memiliki kolom date atau tanggal."
            )
        frame[date_column] = pd.to_datetime(frame[date_column], errors="coerce")
        frame = frame.dropna(subset=[date_column]).set_index(date_column).sort_index()
        if not frame.empty:
            frame = frame[~frame.index.duplicated(keep="last")]
            frame = frame.reindex(pd.date_range(frame.index.min(), frame.index.max(), freq="D"))
        self._frames[commodity] = frame
        return frame

    @staticmethod
    def _market_column(frame: pd.DataFrame, market: str) -> str:
        aliases = {market, market.replace("pasar_", "Pasar_")}
        for column in frame.columns:
            normalized = str(column).strip().lower().replace(" ", "_")
            if normalized in {alias.lower() for alias in aliases}:
                return column
        raise DataUnavailableError(f"Kolom harga untuk pasar {market} tidak ditemukan.")

    def get_series(self, commodity: str, market: str) -> pd.Series:
        frame = self._load(commodity)
        column = self._market_column(frame, market)
        series = pd.to_numeric(frame[column], errors="coerce").replace(0, np.nan)
        series = series * self.unit_scale
        series = series.interpolate(method="linear", limit=2, limit_area="inside")
        return series.astype(float)

    def last_data_date(self, commodity: str, market: str) -> date:
        series = self.get_series(commodity, market).dropna()
        if series.empty:
            raise DataUnavailableError(
                f"Tidak ada data valid untuk {commodity}/{market}."
            )
        return series.index[-1].date()


class MySQLPriceRepository(PriceRepository):
    def __init__(self, config) -> None:
        self.config = config

    def _connect(self):
        return pymysql.connect(
            host=self.config["MYSQL_HOST"],
            port=self.config["MYSQL_PORT"],
            database=self.config["MYSQL_DATABASE"],
            user=self.config["MYSQL_USER"],
            password=self.config["MYSQL_PASSWORD"],
            connect_timeout=self.config["MYSQL_CONNECT_TIMEOUT"],
        )

    def _load(self, commodity: str, market: str) -> pd.Series:
        connection = None
        cursor = None
        try:
            connection = self._connect()
            cursor = connection.cursor()
            cursor.execute(
                """
                SELECT ph.price_date, ph.price
                FROM price_history AS ph
                INNER JOIN commodities AS c ON c.id = ph.commodity_id
                INNER JOIN markets AS m ON m.id = ph.market_id
                WHERE c.slug = %s AND m.slug = %s
                ORDER BY ph.price_date ASC
                """,
                (commodity, market),
            )
            rows = cursor.fetchall()
        except MySQLError as error:
            raise DataUnavailableError(f"Gagal membaca database MySQL: {error}") from error
        finally:
            if cursor is not None:
                cursor.close()
            if connection is not None:
                connection.close()

        if not rows:
            raise DataUnavailableError(
                f"Tidak ada data harga untuk {commodity}/{market} di database."
            )
        frame = pd.DataFrame(rows, columns=["date", "price"])
        frame["date"] = pd.to_datetime(frame["date"], errors="coerce")
        frame["price"] = pd.to_numeric(frame["price"], errors="coerce")
        frame = frame.dropna(subset=["date"]).set_index("date").sort_index()
        frame = frame[~frame.index.duplicated(keep="last")]
        frame["price"] = frame["price"].replace(0, np.nan)
        frame["price"] = frame["price"].interpolate(method="linear", limit=2, limit_area="inside")
        return frame["price"].astype(float)

    def get_series(self, commodity: str, market: str) -> pd.Series:
        return self._load(commodity, market)

    def last_data_date(self, commodity: str, market: str) -> date:
        series = self.get_series(commodity, market).dropna()
        if series.empty:
            raise DataUnavailableError(
                f"Tidak ada data valid untuk {commodity}/{market} di database."
            )
        return series.index[-1].date()