"""POS Analytics service — upload, compare, anomali, golden-hours.""" from __future__ import annotations from collections import defaultdict from datetime import date, datetime, timedelta, timezone from decimal import Decimal from sqlalchemy import delete, func, select from sqlalchemy.ext.asyncio import AsyncSession from app.models.pos_analytics import PosUploadBatch, PosTransaksiPertamina from app.models.spbu import Shift, Nozzle from app.models.product import Produk import httpx from app.core.config import settings from app.schemas.pos_analytics import ( PosUploadResponse, PosExistingBatch, PosCompareDayRow, PosCompareResponse, PosAnomaliItem, PosAnomaliResponse, PosHourlyRow, PosIslandRow, PosIslandHourlyRow, PosGoldenHoursResponse, PosShiftSummaryRow, PosGoldenHoursSummary, PosBatchListItem, PosTransaksiItem, PosBatchTransaksiResponse, GoldenHoursInsightResponse, ) from app.utils.pos_excel_parser import parse_xlsx_to_rows, parse_csv_to_rows, validate_produk_match, PRODUK_MAP, _normalize_spbu_no from app.utils.audit import log_action async def _get_spbu(db: AsyncSession, spbu_id: int): from app.models.spbu import Spbu return (await db.execute(select(Spbu).where(Spbu.id == spbu_id))).scalar_one_or_none() async def _get_produk(db: AsyncSession, produk_id: int) -> Produk: produk = (await db.execute(select(Produk).where(Produk.id == produk_id))).scalar_one_or_none() if not produk: raise ValueError("Produk tidak ditemukan") return produk async def _get_shifts(db: AsyncSession, spbu_id: int) -> list[Shift]: result = await db.execute( select(Shift).where(Shift.spbu_id == spbu_id, Shift.deleted_at.is_(None)) ) return list(result.scalars().all()) def _resolve_shift(jam, shifts: list[Shift]) -> int | None: """Return shift_id for a given time, handling overnight shifts.""" for shift in shifts: mulai = shift.jam_mulai selesai = shift.jam_selesai if selesai >= mulai: # Normal shift: e.g. 06:00 – 14:00 if mulai <= jam < selesai: return shift.id else: # Overnight shift: e.g. 22:00 – 06:00 if jam >= mulai or jam < selesai: return shift.id return None async def get_existing_batch( db: AsyncSession, spbu_id: int, produk_id: int, tanggal_mulai: date, tanggal_selesai: date, ) -> PosUploadBatch | None: result = await db.execute( select(PosUploadBatch).where( PosUploadBatch.spbu_id == spbu_id, PosUploadBatch.produk_id == produk_id, PosUploadBatch.tanggal_mulai == tanggal_mulai, PosUploadBatch.tanggal_selesai == tanggal_selesai, ) ) return result.scalar_one_or_none() async def upload_pos_xlsx( db: AsyncSession, spbu_id: int, produk_id: int, file_bytes: bytes, user_id: int, force_replace: bool = False, file_ext: str = "xlsx", ) -> PosUploadResponse: """Parse and store Pertamina POS file (xlsx or csv). Replaces existing batch if force_replace=True.""" produk = await _get_produk(db, produk_id) spbu = await _get_spbu(db, spbu_id) # Parse file — route by extension if file_ext == "csv": rows = parse_csv_to_rows(file_bytes) else: rows = parse_xlsx_to_rows(file_bytes) # Extract parser metadata from first row meta = rows[0].pop("_meta", {}) suspicious_count: int = meta.get("skipped_invalid", 0) no_spbu_from_file: str | None = meta.get("no_spbu_from_file") # Validation #4: SPBU number in file must match active SPBU if spbu and no_spbu_from_file: if _normalize_spbu_no(no_spbu_from_file) != _normalize_spbu_no(spbu.nomor_pertamina): raise ValueError( f"File bukan milik SPBU ini. " f"No SPBU di file: '{no_spbu_from_file}', " f"No SPBU aktif: '{spbu.nomor_pertamina}'" ) # Validate produk match — check first row first_produk_raw = rows[0]["produk_raw"] try: validate_produk_match(first_produk_raw, produk.kode) except ValueError as e: raise ValueError(str(e)) # Detect date range tanggal_mulai = min(r["tanggal"] for r in rows) tanggal_selesai = max(r["tanggal"] for r in rows) # Check existing batch existing = await get_existing_batch(db, spbu_id, produk_id, tanggal_mulai, tanggal_selesai) if existing and not force_replace: raise ValueError( f"duplicate_batch::{existing.id}::{existing.total_rows}" f"::{existing.tanggal_mulai}::{existing.tanggal_selesai}" f"::{existing.created_at.isoformat()}" ) if existing: await db.execute(delete(PosUploadBatch).where(PosUploadBatch.id == existing.id)) await db.flush() # Get shifts for shift resolution shifts = await _get_shifts(db, spbu_id) # Bulk insert transactions (deduplicate in-memory) skipped = 0 txns = [] seen: set[tuple] = set() for r in rows: ts = datetime.combine(r["tanggal"], r["jam"]).replace(tzinfo=timezone(timedelta(hours=7))) dedup_key = (spbu_id, ts, r["no_dispenser"]) if dedup_key in seen: skipped += 1 continue seen.add(dedup_key) txns.append(PosTransaksiPertamina( spbu_id=spbu_id, shift_id=_resolve_shift(r["jam"], shifts), tanggal=r["tanggal"], jam=r["jam"], code_mor=r["code_mor"], province=r["province"], city=r["city"], no_spbu_pertamina=r["no_spbu_pertamina"], no_nozzle=r["no_nozzle"], no_dispenser=r["no_dispenser"], produk_raw=r["produk_raw"], volume=r["volume"], nilai=r["nilai"], operator_name=r["operator_name"], no_hp=r["no_hp"], delivery_type=r["delivery_type"], plat_nomor=r["plat_nomor"], nik=r["nik"], instansi=r["instansi"], timestamp=ts, island_raw=r["island_raw"], nozzle_raw=r["nozzle_raw"], )) # Compute totals from deduplicated txns total_volume = sum(t.volume for t in txns) total_nilai = sum(t.nilai for t in txns) # Create batch with correct post-dedup totals now = datetime.now(timezone.utc) batch = PosUploadBatch( spbu_id=spbu_id, produk_id=produk_id, uploaded_by_id=user_id, tanggal_mulai=tanggal_mulai, tanggal_selesai=tanggal_selesai, total_rows=len(txns), total_volume=total_volume, total_nilai=total_nilai, created_at=now, ) db.add(batch) await db.flush() # get batch.id # Assign batch_id to all txns for t in txns: t.batch_id = batch.id db.add_all(txns) await log_action(db, user_id=user_id, spbu_id=spbu_id, aksi="import", modul="pos_upload", object_id=batch.id, detail={"inserted": len(txns), "skipped": skipped, "produk": produk.nama}) await db.commit() return PosUploadResponse( batch_id=batch.id, total_rows=len(txns), tanggal_mulai=tanggal_mulai, tanggal_selesai=tanggal_selesai, total_volume=batch.total_volume, total_nilai=batch.total_nilai, skipped=skipped, suspicious_count=suspicious_count, ) async def list_batches(db: AsyncSession, spbu_id: int) -> list[PosBatchListItem]: from app.models.user import User result = await db.execute( select(PosUploadBatch) .where(PosUploadBatch.spbu_id == spbu_id) .order_by(PosUploadBatch.created_at.desc()) ) batches = list(result.scalars().all()) produk_ids = list({b.produk_id for b in batches}) user_ids = list({b.uploaded_by_id for b in batches if b.uploaded_by_id}) produk_map = { p.id: p for p in ( await db.execute(select(Produk).where(Produk.id.in_(produk_ids))) ).scalars() } if produk_ids else {} user_map = { u.id: u for u in ( await db.execute(select(User).where(User.id.in_(user_ids))) ).scalars() } if user_ids else {} items = [] for b in batches: produk = produk_map.get(b.produk_id) uploader = user_map.get(b.uploaded_by_id) if b.uploaded_by_id else None items.append(PosBatchListItem( id=b.id, produk_id=b.produk_id, produk_nama=produk.nama if produk else "", tanggal_mulai=b.tanggal_mulai, tanggal_selesai=b.tanggal_selesai, total_rows=b.total_rows, total_volume=b.total_volume, total_nilai=b.total_nilai, created_at=b.created_at, uploaded_by_name=uploader.name if uploader else None, )) return items async def delete_batch(db: AsyncSession, spbu_id: int, batch_id: int, user_id: int) -> None: batch = (await db.execute( select(PosUploadBatch).where( PosUploadBatch.id == batch_id, PosUploadBatch.spbu_id == spbu_id, ) )).scalar_one_or_none() if not batch: raise ValueError("Batch tidak ditemukan") await db.execute(delete(PosUploadBatch).where(PosUploadBatch.id == batch_id)) await log_action(db, user_id=user_id, spbu_id=spbu_id, aksi="delete", modul="pos_upload", object_id=batch_id) await db.commit() async def get_compare( db: AsyncSession, spbu_id: int, produk_id: int, tanggal_mulai: date, tanggal_selesai: date, ) -> PosCompareResponse: """Compare daily POS totals vs laporan_shift totals for the same produk.""" from app.models.operational import LaporanShift, PenjualanNozzle produk = await _get_produk(db, produk_id) # POS totals per day pos_q = await db.execute( select( PosTransaksiPertamina.tanggal, func.sum(PosTransaksiPertamina.volume).label("vol"), func.sum(PosTransaksiPertamina.nilai).label("nil"), ) .where( PosTransaksiPertamina.spbu_id == spbu_id, PosTransaksiPertamina.tanggal >= tanggal_mulai, PosTransaksiPertamina.tanggal <= tanggal_selesai, ) .group_by(PosTransaksiPertamina.tanggal) .order_by(PosTransaksiPertamina.tanggal) ) pos_by_day: dict[date, dict] = { row.tanggal: {"vol": row.vol or Decimal("0"), "nil": row.nil or Decimal("0")} for row in pos_q } # Laporan totals per day — sum nozzle volume for nozzles with this produk_id lap_q = await db.execute( select( func.date(LaporanShift.tanggal).label("tanggal"), func.sum(PenjualanNozzle.volume).label("vol"), ) .join(PenjualanNozzle, PenjualanNozzle.laporan_shift_id == LaporanShift.id) .join(Nozzle, Nozzle.id == PenjualanNozzle.nozzle_id) .where( LaporanShift.spbu_id == spbu_id, LaporanShift.status.in_(["submitted", "for_approval", "approved"]), func.date(LaporanShift.tanggal) >= tanggal_mulai, func.date(LaporanShift.tanggal) <= tanggal_selesai, Nozzle.produk_id == produk_id, ) .group_by(func.date(LaporanShift.tanggal)) ) lap_by_day: dict[date, Decimal] = { row.tanggal: row.vol or Decimal("0") for row in lap_q } all_dates = sorted(set(list(pos_by_day.keys()) + list(lap_by_day.keys()))) rows_out = [] for d in all_dates: vol_pos = pos_by_day.get(d, {}).get("vol", Decimal("0")) nil_pos = pos_by_day.get(d, {}).get("nil", Decimal("0")) vol_lap = lap_by_day.get(d, Decimal("0")) selisih = vol_pos - vol_lap selisih_pct = (selisih / vol_lap * 100) if vol_lap else None threshold = produk.losses_threshold_penjualan_pct is_flagged = abs(selisih_pct) > threshold if selisih_pct is not None else False rows_out.append(PosCompareDayRow( tanggal=d, volume_pos=vol_pos, nilai_pos=nil_pos, volume_laporan=vol_lap, selisih_volume=selisih, selisih_pct=selisih_pct, is_flagged=is_flagged, )) return PosCompareResponse( rows=rows_out, total_volume_pos=sum(r.volume_pos for r in rows_out), total_volume_laporan=sum(r.volume_laporan for r in rows_out), ) async def get_anomali( db: AsyncSession, spbu_id: int, tanggal: date, produk_id: int | None = None, ) -> PosAnomaliResponse: """Detect DOUBLE_ISI (same plate < 60 min) and BURST (>3 txn/min per dispenser).""" q = select(PosTransaksiPertamina).where( PosTransaksiPertamina.spbu_id == spbu_id, PosTransaksiPertamina.tanggal == tanggal, ) if produk_id: produk = await _get_produk(db, produk_id) produk_raw_values = [k for k, v in PRODUK_MAP.items() if v == produk.kode] q = q.where(PosTransaksiPertamina.produk_raw.in_(produk_raw_values)) result = await db.execute(q.order_by(PosTransaksiPertamina.timestamp)) txns = list(result.scalars().all()) flagged: list[PosAnomaliItem] = [] # Detection A: same plat + same produk < 60 min by_plat: dict[str, list] = {} for t in txns: if t.plat_nomor: key = f"{t.plat_nomor}::{t.produk_raw}" by_plat.setdefault(key, []).append(t) for key, group in by_plat.items(): group.sort(key=lambda x: x.timestamp) for i in range(1, len(group)): diff = (group[i].timestamp - group[i - 1].timestamp).total_seconds() / 60 if diff < 60: flagged.append(PosAnomaliItem( tipe="DOUBLE_ISI", timestamp=group[i].timestamp, plat_nomor=group[i].plat_nomor, no_dispenser=group[i].no_dispenser, volume=group[i].volume, operator_name=group[i].operator_name, detail=f"Plat {group[i].plat_nomor} isi ulang setelah {diff:.0f} menit", )) # Detection B: burst > 3 txn per minute per dispenser burst_map: dict[str, list] = defaultdict(list) for t in txns: minute_key = t.timestamp.strftime("%Y-%m-%d %H:%M") burst_map[f"{t.no_dispenser}::{minute_key}"].append(t) for key, group in burst_map.items(): if len(group) > 3: t = group[0] flagged.append(PosAnomaliItem( tipe="BURST", timestamp=t.timestamp, plat_nomor=None, no_dispenser=t.no_dispenser, volume=sum(x.volume for x in group), operator_name=t.operator_name, detail=f"{len(group)} transaksi dalam 1 menit di {t.no_dispenser}", )) return PosAnomaliResponse(items=flagged, total=len(flagged)) async def get_golden_hours( db: AsyncSession, spbu_id: int, tanggal_mulai: date, tanggal_selesai: date, produk_id: int | None = None, shift_id: int | None = None, ) -> PosGoldenHoursResponse: """Volume per hour (0-23), per shift, and per island/nozzle (with hourly breakdown).""" base_where = [ PosTransaksiPertamina.spbu_id == spbu_id, PosTransaksiPertamina.tanggal >= tanggal_mulai, PosTransaksiPertamina.tanggal <= tanggal_selesai, ] if produk_id: produk = await _get_produk(db, produk_id) produk_raw_values = [k for k, v in PRODUK_MAP.items() if v == produk.kode] base_where.append(PosTransaksiPertamina.produk_raw.in_(produk_raw_values)) if shift_id: base_where.append(PosTransaksiPertamina.shift_id == shift_id) # Summary: total transaksi + volume summary_q = await db.execute( select( func.count(PosTransaksiPertamina.id).label("cnt"), func.sum(PosTransaksiPertamina.volume).label("vol"), ).where(*base_where) ) summary_row = summary_q.one() # Per-shift breakdown shift_q = await db.execute( select( PosTransaksiPertamina.shift_id, Shift.nama.label("shift_nama"), Shift.jam_mulai.label("jam_mulai"), func.count(PosTransaksiPertamina.id).label("cnt"), func.sum(PosTransaksiPertamina.volume).label("vol"), ) .outerjoin(Shift, Shift.id == PosTransaksiPertamina.shift_id) .where(*base_where) .group_by(PosTransaksiPertamina.shift_id, Shift.nama, Shift.jam_mulai) .order_by(Shift.jam_mulai.asc().nulls_last()) ) by_shift = [ PosShiftSummaryRow( shift_id=row.shift_id, shift_nama=row.shift_nama or "Tidak diketahui", transaksi_count=int(row.cnt), volume=row.vol or Decimal("0"), ) for row in shift_q ] summary = PosGoldenHoursSummary( total_transaksi=int(summary_row.cnt or 0), total_volume=summary_row.vol or Decimal("0"), by_shift=by_shift, ) # Hourly breakdown hourly_q = await db.execute( select( func.extract("hour", PosTransaksiPertamina.timestamp).label("jam"), func.sum(PosTransaksiPertamina.volume).label("vol"), func.count(PosTransaksiPertamina.id).label("cnt"), ) .where(*base_where) .group_by(func.extract("hour", PosTransaksiPertamina.timestamp)) .order_by(func.extract("hour", PosTransaksiPertamina.timestamp)) ) hourly = [ PosHourlyRow(jam=int(row.jam), volume=row.vol or Decimal("0"), transaksi_count=int(row.cnt)) for row in hourly_q ] # Island hourly: grouped by (island_raw, nozzle_raw, hour) island_hourly_q = await db.execute( select( PosTransaksiPertamina.island_raw, PosTransaksiPertamina.nozzle_raw, func.extract("hour", PosTransaksiPertamina.timestamp).label("jam"), func.sum(PosTransaksiPertamina.volume).label("vol"), func.count(PosTransaksiPertamina.id).label("cnt"), ) .where(*base_where) .group_by( PosTransaksiPertamina.island_raw, PosTransaksiPertamina.nozzle_raw, func.extract("hour", PosTransaksiPertamina.timestamp), ) .order_by( PosTransaksiPertamina.island_raw, PosTransaksiPertamina.nozzle_raw, func.extract("hour", PosTransaksiPertamina.timestamp), ) ) island_hourly: dict[tuple, list[PosIslandHourlyRow]] = defaultdict(list) for row in island_hourly_q: key = (row.island_raw or "Unknown", row.nozzle_raw) island_hourly[key].append( PosIslandHourlyRow(jam=int(row.jam), volume=row.vol or Decimal("0"), transaksi_count=int(row.cnt)) ) # Island/nozzle breakdown island_q = await db.execute( select( PosTransaksiPertamina.island_raw, PosTransaksiPertamina.nozzle_raw, func.sum(PosTransaksiPertamina.volume).label("vol"), func.count(PosTransaksiPertamina.id).label("cnt"), ) .where(*base_where) .group_by(PosTransaksiPertamina.island_raw, PosTransaksiPertamina.nozzle_raw) .order_by(func.sum(PosTransaksiPertamina.volume).desc()) ) by_island = [ PosIslandRow( island_raw=row.island_raw or "Unknown", nozzle_raw=row.nozzle_raw, volume=row.vol or Decimal("0"), transaksi_count=int(row.cnt), hourly=island_hourly.get((row.island_raw or "Unknown", row.nozzle_raw), []), ) for row in island_q ] return PosGoldenHoursResponse(summary=summary, hourly=hourly, by_island=by_island) async def get_batch_transactions( db: AsyncSession, spbu_id: int, batch_id: int, page: int = 1, per_page: int = 50, search: str | None = None, ) -> PosBatchTransaksiResponse: """List transactions for a specific batch with optional plat_nomor search.""" batch = (await db.execute( select(PosUploadBatch).where( PosUploadBatch.id == batch_id, PosUploadBatch.spbu_id == spbu_id, ) )).scalar_one_or_none() if not batch: raise ValueError("Batch tidak ditemukan") base_where = [PosTransaksiPertamina.batch_id == batch_id] if search: base_where.append(PosTransaksiPertamina.plat_nomor.ilike(f"%{search}%")) total = (await db.execute( select(func.count(PosTransaksiPertamina.id)).where(*base_where) )).scalar_one() result = await db.execute( select(PosTransaksiPertamina) .where(*base_where) .order_by(PosTransaksiPertamina.timestamp) .offset((page - 1) * per_page) .limit(per_page) ) txns = list(result.scalars().all()) return PosBatchTransaksiResponse( items=[ PosTransaksiItem( id=t.id, timestamp=t.timestamp, tanggal=t.tanggal, plat_nomor=t.plat_nomor, no_dispenser=t.no_dispenser, island_raw=t.island_raw, nozzle_raw=t.nozzle_raw, volume=t.volume, nilai=t.nilai, operator_name=t.operator_name, no_hp=t.no_hp, nik=t.nik, instansi=t.instansi, delivery_type=t.delivery_type, ) for t in txns ], total=total, page=page, per_page=per_page, ) async def get_golden_hours_insight( db: AsyncSession, spbu_id: int, tanggal_mulai: date, tanggal_selesai: date, produk_id: int, ) -> GoldenHoursInsightResponse: """Generate AI sales analysis & recommendations using Gemini.""" api_key = settings.GEMINI_API_KEY model = settings.GEMINI_MODEL if not api_key: raise ValueError("GEMINI_API_KEY not configured") # Fetch produk_nama first, then run golden hours (avoids double session call) produk = await _get_produk(db, produk_id) produk_nama = produk.nama data = await get_golden_hours(db, spbu_id, tanggal_mulai, tanggal_selesai, produk_id) periode = f"{tanggal_mulai} s/d {tanggal_selesai}" total_vol = float(data.summary.total_volume) total_txn = data.summary.total_transaksi # Compute derived metrics for richer context avg_vol_per_txn = total_vol / total_txn if total_txn else 0 sorted_hourly = sorted(data.hourly, key=lambda h: float(h.volume), reverse=True) top3_jam = sorted_hourly[:3] bottom3_jam = sorted_hourly[-3:] if len(sorted_hourly) >= 3 else [] shift_lines = "\n".join( f" - {s.shift_nama}: {s.transaksi_count} txn | {float(s.volume):.0f} L " f"({float(s.volume)/total_vol*100:.1f}% dari total)" if total_vol else "" for s in data.summary.by_shift ) hourly_lines = "\n".join( f" - {h.jam:02d}:00–{(h.jam+1)%24:02d}:00: {float(h.volume):.0f} L | {h.transaksi_count} txn | " f"avg {float(h.volume)/h.transaksi_count:.1f} L/txn" if h.transaksi_count else f" - {h.jam:02d}:00: {float(h.volume):.0f} L" for h in sorted_hourly ) island_lines = "\n".join( f" - {row.island_raw}{' – ' + row.nozzle_raw if row.nozzle_raw else ''}: " f"{float(row.volume):.0f} L | {row.transaksi_count} txn | " f"peak: " + ", ".join( f"{h.jam:02d}:00 ({float(h.volume):.0f}L)" for h in sorted(row.hourly, key=lambda h: float(h.volume), reverse=True)[:3] ) for row in data.by_island[:15] ) data_text = f"""DATA PENJUALAN SPBU — {produk_nama} | {periode} RINGKASAN: - Total transaksi: {total_txn} | Total volume: {total_vol:.0f} L - Rata-rata per transaksi: {avg_vol_per_txn:.1f} L BREAKDOWN PER SHIFT: {shift_lines} VOLUME PER JAM (semua jam, urut terbesar): {hourly_lines} VOLUME PER ISLAND/NOZZLE: {island_lines} """ system_prompt = """Kamu adalah konsultan operasional SPBU senior dengan pengalaman 15 tahun di lapangan. Tugasmu: analisis data penjualan BBM dan berikan rekomendasi operasional yang SANGAT SPESIFIK dan bisa langsung dieksekusi manajer SPBU hari ini. ATURAN KETAT: 1. JANGAN rangkum ulang angka — manajer sudah baca tabelnya 2. INTERPRETASIKAN pola: apa artinya puncak di jam X? kenapa island Y sepi? 3. Setiap rekomendasi HARUS menyebut: jam konkret, nama island/nozzle, jumlah orang, atau tindakan spesifik 4. Gunakan Bahasa Indonesia, nada tegas seperti briefing operasional FORMAT OUTPUT WAJIB (lengkap semua section): ## Temuan Utama - [3-4 poin insight kunci: pola waktu, ketimpangan antar island, efisiensi per txn] ## Rekomendasi Penjadwalan Shift - Shift mana yang perlu tambah/kurang operator dan berapa jumlahnya - Jam berapa harus ada supervisor di lapangan - Apakah jam kerja shift perlu disesuaikan dengan pola puncak ## Rekomendasi per Island & Nozzle - Island/nozzle mana yang perlu prioritas (selalu ada petugas, tidak tutup) - Island/nozzle mana yang bisa dikurangi operatornya di jam sepi - Apakah ada nozzle yang underperform dan perlu investigasi ## Strategi Jam Sepi - Jam-jam mana yang volume rendah dan apa yang bisa dilakukan (maintenance, training, restocking) - Apakah layak menutup sebagian nozzle di jam tertentu untuk efisiensi ## Estimasi Dampak - Jika rekomendasi dijalankan, kira-kira berapa persen peningkatan efisiensi atau volume yang bisa dicapai""" url = f"https://generativelanguage.googleapis.com/v1beta/models/{model}:generateContent?key={api_key}" body = { "systemInstruction": {"parts": [{"text": system_prompt}]}, "contents": [{"role": "user", "parts": [{"text": data_text}]}], "generationConfig": {"temperature": 0.5, "maxOutputTokens": 8192}, } try: async with httpx.AsyncClient(timeout=90) as client: resp = await client.post(url, json=body) resp.raise_for_status() result = resp.json() except httpx.HTTPStatusError as e: raise ValueError(f"Gemini API error {e.response.status_code}: {e.response.text[:200]}") from e candidates = result.get("candidates", []) if not candidates: raise ValueError("Gemini returned no candidates") parts = candidates[0].get("content", {}).get("parts", []) insight = "".join(p.get("text", "") for p in parts if "text" in p).strip() if not insight: raise ValueError("Gemini returned empty insight") return GoldenHoursInsightResponse(insight=insight)