""" Anomali detection utilities. Called after laporan shift / stock adjustment is submitted or approved. Creates AnomaliRecord entries for flagging — does not block transactions. """ from decimal import Decimal from typing import Any from sqlalchemy import select, func from sqlalchemy.ext.asyncio import AsyncSession from app.models.anomali import AnomaliRecord, StatusAnomali, TipeAnomali from app.models.operational import LaporanShift, PenjualanNozzle from app.models.operational import StockAdjustment, StockAdjustmentItem from app.models.product import Produk from app.models.spbu import Nozzle, Spbu, Tangki async def _create_anomali( db: AsyncSession, spbu_id: int, tipe: str, detail: dict[str, Any], deskripsi: str | None = None, produk_id: int | None = None, nozzle_id: int | None = None, laporan_shift_id: int | None = None, plat_nomor: str | None = None, ) -> None: """Insert an anomaly record (idempotent-friendly: always creates new).""" record = AnomaliRecord( spbu_id=spbu_id, tipe=tipe, laporan_shift_id=laporan_shift_id, produk_id=produk_id, nozzle_id=nozzle_id, plat_nomor=plat_nomor, deskripsi=deskripsi, detail=detail, status=StatusAnomali.NEW, ) db.add(record) async def check_meter_discrepancy( db: AsyncSession, laporan: LaporanShift, ) -> None: """ METER_DISCREPANCY: fired when |manual_volume - digital_volume| > threshold %. Threshold from SPBU config: teller_discrepancy_threshold_pct (default 0.3%). Minimum absolute difference: 1 liter. """ spbu_result = await db.execute(select(Spbu).where(Spbu.id == laporan.spbu_id)) spbu = spbu_result.scalar_one_or_none() if not spbu: return threshold_pct = Decimal(str(spbu.teller_discrepancy_threshold_pct or "0.300")) for pn in (laporan.penjualan_nozzle or []): manual_vol = (pn.teller_akhir_manual or Decimal(0)) - (pn.teller_awal_manual or Decimal(0)) digital_vol = (pn.teller_akhir_digital or Decimal(0)) - (pn.teller_awal_digital or Decimal(0)) if pn.flag_reset_teller: continue # skip reset cases if manual_vol <= 0 and digital_vol <= 0: continue reference = max(manual_vol, digital_vol) if reference <= 0: continue diff = abs(manual_vol - digital_vol) if diff < 1: continue # minimum 1 liter pct = (diff / reference) * 100 if pct > threshold_pct: nozzle_nama = pn.nozzle.nama if pn.nozzle else f"Nozzle #{pn.nozzle_id}" await _create_anomali( db, spbu_id=laporan.spbu_id, tipe=TipeAnomali.METER_DISCREPANCY, laporan_shift_id=laporan.id, nozzle_id=pn.nozzle_id, produk_id=pn.nozzle.produk_id if pn.nozzle else None, deskripsi=( f"Selisih teller manual vs digital {nozzle_nama}: " f"{round(pct, 3)}% ({diff:.3f} L) melebihi threshold {threshold_pct}%" ), detail={ "laporan_shift_id": laporan.id, "nozzle_id": pn.nozzle_id, "nozzle_nama": nozzle_nama, "manual_volume": str(manual_vol), "digital_volume": str(digital_vol), "selisih": str(diff), "selisih_pct": str(round(pct, 3)), "threshold_pct": str(threshold_pct), "tanggal": str(laporan.tanggal), "shift_id": laporan.shift_id, }, ) async def check_losses_exceeded( db: AsyncSession, spbu_id: int, tangki_id: int, stok_teoritis: Decimal, stok_aktual: Decimal, total_penjualan: Decimal, total_penerimaan: Decimal, tanggal: str, ) -> None: """ LOSSES_EXCEEDED: fired when losses per tank exceed threshold. Thresholds from master_produk: losses_threshold_penjualan_pct, losses_threshold_penerimaan_pct. """ losses = stok_aktual - stok_teoritis if losses >= 0: return # gain or balanced, no anomaly abs_losses = abs(losses) # Get tangki → produk → thresholds tangki_result = await db.execute( select(Tangki).where(Tangki.id == tangki_id) ) tangki = tangki_result.scalar_one_or_none() if not tangki or not tangki.produk_id: return produk_result = await db.execute( select(Produk).where(Produk.id == tangki.produk_id) ) produk = produk_result.scalar_one_or_none() if not produk: return triggered = False # Check penjualan threshold if total_penjualan > 0: penjualan_threshold = Decimal(str(produk.losses_threshold_penjualan_pct or "0.500")) penjualan_pct = (abs_losses / total_penjualan) * 100 if penjualan_pct > penjualan_threshold: triggered = True # Check penerimaan threshold if total_penerimaan > 0: penerimaan_threshold = Decimal(str(produk.losses_threshold_penerimaan_pct or "0.150")) penerimaan_pct = (abs_losses / total_penerimaan) * 100 if penerimaan_pct > penerimaan_threshold: triggered = True if triggered: await _create_anomali( db, spbu_id=spbu_id, tipe=TipeAnomali.LOSSES_EXCEEDED, produk_id=tangki.produk_id, detail={ "tangki_id": tangki_id, "tangki_nama": tangki.nama, "produk_nama": produk.nama, "stok_teoritis": str(stok_teoritis), "stok_aktual": str(stok_aktual), "losses": str(losses), "total_penjualan": str(total_penjualan), "total_penerimaan": str(total_penerimaan), "tanggal": tanggal, }, ) async def check_negative_stock( db: AsyncSession, spbu_id: int, tangki_id: int, stok_teoritis: Decimal, tanggal: str, ) -> None: """NEGATIVE_STOCK: fired when theoretical stock goes below zero.""" if stok_teoritis >= 0: return tangki_result = await db.execute( select(Tangki).where(Tangki.id == tangki_id) ) tangki = tangki_result.scalar_one_or_none() await _create_anomali( db, spbu_id=spbu_id, tipe=TipeAnomali.NEGATIVE_STOCK, produk_id=tangki.produk_id if tangki else None, detail={ "tangki_id": tangki_id, "tangki_nama": tangki.nama if tangki else None, "stok_teoritis": str(stok_teoritis), "tanggal": tanggal, }, ) async def run_penjualan_checks(db: AsyncSession, laporan: LaporanShift) -> None: """Run all applicable anomali checks after a laporan shift is submitted/approved.""" await check_meter_discrepancy(db, laporan) async def run_rekonsiliasi_checks( db: AsyncSession, spbu_id: int, tangki_id: int, stok_teoritis: Decimal, stok_aktual: Decimal, total_penjualan: Decimal, total_penerimaan: Decimal, tanggal: str, ) -> None: """Run anomali checks after rekonsiliasi calculation.""" await check_negative_stock(db, spbu_id, tangki_id, stok_teoritis, tanggal) await check_losses_exceeded( db, spbu_id, tangki_id, stok_teoritis, stok_aktual, total_penjualan, total_penerimaan, tanggal, )