forked from Support/sp-hydra-veil-gui
128 lines
4.3 KiB
Python
128 lines
4.3 KiB
Python
# core
|
|
from core.services.subscriptions.check_payment_loop import CheckPaidTimer
|
|
from core.services.subscriptions.subscriptions_by_protocol import create_invoice, save_billing_code, get_profile_subscription
|
|
from core.models.enum_models import PaymentType
|
|
from core.models.Result import Result, ResultError
|
|
from core.errors.logger import logger
|
|
|
|
# gui
|
|
from gui.v2.infrastructure.ThreadSafetyTool import ThreadSafetyTool
|
|
from gui.v2.infrastructure.take_plumbing_action import take_action
|
|
from gui.v2.infrastructure.setup_observers import connection_observer
|
|
|
|
# generic
|
|
from typing import Callable, Optional
|
|
|
|
def gui_starts_invoice(profile_id: int, invoice_ui_caller: Callable, timer_resume_caller: Callable) -> str | None:
|
|
"""Create a new invoice if we don't have one yet, or re-use the saved data. Returns any errors as strings for UI to display."""
|
|
|
|
subscription = get_profile_subscription(profile_id)
|
|
|
|
#----------------------------
|
|
# INVOICE DATA
|
|
#----------------------------
|
|
take_action(
|
|
action=create_invoice,
|
|
ui_caller=invoice_ui_caller,
|
|
profile_id=profile_id,
|
|
connection_observer=connection_observer
|
|
)
|
|
|
|
#----------------------------
|
|
# FILTER
|
|
#----------------------------
|
|
if not subscription:
|
|
logger.info(f"Started a New Invoice for Profile {profile_id}..")
|
|
return None
|
|
|
|
if not subscription.billing_pending:
|
|
logger.error("This function should not have been called if the invoice is already paid.")
|
|
return "Invoice already paid."
|
|
|
|
if not subscription.billing_code:
|
|
return "Previous subscription didn't have a billing code."
|
|
|
|
#----------------------------
|
|
# CHECK PAID ON PRE-EXISTING INVOICE
|
|
#----------------------------
|
|
logger.info("Resuming a pre-existing payment..")
|
|
|
|
# Setup PyQt6 GUI as mainthread to bring back to:
|
|
pyqt_threader = ThreadSafetyTool()
|
|
|
|
# setup the timer with the callback:
|
|
# it doesn't return result here. It calls the view_callback function.
|
|
check_paid_timer = CheckPaidTimer(
|
|
view_callback=timer_resume_caller,
|
|
billing_code=subscription.billing_code,
|
|
ui_dispatcher=pyqt_threader,
|
|
payment_type=PaymentType.CRYPTO,
|
|
connection_observer=connection_observer,
|
|
profile_id=profile_id
|
|
)
|
|
return None
|
|
|
|
|
|
|
|
def save_and_begin_timer(
|
|
ticketing: bool,
|
|
gui_callback: Callable,
|
|
billing_code: str,
|
|
profile_id: Optional[int] = None,
|
|
expires_at: Optional[str] = None,
|
|
) -> bool:
|
|
"""
|
|
First saves the billing code & profile expiration, if it's an individual (non-ticket) crypto-based one.
|
|
Then it Starts a loop of checking if the invoice is paid and saves it.
|
|
"""
|
|
|
|
if ticketing:
|
|
payment_type = PaymentType.TICKETS
|
|
logger.error("Not yet setup for ticketing")
|
|
return False
|
|
|
|
# NON-ticketing:
|
|
elif profile_id and expires_at:
|
|
payment_type = PaymentType.CRYPTO
|
|
saved = save_billing_code(profile_id=profile_id, billing_code=billing_code, expires_at=expires_at)
|
|
if not saved.valid:
|
|
return False
|
|
else:
|
|
logger.error("You need a profile id and expires_at for individual profiles")
|
|
return False
|
|
|
|
# Setup PyQt6 GUI as mainthread to bring back to:
|
|
pyqt_threader = ThreadSafetyTool()
|
|
|
|
# setup the timer with the callback:
|
|
# it doesn't return result here. It calls the gui_callback function.
|
|
check_paid_timer = CheckPaidTimer(
|
|
view_callback=gui_callback,
|
|
billing_code=billing_code,
|
|
ui_dispatcher=pyqt_threader,
|
|
payment_type=payment_type,
|
|
connection_observer=connection_observer,
|
|
profile_id=profile_id
|
|
)
|
|
logger.info("starting polling from GUI action..")
|
|
check_paid_timer.start_polling()
|
|
return True # this is just that it started
|
|
|
|
|
|
def stop_payment_timer() -> tuple:
|
|
"""Gets the single instance of the payment checker timer, and stops it."""
|
|
try:
|
|
check_paid_timer = CheckPaidTimer.get_instance()
|
|
if check_paid_timer:
|
|
check_paid_timer.stop_polling()
|
|
|
|
content_to_display = "Stopped! You can resume a different profile's invoice now"
|
|
return True, content_to_display
|
|
|
|
except Exception as e:
|
|
error_msg = f"Unable to stop the payment timer: {str(e)}"
|
|
logger.error(error_msg)
|
|
return False, error_msg
|
|
|
|
|
|
|