Functions added for wiping tickets when expired (the entire ticket cycle). Also general error fixing for the payments flow

core/errors/translate_error.py

@@ -65,24 +65,29 @@ def translate_error(error: Type[Exception], profile_id: int) -> Result:
         return Result(valid=False, error_type=ResultError.MISSING_SOFTWARE, message=error_msg)
 
     else:
-        # Extract the traceback from the exception object
+        error_msg = get_traceback_line(error)
-        tb = error.__traceback__
-        
-        # extract_tb returns a list of frames; the LAST one [-1] is where the error happened
-        summary = traceback.extract_tb(tb)
-        if summary:
-            last_frame = summary[-1]
-            func_name = last_frame.name       # The function name
-            line_no = last_frame.lineno      # The line number
-            file_name = last_frame.filename  # The file path
-        else:
-            func_name, line_no, file_name = "Unknown", "Unknown", "Unknown"
-
-        error_msg = f"Error in {func_name} at {file_name}:{line_no} -> {str(error)}"
-        logger.error(error_msg)
         return Result(valid=False, error_type=ResultError.UNKNOWN, message=error_msg)
 
 
+def get_traceback_line(error) -> str:
+    # Extract the traceback from the exception object
+    tb = error.__traceback__
+    
+    # extract_tb returns a list of frames; the LAST one [-1] is where the error happened
+    summary = traceback.extract_tb(tb)
+    if summary:
+        last_frame = summary[-1]
+        func_name = last_frame.name       # The function name
+        line_no = last_frame.lineno      # The line number
+        file_name = last_frame.filename  # The file path
+    else:
+        func_name, line_no, file_name = "Unknown", "Unknown", "Unknown"
+
+    error_msg = f"Error in {func_name} at {file_name}:{line_no} -> {str(error)}"
+    logger.error(error_msg)
+    return error_msg
+
+
 @type_checker
 def get_ignored_exceptions(ignore_state_conflict: Optional[bool] = False):
     ignored_as_list = []

core/infrastructure/plumbing.py

@@ -1,7 +1,7 @@
 from core.infrastructure.thread_dispatch import MainThreadDispatcher
 from core.infrastructure.error_popup import PopupProtocol
 from core.infrastructure.service_registry import ServiceLocator
-
+from core.errors.translate_error import get_traceback_line
 from core.models.Result import Result, ResultError
 # from core.infrastructure.action_registry import ENUM_MAP
 from core.errors.logger import logger
@@ -66,8 +66,9 @@ class ThreadPlumbing:
 
         except Exception as e:
             error_msg = f"There was an error: {str(e)}"
-            # notify(error_msg)
             logger.error(error_msg)
+            stack_trace = get_traceback_line(e)
+            logger.error(stack_trace)
             result = Result(valid=False, error_type=ResultError.UNKNOWN, message=str(e))
 
         finally:
@@ -117,7 +118,7 @@ class ThreadPlumbing:
         # otherwise return it as is:
         else:
             if self.ui_caller:
-                self.ui_caller(passed_result)
+                self.ui_caller(result)
 
 
     def _on_popup_answer(self, user_choice: bool, passed_result: Result):

core/models/Result.py

@@ -19,7 +19,7 @@ class ResultError(Enum):
     DATABASE = "database"
     PERMISSION = "permission"
     SUBSCRIPTION = "subscription"
-    EXPIRED = "expired" # meant for invoice
+    EXPIRED_INVOICE = "expired_invoice" # meant for invoice
     UNPAID = "unpaid"
     EXPIRED_SUB = "expired_sub"
     ACTIVE_BILLING = "active_billing"
@@ -29,6 +29,7 @@ class ResultError(Enum):
     NO_TICKET = "no_ticket"
     RANDOM_TICKETS_OFF = "random_tickets_off"
     TICKET_SETUP = "ticket_setup"
+    TICKET_EXPIRED = "ticket_expired"
     ALREADY_EXISTS = "already_exists"
     BILLING_CODE_EXISTS = "billing_code_exists"
     PROCESS_GOT_KILLED = "process_got_killed"

core/models/manage/server_reply_tools.py

@@ -23,7 +23,7 @@ TRANSLATION_DICT = {
     ServerError.MISSING_DATA: ResultError.INVALID_INPUT,
     ServerError.UNPAID: ResultError.SUBSCRIPTION,
     ServerError.SERVER_SIDE: ResultError.SERVER_SIDE,
-    ServerError.EXPIRED: ResultError.SUBSCRIPTION
+    ServerError.EXPIRED: ResultError.EXPIRED_INVOICE
 }
 
 def format_server_response(api_response: ApiResponse|dict, desired_type: str) -> Result:

core/services/payment_phase/payment_state.py

