"""End-to-End reconciliation service — business logic for actual losses tracking.""" from __future__ import annotations from datetime import date from decimal import Decimal from sqlalchemy import func, select from sqlalchemy.exc import IntegrityError, SQLAlchemyError from sqlalchemy.ext.asyncio import AsyncSession from app.models.end_to_end import EndToEndCycle, StatusEndToEnd from app.models.operational import LaporanShift, PenjualanNozzle from app.models.penerimaan import Penerimaan, PenerimaanItem from app.models.spbu import Nozzle, Tangki from app.repositories import end_to_end_repository from app.schemas.end_to_end import ( EndToEndClose, EndToEndCreate, EndToEndDetailResponse, EndToEndResponse, ) # --------------------------------------------------------------------------- # Internal helpers # --------------------------------------------------------------------------- def _build_response(c: EndToEndCycle) -> EndToEndResponse: """Build an EndToEndResponse from an ORM object.""" return EndToEndResponse( id=c.id, spbu_id=c.spbu_id, tangki_id=c.tangki_id, tangki_nama=c.tangki.nama if c.tangki else None, produk_nama=c.tangki.produk.nama if c.tangki and hasattr(c.tangki, "produk") and c.tangki.produk else None, status=c.status, tanggal_mulai=c.tanggal_mulai, tanggal_selesai=c.tanggal_selesai, dead_stock_awal=c.dead_stock_awal, dead_stock_akhir=c.dead_stock_akhir, total_penerimaan=c.total_penerimaan, total_penjualan=c.total_penjualan, total_pemindahan_in=c.total_pemindahan_in, total_pemindahan_out=c.total_pemindahan_out, losses_aktual=c.losses_aktual, losses_pct=c.losses_pct, started_by_name=c.started_by.name if c.started_by else None, closed_by_name=c.closed_by.name if c.closed_by else None, created_at=c.created_at, updated_at=c.updated_at, ) async def _calculate_running_totals( db: AsyncSession, tangki_id: int, start_date: date, end_date: date | None = None, ) -> dict: """Calculate live totals from penerimaan and penjualan data. Returns dict with keys: running_penerimaan, running_penjualan, running_pemindahan_in, running_pemindahan_out. """ effective_end = end_date or date.today() # Total penerimaan: SUM of PenerimaanItem.volume_diterima for this tangki pen_q = ( select(func.coalesce(func.sum(PenerimaanItem.volume_diterima), Decimal("0"))) .join(Penerimaan, PenerimaanItem.penerimaan_id == Penerimaan.id) .where( PenerimaanItem.tangki_id == tangki_id, Penerimaan.tanggal >= start_date, Penerimaan.tanggal <= effective_end, ) ) pen_result = await db.execute(pen_q) running_penerimaan = pen_result.scalar_one() or Decimal("0") # Total penjualan: SUM of PenjualanNozzle.volume for nozzles sourced from this tangki jual_q = ( select(func.coalesce(func.sum(PenjualanNozzle.volume), Decimal("0"))) .join(LaporanShift, PenjualanNozzle.laporan_shift_id == LaporanShift.id) .join(Nozzle, PenjualanNozzle.nozzle_id == Nozzle.id) .where( Nozzle.tangki_id == tangki_id, LaporanShift.tanggal >= start_date, LaporanShift.tanggal <= effective_end, ) ) jual_result = await db.execute(jual_q) running_penjualan = jual_result.scalar_one() or Decimal("0") # Pemindahan: not yet implemented — hardcode 0 return { "running_penerimaan": running_penerimaan, "running_penjualan": running_penjualan, "running_pemindahan_in": Decimal("0"), "running_pemindahan_out": Decimal("0"), } # --------------------------------------------------------------------------- # Public API # --------------------------------------------------------------------------- async def list_cycles( db: AsyncSession, spbu_id: int, tangki_id: int | None = None, status: str | None = None, skip: int = 0, limit: int = 50, ) -> tuple[list[EndToEndResponse], int]: """Return paginated list of E2E cycles for an SPBU.""" rows, total = await end_to_end_repository.get_all( db, spbu_id, tangki_id=tangki_id, status=status, skip=skip, limit=limit ) return [_build_response(c) for c in rows], total async def start_cycle( db: AsyncSession, spbu_id: int, user_id: int, data: EndToEndCreate, ) -> EndToEndResponse: """Start a new E2E cycle for a tank.""" # Validate tangki belongs to this SPBU tangki = await db.get(Tangki, data.tangki_id) if tangki is None or tangki.spbu_id != spbu_id: raise ValueError("Tangki tidak ditemukan untuk SPBU ini") if tangki.is_active is False: raise ValueError("Tangki tidak aktif") # Check no open cycle existing = await end_to_end_repository.get_open_by_tangki(db, data.tangki_id) if existing is not None: raise ValueError( f"Tangki '{tangki.nama}' sudah memiliki siklus E2E yang sedang berjalan " f"(dimulai {existing.tanggal_mulai})" ) try: cycle = await end_to_end_repository.create( db, { "spbu_id": spbu_id, "tangki_id": data.tangki_id, "status": StatusEndToEnd.OPEN, "tanggal_mulai": date.today(), "dead_stock_awal": data.dead_stock_awal, "started_by_id": user_id, }, ) await db.commit() except IntegrityError: await db.rollback() raise ValueError("Siklus E2E konflik atau sudah ada") except SQLAlchemyError: await db.rollback() raise return _build_response(cycle) async def get_cycle_detail( db: AsyncSession, spbu_id: int, cycle_id: int, ) -> EndToEndDetailResponse: """Get full detail of an E2E cycle with running totals.""" cycle = await end_to_end_repository.get_by_id(db, cycle_id, spbu_id) if cycle is None: raise ValueError("Siklus E2E tidak ditemukan") base = _build_response(cycle) if cycle.status == StatusEndToEnd.OPEN: # Live calculation totals = await _calculate_running_totals( db, cycle.tangki_id, cycle.tanggal_mulai ) else: # Frozen — use stored values totals = { "running_penerimaan": cycle.total_penerimaan, "running_penjualan": cycle.total_penjualan, "running_pemindahan_in": cycle.total_pemindahan_in, "running_pemindahan_out": cycle.total_pemindahan_out, } return EndToEndDetailResponse( **base.model_dump(), **totals, ) async def close_cycle( db: AsyncSession, spbu_id: int, cycle_id: int, user_id: int, data: EndToEndClose, ) -> EndToEndDetailResponse: """Close an open E2E cycle, freeze totals and calculate losses.""" cycle = await end_to_end_repository.get_by_id(db, cycle_id, spbu_id) if cycle is None: raise ValueError("Siklus E2E tidak ditemukan") if cycle.status != StatusEndToEnd.OPEN: raise ValueError("Siklus sudah ditutup") # Calculate final totals totals = await _calculate_running_totals( db, cycle.tangki_id, cycle.tanggal_mulai ) total_in = totals["running_penerimaan"] + totals["running_pemindahan_in"] total_out = totals["running_penjualan"] + totals["running_pemindahan_out"] dead_stock_delta = data.dead_stock_akhir - cycle.dead_stock_awal losses_aktual = total_in - total_out - dead_stock_delta losses_pct = ( (losses_aktual / totals["running_penerimaan"] * 100) if totals["running_penerimaan"] > 0 else Decimal("0") ) try: updated = await end_to_end_repository.update( db, cycle, { "status": StatusEndToEnd.CLOSED, "tanggal_selesai": date.today(), "dead_stock_akhir": data.dead_stock_akhir, "total_penerimaan": totals["running_penerimaan"], "total_penjualan": totals["running_penjualan"], "total_pemindahan_in": totals["running_pemindahan_in"], "total_pemindahan_out": totals["running_pemindahan_out"], "losses_aktual": losses_aktual, "losses_pct": losses_pct, "closed_by_id": user_id, }, ) await db.commit() except SQLAlchemyError: await db.rollback() raise return await get_cycle_detail(db, spbu_id, updated.id)