"""Hizmet bakiyelerini takip ederek kritik seviyede güvenli kapanış başlatan asenkron modül.""" from __future__ import annotations import asyncio import logging import re import threading from typing import Any, Dict, Optional from anymessage import fetch_balance as anymessage_fetch_balance from captcha_service import get_balance as captcha_fetch_balance from castle_solver import CastleSolverError, fetch_balance as castle_fetch_balance from disposablemail import DisposableMailAPI, DisposableMailAPIError from xgen import ( is_global_shutdown_requested, request_global_shutdown, ) LOGGER = logging.getLogger(__name__) DEFAULT_BALANCE_THRESHOLD = 0.01 _DEFAULT_INTERVAL_SECONDS = 60.0 _DEFAULT_INITIAL_DELAY_SECONDS = 3.0 def _is_feature_enabled(value: Any) -> bool: if isinstance(value, bool): return value if isinstance(value, (int, float)): return value != 0 if isinstance(value, str): lowered = value.strip().lower() return lowered in {"1", "true", "yes", "on"} return False def _coerce_float(value: Any) -> Optional[float]: """Girdi değerini kayan noktalı sayıya dönüştürmeye çalış.""" if value is None: return None if isinstance(value, (int, float)): return float(value) if isinstance(value, str): cleaned = value.strip() if not cleaned: return None match = re.search(r"[-+]?\d+(?:[.,]\d+)?", cleaned) if not match: return None number_text = match.group(0).replace(",", ".") try: return float(number_text) except ValueError: return None return None class BalanceMonitor: """Code solver ve mail servislerinin bakiyelerini düzenli aralıklarla denetler.""" def __init__( self, config: Dict[str, Any], *, interval: float = _DEFAULT_INTERVAL_SECONDS, threshold: float = DEFAULT_BALANCE_THRESHOLD, ) -> None: self._interval = max(float(interval), 1.0) self._threshold = float(threshold) initial_delay = config.get("balance_initial_delay_seconds", _DEFAULT_INITIAL_DELAY_SECONDS) try: self._initial_delay = max(float(initial_delay), 0.0) except (TypeError, ValueError): self._initial_delay = _DEFAULT_INITIAL_DELAY_SECONDS self._code_solver_api_key = str(config.get("code_solver_api_key") or "").strip() self._mail_service_name = str(config.get("mail_service") or "").strip().lower() self._mail_api: Optional[DisposableMailAPI] = None if self._mail_service_name == "disposablemail": try: self._mail_api = DisposableMailAPI(config) except DisposableMailAPIError: LOGGER.exception( "DisposableMail client could not be created; balance monitoring will be disabled." ) self._mail_service_name = "" castle_config = config.get("castlesolver.cc") self._castle_solver_api_key: str = "" self._castle_balance_timeout = 30 if isinstance(castle_config, dict) and _is_feature_enabled( castle_config.get("enabled", True) ): api_key_value = str(castle_config.get("api_key") or "").strip() if api_key_value: self._castle_solver_api_key = api_key_value timeout_value = castle_config.get("balance_timeout") or castle_config.get( "timeout" ) try: self._castle_balance_timeout = int(timeout_value or 30) except (TypeError, ValueError): self._castle_balance_timeout = 30 self._loop: Optional[asyncio.AbstractEventLoop] = None self._thread: Optional[threading.Thread] = None self._stop_event: Optional[asyncio.Event] = None self._task: Optional[asyncio.Task[None]] = None # --- Genel yardımcılar --- @staticmethod def parse_code_solver_balance(response: Dict[str, Any]) -> Optional[float]: """Code solver yanıtından bakiye bilgisini çıkar.""" for key in ("balanceUsd", "balanceUSD", "BalanceUsd", "BalanceUSD"): if key in response: value = _coerce_float(response.get(key)) if value is not None: return value return None @staticmethod def parse_castle_solver_balance(response: Dict[str, Any]) -> Optional[float]: """CastleSolver yanıtından bakiye bilgisini çıkar.""" for key in ("balance", "Balance", "balanceUsd", "balanceUSD", "BalanceUsd", "BalanceUSD"): if key in response: value = _coerce_float(response.get(key)) if value is not None: return value return None @staticmethod def parse_mail_balance(service_name: str, response: Dict[str, Any]) -> Optional[float]: """Mail servisi yanıtından kredi bilgisini çıkar.""" lowered = service_name.strip().lower() if not lowered: return None if lowered == "disposablemail": value = response.get("data") return _coerce_float(value) value = response.get("balance") or response.get("Balance") return _coerce_float(value) # --- Yaşam döngüsü --- def start(self) -> None: """Arka planda bakiye denetimini başlat.""" if self._thread and self._thread.is_alive(): return if is_global_shutdown_requested(): LOGGER.info("Global shutdown already requested; balance monitor will not start.") return self._loop = asyncio.new_event_loop() self._stop_event = asyncio.Event() self._task = self._loop.create_task(self._run()) def _run_loop(loop: asyncio.AbstractEventLoop) -> None: asyncio.set_event_loop(loop) try: loop.run_until_complete(self._task) except asyncio.CancelledError: pass # Görev iptali beklenen bir durumdur except Exception: # pragma: no cover - beklenmeyen istisnalar için LOGGER.exception("Balance monitor terminated unexpectedly.") finally: tasks = asyncio.all_tasks(loop) for task in tasks: task.cancel() loop.run_until_complete(loop.shutdown_asyncgens()) loop.close() self._thread = threading.Thread( target=_run_loop, name="BalanceMonitor", args=(self._loop,), daemon=True, ) self._thread.start() LOGGER.info( "Balance monitor started; checking every %.0f seconds (threshold=%.3f).", self._interval, self._threshold, ) def stop(self) -> None: """Arka plandaki bakiye denetimini durdur.""" if not self._loop: return if self._stop_event and not self._stop_event.is_set(): self._loop.call_soon_threadsafe(self._stop_event.set) if self._task: self._loop.call_soon_threadsafe(self._task.cancel) if self._thread: self._thread.join(timeout=self._interval + 5) self._loop = None self._task = None self._thread = None self._stop_event = None # --- İç akış --- async def _run(self) -> None: """Bakiye denetim döngüsü.""" if not self._stop_event: return first_run = True while not self._stop_event.is_set(): if first_run: first_run = False if self._initial_delay > 0: try: await asyncio.wait_for( self._stop_event.wait(), timeout=self._initial_delay, ) break except asyncio.TimeoutError: pass await asyncio.to_thread(self._check_balances_once) if is_global_shutdown_requested(): break try: await asyncio.wait_for(self._stop_event.wait(), timeout=self._interval) except asyncio.TimeoutError: continue def _check_balances_once(self) -> None: """Code solver ve mail servislerini tek seferde kontrol et.""" self._check_code_solver_balance() self._check_mail_service_balance() def _check_code_solver_balance(self) -> None: if not self._castle_solver_api_key and not self._code_solver_api_key: LOGGER.warning( "No configuration found for code solver balance; skipping balance check." ) return castle_response: Optional[Dict[str, Any]] = None if self._castle_solver_api_key: try: castle_response = castle_fetch_balance( self._castle_solver_api_key, timeout=self._castle_balance_timeout, ) LOGGER.info("Periodic CastleSolver balance response: %s", castle_response) except CastleSolverError as exc: LOGGER.warning("Failed to query CastleSolver balance: %s", exc) except Exception: LOGGER.exception("Unexpected error while querying CastleSolver balance.") if isinstance(castle_response, dict): castle_balance = self.parse_castle_solver_balance(castle_response) if castle_balance is not None: LOGGER.info("CastleSolver balance updated: %.4f USD", castle_balance) if castle_balance < self._threshold: message = ( f"CastleSolver balance is critical ({castle_balance:.4f} USD). " "Application will shut down safely." ) LOGGER.error(message) request_global_shutdown(message) else: LOGGER.warning( "Could not read valid value from CastleSolver balance response: %s", castle_response, ) if not self._code_solver_api_key: return try: response = captcha_fetch_balance(self._code_solver_api_key) except Exception: LOGGER.exception("Error while querying code solver balance.") return LOGGER.info("Periodic code solver balance response: %s", response) balance = self.parse_code_solver_balance(response) if isinstance(response, dict) else None if balance is None: LOGGER.warning( "Could not read valid value from code solver balance response: %s", response, ) return LOGGER.info("Code solver balance updated: %.4f USD", balance) if balance < self._threshold: message = ( f"Code solver balance is critical ({balance:.4f} USD). " "Application will shut down safely." ) LOGGER.error(message) request_global_shutdown(message) def _check_mail_service_balance(self) -> None: if not self._mail_service_name or self._mail_service_name == "manual": return response: Optional[Dict[str, Any]] = None try: if self._mail_service_name == "disposablemail" and self._mail_api: response = self._mail_api.get_balance() else: response = anymessage_fetch_balance() except DisposableMailAPIError: LOGGER.exception("Error while querying DisposableMail balance.") return except Exception: LOGGER.exception("Error while querying mail service balance.") return LOGGER.info( "Periodic mail service (%s) balance response: %s", self._mail_service_name, response, ) balance = ( self.parse_mail_balance(self._mail_service_name, response) if isinstance(response, dict) else None ) if balance is None: LOGGER.warning( "Could not read valid balance from mail service (%s) response: %s", self._mail_service_name, response, ) return LOGGER.info( "Mail service (%s) balance updated: %.4f credits", self._mail_service_name, balance, ) if balance < self._threshold: message = ( f"Mail service ({self._mail_service_name}) balance is critical " f"({balance:.4f}). Application will shut down safely." ) LOGGER.error(message) request_global_shutdown(message)