@@ -130,14 +130,14 @@ class StatePersistenceMixin(ValidatedDataclass):
                 setattr(self, field.name, value)
 
 
-        def wipe_data_on_disk(self):
+    def wipe_data_on_disk(self):
-            logger.info("Starting wipe of pre-existing data")
+        logger.info("Starting wipe of pre-existing data")
 
-            if not os.path.exists(self._data_file):
+        if not os.path.exists(self._data_file):
-                logger.info("File does NOT exist already")
+            logger.info("File does NOT exist already")
-                return
+            return
 
-            return delete_file(self._data_file)
+        return delete_file(self._data_file)
 
 
     def exists_already(self) -> bool:
@@ -215,6 +215,16 @@ class PaymentData(StatePersistenceMixin):
         
         self.load()            
 
+    def wipe(self):
+        """Wipe all data and start over."""
+        wiped = self.wipe_data_on_disk()
+        self.currency = None
+        self.payment_type = None
+        self.plan_id = None
+        self.location_id = None
+        self.protocol_choice = None
+
+
 @singleton
 @dataclass
 class OverallPaymentCycle():

core/services/payment_phase/ticket_cycle.py

@@ -0,0 +1,42 @@
+from core.services.subscriptions.subscriptions_by_protocol import wipe_billing_cycle
+from core.services.payment_phase.payment_state import OverallPaymentCycle, TicketPaymentData
+from core.controllers.ConfigurationController import ConfigurationController
+from core.services.prepare_tickets.ticket_tracker import ticket_tracker_exists, delete_ticket_data
+from core.utils.basic_operations.file_tools import delete_file
+from core.Constants import Constants
+from core.models.Result import Result, ResultError
+from core.errors.logger import logger
+
+def wipe_ticket_cycle() -> Result:
+    """Wipe the entire global ticketing cycle & all associated tickets."""
+
+    # global temp billing in memory:
+    ticket_config = TicketPaymentData()
+    ticket_config.wipe()
+
+    # wipe both file and in memory cycle:
+    wiped_whole_cycle = wipe_billing_cycle()
+
+    # wipe tickets tracker:
+    if ticket_tracker_exists():
+        deleted_tracker = delete_file(Constants.TICKET_TRACKER_PATH)
+    else:
+        deleted_tracker = True
+
+    # wipe tickets themselves:
+    wiped_literal_tickets = delete_ticket_data()
+
+    # wipe global "use" config:
+    config_wiped = ConfigurationController.tickets_setup(False)
+    random_stopped = ConfigurationController.change_random_ticket(False)
+
+    if (
+        wiped_whole_cycle.valid and
+        config_wiped and
+        random_stopped and
+        deleted_tracker and
+        wiped_literal_tickets
+    ):
+        return Result(valid=True)
+    else:
+        return Result(valid=False, error_type=ResultError.FILE_SYSTEM, message="Unable to wipe ticket billing cycle.")

core/services/subscriptions/subscriptions_by_protocol.py

@@ -1,6 +1,7 @@
 from core.services.payment_phase.check_payment_loop import CheckPaidTimer
 from core.models.manage.server_reply_tools import api_call_and_extraction, ApiCall
 from core.controllers.ProfileController import ProfileController
+from core.controllers.ConfigurationController import ConfigurationController
 from core.models.BaseProfile import BaseProfile as Profile
 from core.services.payment_phase.payment_state import OverallPaymentCycle, PaymentData, TicketPaymentData
 from core.models.enum_models import ProtocolChoice, PaymentCategory
