hydraveil-gui/gui/v2/actions/menu_coordinator.py

# core
from core.controllers.ConfigurationController import ConfigurationController
# from core.services.prepare_tickets.ticket_tracker import get_a_random_ticket
from core.services.profiles.profile_connections import enable_profile, disable_profile
from core.controllers.ProfileController import ProfileController
from core.models.Configuration import Configuration, ConnectionChoice
from core.services.subscriptions.subscriptions_by_protocol import get_profile_subscription, start_new_payment_cycle, wipe_subscription, evaluate_repeat_cycle
from core.services.payment_phase.pre_existing import check_then_save_billing_code
from core.utils.check_if_time_lapsed import check_if_time_lapsed
from core.models.enum_models import PaymentCategory
from core.utils.type_checker import type_checker
from core.models.Result import Result, ResultError
from core.errors.logger import logger
from core.errors.translate_error import translate_error, get_ignored_exceptions
from core.services.payment_phase.payment_state import OverallPaymentCycle

# gui
# from gui.v2.actions.payment_actions import wipe_expired_subscription
from gui.v2.actions.ticket_actions import dispatch_ticketing
from gui.v2.infrastructure.action_registry import register_action
from gui.v2.infrastructure.take_plumbing_action import take_action
from gui.v2.infrastructure.screen_size import get_max_screensize
from gui.v2.infrastructure.setup_observers import (
    application_version_observer, connection_observer,
    profile_observer, ticket_observer,
)
from gui.v2.actions.error_handlers import navigate_to_new_sub_page

# generic
from typing import Callable, Optional
from functools import partial

@type_checker
def get_profile(profile_id: int):
    profile = ProfileController.get(int(profile_id))
    if not profile:
        return Result(valid=False, error_type=ResultError.INVALID_INPUT, message=f"There is no profile with the id {profile_id}")
    else:
        return Result(valid=True, data=profile)


def dispatch_individual(profile_id: int, ui_caller: Callable) -> Result:
    payment_cycle = OverallPaymentCycle()

    print(f"payment_cycle is {payment_cycle}")

    if payment_cycle.active:
        same_result = evaluate_repeat_cycle(payment_cycle=payment_cycle, profile_id=profile_id)
        if not same_result.valid:
            return same_result
        else:
            error_msg = "There's an error in the dispatch_individual function"
            logger.error(f"This shouldn't be going off.  {error_msg}")
            return Result(valid=False, error_type=ResultError.NAVIGATE_TO, navigate_to="menu", message=error_msg)
    else:
        # Starting fresh.  Pick a billing type:
        return navigate_to_new_sub_page()


def filter_subscription(profile_id: int, passed_ui_caller: Callable) -> Result:
    """Evaluate if this profile has a subscription that's ready to connect."""

    logger.info(f"Checking the subscription on profile id {profile_id}")
    subscription = get_profile_subscription(profile_id=profile_id)

    #-----------------------------
    # No Subscription
    #-----------------------------
    if not subscription or not subscription.billing_code:
        tickets_setup = ConfigurationController.are_tickets_setup()
        if tickets_setup:
            # already opted into ticketing:
            return dispatch_ticketing(profile_id=profile_id, passed_ui_caller=passed_ui_caller)
        else:
            return dispatch_individual(profile_id=profile_id, ui_caller=passed_ui_caller)

    #-----------------------------
    # Pending Subscription
    #-----------------------------
    if subscription.billing_pending:
        start_result = start_new_payment_cycle(category=PaymentCategory.INDIVIDUAL, profile_id=profile_id)
        if start_result.valid:
            logger.info("creating a new payment cycle was valid, it worked.")
            return Result(valid=False, error_type=ResultError.NAVIGATE_TO, navigate_to="payment_details", message="Please finish your payment.")
        else:
            logger.info(f"Returning the error of start_result {start_result}")
            return start_result

    #-----------------------------
    # Expired Subscription
    #-----------------------------
    expiration_date = subscription.expires_at
    if not expiration_date:
        # auto check tickets would be here.
        return navigate_to_new_sub_page()

    expired = check_if_time_lapsed(expiration_date)
    if expired:
        function_choice = partial(wipe_subscription, profile_id=profile_id)
        return Result(valid=False, function_choice=function_choice, error_type=ResultError.EXPIRED_SUB, message="This subscription is expired, would you like to wipe it and get a new one?")

    #-----------------------------
    # Proceed with Valid Sub
    #-----------------------------
    logger.info(f"Profile id {profile_id} has a valid subscription ready to go..")
    return Result(valid=True)


@register_action(ResultError.TOR_NEEDED)
def switch_to_tor_connection() -> Result:
    ConfigurationController.set_connection_enum(ConnectionChoice.TOR)
    return Result(valid=True)

