NovaBryan/EVE_Dump
Log | Files | 📦 下载最新源码 (.zip) | Refs | README

upsell_neocom_controller.py (4182B)

      1 #Embedded file name: C:\BuildAgent\work\20d73f1b24eed06e\eve\release\V24.01\eve\client\script\ui\clonegrade\upsell_neocom_controller.py
      2 import logging
      3 import threadutils
      4 import uthread2
      5 from eve.client.script.ui.omega.upsell.neocomButton import OmegaUpsellFixedNeocomButtonExtension
      6 from omega.client.clone_grade_svc import get_clone_grade_svc
      7 from omega.client.const import LOG_ENTRY_NAME
      8 import omega.client.signal as omega_signals
      9 from stackless_response_router.exceptions import TimeoutException
     10 from omega.client.mock.omega_state import mocked_long_lapsing
     11 logger = logging.getLogger(__name__)
     12 
     13 class UpsellNeocomController(object):
     14 
     15     def __init__(self):
     16         self._neocom_button_extension = None
     17         self._update_thread = None
     18         omega_signals.on_omega_license_acquired.connect(self._on_omega_license_changed)
     19         omega_signals.on_omega_license_changed.connect(self._on_omega_license_changed)
     20         omega_signals.on_omega_license_expired.connect(self._on_omega_license_changed)
     21         omega_signals.on_omega_recurring_changed.connect(self._on_omega_license_changed)
     22         mocked_long_lapsing.on_change.connect(self._on_mocked_omega_lapsing_changed)
     23 
     24     def __del__(self):
     25         omega_signals.on_omega_license_acquired.disconnect(self._on_omega_license_changed)
     26         omega_signals.on_omega_license_changed.disconnect(self._on_omega_license_changed)
     27         omega_signals.on_omega_license_expired.disconnect(self._on_omega_license_changed)
     28         omega_signals.on_omega_recurring_changed.disconnect(self._on_omega_license_changed)
     29         mocked_long_lapsing.on_change.connect(self._on_mocked_omega_lapsing_changed)
     30         self._kill_update_thread()
     31 
     32     def _kill_update_thread(self):
     33         if self._update_thread is not None:
     34             self._update_thread.kill()
     35             self._update_thread = None
     36 
     37     def update_neocom_button(self):
     38         self._kill_update_thread()
     39         self._update_thread = uthread2.StartTasklet(self._update_neocom_button)
     40 
     41     def _update_neocom_button(self):
     42         log_info = '%s - Updating NEOCOM button: ' % LOG_ENTRY_NAME
     43         try:
     44             is_omega = get_clone_grade_svc().is_omega_without_grace_period()
     45             log_info += 'user is Omega, ' if is_omega else 'user is Alpha, '
     46             is_about_to_lapse = get_clone_grade_svc().is_omega_about_to_lapse()
     47             log_info += 'about to lapse, ' if is_about_to_lapse else 'not about to lapse, '
     48             if not is_omega or is_about_to_lapse:
     49                 log_info += 'so we need to show the button in the NEOCOM.'
     50                 self._register_neocom_button_extension()
     51             else:
     52                 log_info += 'so we need to hide the button in the NEOCOM.'
     53                 self._unregister_neocom_button_extension()
     54         except TimeoutException:
     55             log_info += "Could not get Omega status - timed out. If not using mocks, this isn't expected to work in dev tier."
     56         finally:
     57             logger.info(log_info)
     58 
     59     @threadutils.threaded
     60     def _register_neocom_button_extension(self):
     61         if self._neocom_button_extension is not None:
     62             if not self._unregister_neocom_button_extension():
     63                 return
     64         logger.info('%s - Registering NEOCOM button extension' % LOG_ENTRY_NAME)
     65         self._neocom_button_extension = OmegaUpsellFixedNeocomButtonExtension()
     66         sm.GetService('neocom').RegisterFixedButtonExtension(self._neocom_button_extension)
     67 
     68     def _unregister_neocom_button_extension(self):
     69         if self._neocom_button_extension is not None:
     70             try:
     71                 logger.info('%s - Unregistering NEOCOM button extension' % LOG_ENTRY_NAME)
     72                 sm.GetService('neocom').UnregisterFixedButtonExtension(self._neocom_button_extension)
     73                 return True
     74             except ValueError:
     75                 return False
     76             finally:
     77                 self._neocom_button_extension = None
     78 
     79     def _on_omega_license_changed(self, *args):
     80         self._kill_update_thread()
     81         self._update_neocom_button()
     82 
     83     def _on_mocked_omega_lapsing_changed(self, *args):
     84         self._kill_update_thread()
     85         self._update_neocom_button()