@@ -235,10 +236,30 @@ def return_pre_existing_invoice(billing_code: str, connection_observer: Connecti
     url = f'{Constants.SP_BASE_URL}/v2/get_invoice'
 
     result = _api_call_for_invoice(url=url, payload=payload, connection_observer=connection_observer)
+
+    #-----------------------------
+    # EXPIRED? CLARIFY ON WHAT'S EXPIRED (TICKETS VS INDIVIDUAL)
+    #-----------------------------
+    if not result.valid and result.error_type == ResultError.EXPIRED_INVOICE:
+        # TICKET
+        if category == PaymentCategory.TICKET:
+            result.error_type = ResultError.TICKET_EXPIRED
+            result.message = "This ticket billing is expired.  Do you want to wipe the ticket config & literal tickets and start over?"
+            return result
+
+        # INDIVIDUAL
+        if category == PaymentCategory.INDIVIDUAL:
+            result.error_type = ResultError.EXPIRED_INVOICE
+            result.message = "This invoice is expired.  Do you want to wipe it and start over?"
+            return result
+
     return result
 
 
 
+
+
+
 def get_ticket_invoice(connection_observer: ConnectionObserver) -> Result:
     """Evaluates if it needs a new ticket invoice, and then returns pre-existing"""
 
@@ -429,7 +450,7 @@ def get_current_billing_data(category: Optional[PaymentCategory] = None) -> Resu
     return Result(valid=True, data=billing_data)
 
 
-def wipe_billing_cycle() -> Result:
+def wipe_billing_cycle(inserted_profile_id: Optional[int] = None) -> Result:
     """Wipes the entire overall cycle"""
     logger.info("Wiping the overall cycle, so this cycle no longer blocks other new ones")
 
@@ -446,10 +467,38 @@ def wipe_billing_cycle() -> Result:
 
     # Overall State:
     payment_cycle = OverallPaymentCycle()
-    if payment_cycle:
+    if not payment_cycle:
-        payment_cycle.wipe()
+        return Result(valid=True)
 
-    # This has to return a Result object, even if not checked, because of it's use in plumbing,
+    # if no profile id was given, wipe it and move on:
+    if not inserted_profile_id:
+        payment_cycle.wipe()
+        return Result(valid=True)
+
+    # from here on assume we were given a profile id
+    return delete_individual_cycle(payment_cycle=payment_cycle, inserted_profile_id=inserted_profile_id)
+
+
+def delete_individual_cycle(payment_cycle: OverallPaymentCycle, inserted_profile_id: int) -> Result:
+
+    logger.info("Deleting an individual profile")
+    billing_data = payment_cycle.billing_data
+    if not billing_data:
+        error_msg = "No billing data to wipe.  Did you already wipe it?"
+        logger.error(error_msg)
+        return Result(valid=False, error_type=ResultError.MISSING_DATA, message=error_msg)
+
+    cycle_profile_id = billing_data.instance_id
+
+    if inserted_profile_id != cycle_profile_id:
+        error_msg = f"You're trying to wipe the wrong profile.  You hit delete {inserted_profile_id} but the overall cycle is for {cycle_profile_id}"
+        logger.error(error_msg)
+        return Result(valid=False, error_type=ResultError.INVALID_INPUT, message=error_msg)
+
+    payment_cycle.wipe()
+
+    individual_cycle = PaymentData(instance_id=inserted_profile_id)
+    individual_cycle.wipe()
     return Result(valid=True)
 
 
@@ -572,6 +621,6 @@ def return_pre_existing_payment_data(profile_id: int, subscription: Subscription
         return Result(valid=False, message=error_msg, error_type=ResultError.INVALID_INPUT)
 
 
-def evaluate_ticketing():
+def evaluate_ticketing() -> bool:
     setup = ConfigurationController.are_tickets_setup()
-    
+    return setup    

core/utils/type_checker.py

@@ -1,57 +1,49 @@
 from core.models.Result import Result, ResultError
 from core.errors.logger import logger
 
-
 import inspect
-from functools import wraps
+from functools import wraps, partial
-from typing import get_origin, get_args, Union
+from typing import get_origin, get_args, Union, Any, Literal, Callable, Annotated
+from collections.abc import Callable as ABCCallable
 import types  # Required for UnionType (the | operator)
 
 def _get_type_error_details(check_type, value, origin, args_types):
-    """
-    Helper to construct human-readable names for the expected and actual types.
-    """
-    # Case 1: User used Type[T] but passed an instance of T
     if origin is type:
         inner_name = getattr(args_types[0], '__name__', 'the specified type')
         return f"a class/type (subclass of {inner_name})", f"an instance of {type(value).__name__}"
     
-    # Case 2: check_type is a tuple (common for Unions)
     if isinstance(check_type, tuple):
         type_names = ", ".join([getattr(t, '__name__', str(t)) for t in check_type])
         return f"one of ({type_names})", f"type {type(value).__name__}"
     
-    # Case 3: Standard single type
     type_name = getattr(check_type, '__name__', str(check_type))
     return type_name, f"type {type(value).__name__}"
 
 def _validate_inner_types(value, origin, args_types, func, name):
-    """
-    Helper to handle deep validation for generics like List[int] or Type[T].
-    Returns (is_valid, error_message).
-    """
     if not args_types:
         return True, None
 
     inner_type = args_types[0]
-
+    if inner_type is Any:
-    # Handle Type[T]: value must be a subclass of T
+        return True, None
-    if origin is type:
+
-        if not issubclass(value, inner_type):
+    if origin is type:
-            inner_name = getattr(inner_type, '__name__', str(inner_type))
+        try:
-            return False, f"For function {func}, Argument '{name}' must be a subclass of {inner_name}, got {value.__name__}"
+            if not issubclass(value, inner_type):
+                inner_name = getattr(inner_type, '__name__', str(inner_type))
+                return False, f"For function {func}, Argument '{name}' must be a subclass of {inner_name}, got {value.__name__}"
+        except TypeError:
+            return False, f"For function {func}, Argument '{name}' expected a class, got {type(value).__name__}"
         return True, None
 
-    # Handle Collections: list, set, tuple
     if isinstance(value, (list, set, tuple)):
         for item in value:
+            if inner_type is Any:
+                continue
             inner_origin = get_origin(inner_type)
-            
-            # Handle Unions inside collections (e.g., List[Union[int, str]])
             if inner_origin is Union or (hasattr(types, "UnionType") and inner_origin is types.UnionType):
                 if not isinstance(item, get_args(inner_type)):
                     return False, f"For function {func}, Argument '{name}' contains element {type(item).__name__}, expected {inner_type}"
-            # Handle standard types inside collections
             elif not isinstance(item, inner_type):
                 inner_name = getattr(inner_type, '__name__', str(inner_type))
                 return False, f"For function {func}, Argument '{name}' contains element {type(item).__name__}, expected {inner_name}"
@@ -63,7 +55,6 @@ def type_checker(func):
     def wrapper(*args, **kwargs):
         sig = inspect.signature(func)
         
-        # 1. Bind arguments to names to handle both positional and keyword args
         try:
             bound_args = sig.bind(*args, **kwargs)
             bound_args.apply_defaults()
@@ -72,31 +63,70 @@ def type_checker(func):
             logger.error(error_msg)
             return Result(valid=False, error_type=ResultError.INVALID_INPUT, message=error_msg)
 
-        # 2. Validate each argument against its type hint
         for name, value in bound_args.arguments.items():
             expected_type = sig.parameters[name].annotation
-            
-            # Skip if no type hint is provided
             if expected_type is inspect.Parameter.empty:
                 continue
             
+            # --- PHASE 1: SANITIZATION ---
+            # 1. Handle Any (always valid)
+            if expected_type is Any:
+                continue
+            
             origin = get_origin(expected_type)
             args_types = get_args(expected_type)
-            
+
-            # Determine the type to use for isinstance() 
+            # 2. Handle Annotated[T, ...] (peel the wrapper to get T)
-            # Handles both typing.Union and the Python 3.10+ | operator (UnionType)
+            if origin is Annotated:
+                expected_type = args_types[0]
+                origin = get_origin(expected_type)
+                args_types = get_args(expected_type)
+
+            # 3. Handle Literal (check if value is one of the allowed literals)
+            if origin is Literal:
+                if value not in args_types:
+                    error_msg = f"For function {func}, Argument '{name}' must be one of {args_types}, got {value}"
+                    logger.error(error_msg)
+                    return Result(valid=False, error_type=ResultError.INVALID_INPUT, message=error_msg)
+                continue
+
+            # --- PHASE 2: TYPE DETERMINATION ---
             is_union = origin is Union or (hasattr(types, "UnionType") and origin is types.UnionType)
-            check_type = args_types if is_union else (origin if origin is not None else expected_type)
+            is_callable_hint = (expected_type is Callable or expected_type is ABCCallable or origin is ABCCallable)
+
+            if is_union:
+                # Sanitize Union tuple: replace typing.Callable with ABCCallable
+                check_type = tuple(ABCCallable if t is Callable else t for t in args_types)
+            else:
+                check_type = origin if origin is not None else expected_type
+
+            # --- PHASE 3: VALIDATION ---
+            is_valid = True
             
-            # Basic type validation
+            if is_callable_hint and not is_union:
-            if not isinstance(value, check_type):
+                # Use callable() for partials, lambdas, and functions
+                if not callable(value):
+                    is_valid = False
+            elif not is_union and check_type is Callable:
+                is_valid = callable(value)
+            else:
+                # FINAL SAFETY CHECK:
+                # Only call isinstance if check_type is actually a type or a tuple of types.
+                # This prevents the "arg 2 must be a type" crash for any remaining SpecialForms.
+                if isinstance(check_type, (type, tuple)):
+                    if not isinstance(value, check_type):
+                        is_valid = False
+                else:
+                    # If we reached here and it's not a type/tuple, it's a SpecialForm we can't check.
+                    # We treat it as valid to avoid crashing.
+                    is_valid = True
+
+            if not is_valid:
                 type_name, actual_name = _get_type_error_details(check_type, value, origin, args_types)
                 error_msg = f"For function {func}, Argument '{name}' must be {type_name}, got {actual_name}"
                 logger.error(error_msg)
                 return Result(valid=False, error_type=ResultError.INVALID_INPUT, message=error_msg)
             
-            # Deep validation for generics (e.g., List[int], Type[MyClass])
-            # We only do this if it's not a Union, as Unions were handled by the basic isinstance check above
             if origin is not None and not is_union:
                 is_valid, error_msg = _validate_inner_types(value, origin, args_types, func, name)
                 if not is_valid: