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
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

# gui
from gui.v2.actions.ticket_actions import get_ticket_data, use_a_ticket
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,
)

# generic
from typing import Callable, Optional


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)


@register_action(ResultError.SUBSCRIPTION)
def navigate_to_new_sub_page() -> Result:
    return Result(valid=False, error_type=ResultError.NAVIGATE_TO, navigate_to="id")


@register_action(ResultError.EXPIRED_SUB)
def wipe_expired_subscription(profile_id: int) -> Result:
    """Call core to wipe a subscription, then navigate to the billing page"""
    wiped = wipe_subscription(profile_id)
    if wiped.valid:
        return navigate_to_new_sub_page()
    else:
        return wiped


@register_action(ResultError.NO_TICKET)
def start_and_navigate_to_ticketing():
    """Starts a new ticketing cycle. (assumes the user opted in)."""
    start_result = start_new_payment_cycle(category=PaymentCategory.TICKET)
    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="plan_picker")
    else:
        logger.info(f"Returning the error of start_result {start_result}")
        return start_result



def random_ticket_dispatch(profile_id: int, passed_ui_caller: Callable) -> Result:
    """
    User explicitly wants a random ticket.
    The reason this is in GUI repo, is because use_a_ticket is.
    """

    random_ticket_result = get_a_random_ticket()

    if not random_ticket_result.valid:
        return random_ticket_result

    which_ticket = random_ticket_result.data

    # this is a redundant sanity check:
    if not which_ticket:
        Result(valid=False, error_type=ResultError.NO_TICKET, message="No tickets left.  Would you like to get some?")

    # calls GUI repo function:
    return use_a_ticket(profile_id=profile_id, which_ticket=which_ticket, ui_caller=passed_ui_caller)


def dispatch_ticketing(profile_id: int, passed_ui_caller: Callable) -> Result:
    """Evaluate if a random ticket will be used."""

    use_random_setting = ConfigurationController.use_random()

    if use_random_setting:
        return random_ticket_dispatch(profile_id=profile_id, passed_ui_caller=passed_ui_caller)

    # User does NOT want an automated selection, so they can pick from their tickets:
    do_they_have_tickets = get_ticket_data()

    if do_they_have_tickets.valid:
        return Result(valid=False, error_type=ResultError.NAVIGATE_TO, navigate_to="ticket_or_billing_choice")

    else:
        # user can decide if they want to setup tickets or do an individual profile:
        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:
            # Starting fresh.  Pick a billing type:
            return navigate_to_new_sub_page()

    #-----------------------------
    # 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:
        return Result(valid=False, 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)


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,
    ignore_state_conflict: Optional[bool] = False
    ) -> 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=ignore_state_conflict)

    # 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,
        ignore_state_conflict=True
    )


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)