"""Periodic live-status reconciliation for Billing, customers and hotspot vouchers. Voucher lifecycle rules (independent triggers, either one is enough): 1. Duration/jatah habis -> hapus dari RouterOS + database. 2. Masa berlaku habis -> hapus dari RouterOS + database. Status juga disinkronkan LIVE dari RouterOS (uptime / limit-uptime / active / disabled) supaya DataTables voucher selalu mencerminkan kondisi perangkat. """ import asyncio import logging import os import re _log = logging.getLogger("bestweb.voucher_monitor") from sqlalchemy import select from datetime import datetime, timezone, timedelta from app.core.database import AsyncSessionLocal from app.core.onboarding import refresh_customer_online_status from app.models.models import Voucher, VoucherBatch, Customer, MikrotikDevice, HotspotProfile from app.models.models import ( Customer, DeviceCredential, MikrotikDevice, Voucher, ) from app.core.mikrotik import RealMikrotikAdapter from app.core.secrets import decrypt_secret_strict, SecretDecryptionError def poll_interval() -> int: try: return max(10, int(os.getenv("BESTWEB_STATUS_POLL_SECONDS", "30"))) except ValueError: return 30 def voucher_poll_interval() -> int: try: return max(5, int(os.getenv("BESTWEB_VOUCHER_POLL_SECONDS", "10"))) except ValueError: return 10 def parse_uptime_seconds(value: str | None) -> int: """Parse RouterOS uptime/limit-uptime into seconds. Accepts 1d02:03:04, 02:03:04, 3h04m, 2w, 5m, 45s and bare digits. """ raw = str(value or "").strip().lower() if not raw: return 0 # RouterOS uptime: [Nd]HH:MM:SS (4 parts = days + h:m:s, 3 parts = h:m:s) if ":" in raw: days = 0 body = raw day_match = re.match(r"^(\d+)d(.*)$", raw) if day_match: days = int(day_match.group(1)) body = day_match.group(2) parts = body.split(":") try: nums = [int(float(p)) for p in parts] except ValueError: return 0 while len(nums) < 3: nums.insert(0, 0) if len(nums) >= 4: days += nums[0] nums = nums[-3:] return days * 86400 + nums[0] * 3600 + nums[1] * 60 + nums[2] match = re.fullmatch( r"(?:(\d+)w)?(?:(\d+)d)?(?:(\d+)h)?(?:(\d+)m)?(?:(\d+)s)?", raw ) if match and any(match.groups()): w, d, h, m, s = (int(g) if g else 0 for g in match.groups()) return w * 604800 + d * 86400 + h * 3600 + m * 60 + s try: return int(float(raw)) except ValueError: return 0 def as_utc(value: datetime | None) -> datetime | None: """SQLite drops tzinfo, so treat naive timestamps as UTC before comparing.""" if value is None: return None if value.tzinfo is None: return value.replace(tzinfo=timezone.utc) return value.astimezone(timezone.utc) async def _device_adapters(db) -> dict[int, RealMikrotikAdapter]: """Build one adapter per active MikroTik device that has valid credentials. Raises SecretDecryptionError when a stored secret cannot be decrypted with the active encryption key, so the caller can surface it instead of silently pushing ciphertext to the device. """ devices = (await db.execute( select(MikrotikDevice).where(MikrotikDevice.is_active == True) )).scalars().all() adapters: dict[int, RealMikrotikAdapter] = {} for mk in devices: if not mk.credential_id: continue cred = await db.get(DeviceCredential, mk.credential_id) if not (cred and cred.is_active): continue password = decrypt_secret_strict( cred.encrypted_secret, label=f"Kredensial {cred.name}" ) adapters[mk.id] = RealMikrotikAdapter( host=mk.host, username=cred.username, password=password, api_port=mk.api_port or 8728, ) return adapters def _routeros_bytes(value) -> int: """Parse RouterOS byte counters like '123456' or '1.2MiB' -> int bytes.""" if value in (None, "", b""): return 0 text = str(value).strip() if text.isdigit(): return int(text) units = {"kib": 1024, "mib": 1024 ** 2, "gib": 1024 ** 3, "kb": 1000, "mb": 1000 ** 2, "gb": 1000 ** 3} lowered = text.lower().replace(" ", "") for suffix, mult in units.items(): if lowered.endswith(suffix): try: return int(float(lowered[: -len(suffix)]) * mult) except ValueError: return 0 try: return int(float(lowered)) except ValueError: return 0 async def reconcile_vouchers() -> None: """Sync voucher status/deletion against live RouterOS state.""" async with AsyncSessionLocal() as db: now = datetime.now(timezone.utc) # `reserved` = voucher dipegang order online yang belum dibayar. Voucher # itu belum masuk servis, jadi harus TIDAK dinilai terhadap state RouterOS: # kalau ikut di loop, pembeli yang kebetulan login duluan tidak akan # pernah transisi ke `used`, dan bisa ikut kedaluwarsa oleh aturan masa berlaku. vouchers = ( await db.execute(select(Voucher).where(Voucher.status != "reserved")) ).scalars().all() if not vouchers: return try: adapters = await _device_adapters(db) except SecretDecryptionError as exc: _log.error("voucher reconcile stopped: %s", exc) raise except Exception: return # Cache RouterOS state per device so we hit the API once per device. remote_by_device: dict[int, dict[str, dict]] = {} device_synced: dict[int, bool] = {} changed = False for device_id, adapter in adapters.items(): try: remote_rows = await asyncio.to_thread(adapter.list_hotspot_users) remote_by_device[device_id] = {row["name"]: row for row in remote_rows} device_synced[device_id] = True except Exception: # Device offline or credentials rejected: skip status sync for # this cycle rather than blocking the expiry rule below. remote_by_device[device_id] = {} device_synced[device_id] = False _log.warning( "voucher reconcile: RouterOS read failed for device %s", device_id ) else: device = await db.get(MikrotikDevice, device_id) if device is not None and device.last_sync_at != now: device.last_sync_at = now changed = True for v in vouchers: username = (v.username or v.code or "").strip() device_id = v.device_id batch = await db.get(VoucherBatch, v.batch_id) remote = remote_by_device.get(device_id, {}).get(username) if device_id else None is_synced = device_synced.get(device_id, False) # Condition 2: absolute expiry (masa berlaku) reached ONLY for vouchers that have been activated/used. # Vouchers that are simply sold/available should NOT expire based on creation time, because # validity starts on first login! Only if `used_at` is set, or if it has activity. expires_at = as_utc(v.expires_at) # If voucher was never used, its validity countdown has not started yet. if v.used_at: expired = bool(expires_at and expires_at <= now) else: expired = False # Condition 1: quota/jatah reached. RouterOS `uptime` resets to 0 on every # logout, so a single read cannot represent total usage. We keep a # cumulative accumulator and add the delta of each poll, which # makes logout/login cycles consume from the SAME remaining quota # exactly like a real hotspot voucher. duration_exhausted = False if remote: session_seconds = parse_uptime_seconds(remote.get("uptime")) limit_seconds = parse_uptime_seconds(remote.get("limit_uptime")) or parse_uptime_seconds(v.limit_uptime) if session_seconds < (v.last_poll_uptime or 0): # Uptime went backwards => RouterOS started a new session. # The previous session's final reading was already banked # into used_seconds, so only start counting the new one. v.last_poll_uptime = 0 delta = session_seconds - (v.last_poll_uptime or 0) if delta > 0: v.used_seconds = int(v.used_seconds or 0) + delta v.last_poll_uptime = session_seconds changed = True if limit_seconds > 0 and int(v.used_seconds or 0) >= limit_seconds: duration_exhausted = True # If user is gone from a successfully synced device, they are definitely exhausted. gone_from_device = is_synced and remote is None and v.status == "used" # Live status sync from device. # A voucher counts as USED if it is online now OR it carries any # traffic accounting. RouterOS resets `uptime` to 0 on logout, so # relying on uptime alone made a logged-in-then-logged-out voucher # look untouched and the DataTables status never flipped. has_traffic = _routeros_bytes(remote and remote.get("bytes_in")) or _routeros_bytes( remote and remote.get("bytes_out") ) if remote: if remote.get("active") or parse_uptime_seconds(remote.get("uptime")) > 0 or has_traffic: if v.status in ("sold", "available"): v.status = "used" v.used_at = v.used_at or now # Validity starts at FIRST LOGIN. Prefer the precise # duration_seconds (minute/hour granularity chosen in the # profile form); duration_days is the fallback. validity = timedelta( seconds=int(batch.duration_seconds) ) if batch and batch.duration_seconds else timedelta( days=int(batch.duration_days) ) if batch and batch.duration_days else None if validity: v.expires_at = v.used_at + validity changed = True _log.info( "voucher %s marked used (live RouterOS activity)", username ) # NOTE: status "used" is deliberately sticky. RouterOS resets # `uptime` and `bytes-in/out` to 0 when a session ends, so a # revert rule here made vouchers flap used -> sold -> used # every cycle and never settled in the DataTables. if not (expired or duration_exhausted or gone_from_device): continue adapter = adapters.get(device_id) if device_id else None if adapter and username: try: await asyncio.to_thread(adapter.delete_hotspot_user, username) except Exception: _log.warning("voucher %s: RouterOS delete failed", username) # Archive before delete from app.models.models import VoucherArchive reason = ( "expired" if expired else ("duration_exhausted" if duration_exhausted else "gone_from_device") ) db.add( VoucherArchive( voucher_id=v.id, code=v.code, username=v.username, profile=v.profile, price=v.price, status=v.status, sold_at=v.sold_at, used_at=v.used_at, used_seconds=int(v.used_seconds or 0), deleted_at=now, reason=reason, batch_id=v.batch_id, agent_id=v.agent_id, ) ) _log.info("voucher %s removed (%s) from DB and device", username, reason) await db.delete(v) changed = True # NOTE: this commit MUST stay inside the `async with` block. Sitting # outside it meant the session was already closed (implicit rollback) # before commit ran, so archives and deletions were silently discarded. if changed: await db.commit() async def monitor_loop() -> None: """Keep billing status aligned with live OLT/MikroTik state and expire vouchers.""" while True: try: async with AsyncSessionLocal() as db: customers = (await db.execute( select(Customer).where( Customer.is_archived == False, Customer.pppoe_username.is_not(None), Customer.mikrotik_device_id.is_not(None), Customer.olt_device_id.is_not(None), ) )).scalars().all() for customer in customers: try: await refresh_customer_online_status(customer, db) except Exception: await db.rollback() except asyncio.CancelledError: raise except Exception: pass await asyncio.sleep(poll_interval()) async def voucher_monitor_loop() -> None: """Fast lane voucher reconciliation so UI reflects RouterOS state live.""" while True: try: await reconcile_vouchers() except asyncio.CancelledError: raise except Exception: _log.exception("voucher reconcile cycle failed") await asyncio.sleep(voucher_poll_interval()) def voucher_reconcile_health() -> dict: """Report whether live voucher reconciliation can actually reach devices.""" import asyncio as _asyncio from app.core.database import AsyncSessionLocal as _Session async def probe(): async with _Session() as db: devices = (await db.execute( select(MikrotikDevice).where(MikrotikDevice.is_active == True) )).scalars().all() report = [] for mk in devices: item = { "device": mk.name, "host": mk.host, "port": mk.api_port or 8728, "credential": None, "ok": False, "error": None, } try: adapters = await _device_adapters(db) adapter = adapters.get(mk.id) if adapter is None: item["error"] = "Kredensial tidak aktif / belum ditugaskan" else: rows = await _asyncio.to_thread(adapter.list_hotspot_users) item["credential"] = adapter.username item["ok"] = True item["users"] = len(rows) except Exception as exc: item["error"] = str(exc) report.append(item) return report try: devices = _asyncio.run(probe()) except Exception as exc: return {"ok": False, "error": str(exc), "devices": []} return { "ok": all(d["ok"] for d in devices) if devices else False, "devices": devices, }