187 lines
7 KiB
Python
187 lines
7 KiB
Python
from __future__ import annotations
|
|
from typing import TYPE_CHECKING
|
|
|
|
if TYPE_CHECKING:
|
|
from core.observers.TicketObserver import TicketObserver
|
|
from core.essentials.observers.ConnectionObserver import ConnectionObserver
|
|
|
|
from core.Constants import Constants
|
|
from core.observers.BaseObserver import BaseObserver
|
|
from core.services.using_tickets.use_ticket_orchestrator import use_ticket_orchestrator
|
|
from core.services.prepare_tickets.ticket_tracker import (
|
|
get_data_for_a_single_ticket,
|
|
does_ticket_tracker_exist,
|
|
)
|
|
from core.services.helpers.does_ticket_file_exist import does_ticket_file_exist
|
|
from core.services.helpers.get_value_from_config import get_value_from_config
|
|
from core.utils.basic_operations.does_file_exist import does_file_exist
|
|
from core.utils.basic_operations.write_or_read_from_json import update_json
|
|
from core.services.prepare_tickets.ticket_tracker import get_all_unused_tickets
|
|
import random
|
|
|
|
|
|
def modify_random_tickets_setting(
|
|
on_or_off: str,
|
|
ticket_observer: TicketObserver,
|
|
) -> dict:
|
|
|
|
choices = ["on", "off"]
|
|
if on_or_off not in choices:
|
|
ticket_observer.notify("failed_input", None)
|
|
return {"valid": False, "message": f"Invalid choice for turning on or off"}
|
|
|
|
billing_folder = Constants.HV_TICKETING_CONFIG_HOME
|
|
filepath = f"{billing_folder}/billing_choices.json"
|
|
does_billing_config_exist = does_file_exist(filepath)
|
|
|
|
if does_billing_config_exist == False:
|
|
notification = "First setup the Tickets before picking use"
|
|
ticket_observer.notify("failed_input", subject=notification)
|
|
return {"valid": False, "message": notification}
|
|
else:
|
|
try:
|
|
if on_or_off == "on":
|
|
update_json(filepath, "use_random", True)
|
|
return {"valid": True}
|
|
elif on_or_off == "off":
|
|
update_json(filepath, "use_random", False)
|
|
return {"valid": True}
|
|
else:
|
|
ticket_observer.notify("failed_input", None)
|
|
return {
|
|
"valid": False,
|
|
"message": f"Invalid choice for turning on or off",
|
|
}
|
|
except:
|
|
notification = f"Error with modifying config file. Check {filepath}"
|
|
ticket_observer.notify("error", subject=notification)
|
|
return {"valid": False, "message": notification}
|
|
|
|
|
|
"""
|
|
# this has 3 possible results:
|
|
1) getting a random ticket number for 'which_ticket'
|
|
2) the user has the config set to manual, so 'which_ticket' is 'None' and error_msg to None
|
|
3) an error message on reading the config file (variable 'error_msg' is not 'none')
|
|
"""
|
|
|
|
|
|
def do_we_use_a_random_ticket(ticket_observer: TicketObserver) -> tuple:
|
|
# check if they have it on random mode:
|
|
config_data = get_value_from_config("use_random")
|
|
|
|
# if the 'value' key is in the config, that means it successfully read the config.
|
|
if "value" in config_data:
|
|
random_setting = config_data["value"]
|
|
|
|
# they want a random ticket
|
|
if random_setting == True:
|
|
data_results = pick_a_random_ticket(ticket_observer)
|
|
|
|
if "random_ticket" in data_results:
|
|
which_ticket = data_results["random_ticket"]
|
|
error_msg = None
|
|
return which_ticket, error_msg
|
|
else:
|
|
which_ticket = "error"
|
|
if "message" in data_results:
|
|
error_msg = data_results["message"]
|
|
else:
|
|
error_msg = (
|
|
"Missing or Invalid Data. Unable to get unused ticket list."
|
|
)
|
|
return which_ticket, error_msg
|
|
|
|
# if it read the config, but the value is false:
|
|
else:
|
|
which_ticket = None
|
|
error_msg = None
|
|
return which_ticket, error_msg
|
|
|
|
# this is a problem with reading the config itself:
|
|
else:
|
|
which_ticket = "error"
|
|
error_msg = "There is an error with the config file, or no config. Are you sure you have tickets?"
|
|
ticket_observer.notify("failed_input", subject=error_msg)
|
|
return which_ticket, error_msg
|
|
|
|
|
|
def get_unused_tickets(ticket_observer: TicketObserver) -> dict:
|
|
# does the file keeping track of ALL tickets exist:
|
|
does_the_file_exist, path_of_file = does_ticket_tracker_exist()
|
|
if does_the_file_exist == False:
|
|
error_msg = f"The ticket tracker organizer file does not exist. Check the folder {path_of_file}"
|
|
ticket_observer.notify("failed_input", subject=error_msg)
|
|
return {"valid": False, "message": error_msg}
|
|
|
|
# Use the Model:
|
|
unused_tickets = get_all_unused_tickets()
|
|
|
|
return unused_tickets
|
|
|
|
|
|
"""
|
|
use_ticket function requires:
|
|
|
|
-Knowing which ticket is being used.
|
|
-Knowing which location it's being assigned to.
|
|
-The ticket file completed.
|
|
-Prior registration on the server.
|
|
"""
|
|
|
|
|
|
def use_ticket(
|
|
which_ticket: int,
|
|
which_location: str,
|
|
ticket_observer: TicketObserver,
|
|
connection_observer: ConnectionObserver,
|
|
) -> dict:
|
|
|
|
which_ticket = str(which_ticket) # type: ignore
|
|
|
|
# does the ticket's file exist:
|
|
ticket_exists = does_ticket_file_exist(which_ticket)
|
|
if ticket_exists == False:
|
|
error_msg = f"The ticket file does not exist in the correct folder."
|
|
ticket_observer.notify("failed_input", subject=error_msg)
|
|
return {"valid": False, "message": error_msg}
|
|
|
|
# does the file keeping track of ALL tickets exist:
|
|
does_the_file_exist, path_of_file = does_ticket_tracker_exist()
|
|
if does_the_file_exist == False:
|
|
error_msg = f"The ticket tracker organizer file does not exist. Check the folder {path_of_file}"
|
|
ticket_observer.notify("failed_input", subject=error_msg)
|
|
return {"valid": False, "message": error_msg}
|
|
|
|
# is the ticket used?
|
|
try:
|
|
status, location, subscription = get_data_for_a_single_ticket(which_ticket)
|
|
if status == "used":
|
|
error_msg = f"Ticket is already tied to {location} with the subscription {subscription}"
|
|
ticket_observer.notify("failed_input", subject=error_msg)
|
|
return {"valid": False, "message": error_msg}
|
|
except:
|
|
error_msg = f"Your local ticket tracker has no value for ticket {which_ticket}"
|
|
return {"valid": False, "message": error_msg}
|
|
|
|
# the actual work here, everything else is just handling:
|
|
ticket_observer.notify("connecting", "Connecting..")
|
|
reply = use_ticket_orchestrator(which_ticket, which_location, connection_observer)
|
|
|
|
return reply
|
|
|
|
|
|
def pick_a_random_ticket(ticket_observer: TicketObserver) -> dict:
|
|
ticket_data = get_unused_tickets(ticket_observer)
|
|
|
|
if "valid" in ticket_data:
|
|
if ticket_data["valid"] == True:
|
|
list_of_unused_tickets = ticket_data["data"]
|
|
|
|
random_ticket = random.choice(list_of_unused_tickets)
|
|
return {"valid": True, "random_ticket": random_ticket}
|
|
else:
|
|
return ticket_data
|
|
else:
|
|
error_msg = "Missing or Invalid Data. Unable to get unused ticket list."
|
|
return {"valid": False, "message": error_msg}
|