"""Telegram notification service. All notify_* functions are fire-and-forget — call via FastAPI BackgroundTasks. They open their own DB session so the request lifecycle doesn't affect them. Errors are logged and swallowed; Telegram failures must never break the main flow. """ from __future__ import annotations import json import logging from collections import defaultdict from datetime import datetime, timedelta, timezone from decimal import Decimal from pathlib import Path import httpx from sqlalchemy import select from sqlalchemy.orm import selectinload from app.core.config import settings from app.core.database import AsyncSessionLocal from app.models.telegram_config import TelegramConfig logger = logging.getLogger(__name__) TELEGRAM_API = "https://api.telegram.org/bot{token}/{method}" WIB = timezone(timedelta(hours=7)) # ── Low-level helpers ────────────────────────────────────────────────────────── def _api(method: str) -> str: return TELEGRAM_API.format(token=settings.TELEGRAM_BOT_TOKEN, method=method) def _fmt_rp(value: Decimal | float | int) -> str: return f"Rp {int(value):,}".replace(",", ".") def _fmt_vol(value: Decimal | float) -> str: return f"{float(value):,.1f} L".replace(",", "X").replace(".", ",").replace("X", ".") def _wib(dt) -> str: if not dt: return "—" return dt.astimezone(WIB).strftime("%d %b %Y, %H:%M WIB") def _read_file_bytes(stored_url: str) -> bytes | None: """Read file bytes from local storage. Returns None if file not found or GDrive.""" if not stored_url: return None if stored_url.startswith("https://"): return None # GDrive — skip try: rel = stored_url.lstrip("/") path = Path(rel) if path.exists(): return path.read_bytes() except Exception as e: logger.warning("telegram: failed to read file %s — %s", stored_url, e) return None async def _send_text(chat_id: str, thread_id: int | None, text: str) -> None: payload: dict = {"chat_id": chat_id, "text": text, "parse_mode": "HTML"} if thread_id: payload["message_thread_id"] = thread_id async with httpx.AsyncClient(timeout=10) as client: resp = await client.post(_api("sendMessage"), json=payload) resp.raise_for_status() async def _send_document( chat_id: str, thread_id: int | None, caption: str, filename: str, file_bytes: bytes, content_type: str = "application/pdf", ) -> None: """Send a file document to a Telegram chat.""" data_fields: dict = {"chat_id": chat_id, "caption": caption, "parse_mode": "HTML"} if thread_id: data_fields["message_thread_id"] = str(thread_id) files = {"document": (filename, file_bytes, content_type)} async with httpx.AsyncClient(timeout=30) as client: resp = await client.post(_api("sendDocument"), data=data_fields, files=files) resp.raise_for_status() async def _send_media_group( chat_id: str, thread_id: int | None, caption: str, file_bytes_list: list[tuple[str, bytes]], ) -> None: """Send up to 10 photos as a media group. First photo gets the caption.""" if not file_bytes_list: return media = [] files: dict = {} for i, (filename, data) in enumerate(file_bytes_list[:10]): key = f"photo_{i}" media.append({ "type": "photo", "media": f"attach://{key}", **({"caption": caption, "parse_mode": "HTML"} if i == 0 else {}), }) files[key] = (filename, data, "image/jpeg") data_fields: dict = {"chat_id": chat_id, "media": json.dumps(media)} if thread_id: data_fields["message_thread_id"] = str(thread_id) async with httpx.AsyncClient(timeout=30) as client: resp = await client.post(_api("sendMediaGroup"), data=data_fields, files=files) resp.raise_for_status() def _collect_photos(urls: list[str | None]) -> list[tuple[str, bytes]]: result: list[tuple[str, bytes]] = [] for url in urls: if not url: continue data = _read_file_bytes(url) if data: result.append((url.split("/")[-1], data)) return result # ── Config loader ────────────────────────────────────────────────────────────── async def _get_config(spbu_id: int) -> TelegramConfig | None: async with AsyncSessionLocal() as db: result = await db.execute( select(TelegramConfig).where( TelegramConfig.spbu_id == spbu_id, TelegramConfig.is_active.is_(True), ) ) return result.scalar_one_or_none() def _link(path: str) -> str: return f'Buka →' def _spbu_header(spbu) -> str: if spbu is None: return "SPBU" name = getattr(spbu, "name", "SPBU") nomor = getattr(spbu, "nomor_pertamina", "") return f"{name} — {nomor}" if nomor else name async def _load_spbu(db, spbu_id: int): from app.models.spbu import Spbu result = await db.execute(select(Spbu).where(Spbu.id == spbu_id)) return result.scalar_one_or_none() # ── Generic unlock helper ────────────────────────────────────────────────────── def _unlock_text(icon: str, modul: str, spbu_hdr: str, tanggal_str: str, extra: str, unlocked_by: str, alasan: str, unlocked_at) -> str: return ( f"{icon} {modul} Di-Unlock\n\n" f"🏪 {spbu_hdr}\n" f"📅 {tanggal_str}\n" f"{extra}" f"🔓 Di-unlock oleh : {unlocked_by}\n" f"📝 Alasan : {alasan}\n" f"🕐 Waktu : {_wib(unlocked_at)}" ) # ── Penjualan ────────────────────────────────────────────────────────────────── async def notify_penjualan_submitted(spbu_id: int, laporan_id: int) -> None: if not settings.TELEGRAM_BOT_TOKEN or not settings.is_production: return try: cfg = await _get_config(spbu_id) if not cfg or not cfg.chat_operasional_id or not cfg.thread_penjualan_id: return await _do_notify_penjualan(cfg, spbu_id, laporan_id, event="submitted") except Exception as e: logger.error("telegram notify_penjualan_submitted error: %s", e) async def notify_penjualan_approved(spbu_id: int, laporan_id: int) -> None: if not settings.TELEGRAM_BOT_TOKEN or not settings.is_production: return try: cfg = await _get_config(spbu_id) if not cfg or not cfg.chat_operasional_id or not cfg.thread_penjualan_id: return await _do_notify_penjualan(cfg, spbu_id, laporan_id, event="approved") except Exception as e: logger.error("telegram notify_penjualan_approved error: %s", e) async def notify_penjualan_unlocked(spbu_id: int, laporan_id: int) -> None: if not settings.TELEGRAM_BOT_TOKEN or not settings.is_production: return try: cfg = await _get_config(spbu_id) if not cfg or not cfg.chat_operasional_id or not cfg.thread_penjualan_id: return from app.models.operational import LaporanShift async with AsyncSessionLocal() as db: result = await db.execute( select(LaporanShift) .where(LaporanShift.id == laporan_id, LaporanShift.spbu_id == spbu_id) .options(selectinload(LaporanShift.spbu), selectinload(LaporanShift.shift), selectinload(LaporanShift.unlocked_by)) ) laporan = result.scalar_one_or_none() if not laporan: return shift_nama = laporan.shift.nama if laporan.shift else "—" tanggal_str = laporan.tanggal.strftime("%d %b %Y") if laporan.tanggal else "—" text = _unlock_text( "🔓", "Laporan Shift", _spbu_header(laporan.spbu), f"{tanggal_str} · {shift_nama}", "", laporan.unlocked_by.name if laporan.unlocked_by else "—", laporan.unlock_reason or "—", laporan.unlocked_at, ) text += f"\n🔗 {_link(f'/penjualan/{laporan_id}')}" await _send_text(cfg.chat_operasional_id, cfg.thread_penjualan_id, text) except Exception as e: logger.error("telegram notify_penjualan_unlocked error: %s", e) async def _do_notify_penjualan( cfg: TelegramConfig, spbu_id: int, laporan_id: int, event: str ) -> None: from app.models.operational import LaporanShift, PenjualanNozzle from app.models.spbu import Nozzle async with AsyncSessionLocal() as db: result = await db.execute( select(LaporanShift) .where(LaporanShift.id == laporan_id, LaporanShift.spbu_id == spbu_id) .options( selectinload(LaporanShift.spbu), selectinload(LaporanShift.shift), selectinload(LaporanShift.submitted_by), selectinload(LaporanShift.reviewed_by), selectinload(LaporanShift.penjualan_nozzle).selectinload( PenjualanNozzle.nozzle ).selectinload(Nozzle.produk), selectinload(LaporanShift.island_fotos), ) ) laporan = result.scalar_one_or_none() if not laporan: return spbu_hdr = _spbu_header(laporan.spbu) shift_nama = laporan.shift.nama if laporan.shift else "—" jam_mulai = laporan.shift.jam_mulai.strftime("%H:%M") if laporan.shift else "" jam_selesai = laporan.shift.jam_selesai.strftime("%H:%M") if laporan.shift else "" tanggal_str = laporan.tanggal.strftime("%A, %d %b %Y") if laporan.tanggal else "—" produk_totals: dict[str, dict] = defaultdict(lambda: {"volume": Decimal(0), "nilai": Decimal(0)}) for pn in laporan.penjualan_nozzle: nama = (pn.nozzle.produk.nama if pn.nozzle and pn.nozzle.produk else "?") produk_totals[nama]["volume"] += pn.volume produk_totals[nama]["nilai"] += pn.nilai total_volume = sum(v["volume"] for v in produk_totals.values()) total_nilai = sum(v["nilai"] for v in produk_totals.values()) produk_lines = "\n".join( f"⛽ {nama:<12} {_fmt_vol(v['volume']):<14} {_fmt_rp(v['nilai'])}" for nama, v in produk_totals.items() ) if event == "submitted": kas_tunai = ( laporan.kas_100k * 100000 + laporan.kas_50k * 50000 + laporan.kas_20k * 20000 + laporan.kas_10k * 10000 + laporan.kas_5k * 5000 + laporan.kas_2k * 2000 + laporan.kas_1k * 1000 + laporan.kas_logam ) operator_name = laporan.submitted_by.name if laporan.submitted_by else "—" text = ( f"📋 Laporan Shift Masuk\n\n" f"🏪 {spbu_hdr}\n" f"📅 {tanggal_str}\n" f"⏰ {shift_nama} ({jam_mulai} – {jam_selesai})\n" f"👤 Operator: {operator_name}\n\n" f"──────────────────────\n" f"PENJUALAN PER PRODUK\n" f"──────────────────────\n" f"{produk_lines}\n" f"──────────────────────\n" f"📦 Total Volume {_fmt_vol(total_volume)}\n" f"💵 Total Nilai {_fmt_rp(total_nilai)}\n" f"──────────────────────\n\n" f"💳 KAS & PEMBAYARAN\n" f"Tunai {_fmt_rp(kas_tunai)}\n" f"Kartu {_fmt_rp(laporan.pembayaran_kartu)}\n" f"QR Code {_fmt_rp(laporan.pembayaran_qr)}\n" f"Instansi {_fmt_rp(laporan.pembayaran_instansi)}\n\n" f"🕐 Submit: {_wib(laporan.submitted_at)}\n" f"🔗 {_link(f'/penjualan/{laporan_id}')}" ) await _send_text(cfg.chat_operasional_id, cfg.thread_penjualan_id, text) photos = _collect_photos([f.foto_url for f in laporan.island_fotos]) if photos: caption = f"📸 Foto Island — {spbu_hdr}\n{shift_nama} · {tanggal_str}" await _send_media_group(cfg.chat_operasional_id, cfg.thread_penjualan_id, caption, photos) elif event == "approved": approver_name = laporan.reviewed_by.name if laporan.reviewed_by else "—" operator_name = laporan.submitted_by.name if laporan.submitted_by else "—" text = ( f"✅ Laporan Shift Disetujui\n\n" f"🏪 {spbu_hdr}\n" f"📅 {tanggal_str} · {shift_nama}\n\n" f"──────────────────────\n" f"💵 Total Nilai {_fmt_rp(total_nilai)}\n" f"📦 Total Volume {_fmt_vol(total_volume)}\n" f"──────────────────────\n" f"{produk_lines}\n\n" f"👤 Operator : {operator_name}\n" f"✍️ Disetujui : {approver_name}\n" f"🕐 Waktu : {_wib(laporan.reviewed_at)}\n" f"🔗 {_link(f'/penjualan/{laporan_id}')}" ) await _send_text(cfg.chat_operasional_id, cfg.thread_penjualan_id, text) # ── Penerimaan ───────────────────────────────────────────────────────────────── async def notify_penerimaan_submitted(spbu_id: int, penerimaan_id: int) -> None: if not settings.TELEGRAM_BOT_TOKEN or not settings.is_production: return try: cfg = await _get_config(spbu_id) if not cfg or not cfg.chat_operasional_id or not cfg.thread_penerimaan_id: return await _do_notify_penerimaan(cfg, spbu_id, penerimaan_id, event="submitted") except Exception as e: logger.error("telegram notify_penerimaan_submitted error: %s", e) async def notify_penerimaan_approved(spbu_id: int, penerimaan_id: int) -> None: if not settings.TELEGRAM_BOT_TOKEN or not settings.is_production: return try: cfg = await _get_config(spbu_id) if not cfg or not cfg.chat_operasional_id or not cfg.thread_penerimaan_id: return await _do_notify_penerimaan(cfg, spbu_id, penerimaan_id, event="approved") except Exception as e: logger.error("telegram notify_penerimaan_approved error: %s", e) async def _do_notify_penerimaan( cfg: TelegramConfig, spbu_id: int, penerimaan_id: int, event: str ) -> None: from app.models.penerimaan import Penerimaan, PenerimaanFoto, PenerimaanItem async with AsyncSessionLocal() as db: result = await db.execute( select(Penerimaan) .where(Penerimaan.id == penerimaan_id, Penerimaan.spbu_id == spbu_id) .options( selectinload(Penerimaan.submitted_by), selectinload(Penerimaan.reviewed_by), selectinload(Penerimaan.items).selectinload(PenerimaanItem.produk), selectinload(Penerimaan.fotos), ) ) p = result.scalar_one_or_none() if not p: return spbu = await _load_spbu(db, spbu_id) spbu_hdr = _spbu_header(spbu) tanggal_str = p.tanggal.strftime("%d %b %Y") if p.tanggal else "—" total_diterima = sum(item.volume_diterima for item in p.items) def _penerimaan_line(item) -> str: produk = item.produk.nama if item.produk else "?" diterima = _fmt_vol(item.volume_diterima) expected = _fmt_vol(item.volume_expected) if item.volume_expected else "—" return f"🛢 {produk:<12} {diterima:<16} exp: {expected}" item_lines = "\n".join(_penerimaan_line(item) for item in p.items) if event == "submitted": operator_name = p.submitted_by.name if p.submitted_by else "—" no_polisi = p.no_polisi or "—" text = ( f"🚚 Penerimaan BBM Masuk\n\n" f"🏪 {spbu_hdr}\n" f"📅 {tanggal_str}\n" f"🚛 Kendaraan: {no_polisi}\n" f"👤 Petugas: {operator_name}\n\n" f"──────────────────────\n" f"VOLUME DITERIMA\n" f"──────────────────────\n" f"{item_lines}\n" f"──────────────────────\n" f"📦 Total: {_fmt_vol(total_diterima)}\n\n" f"🕐 Submit: {_wib(p.submitted_at)}\n" f"🔗 {_link(f'/penerimaan/{penerimaan_id}')}" ) await _send_text(cfg.chat_operasional_id, cfg.thread_penerimaan_id, text) vehicle_urls = [f.url for f in p.fotos if f.tipe in ("truck", "surat_jalan", "compartment_buka", "compartment_kosong")] vehicle_photos = _collect_photos(vehicle_urls) if vehicle_photos: caption = f"📸 Foto Kendaraan & Kompartemen — {spbu_hdr}\n{tanggal_str}" await _send_media_group(cfg.chat_operasional_id, cfg.thread_penerimaan_id, caption, vehicle_photos) dipstick_urls = [f.url for f in p.fotos if f.tipe in ("dipstick_sebelum", "dipstick_sesudah", "stick_awal", "stick_akhir")] dipstick_photos = _collect_photos(dipstick_urls) if dipstick_photos: caption = f"📏 Foto Dipstick Tangki — {spbu_hdr}\n{tanggal_str}" await _send_media_group(cfg.chat_operasional_id, cfg.thread_penerimaan_id, caption, dipstick_photos) elif event == "approved": approver_name = p.reviewed_by.name if p.reviewed_by else "—" operator_name = p.submitted_by.name if p.submitted_by else "—" text = ( f"✅ Penerimaan BBM Disetujui\n\n" f"🏪 {spbu_hdr}\n" f"📅 {tanggal_str}\n\n" f"──────────────────────\n" f"{item_lines}\n" f"──────────────────────\n" f"📦 Total: {_fmt_vol(total_diterima)}\n\n" f"👤 Petugas : {operator_name}\n" f"✍️ Disetujui : {approver_name}\n" f"🕐 Waktu : {_wib(p.reviewed_at)}\n" f"🔗 {_link(f'/penerimaan/{penerimaan_id}')}" ) await _send_text(cfg.chat_operasional_id, cfg.thread_penerimaan_id, text) async def notify_penerimaan_unlocked(spbu_id: int, penerimaan_id: int) -> None: if not settings.TELEGRAM_BOT_TOKEN or not settings.is_production: return try: cfg = await _get_config(spbu_id) if not cfg or not cfg.chat_operasional_id or not cfg.thread_penerimaan_id: return from app.models.penerimaan import Penerimaan async with AsyncSessionLocal() as db: result = await db.execute( select(Penerimaan) .where(Penerimaan.id == penerimaan_id, Penerimaan.spbu_id == spbu_id) .options(selectinload(Penerimaan.unlocked_by)) ) p = result.scalar_one_or_none() if not p: return spbu = await _load_spbu(db, spbu_id) tanggal_str = p.tanggal.strftime("%d %b %Y") if p.tanggal else "—" no_polisi = f"🚛 Kendaraan : {p.no_polisi}\n" if p.no_polisi else "" text = _unlock_text( "🔓", "Penerimaan BBM", _spbu_header(spbu), tanggal_str, no_polisi, p.unlocked_by.name if p.unlocked_by else "—", p.unlock_reason or "—", p.unlocked_at, ) text += f"\n🔗 {_link(f'/penerimaan/{penerimaan_id}')}" await _send_text(cfg.chat_operasional_id, cfg.thread_penerimaan_id, text) except Exception as e: logger.error("telegram notify_penerimaan_unlocked error: %s", e) async def notify_penerimaan_losses_alert(spbu_id: int, penerimaan_id: int) -> None: """Alert ke thread_penerimaan_id jika ada losses pengiriman melebihi threshold.""" if not settings.TELEGRAM_BOT_TOKEN or not settings.is_production: return try: cfg = await _get_config(spbu_id) if not cfg or not cfg.chat_operasional_id or not cfg.thread_penerimaan_id: return await _do_notify_penerimaan_losses(cfg, spbu_id, penerimaan_id) except Exception as e: logger.error("telegram notify_penerimaan_losses_alert error: %s", e) async def _do_notify_penerimaan_losses( cfg: TelegramConfig, spbu_id: int, penerimaan_id: int ) -> None: from app.models.penerimaan import Penerimaan, PenerimaanItem from app.models.product import Produk async with AsyncSessionLocal() as db: result = await db.execute( select(Penerimaan) .where(Penerimaan.id == penerimaan_id, Penerimaan.spbu_id == spbu_id) .options( selectinload(Penerimaan.items).selectinload(PenerimaanItem.tangki), ) ) p = result.scalar_one_or_none() if not p: return spbu = await _load_spbu(db, spbu_id) # Fetch thresholds per produk produk_ids = [ item.tangki.produk_id for item in p.items if item.tangki and item.tangki.produk_id ] produk_map: dict[int, Produk] = {} if produk_ids: pr = await db.execute(select(Produk).where(Produk.id.in_(produk_ids))) for prod in pr.scalars().all(): produk_map[prod.id] = prod # Extract item data before session closes item_snapshots = [] for item in p.items: tangki = item.tangki if not tangki or not tangki.produk_id: continue produk = produk_map.get(tangki.produk_id) item_snapshots.append({ "volume_diterima": item.volume_diterima, "volume_expected": item.volume_expected, "tangki_produk_id": tangki.produk_id, "tangki_nama": tangki.nama, "produk_nama": produk.nama if produk else "?", "threshold": Decimal(str(produk.losses_threshold_penerimaan_pct if produk else "0.150")), }) submitted_at = p.submitted_at spbu_header = _spbu_header(spbu) # Build exceed lines (outside session — using plain dicts) exceed_lines = [] for snap in item_snapshots: volume_diterima = snap["volume_diterima"] volume_expected = snap["volume_expected"] if volume_diterima is None or volume_expected is None or volume_expected <= 0: continue losses = volume_diterima - volume_expected if losses >= 0: continue abs_losses = abs(losses) losses_pct = (abs_losses / volume_expected) * 100 threshold = snap["threshold"] if losses_pct <= threshold: continue exceed_lines.append( f"🛢 {snap['produk_nama']} → {snap['tangki_nama']}\n" f" DO: {_fmt_vol(volume_expected)} | Diterima: {_fmt_vol(volume_diterima)}\n" f" Losses: -{_fmt_vol(abs_losses)} (-{round(losses_pct, 3)}%) ⚠️ batas {threshold}%" ) if not exceed_lines: return tanggal_str = _wib(submitted_at) if submitted_at else "—" items_text = "\n\n".join(exceed_lines) text = ( f"⚠️ Losses Pengiriman BBM\n" f"🏪 {spbu_header}\n\n" f"{items_text}\n\n" f"📅 {tanggal_str}\n" f"🔗 {_link(f'/penerimaan/{penerimaan_id}')}" ) await _send_text(cfg.chat_operasional_id, cfg.thread_penerimaan_id, text) # ── Stock Adjustment ─────────────────────────────────────────────────────────── async def notify_stock_submitted(spbu_id: int, adj_id: int) -> None: if not settings.TELEGRAM_BOT_TOKEN or not settings.is_production: return try: cfg = await _get_config(spbu_id) if not cfg or not cfg.chat_operasional_id or not cfg.thread_stock_id: return await _do_notify_stock(cfg, spbu_id, adj_id, event="submitted") except Exception as e: logger.error("telegram notify_stock_submitted error: %s", e) async def notify_stock_approved(spbu_id: int, adj_id: int) -> None: if not settings.TELEGRAM_BOT_TOKEN or not settings.is_production: return try: cfg = await _get_config(spbu_id) if not cfg or not cfg.chat_operasional_id or not cfg.thread_stock_id: return await _do_notify_stock(cfg, spbu_id, adj_id, event="approved") except Exception as e: logger.error("telegram notify_stock_approved error: %s", e) async def _do_notify_stock( cfg: TelegramConfig, spbu_id: int, adj_id: int, event: str ) -> None: from app.models.operational import StockAdjustment, StockAdjustmentItem, StockAdjustmentItemFoto from app.models.spbu import Tangki async with AsyncSessionLocal() as db: result = await db.execute( select(StockAdjustment) .where(StockAdjustment.id == adj_id, StockAdjustment.spbu_id == spbu_id) .options( selectinload(StockAdjustment.spbu), selectinload(StockAdjustment.shift), selectinload(StockAdjustment.submitted_by), selectinload(StockAdjustment.reviewed_by), selectinload(StockAdjustment.items).selectinload(StockAdjustmentItem.tangki).selectinload(Tangki.produk), selectinload(StockAdjustment.items).selectinload(StockAdjustmentItem.fotos), ) ) adj = result.scalar_one_or_none() if not adj: return spbu_hdr = _spbu_header(adj.spbu) shift_nama = adj.shift.nama if adj.shift else "—" tanggal_str = adj.tanggal.strftime("%d %b %Y") if adj.tanggal else "—" def _item_line(item: StockAdjustmentItem) -> str: tangki_nama = item.tangki.nama if item.tangki else "?" produk_nama = (item.tangki.produk.nama if item.tangki and item.tangki.produk else "") label = f"{tangki_nama} - {produk_nama}" if produk_nama else tangki_nama dipstick = item.dipstick_manual_mm or item.dipstick_digital_mm height_str = f"{int(dipstick)} mm" if dipstick is not None else "— mm" return f"🛢 {label:<22} {height_str:<10} {_fmt_vol(item.volume_final_liter or 0)}" item_lines = "\n".join(_item_line(item) for item in adj.items) if event == "submitted": operator_name = adj.submitted_by.name if adj.submitted_by else "—" text = ( f"📊 Sounding Stok Masuk\n\n" f"🏪 {spbu_hdr}\n" f"📅 {tanggal_str} · {shift_nama}\n" f"👤 Operator: {operator_name}\n\n" f"──────────────────────\n" f"VOLUME PER TANGKI\n" f"──────────────────────\n" f"{item_lines}\n" f"──────────────────────\n\n" f"🕐 Submit: {_wib(adj.submitted_at)}\n" f"🔗 {_link(f'/stock/{adj_id}')}" ) await _send_text(cfg.chat_operasional_id, cfg.thread_stock_id, text) foto_urls = [f.url for item in adj.items for f in item.fotos] photos = _collect_photos(foto_urls) if photos: caption = f"📸 Foto Sounding — {spbu_hdr}\n{shift_nama} · {tanggal_str}" await _send_media_group(cfg.chat_operasional_id, cfg.thread_stock_id, caption, photos) elif event == "approved": approver_name = adj.reviewed_by.name if adj.reviewed_by else "—" operator_name = adj.submitted_by.name if adj.submitted_by else "—" text = ( f"✅ Sounding Stok Disetujui\n\n" f"🏪 {spbu_hdr}\n" f"📅 {tanggal_str} · {shift_nama}\n\n" f"──────────────────────\n" f"{item_lines}\n" f"──────────────────────\n\n" f"👤 Operator : {operator_name}\n" f"✍️ Disetujui : {approver_name}\n" f"🕐 Waktu : {_wib(adj.reviewed_at)}\n" f"🔗 {_link(f'/stock/{adj_id}')}" ) await _send_text(cfg.chat_operasional_id, cfg.thread_stock_id, text) async def notify_stock_unlocked(spbu_id: int, adj_id: int) -> None: if not settings.TELEGRAM_BOT_TOKEN or not settings.is_production: return try: cfg = await _get_config(spbu_id) if not cfg or not cfg.chat_operasional_id or not cfg.thread_stock_id: return from app.models.operational import StockAdjustment async with AsyncSessionLocal() as db: result = await db.execute( select(StockAdjustment) .where(StockAdjustment.id == adj_id, StockAdjustment.spbu_id == spbu_id) .options(selectinload(StockAdjustment.spbu), selectinload(StockAdjustment.shift), selectinload(StockAdjustment.unlocked_by)) ) adj = result.scalar_one_or_none() if not adj: return shift_nama = adj.shift.nama if adj.shift else "—" tanggal_str = adj.tanggal.strftime("%d %b %Y") if adj.tanggal else "—" text = _unlock_text( "🔓", "Sounding Stok", _spbu_header(adj.spbu), f"{tanggal_str} · {shift_nama}", "", adj.unlocked_by.name if adj.unlocked_by else "—", adj.unlock_reason or "—", adj.unlocked_at, ) text += f"\n🔗 {_link(f'/stock/{adj_id}')}" await _send_text(cfg.chat_operasional_id, cfg.thread_stock_id, text) except Exception as e: logger.error("telegram notify_stock_unlocked error: %s", e) # ── Penebusan ────────────────────────────────────────────────────────────────── async def notify_penebusan_submitted(spbu_id: int, penebusan_id: int) -> None: """Fire when penebusan DRAFT → SUBMITTED.""" if not settings.TELEGRAM_BOT_TOKEN or not settings.is_production: return try: cfg = await _get_config(spbu_id) if not cfg or not cfg.chat_finansial_id or not cfg.thread_penebusan_id: return from app.models.penebusan import Penebusan, PenebusanItem async with AsyncSessionLocal() as db: result = await db.execute( select(Penebusan) .where(Penebusan.id == penebusan_id, Penebusan.spbu_id == spbu_id) .options(selectinload(Penebusan.spbu), selectinload(Penebusan.items).selectinload(PenebusanItem.produk)) ) pnb = result.scalar_one_or_none() if not pnb: return spbu_hdr = _spbu_header(pnb.spbu) tanggal_str = pnb.tanggal.strftime("%d %b %Y") if pnb.tanggal else "—" item_lines = "\n".join( f"⛽ {item.produk.nama if item.produk else '?':<12} {_fmt_vol(item.volume_pesan)}" for item in pnb.items ) total_vol = sum(item.volume_pesan for item in pnb.items) no_so_line = f"📄 No. SO: {pnb.no_so}" if pnb.no_so else "📄 No. SO: (belum ada)" text = ( f"📤 Penebusan BBM Disubmit\n\n" f"🏪 {spbu_hdr}\n" f"📅 {tanggal_str}\n" f"🔖 Booking: {pnb.booking_code}\n" f"{no_so_line}\n\n" f"──────────────────────\n" f"{item_lines}\n" f"──────────────────────\n" f"📦 Total: {_fmt_vol(total_vol)}\n" f"💵 Total Nilai: {_fmt_rp(pnb.total)}\n" f"🔗 {_link(f'/penebusan/{penebusan_id}')}" ) await _send_text(cfg.chat_finansial_id, cfg.thread_penebusan_id, text) except Exception as e: logger.error("telegram notify_penebusan_submitted error: %s", e) async def notify_penebusan_partial_received(spbu_id: int, penerimaan_id: int) -> None: """Fire on every penerimaan submit — shows cumulative progress. Skip if FULLY_RECEIVED (handled by completed).""" if not settings.TELEGRAM_BOT_TOKEN or not settings.is_production: return try: cfg = await _get_config(spbu_id) if not cfg or not cfg.chat_finansial_id or not cfg.thread_penebusan_id: return from app.models.penebusan import Penebusan, PenebusanItem, StatusPenebusan from app.models.penerimaan import Penerimaan async with AsyncSessionLocal() as db: pr = await db.execute( select(Penerimaan).where(Penerimaan.id == penerimaan_id) .options( selectinload(Penerimaan.penebusan).selectinload(Penebusan.spbu), selectinload(Penerimaan.penebusan).selectinload(Penebusan.items).selectinload(PenebusanItem.produk), selectinload(Penerimaan.items), ) ) penerimaan = pr.scalar_one_or_none() if not penerimaan: return pnb = penerimaan.penebusan if not pnb: return if pnb.status == StatusPenebusan.FULLY_RECEIVED: return # notify_penebusan_completed handles this spbu_hdr = _spbu_header(pnb.spbu) tanggal_str = pnb.tanggal.strftime("%d %b %Y") if pnb.tanggal else "—" total_pesan = sum(item.volume_pesan for item in pnb.items) total_diterima = sum(item.volume_diterima for item in pnb.items) item_lines = "\n".join( f"⛽ {item.produk.nama if item.produk else '?':<12} " f"terima {_fmt_vol(item.volume_diterima):<16} / pesan {_fmt_vol(item.volume_pesan)}" for item in pnb.items ) text = ( f"🚚 Penerimaan BBM Masuk\n\n" f"🏪 {spbu_hdr}\n" f"📅 {tanggal_str}\n" f"🔖 Booking: {pnb.booking_code}\n" f"📄 No. SO: {pnb.no_so or '—'}\n\n" f"──────────────────────\n" f"{item_lines}\n" f"──────────────────────\n" f"📦 Total diterima: {_fmt_vol(total_diterima)} / {_fmt_vol(total_pesan)}\n" f"🔗 {_link(f'/penebusan/{pnb.id}')}" ) await _send_text(cfg.chat_finansial_id, cfg.thread_penebusan_id, text) except Exception as e: logger.error("telegram notify_penebusan_partial_received error: %s", e) async def notify_penebusan_completed(spbu_id: int, penerimaan_id: int) -> None: """Fire when all volumes received — penebusan status becomes FULLY_RECEIVED. Called from penerimaan submit; checks internally before sending.""" if not settings.TELEGRAM_BOT_TOKEN or not settings.is_production: return try: cfg = await _get_config(spbu_id) if not cfg or not cfg.chat_finansial_id or not cfg.thread_penebusan_id: return from app.models.penebusan import Penebusan, PenebusanItem, StatusPenebusan from app.models.penerimaan import Penerimaan async with AsyncSessionLocal() as db: pr = await db.execute( select(Penerimaan).where(Penerimaan.id == penerimaan_id) .options(selectinload(Penerimaan.penebusan).selectinload(Penebusan.spbu), selectinload(Penerimaan.penebusan).selectinload(Penebusan.items).selectinload(PenebusanItem.produk)) ) penerimaan = pr.scalar_one_or_none() if not penerimaan: return pnb = penerimaan.penebusan if not pnb or pnb.status != StatusPenebusan.FULLY_RECEIVED: return spbu_hdr = _spbu_header(pnb.spbu) tanggal_str = pnb.tanggal.strftime("%d %b %Y") if pnb.tanggal else "—" total_vol = sum(item.volume_diterima for item in pnb.items) text = ( f"🎉 DO Selesai — Semua Volume Diterima\n\n" f"🏪 {spbu_hdr}\n" f"📅 {tanggal_str}\n" f"🔖 Booking : {pnb.booking_code}\n" f"📄 No. SO : {pnb.no_so or '—'}\n" f"📦 Total : {_fmt_vol(total_vol)}\n\n" f"✅ Semua volume sudah diterima\n" f"🔗 {_link(f'/penebusan/{pnb.id}')}" ) await _send_text(cfg.chat_finansial_id, cfg.thread_penebusan_id, text) except Exception as e: logger.error("telegram notify_penebusan_completed error: %s", e) # ── Penyetoran ───────────────────────────────────────────────────────────────── async def notify_penyetoran_submitted(spbu_id: int, batch_id: int) -> None: if not settings.TELEGRAM_BOT_TOKEN or not settings.is_production: return try: cfg = await _get_config(spbu_id) if not cfg or not cfg.chat_finansial_id or not cfg.thread_penyetoran_id: return await _do_notify_penyetoran(cfg, spbu_id, batch_id, event="submitted") except Exception as e: logger.error("telegram notify_penyetoran_submitted error: %s", e) async def notify_penyetoran_approved(spbu_id: int, batch_id: int) -> None: if not settings.TELEGRAM_BOT_TOKEN or not settings.is_production: return try: cfg = await _get_config(spbu_id) if not cfg or not cfg.chat_finansial_id or not cfg.thread_penyetoran_id: return await _do_notify_penyetoran(cfg, spbu_id, batch_id, event="approved") except Exception as e: logger.error("telegram notify_penyetoran_approved error: %s", e) async def _do_notify_penyetoran( cfg: TelegramConfig, spbu_id: int, batch_id: int, event: str ) -> None: from app.models.penyetoran import Penyetoran, PenyetoranBatch from app.models.operational import LaporanShift from app.models.spbu import Shift async with AsyncSessionLocal() as db: result = await db.execute( select(PenyetoranBatch) .where(PenyetoranBatch.id == batch_id, PenyetoranBatch.spbu_id == spbu_id) .options( selectinload(PenyetoranBatch.spbu), selectinload(PenyetoranBatch.submitted_by), selectinload(PenyetoranBatch.reviewed_by), selectinload(PenyetoranBatch.items) .selectinload(Penyetoran.laporan_shift) .selectinload(LaporanShift.shift), ) ) batch = result.scalar_one_or_none() if not batch: return spbu_hdr = _spbu_header(batch.spbu) tgl_from = batch.tanggal_from.strftime("%d %b") if batch.tanggal_from else "—" tgl_to = batch.tanggal_to.strftime("%d %b %Y") if batch.tanggal_to else "—" periode = f"{tgl_from} – {tgl_to}" def _setor_line(item: Penyetoran) -> str: tgl = item.tanggal.strftime("%d %b") if item.tanggal else "?" shift = item.laporan_shift.shift.nama if item.laporan_shift and item.laporan_shift.shift else "—" return f"📅 {tgl} · {shift:<12} {_fmt_rp(item.total_penjualan)}" item_lines = "\n".join( _setor_line(item) for item in sorted(batch.items, key=lambda x: (x.tanggal or "", x.shift_id or 0)) ) if event == "submitted": submitter_name = batch.submitted_by.name if batch.submitted_by else "—" text = ( f"💰 Batch Penyetoran Masuk\n\n" f"🏪 {spbu_hdr}\n" f"📅 Periode: {periode}\n" f"👤 Disubmit: {submitter_name}\n\n" f"──────────────────────\n" f"DETAIL PER TANGGAL\n" f"──────────────────────\n" f"{item_lines}\n" f"──────────────────────\n" f"💵 Total: {_fmt_rp(batch.total_amount)}\n\n" f"🕐 Submit: {_wib(batch.submitted_at)}\n" f"🔗 {_link(f'/penyetoran/{batch_id}')}" ) await _send_text(cfg.chat_finansial_id, cfg.thread_penyetoran_id, text) elif event == "approved": approver_name = batch.reviewed_by.name if batch.reviewed_by else "—" submitter_name = batch.submitted_by.name if batch.submitted_by else "—" text = ( f"✅ Penyetoran Disetujui\n\n" f"🏪 {spbu_hdr}\n" f"📅 Periode: {periode}\n" f"💵 Total: {_fmt_rp(batch.total_amount)}\n\n" f"👤 Disubmit : {submitter_name}\n" f"✍️ Disetujui: {approver_name}\n" f"🕐 Waktu : {_wib(batch.reviewed_at)}\n" f"🔗 {_link(f'/penyetoran/{batch_id}')}" ) await _send_text(cfg.chat_finansial_id, cfg.thread_penyetoran_id, text) async def notify_penyetoran_unlocked(spbu_id: int, batch_id: int) -> None: if not settings.TELEGRAM_BOT_TOKEN or not settings.is_production: return try: cfg = await _get_config(spbu_id) if not cfg or not cfg.chat_finansial_id or not cfg.thread_penyetoran_id: return from app.models.penyetoran import PenyetoranBatch async with AsyncSessionLocal() as db: result = await db.execute( select(PenyetoranBatch) .where(PenyetoranBatch.id == batch_id, PenyetoranBatch.spbu_id == spbu_id) .options(selectinload(PenyetoranBatch.spbu), selectinload(PenyetoranBatch.unlocked_by)) ) batch = result.scalar_one_or_none() if not batch: return tgl_from = batch.tanggal_from.strftime("%d %b") if batch.tanggal_from else "—" tgl_to = batch.tanggal_to.strftime("%d %b %Y") if batch.tanggal_to else "—" text = _unlock_text( "🔓", "Batch Penyetoran", _spbu_header(batch.spbu), f"{tgl_from} – {tgl_to}", f"💵 Total: {_fmt_rp(batch.total_amount)}\n", batch.unlocked_by.name if batch.unlocked_by else "—", batch.unlock_reason or "—", batch.unlocked_at, ) text += f"\n🔗 {_link(f'/penyetoran/{batch_id}')}" await _send_text(cfg.chat_finansial_id, cfg.thread_penyetoran_id, text) except Exception as e: logger.error("telegram notify_penyetoran_unlocked error: %s", e) # ── Expenses ─────────────────────────────────────────────────────────────────── async def notify_expense_submitted(spbu_id: int, expense_id: int) -> None: if not settings.TELEGRAM_BOT_TOKEN or not settings.is_production: return try: cfg = await _get_config(spbu_id) if not cfg or not cfg.chat_finansial_id or not cfg.thread_expense_id: return await _do_notify_expense(cfg, spbu_id, expense_id, event="submitted") except Exception as e: logger.error("telegram notify_expense_submitted error: %s", e) async def notify_expense_approved(spbu_id: int, expense_id: int) -> None: if not settings.TELEGRAM_BOT_TOKEN or not settings.is_production: return try: cfg = await _get_config(spbu_id) if not cfg or not cfg.chat_finansial_id or not cfg.thread_expense_id: return await _do_notify_expense(cfg, spbu_id, expense_id, event="approved") except Exception as e: logger.error("telegram notify_expense_approved error: %s", e) async def _do_notify_expense( cfg: TelegramConfig, spbu_id: int, expense_id: int, event: str ) -> None: from app.models.expenses import Expense async with AsyncSessionLocal() as db: result = await db.execute( select(Expense) .where(Expense.id == expense_id, Expense.spbu_id == spbu_id) .options( selectinload(Expense.kategori), selectinload(Expense.submitted_by), selectinload(Expense.reviewed_by), ) ) exp = result.scalar_one_or_none() if not exp: return spbu = await _load_spbu(db, spbu_id) spbu_hdr = _spbu_header(spbu) tanggal_str = exp.tanggal.strftime("%d %b %Y") if exp.tanggal else "—" kategori_nama = exp.kategori.nama if exp.kategori else "—" keterangan = exp.keterangan or "—" if event == "submitted": operator_name = exp.submitted_by.name if exp.submitted_by else "—" text = ( f"🧾 Pengeluaran (Expense) Masuk\n\n" f"🏪 {spbu_hdr}\n" f"📅 {tanggal_str}\n" f"🏷 Kategori: {kategori_nama}\n" f"💵 Jumlah: {_fmt_rp(exp.jumlah)}\n" f"📝 Keterangan: {keterangan}\n" f"👤 Disubmit: {operator_name}\n\n" f"🕐 Submit: {_wib(exp.submitted_at)}\n" f"🔗 {_link(f'/expenses/{expense_id}')}" ) await _send_text(cfg.chat_finansial_id, cfg.thread_expense_id, text) elif event == "approved": approver_name = exp.reviewed_by.name if exp.reviewed_by else "—" operator_name = exp.submitted_by.name if exp.submitted_by else "—" text = ( f"✅ Expense Disetujui\n\n" f"🏪 {spbu_hdr}\n" f"📅 {tanggal_str}\n" f"🏷 {kategori_nama} {_fmt_rp(exp.jumlah)}\n" f"📝 {keterangan}\n\n" f"👤 Disubmit : {operator_name}\n" f"✍️ Disetujui: {approver_name}\n" f"🕐 Waktu : {_wib(exp.reviewed_at)}\n" f"🔗 {_link(f'/expenses/{expense_id}')}" ) await _send_text(cfg.chat_finansial_id, cfg.thread_expense_id, text) # ── PB (Permintaan Barang) ───────────────────────────────────────────────────── async def notify_pb_submitted(spbu_id: int, pb_id: int) -> None: if not settings.TELEGRAM_BOT_TOKEN or not settings.is_production: return try: cfg = await _get_config(spbu_id) if not cfg or not cfg.chat_finansial_id or not cfg.thread_pb_id: return await _do_notify_pb(cfg, spbu_id, pb_id, event="submitted") except Exception as e: logger.error("telegram notify_pb_submitted error: %s", e) async def notify_pb_reviewed(spbu_id: int, pb_id: int) -> None: if not settings.TELEGRAM_BOT_TOKEN or not settings.is_production: return try: cfg = await _get_config(spbu_id) if not cfg or not cfg.chat_finansial_id or not cfg.thread_pb_id: return await _do_notify_pb(cfg, spbu_id, pb_id, event="reviewed") except Exception as e: logger.error("telegram notify_pb_reviewed error: %s", e) async def _do_notify_pb( cfg: TelegramConfig, spbu_id: int, pb_id: int, event: str ) -> None: from app.models.pb import PB, PBItem async with AsyncSessionLocal() as db: result = await db.execute( select(PB) .where(PB.id == pb_id, PB.spbu_id == spbu_id) .options( selectinload(PB.created_by), selectinload(PB.reviewed_by), selectinload(PB.items), ) ) pb = result.scalar_one_or_none() if not pb: return spbu = await _load_spbu(db, spbu_id) spbu_hdr = _spbu_header(spbu) total_estimasi = sum( (item.qty_requested * item.harga_satuan) for item in pb.items ) item_lines = "\n".join( f"• {item.nama_item} ({item.qty_requested:g} {item.satuan}) — {_fmt_rp(item.qty_requested * item.harga_satuan)}" for item in pb.items ) if event == "submitted": creator_name = pb.created_by.name if pb.created_by else "—" text = ( f"🛒 Permintaan Barang Baru\n\n" f"🏪 {spbu_hdr}\n" f"📋 {pb.nomor}\n" f"📝 {pb.judul}\n" f"👤 Dibuat: {creator_name}\n\n" f"──────────────────────\n" f"ITEM ({len(pb.items)} item)\n" f"──────────────────────\n" f"{item_lines}\n" f"──────────────────────\n" f"💵 Total Estimasi: {_fmt_rp(total_estimasi)}\n\n" f"🕐 Submit: {_wib(pb.submitted_at)}\n" f"🔗 {_link(f'/pb/{pb_id}')}" ) await _send_text(cfg.chat_finansial_id, cfg.thread_pb_id, text) elif event == "reviewed": reviewer_name = pb.reviewed_by.name if pb.reviewed_by else "—" approved_count = sum(1 for i in pb.items if i.status == "approved") rejected_count = sum(1 for i in pb.items if i.status == "rejected") total_approved = sum( (i.qty_approved or i.qty_requested) * i.harga_satuan for i in pb.items if i.status == "approved" ) catatan = pb.catatan_head or "—" text = ( f"📋 PB Direview\n\n" f"🏪 {spbu_hdr}\n" f"📋 {pb.nomor} — {pb.judul}\n\n" f"✅ Disetujui : {approved_count} item — {_fmt_rp(total_approved)}\n" f"❌ Ditolak : {rejected_count} item\n" f"📝 Catatan : {catatan}\n\n" f"✍️ Reviewer : {reviewer_name}\n" f"🕐 Waktu : {_wib(pb.reviewed_at)}\n" f"🔗 {_link(f'/pb/{pb_id}')}" ) await _send_text(cfg.chat_finansial_id, cfg.thread_pb_id, text) # ── Absensi ──────────────────────────────────────────────────────────────────── async def notify_absensi_uploaded(spbu_id: int, absensi_id: int) -> None: if not settings.TELEGRAM_BOT_TOKEN or not settings.is_production: return try: cfg = await _get_config(spbu_id) if not cfg or not cfg.chat_ga_id or not cfg.thread_absensi_id: return await _do_notify_absensi(cfg, spbu_id, absensi_id) except Exception as e: logger.error("telegram notify_absensi_uploaded error: %s", e) async def _do_notify_absensi(cfg: TelegramConfig, spbu_id: int, absensi_id: int) -> None: from app.models.general_affairs import Absensi async with AsyncSessionLocal() as db: result = await db.execute( select(Absensi) .where(Absensi.id == absensi_id, Absensi.spbu_id == spbu_id) .options( selectinload(Absensi.spbu), # Absensi has spbu relationship selectinload(Absensi.shift), ) ) a = result.scalar_one_or_none() if not a: return spbu_hdr = _spbu_header(a.spbu) tanggal_str = a.tanggal.strftime("%d %b %Y") if a.tanggal else "—" shift_nama = a.shift.nama if a.shift else "—" exif_waktu = f"\n🕐 EXIF Waktu: {a.foto_eksif_waktu}" if a.foto_eksif_waktu else "" text = ( f"📍 Absensi Terupload\n\n" f"🏪 {spbu_hdr}\n" f"📅 {tanggal_str} · {shift_nama}" f"{exif_waktu}\n" f"🔗 {_link(f'/general-affairs/absensi?date={a.tanggal.isoformat()}')}" ) await _send_text(cfg.chat_ga_id, cfg.thread_absensi_id, text) if a.foto_url: photos = _collect_photos([a.foto_url]) if photos: caption = f"📸 Foto Absensi — {spbu_hdr}\n{shift_nama} · {tanggal_str}" await _send_media_group(cfg.chat_ga_id, cfg.thread_absensi_id, caption, photos) # ── Housekeeping ─────────────────────────────────────────────────────────────── async def notify_housekeeping_submitted(spbu_id: int, hk_id: int) -> None: if not settings.TELEGRAM_BOT_TOKEN or not settings.is_production: return try: cfg = await _get_config(spbu_id) if not cfg or not cfg.chat_ga_id or not cfg.thread_housekeeping_id: return await _do_notify_housekeeping(cfg, spbu_id, hk_id) except Exception as e: logger.error("telegram notify_housekeeping_submitted error: %s", e) async def _do_notify_housekeeping(cfg: TelegramConfig, spbu_id: int, hk_id: int) -> None: from app.models.general_affairs import Housekeeping, HousekeepingFoto, HousekeepingItem async with AsyncSessionLocal() as db: result = await db.execute( select(Housekeeping) .where(Housekeeping.id == hk_id, Housekeeping.spbu_id == spbu_id) .options( selectinload(Housekeeping.items), selectinload(Housekeeping.fotos), ) ) hk = result.scalar_one_or_none() if not hk: return spbu = await _load_spbu(db, spbu_id) spbu_hdr = _spbu_header(spbu) tanggal_str = hk.tanggal.strftime("%d %b %Y") if hk.tanggal else "—" n_area = len(hk.items) n_foto = len(hk.fotos) if hk.items: area_lines = "\n".join(f"• {item.deskripsi}" for item in hk.items[:10]) if n_area > 10: area_lines += f"\n… dan {n_area - 10} area lainnya" else: area_lines = f"📷 {n_foto} foto diupload" if n_foto else "—" summary = f"{n_area} area" if n_area else "" if n_foto: summary = f"{summary}, {n_foto} foto" if summary else f"{n_foto} foto" text = ( f"🧹 Housekeeping Disubmit\n\n" f"🏪 {spbu_hdr}\n" f"📅 {tanggal_str}\n\n" f"──────────────────────\n" f"AREA ({summary})\n" f"──────────────────────\n" f"{area_lines}\n" f"🔗 {_link(f'/general-affairs/housekeeping?date={hk.tanggal.isoformat()}')}" ) await _send_text(cfg.chat_ga_id, cfg.thread_housekeeping_id, text) foto_urls = [f.foto_url for f in hk.fotos] photos = _collect_photos(foto_urls) if photos: caption = f"📸 Foto Housekeeping — {spbu_hdr}\n{tanggal_str}" await _send_media_group(cfg.chat_ga_id, cfg.thread_housekeeping_id, caption, photos) # ── Sapras ───────────────────────────────────────────────────────────────────── async def notify_sapras_submitted(spbu_id: int, sapras_id: int) -> None: if not settings.TELEGRAM_BOT_TOKEN or not settings.is_production: return try: cfg = await _get_config(spbu_id) if not cfg or not cfg.chat_ga_id or not cfg.thread_sapras_id: return await _do_notify_sapras(cfg, spbu_id, sapras_id) except Exception as e: logger.error("telegram notify_sapras_submitted error: %s", e) async def _do_notify_sapras(cfg: TelegramConfig, spbu_id: int, sapras_id: int) -> None: from app.models.general_affairs import Sapras, SaprasItem async with AsyncSessionLocal() as db: result = await db.execute( select(Sapras) .where(Sapras.id == sapras_id, Sapras.spbu_id == spbu_id) .options( selectinload(Sapras.items), ) ) sapras = result.scalar_one_or_none() if not sapras: return spbu = await _load_spbu(db, spbu_id) spbu_hdr = _spbu_header(spbu) tanggal_str = sapras.tanggal.strftime("%d %b %Y") if sapras.tanggal else "—" catatan = sapras.catatan or "" kegiatan_lines = "\n".join( f"• {item.kegiatan}" for item in sapras.items[:10] ) if sapras.items else "—" text = ( f"🔧 Sarana & Prasarana Disubmit\n\n" f"🏪 {spbu_hdr}\n" f"📅 {tanggal_str}\n" + (f"📝 {catatan}\n" if catatan else "") + f"\n──────────────────────\n" f"KEGIATAN ({len(sapras.items)} item)\n" f"──────────────────────\n" f"{kegiatan_lines}\n" f"🔗 {_link(f'/general-affairs/sapras?date={sapras.tanggal.isoformat()}')}" ) await _send_text(cfg.chat_ga_id, cfg.thread_sapras_id, text) foto_urls = [u for item in sapras.items for u in [item.foto_sebelum_url, item.foto_sesudah_url] if u] photos = _collect_photos(foto_urls) if photos: caption = f"📸 Foto Sapras — {spbu_hdr}\n{tanggal_str}" await _send_media_group(cfg.chat_ga_id, cfg.thread_sapras_id, caption, photos) # ── POS Upload ───────────────────────────────────────────────────────────────── async def notify_pos_upload(spbu_id: int, batch_id: int) -> None: if not settings.TELEGRAM_BOT_TOKEN or not settings.is_production: return try: cfg = await _get_config(spbu_id) if not cfg or not cfg.chat_operasional_id or not cfg.thread_pos_upload_id: return await _do_notify_pos_upload(cfg, spbu_id, batch_id) except Exception as e: logger.error("telegram notify_pos_upload error: %s", e) async def _do_notify_pos_upload(cfg: TelegramConfig, spbu_id: int, batch_id: int) -> None: from app.models.pos_analytics import PosUploadBatch from app.models.product import Produk from app.models.user import User async with AsyncSessionLocal() as db: batch = (await db.execute( select(PosUploadBatch).where( PosUploadBatch.id == batch_id, PosUploadBatch.spbu_id == spbu_id, ) )).scalar_one_or_none() if not batch: return produk = (await db.execute(select(Produk).where(Produk.id == batch.produk_id))).scalar_one_or_none() uploader: User | None = None if batch.uploaded_by_id: uploader = (await db.execute(select(User).where(User.id == batch.uploaded_by_id))).scalar_one_or_none() spbu = await _load_spbu(db, spbu_id) spbu_hdr = _spbu_header(spbu) produk_nama = produk.nama if produk else "—" tgl_mulai = batch.tanggal_mulai.strftime("%d %b %Y") tgl_selesai = batch.tanggal_selesai.strftime("%d %b %Y") tgl_range = tgl_mulai if tgl_mulai == tgl_selesai else f"{tgl_mulai} – {tgl_selesai}" uploader_name = uploader.full_name if uploader else "—" text = ( f"📊 POS Upload Berhasil\n\n" f"🏪 {spbu_hdr}\n" f"⛽ {produk_nama}\n" f"📅 {tgl_range}\n\n" f"──────────────────────\n" f"Transaksi : {batch.total_rows:,} baris\n" f"Volume : {_fmt_vol(batch.total_volume)}\n" f"Nilai : {_fmt_rp(batch.total_nilai)}\n" f"──────────────────────\n" f"👤 {uploader_name}\n" f"🔗 {_link(f'/pos-upload?batch_id={batch_id}')}" ) await _send_text(cfg.chat_operasional_id, cfg.thread_pos_upload_id, text) async def notify_rekonsiliasi_run( spbu_id: int, tanggal_str: str, result_data: dict ) -> None: """Notif ke thread_rekonsiliasi_id saat rekonsiliasi di-run.""" if not settings.TELEGRAM_BOT_TOKEN or not settings.is_production: return try: cfg = await _get_config(spbu_id) if not cfg or not cfg.chat_operasional_id or not cfg.thread_rekonsiliasi_id: return async with AsyncSessionLocal() as db: spbu = await _load_spbu(db, spbu_id) spbu_header = _spbu_header(spbu) status = result_data.get("status", "pending") if status == "balanced": status_line = "✅ Status: BALANCED — semua tangki dalam batas" else: status_line = "⚠️ Status: PENDING — ada tangki melebihi threshold" items = result_data.get("items", []) tank_lines = [] for item in items: tangki_nama = item.get("tangki_nama") or f"Tangki #{item.get('tangki_id')}" produk_nama = item.get("produk_nama") or "?" losses = Decimal(str(item.get("losses", "0"))) losses_pct = item.get("losses_pct") threshold = item.get("threshold_pct") penerimaan_losses = Decimal(str(item.get("penerimaan_losses_liter", "0"))) penerimaan_losses_pct = item.get("penerimaan_losses_pct") if losses_pct is not None and threshold is not None: exceed = Decimal(str(losses_pct)) > Decimal(str(threshold)) else: exceed = False if losses < 0: losses_icon = "⚠️" if exceed else "✅" pct_str = f"{losses_pct}%" if losses_pct is not None else "—%" losses_line = f"Losses: -{_fmt_vol(abs(losses))} (-{pct_str}) {losses_icon}" else: losses_icon = "⚠️" if exceed else "✅" pct_str = f"{losses_pct}%" if losses_pct is not None else "—%" losses_line = f"Gain: +{_fmt_vol(losses)} (+{pct_str}) {losses_icon}" line = f"🛢 {tangki_nama} · {produk_nama}\n {losses_line}" if penerimaan_losses < 0: pen_pct_str = f"{penerimaan_losses_pct}%" if penerimaan_losses_pct is not None else "—%" line += f"\n Losses Pengiriman: -{_fmt_vol(abs(penerimaan_losses))} (-{pen_pct_str}) ⚠️" tank_lines.append(line) tank_section = "\n\n".join(tank_lines) run_by = result_data.get("run_by_name") or "—" total_nilai = Decimal(str(result_data.get("total_nilai_losses") or "0")) total_penerimaan_nilai = Decimal(str(result_data.get("total_nilai_penerimaan_losses") or "0")) total_gabungan = Decimal(str(result_data.get("total_gabungan_losses") or "0")) rp_section = f"💰 Total Losses: {_fmt_rp(total_nilai)}" if total_penerimaan_nilai != 0: rp_section += ( f"\n💰 Total Losses Pengiriman: {_fmt_rp(total_penerimaan_nilai)}" f"\n💰 Total Gabungan: {_fmt_rp(total_gabungan)}" ) text = ( f"🔄 Rekonsiliasi Harian\n" f"🏪 {spbu_header}\n" f"📅 {tanggal_str}\n\n" f"{status_line}\n\n" f"{tank_section}\n\n" f"{rp_section}\n\n" f"👤 Dijalankan oleh: {run_by}\n" f"🔗 {_link(f'/rekonsiliasi?tanggal={tanggal_str}')}" ) await _send_text(cfg.chat_operasional_id, cfg.thread_rekonsiliasi_id, text) except Exception as e: logger.error("telegram notify_rekonsiliasi_run error: %s", e) async def notify_rekonsiliasi_approved( spbu_id: int, tanggal_str: str, approved_by_name: str, result_data: dict | None = None, ) -> None: """Notif ke thread_rekonsiliasi_id saat rekonsiliasi di-approve, dengan lampiran BA PDF.""" if not settings.TELEGRAM_BOT_TOKEN or not settings.is_production: return try: cfg = await _get_config(spbu_id) if not cfg or not cfg.chat_operasional_id or not cfg.thread_rekonsiliasi_id: return async with AsyncSessionLocal() as db: spbu = await _load_spbu(db, spbu_id) spbu_header = _spbu_header(spbu) waktu_str = _wib(datetime.now(timezone.utc)) caption = ( f"✅ Rekonsiliasi Diapprove\n" f"🏪 {spbu_header}\n" f"📅 {tanggal_str}\n\n" f"👤 Diapprove oleh: {approved_by_name} · {waktu_str}\n" f"🔗 {_link(f'/rekonsiliasi?tanggal={tanggal_str}')}" ) # Try to generate and attach BA PDF if result_data: try: from app.utils.ba_pdf import generate_ba_rekonsiliasi items = result_data.get("items", []) shift_breakdown = result_data.get("shift_breakdown", []) harga_per_tangki = { int(item["tangki_id"]): Decimal(str(item["harga_per_liter"])) for item in items if item.get("tangki_id") is not None and item.get("harga_per_liter") is not None } run_by_name = result_data.get("run_by_name") approved_at = result_data.get("updated_at") pdf_bytes = generate_ba_rekonsiliasi( spbu_name=spbu.name, tanggal=tanggal_str, items=items, shift_breakdown=shift_breakdown or None, harga_per_tangki=harga_per_tangki or None, run_by_name=run_by_name, approved_by_name=approved_by_name, approved_at=approved_at, ) filename = f"BA-Rekonsiliasi-{tanggal_str}.pdf" await _send_document( cfg.chat_operasional_id, cfg.thread_rekonsiliasi_id, caption, filename, pdf_bytes, ) return except Exception as pdf_err: logger.warning("BA PDF generation failed, falling back to text: %s", pdf_err) await _send_text(cfg.chat_operasional_id, cfg.thread_rekonsiliasi_id, caption) except Exception as e: logger.error("telegram notify_rekonsiliasi_approved error: %s", e)