@type_checker
def NOT_need_tor_warnings(profile_id: int) -> Result:
    """
    Purpose:
        Filters if we need to warn the user about a potential Tor leak.

    Context:
        If it's a Tor profile, and the user does NOT have Tor connection enabled, do we need to warn them?

    Returns:
        If we need it, it emits Result as False, with ResultError.TOR_NEEDED
        If not, returns Result as True, to move on.
    """

    # PROFILE
    profile_result = get_profile(profile_id)
    profile = profile_result.data
    
    # Actual Profile is a Tor Profile:
    if profile.connection.code != "tor":
        return Result(valid=True) # no warning needed, it's not even Tor

    # Config shows a non-Tor connection to download apps:
    config = Configuration.get()
    if config.connection != ConnectionChoice.TOR:
        return Result(valid=True) # no warning needed, it's already using Tor to download apps

    # Lacks Application:
    if profile.application_version:
        return Result(valid=True) # no warning needed, they have the app already.

    error_msg = "This app is not yet installed. Since it's a Tor profile, do you want to switch to Tor to download it? (Hit 'No' to proceed over clearweb)"
    return Result(valid=False, error_type=ResultError.TOR_NEEDED, message=error_msg)



@register_action(ResultError.ENDPOINT_VERIFICATION)
def literally_connect(
    profile_id: int
    ) -> Result:

    # Step 1: Get the Profile
    profile_result = get_profile(profile_id)
    if not profile_result:
        return Result(valid=False, error_type=ResultError.INVALID_INPUT, message="This profile does not match one of your configuration files.")

    profile = profile_result.data

    # EXCEPTIONS
    ignored = get_ignored_exceptions(ignore_state_conflict=True)

    # SCREEN SIZE:
    max_resolution = get_max_screensize()
    
    try:
        final_result = enable_profile(
            profile=profile,
            ignore=ignored,
            profile_observer=profile_observer,
            application_version_observer=application_version_observer,
            connection_observer=connection_observer,
            ticket_observer=ticket_observer,
            max_resolution=max_resolution
        )
        if final_result:
            return final_result
        else:
            return Result(valid=True)
    except Exception as error:
        return translate_error(error=error, profile_id=profile.id)



@register_action(ResultError.STATE_CONFLICT)
def ignore_state_conflict(profile_id: int) -> Result:
    return literally_connect(
        profile_id=profile_id
    )


def _connect_profile(profile_id: int, passed_ui_caller: Callable) -> Result:
    """Background Threaded Orchestration to Start a Connection"""

    # Step 1: Subscription filter:
    sub_result = filter_subscription(profile_id=profile_id, passed_ui_caller=passed_ui_caller)
    if not sub_result.valid:
        return sub_result

    # Step 2: Tor filter check:
    result = NOT_need_tor_warnings(profile_id=profile_id)
    if not result.valid:
        return result

    # Step 3: Actual Connection:
    logger.info(f"Profile id {profile_id} is ready to connect..")
    return literally_connect(profile_id=profile_id)


@type_checker
def connect_profile(profile_id: int, ui_caller: Callable) -> Result:
    """Type check before we thread"""
    take_action(
        action=_connect_profile,
        ui_caller=ui_caller,
        profile_id=profile_id,
        passed_ui_caller=ui_caller
    )
    return Result(valid=True)



def connect_with_billing_code(profile_id: int, billing_code: str) -> Result:
    """Apply a pre-existing billing code to a profile"""

    print(f"billing_code is {billing_code}")

    paid_result = check_then_save_billing_code(
        profile_id=profile_id,
        billing_code=billing_code,
        connection_observer=connection_observer
    )

    if not paid_result.valid:
        return paid_result

    return literally_connect(profile_id=profile_id)


@type_checker
def literally_disconnect(
    profile_id: int,
    ignore_state_conflict: Optional[bool] = False    
) -> Result:
    """Get a profile and use core to disable it."""

    # Step 1: Get the Profile
    profile_result = get_profile(profile_id)
    if not profile_result:
        return Result(valid=False, error_type=ResultError.INVALID_INPUT, message="This profile does not match one of your configuration files.")

    profile = profile_result.data

    # EXCEPTIONS
    ignored = get_ignored_exceptions(ignore_state_conflict=ignore_state_conflict)

    try:
        final_result = disable_profile(
            profile=profile,
            ignore=ignored,
            profile_observer=profile_observer,
            connection_observer=connection_observer,
            ticket_observer=ticket_observer
        )
        if final_result:
            return final_result
        else:
            return Result(valid=True)
    except Exception as error:
        return translate_error(error=error, profile_id=profile_id)


def disconnect_profile(profile_id: int, ui_caller: Callable) -> Result:
    """Background Thread to literally_disconnect"""
    take_action(
        action=literally_disconnect,
        ui_caller=ui_caller,
        profile_id=profile_id
    )
    return Result(valid=True)