from datetime import datetime, timezone, timedelta import secrets import io import base64 import asyncio import math from fastapi import APIRouter, Request, Form, Depends, HTTPException from fastapi.responses import HTMLResponse, RedirectResponse, StreamingResponse, JSONResponse from fastapi.templating import Jinja2Templates from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy import select, update, or_, func, case from app.core.database import get_db from app.core.security import require_roles from app.models.models import ( Voucher, VoucherAgent, HotspotProfile, VoucherBatch, MikrotikDevice, DeviceCredential, Site, User, Payment, WalletTransaction, AgentCommission ) from app.core.device_credentials import decrypt_secret from app.core.mikrotik import RealMikrotikAdapter, MikrotikProvisioningError from app.core.voucher_generation import generate_credentials, CredentialPolicyError from app.core.device_credentials import encrypt_secret import os import qrcode router = APIRouter(dependencies=[Depends(require_roles("agen_voucher", "admin_cs", "super_admin"))]) BASE_DIR = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) templates = Jinja2Templates(directory=os.path.join(BASE_DIR, "app/templates")) _batch_action_lock = asyncio.Lock() async def _get_hotspot_device(db: AsyncSession, device_id: int | None = None) -> tuple[MikrotikDevice, DeviceCredential]: if device_id: mk = await db.get(MikrotikDevice, device_id) if not mk or not mk.is_active: raise HTTPException(status_code=404, detail="Perangkat MikroTik tujuan tidak ditemukan atau tidak aktif") else: mk = (await db.execute(select(MikrotikDevice).where(MikrotikDevice.is_active == True).order_by(MikrotikDevice.id))).scalars().first() if not mk or not mk.credential_id: raise HTTPException(status_code=503, detail="MikroTik Hotspot belum terkonfigurasi atau tidak memiliki kredensial") cred = await db.get(DeviceCredential, mk.credential_id) if not (cred and cred.is_active): raise HTTPException(status_code=503, detail="Kredensial MikroTik tidak aktif") return mk, cred async def _hotspot_adapter(db: AsyncSession, device_id: int | None = None) -> RealMikrotikAdapter: mk, cred = await _get_hotspot_device(db, device_id) return RealMikrotikAdapter(host=mk.host, username=cred.username, password=decrypt_secret(cred.encrypted_secret), api_port=mk.api_port or 8728) async def _batch_adapters(db: AsyncSession, vouchers: list[Voucher]) -> dict[int, RealMikrotikAdapter]: device_ids = {v.device_id for v in vouchers if v.device_id} adapters: dict[int, RealMikrotikAdapter] = {} if not device_ids: default_mk, _ = await _get_hotspot_device(db) adapters[default_mk.id] = await _hotspot_adapter(db, default_mk.id) for v in vouchers: v.device_id = default_mk.id return adapters for dev_id in device_ids: adapters[dev_id] = await _hotspot_adapter(db, dev_id) return adapters def _format_validity(uptime_str: str | None, duration_days: int | None = None) -> str: if not uptime_str and duration_days: return f"{duration_days} Hari" if not uptime_str: return "24 Jam" u = str(uptime_str).strip().lower() if u.endswith("h"): return f"{u[:-1]} Jam" elif u.endswith("d"): return f"{u[:-1]} Hari" elif u.endswith("m"): return f"{u[:-1]} Menit" return uptime_str def _generate_qr_data_uri(content: str) -> str: qr = qrcode.QRCode(version=1, box_size=3, border=1) qr.add_data(content) qr.make(fit=True) img = qr.make_image(fill_color="black", back_color="white") buf = io.BytesIO() img.save(buf, format="PNG") b64 = base64.b64encode(buf.getvalue()).decode("ascii") return f"data:image/png;base64,{b64}" @router.get("/vouchers", response_class=HTMLResponse) async def vouchers_page( request: Request, q: str | None = None, batch: str | None = None, profile_id: int | None = None, db: AsyncSession = Depends(get_db), actor: User = Depends(require_roles("agen_voucher", "admin_cs", "super_admin")), ): selected_batch = None if batch: if not batch.isascii() or not batch.isdigit() or int(batch) <= 0: raise HTTPException(status_code=422, detail="ID batch tidak valid") selected_batch = await db.get(VoucherBatch, int(batch)) if selected_batch is None: raise HTTPException(status_code=404, detail="Batch voucher tidak ditemukan") profiles = (await db.execute(select(HotspotProfile).where(HotspotProfile.is_active == True).order_by(HotspotProfile.name))).scalars().all() sites = (await db.execute(select(Site).order_by(Site.name))).scalars().all() devices = (await db.execute(select(MikrotikDevice).where(MikrotikDevice.is_active == True).order_by(MikrotikDevice.name))).scalars().all() batches = (await db.execute(select(VoucherBatch).order_by(VoucherBatch.id.desc()).limit(100))).scalars().all() agents = (await db.execute(select(VoucherAgent).where(VoucherAgent.is_active == True).order_by(VoucherAgent.name))).scalars().all() total_count = await db.scalar(select(func.count(Voucher.id))) or 0 profile_usage = {} for p in profiles: num_batches = await db.scalar(select(func.count(VoucherBatch.id)).where(VoucherBatch.profile_id == p.id)) or 0 num_vouchers = await db.scalar(select(func.count(Voucher.id)).where(Voucher.profile_id == p.id)) or 0 profile_usage[p.id] = {'batches': num_batches, 'vouchers': num_vouchers} protected_profiles = ["default"] return templates.TemplateResponse(request=request, name="vouchers.html", context={ "profiles": profiles, "sites": sites, "devices": devices, "batches": batches, "agents": agents, "current_user": actor, "q": q or "", "batch": int(batch) if batch and batch.isdigit() else None, "selected_batch": selected_batch, "selected_profile_id": profile_id, "total_vouchers": total_count, "profile_usage": profile_usage, "protected_profiles": protected_profiles, }) @router.get("/api/vouchers/batches") async def list_voucher_batches_api( profile_id: int | None = None, db: AsyncSession = Depends(get_db), actor: User = Depends(require_roles("agen_voucher", "admin_cs", "super_admin")), ): stmt = ( select( VoucherBatch.id, VoucherBatch.batch_code, VoucherBatch.profile, VoucherBatch.profile_id, VoucherBatch.comment, VoucherBatch.quantity, func.count(Voucher.id).label("voucher_count"), func.sum(case((Voucher.status == "available", 1), else_=0)).label("available_count"), func.sum(case((Voucher.status == "sold", 1), else_=0)).label("sold_count"), VoucherBatch.created_at, ) .join(Voucher, Voucher.batch_id == VoucherBatch.id) .group_by(VoucherBatch.id) .order_by(VoucherBatch.id.desc()) ) if profile_id: stmt = stmt.where(VoucherBatch.profile_id == profile_id) results = (await db.execute(stmt)).all() batches = [] for r in results: comment_part = r.comment.strip() if r.comment and r.comment.strip() else r.batch_code profile_part = r.profile or "" # Mikhmon format: comment/batch profile [sisa_count] # e.g. YULI-PYH_26SEP26_100pcs P5rb [50] remaining = r.available_count if r.available_count is not None else r.voucher_count label = f"{comment_part} {profile_part} [{remaining}]".strip() batches.append({ "id": r.id, "batch_code": r.batch_code, "profile": r.profile, "profile_id": r.profile_id, "comment": r.comment or "", "display_label": label, "quantity": r.quantity, "voucher_count": r.voucher_count, "available_count": r.available_count or 0, "sold_count": r.sold_count or 0, "created_at": r.created_at.strftime("%Y-%m-%d %H:%M") if r.created_at else "", }) return {"batches": batches} @router.get("/api/vouchers/datatable") async def vouchers_datatable( request: Request, db: AsyncSession = Depends(get_db), actor: User = Depends(require_roles("agen_voucher", "admin_cs", "super_admin")), ): params = request.query_params try: draw = int(params.get("draw", 1)) except (ValueError, TypeError): draw = 1 try: start = max(0, int(params.get("start", 0))) except (ValueError, TypeError): start = 0 try: length = int(params.get("length", 10)) if length <= 0 or length > 100: length = 10 except (ValueError, TypeError): length = 10 search_value = (params.get("search[value]") or params.get("q") or "").strip() batch_param = params.get("batch") or params.get("batch_id") profile_param = params.get("profile_id") status_param = params.get("status") try: order_col_idx = int(params.get("order[0][column]", 0)) except (ValueError, TypeError): order_col_idx = 0 order_dir = params.get("order[0][dir]", "desc").lower() col_map = { 0: Voucher.id, 1: Voucher.username, 2: Voucher.id, 3: Voucher.profile, 4: Voucher.price, 5: Voucher.status, 6: Voucher.batch_id, 7: Voucher.id, } order_col = col_map.get(order_col_idx, Voucher.id) order_clause = order_col.desc() if order_dir == "desc" else order_col.asc() records_total = await db.scalar(select(func.count(Voucher.id))) or 0 filters = [] if batch_param and str(batch_param).isdigit() and int(batch_param) > 0: filters.append(Voucher.batch_id == int(batch_param)) if profile_param and str(profile_param).isdigit() and int(profile_param) > 0: filters.append(Voucher.profile_id == int(profile_param)) if status_param and status_param.strip(): filters.append(Voucher.status == status_param.strip()) if search_value: needle = f"%{search_value}%" filters.append( or_( Voucher.code.ilike(needle), Voucher.username.ilike(needle), Voucher.comment.ilike(needle), Voucher.profile.ilike(needle), Voucher.status.ilike(needle), ) ) count_stmt = select(func.count(Voucher.id)) if filters: count_stmt = count_stmt.where(*filters) records_filtered = await db.scalar(count_stmt) or 0 stmt = select(Voucher) if filters: stmt = stmt.where(*filters) stmt = stmt.order_by(order_clause, Voucher.id.desc()).offset(start).limit(length) vouchers = (await db.execute(stmt)).scalars().all() data = [] for v in vouchers: data.append({ "id": v.id, "username": v.username or v.code, "has_password": bool(v.password_ciphertext), "profile": v.profile or "-", "price": v.price or 0, "price_formatted": f"Rp {v.price:,}".replace(",", ".") if v.price is not None else "Rp 0", "status": v.status, "batch_id": v.batch_id, "expires_at": v.expires_at.strftime('%Y-%m-%d') if v.expires_at else None, "comment": v.comment or "", }) return { "draw": draw, "recordsTotal": records_total, "recordsFiltered": records_filtered, "data": data, } @router.post("/hotspot-profiles") async def create_hotspot_profile( name: str = Form(...), local_profile: str = Form(...), price: int = Form(...), rate_down: int = Form(...), rate_up: int = Form(...), session_timeout: str = Form("12h"), idle_timeout: str = Form("5m"), shared: int = Form(1), expiry_value: int = Form(30), expiry_unit: str = Form("days"), device_id: int = Form(...), db: AsyncSession = Depends(get_db), actor: User = Depends(require_roles("super_admin")), ): name, local_profile = name.strip(), local_profile.strip() if not name or not local_profile or price < 0 or rate_down <= 0 or rate_up <= 0: raise HTTPException(status_code=422, detail="Data Hotspot Profile tidak valid") if not 1 <= shared <= 20 or expiry_value < 1: raise HTTPException(status_code=422, detail="Batas pengguna atau masa berlaku tidak valid") # Calculate days representation if expiry_unit == "hours": voucher_expiry_days = max(1, math.ceil(expiry_value / 24)) else: voucher_expiry_days = max(1, min(expiry_value, 365)) mk, _ = await _get_hotspot_device(db, device_id) adapter = await _hotspot_adapter(db, device_id) rate_limit = f"{rate_down}k/{rate_up}k" try: await asyncio.to_thread(adapter.upsert_hotspot_profile, local_profile, rate_limit, session_timeout, idle_timeout, shared) except MikrotikProvisioningError as exc: raise HTTPException(status_code=502, detail=f"Profile gagal diterapkan ke MikroTik: {exc}") from exc profile = (await db.execute(select(HotspotProfile).where(HotspotProfile.name == name, HotspotProfile.site_id == mk.site_id))).scalars().first() if profile is None: profile = HotspotProfile(name=name, site_id=mk.site_id) db.add(profile) profile.local_profile = local_profile profile.rate_down, profile.rate_up = rate_down, rate_up profile.session_timeout, profile.idle_timeout = session_timeout, idle_timeout profile.shared, profile.price = shared, price profile.voucher_expiry_days = voucher_expiry_days profile.is_active, profile.sync_status = True, "synced" await db.commit() return RedirectResponse("/vouchers?profile_saved=1", status_code=303) @router.get("/api/voucher-targets/{device_id}/profiles") async def load_target_profiles(device_id: int, db: AsyncSession = Depends(get_db)): adapter = await _hotspot_adapter(db, device_id) try: remote_profiles = await asyncio.to_thread(adapter.list_hotspot_profiles) except MikrotikProvisioningError as exc: raise HTTPException(status_code=502, detail="Profile MikroTik tidak dapat dimuat") from exc configured = (await db.execute(select(HotspotProfile).where(HotspotProfile.is_active == True))).scalars().all() by_local = {p.local_profile or p.name: p for p in configured} return {"profiles": [{"name": item.get("name"), "configured": item.get("name") in by_local, "id": by_local[item.get("name")].id if item.get("name") in by_local else None, "price": by_local[item.get("name")].price if item.get("name") in by_local else None, "session_timeout": by_local[item.get("name")].session_timeout if item.get("name") in by_local else None, "voucher_expiry_days": by_local[item.get("name")].voucher_expiry_days if item.get("name") in by_local else None} for item in remote_profiles if item.get("name")]} @router.post("/vouchers/generate") async def generate_vouchers( profile_id: int | None = Form(None), profile: str | None = Form(None), device_id: int | None = Form(None), quantity: int = Form(...), agent_id: int | None = Form(None), user_mode: str = Form("same"), character_length: int = Form(6), character_mode: str = Form("mixed_alnum"), price: int | None = Form(None), duration_days: int | None = Form(None), limit_uptime: str | None = Form(None), limit_idle_time: str | None = Form(None), batch_code: str | None = Form(None), comment: str | None = Form(None), site_id: int | None = Form(None), db: AsyncSession = Depends(get_db), actor: User = Depends(require_roles("agen_voucher", "admin_cs", "super_admin")), ): if not 1 <= quantity <= 400: raise HTTPException(status_code=422, detail="Jumlah voucher harus 1 sampai 400") try: generate_credentials(user_mode, character_length, character_mode) except CredentialPolicyError as exc: raise HTTPException(status_code=422, detail=str(exc)) from exc mk, _ = await _get_hotspot_device(db, device_id) if site_id is not None and mk.site_id is not None and site_id != mk.site_id: raise HTTPException(status_code=422, detail="Site tidak cocok dengan perangkat tujuan") profile_row = await db.get(HotspotProfile, profile_id) if profile_id else None if profile_id and (not profile_row or not profile_row.is_active): raise HTTPException(status_code=422, detail="Hotspot Profile tidak aktif atau tidak ditemukan") if profile_row: profile_name = profile_row.local_profile or profile_row.name resolved_price = int(profile_row.price or 0) resolved_uptime = profile_row.session_timeout or "12h" resolved_idle = profile_row.idle_timeout resolved_days = int(profile_row.voucher_expiry_days or 30) else: profile_name = (profile or "default").strip() resolved_price = int(price or 0) resolved_uptime = (limit_uptime or "12h").strip() resolved_idle = limit_idle_time resolved_days = max(1, min(int(duration_days or 30), 365)) adapter = await _hotspot_adapter(db, mk.id) # Fetch agent details if provided agent = None if agent_id: agent = await db.get(VoucherAgent, agent_id) generated_batch_code = (batch_code or f"B{datetime.now().strftime('%y%m%d%H%M%S')}-{secrets.token_hex(2).upper()}").strip() # Format comment if agent is present and no comment is explicitly typed resolved_comment = (comment or "").strip() if not resolved_comment and agent: resolved_comment = f"{agent.name}-{agent.area or 'GEN'}_{datetime.now().strftime('%d%b%y').upper()}_{quantity}pcs" batch = VoucherBatch( batch_code=generated_batch_code, profile=profile_name, quantity=quantity, price=resolved_price, agent_id=agent.id if agent else None, site_id=mk.site_id, device_id=mk.id, created_by=actor.id, comment=resolved_comment, sync_status="pending", profile_id=profile_row.id if profile_row else None, duration_days=resolved_days, limit_uptime=resolved_uptime, limit_idle_time=resolved_idle, user_mode=user_mode, character_length=character_length, character_mode=character_mode, ) db.add(batch) await db.flush() expires_at = datetime.now(timezone.utc) + timedelta(days=resolved_days) created_vouchers: list[Voucher] = [] generated_creds: list[tuple[str, str]] = [] for _ in range(quantity): while True: u, p = generate_credentials(user_mode, character_length, character_mode) if not any(item[0] == u for item in generated_creds): generated_creds.append((u, p)) break for u, p in generated_creds: v = Voucher( code=u, username=u, password_ciphertext=encrypt_secret(p) if user_mode == "separate" else None, credential_mode=user_mode, credential_length=character_length, character_mode=character_mode, profile=profile_name, profile_id=profile_row.id if profile_row else None, price=resolved_price, status="sold" if agent else "available", agent_id=agent.id if agent else None, sold_at=datetime.now(timezone.utc) if agent else None, expires_at=expires_at, limit_uptime=resolved_uptime, limit_idle_time=resolved_idle, comment=resolved_comment, sync_status="pending", batch_id=batch.id, device_id=mk.id, site_id=mk.site_id, created_by=actor.id, ) db.add(v) created_vouchers.append(v) await db.flush() for idx, v in enumerate(created_vouchers): u, p = generated_creds[idx] comment_text = f"b:{batch.id} c:{v.code}" + (f" ag:{agent.name}" if agent else "") try: await asyncio.to_thread(adapter.add_hotspot_user, u, p, profile_name, resolved_uptime, resolved_idle, comment_text) v.sync_status = "synced" v.sync_updated_at = datetime.now(timezone.utc) except MikrotikProvisioningError as exc: v.sync_status = "failed" batch.sync_status = "failed" await db.commit() raise HTTPException(status_code=502, detail=f"Gagal push user {v.username} ke RouterOS: {exc}") from exc batch.sync_status = "synced" batch.generated_count = quantity await db.commit() return RedirectResponse(url=f"/vouchers?batch={batch.id}&generated=1", status_code=303) @router.get("/vouchers/batches//print", response_class=HTMLResponse) @router.get("/vouchers/batches/print", response_class=HTMLResponse) async def missing_batch_print(): return RedirectResponse(url="/vouchers?notice=select_batch", status_code=303) @router.get("/vouchers/batches/{batch_id}/print", response_class=HTMLResponse, dependencies=[Depends(require_roles("agen_voucher", "admin_cs", "super_admin"))]) async def print_voucher_batch(batch_id: int, request: Request, db: AsyncSession = Depends(get_db)): batch = await db.get(VoucherBatch, batch_id) if batch is None: raise HTTPException(status_code=404, detail="Batch voucher tidak ditemukan") vouchers = (await db.execute(select(Voucher).where(Voucher.batch_id == batch.id).order_by(Voucher.id.asc()))).scalars().all() site_name = "Munara.Id" login_url = "http://munara.net" if batch.site_id: site = await db.get(Site, batch.site_id) if site and site.name: site_name = f"Munara.Id_{site.name}" if "munara" not in site.name.lower() else site.name elif batch.device_id: dev = await db.get(MikrotikDevice, batch.device_id) if dev and dev.name: dev_cleaned = dev.name.replace("MK-SITE-", "").replace("MK-", "") site_name = f"Munara.Id_{dev_cleaned}" if "munara" not in dev.name.lower() else dev.name printable = [] for idx, voucher in enumerate(vouchers): password = None if voucher.password_ciphertext: try: password = decrypt_secret(voucher.password_ciphertext) except Exception: password = None is_same_mode = (voucher.credential_mode == "legacy_same" or voucher.credential_mode == "same" or batch.user_mode == "same" or not password or password == (voucher.username or voucher.code)) validity = _format_validity(voucher.limit_uptime or batch.limit_uptime, batch.duration_days) price_val = voucher.price if voucher.price is not None else batch.price price_formatted = f"Rp {price_val:,.0f}".replace(",", ".") printable.append({ "id": voucher.id, "index": idx + 1, "username": voucher.username or voucher.code, "password": password, "is_same_mode": is_same_mode, "profile": voucher.profile, "price": price_val, "price_formatted": price_formatted, "validity": validity, "comment": voucher.comment or batch.comment or "", "qr_content": voucher.username or voucher.code, "qr_data_uri": _generate_qr_data_uri(voucher.username or voucher.code), }) return templates.TemplateResponse(request=request, name="voucher_batch_print.html", context={ "batch": batch, "site_name": site_name, "login_url": login_url, "vouchers": printable, }) @router.post("/vouchers/batches/{batch_id}/sell") async def sell_voucher_batch( batch_id: int, agent_id: int = Form(...), db: AsyncSession = Depends(get_db), actor: User = Depends(require_roles("agen_voucher", "admin_cs", "super_admin")), ): async with _batch_action_lock: batch = await db.get(VoucherBatch, batch_id) agent = await db.get(VoucherAgent, agent_id) if not batch or not agent or not agent.is_active: raise HTTPException(status_code=404, detail="Batch atau agen tidak ditemukan/tidak aktif") vouchers = (await db.execute(select(Voucher).where(Voucher.batch_id == batch_id, Voucher.status == "available"))).scalars().all() if not vouchers: raise HTTPException(status_code=409, detail="Tidak ada voucher yang tersedia dalam batch ini") now = datetime.now(timezone.utc) for v in vouchers: v.agent_id = agent.id v.status = "sold" v.sold_at = now v.sold_price = v.price batch.agent_id = agent.id await db.commit() return RedirectResponse(url=f"/vouchers?batch={batch_id}&sold={len(vouchers)}", status_code=303) @router.post("/vouchers/batches/{batch_id}/delete") async def delete_voucher_batch( batch_id: int, db: AsyncSession = Depends(get_db), actor: User = Depends(require_roles("agen_voucher", "admin_cs", "super_admin")), ): async with _batch_action_lock: batch = await db.get(VoucherBatch, batch_id) if batch is None: raise HTTPException(status_code=404, detail="Batch voucher tidak ditemukan") vouchers = (await db.execute( select(Voucher).where(Voucher.batch_id == batch_id).order_by(Voucher.id.asc()) )).scalars().all() if vouchers: adapter = await _hotspot_adapter(db, batch.device_id or vouchers[0].device_id) failed: list[str] = [] for voucher in vouchers: username = voucher.username or voucher.code try: if not await asyncio.to_thread(adapter.delete_hotspot_user, username): failed.append(username) except MikrotikProvisioningError: failed.append(username) if failed: raise HTTPException( status_code=502, detail=f"Batch belum dihapus. Gagal menghapus/verifikasi {len(failed)} user di RouterOS: {', '.join(failed[:5])}", ) for voucher in vouchers: await db.delete(voucher) await db.delete(batch) await db.commit() return RedirectResponse(url="/vouchers?batch_deleted=1", status_code=303) @router.post("/vouchers/{voucher_id}/delete") async def delete_single_voucher( voucher_id: int, db: AsyncSession = Depends(get_db), actor: User = Depends(require_roles("agen_voucher", "admin_cs", "super_admin")), ): voucher = await db.get(Voucher, voucher_id) if voucher is None: raise HTTPException(status_code=404, detail="Voucher tidak ditemukan") username = voucher.username or voucher.code if username: try: adapter = await _hotspot_adapter(db, voucher.device_id) await asyncio.to_thread(adapter.delete_hotspot_user, username) except Exception: pass await db.delete(voucher) await db.commit() return RedirectResponse(url="/vouchers?deleted=1", status_code=303) @router.get("/vouchers/{voucher_id}/print", response_class=HTMLResponse, dependencies=[Depends(require_roles("agen_voucher", "admin_cs", "super_admin"))]) async def print_single_voucher(voucher_id: int, request: Request, db: AsyncSession = Depends(get_db)): voucher = await db.get(Voucher, voucher_id) if voucher is None: raise HTTPException(status_code=404, detail="Voucher tidak ditemukan") batch = await db.get(VoucherBatch, voucher.batch_id) if voucher.batch_id else None site_name = "Munara.Id" login_url = "http://munara.net" if voucher.site_id: site = await db.get(Site, voucher.site_id) if site and site.name: site_name = f"Munara.Id_{site.name}" if "munara" not in site.name.lower() else site.name elif voucher.device_id: dev = await db.get(MikrotikDevice, voucher.device_id) if dev and dev.name: dev_cleaned = dev.name.replace("MK-SITE-", "").replace("MK-", "") site_name = f"Munara.Id_{dev_cleaned}" if "munara" not in dev.name.lower() else dev.name password = None if voucher.password_ciphertext: try: password = decrypt_secret(voucher.password_ciphertext) except Exception: password = None is_same_mode = (voucher.credential_mode == "legacy_same" or voucher.credential_mode == "same" or (batch and batch.user_mode == "same") or not password or password == (voucher.username or voucher.code)) validity = _format_validity(voucher.limit_uptime or (batch.limit_uptime if batch else None), batch.duration_days if batch else None) price_val = voucher.price if voucher.price is not None else (batch.price if batch else 0) price_formatted = f"Rp {price_val:,.0f}".replace(",", ".") printable = [{ "id": voucher.id, "index": 1, "username": voucher.username or voucher.code, "password": password, "is_same_mode": is_same_mode, "profile": voucher.profile, "price": price_val, "price_formatted": price_formatted, "validity": validity, "comment": voucher.comment or (batch.comment if batch else ""), "qr_content": voucher.username or voucher.code, "qr_data_uri": _generate_qr_data_uri(voucher.username or voucher.code), }] mock_batch = batch or VoucherBatch(batch_code=f"VOUCHER-{voucher.id}") return templates.TemplateResponse(request=request, name="voucher_batch_print.html", context={ "batch": mock_batch, "site_name": site_name, "login_url": login_url, "vouchers": printable, })