From a7b733db0d1d66d3831fa6ec8983ec31cb1ae626 Mon Sep 17 00:00:00 2001 From: SimplifiedPrivacy Date: Mon, 19 Oct 2026 19:44:23 -0400 Subject: [PATCH] Functions added for wiping tickets when expired (the entire ticket cycle). Also general error fixing for the payments flow --- core/errors/translate_error.py | 35 +++--- core/infrastructure/plumbing.py | 7 +- core/models/Result.py | 3 +- core/models/manage/server_reply_tools.py | 2 +- core/services/payment_phase/payment_state.py | 22 ++-- core/services/payment_phase/ticket_cycle.py | 42 ++++++++ .../subscriptions_by_protocol.py | 61 +++++++++-- core/utils/type_checker.py | 100 ++++++++++++------ 8 files changed, 205 insertions(+), 67 deletions(-) create mode 100644 core/services/payment_phase/ticket_cycle.py diff --git a/core/errors/translate_error.py b/core/errors/translate_error.py index 7e2ae05..967a406 100644 --- a/core/errors/translate_error.py +++ b/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 - 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) + error_msg = get_traceback_line(error) 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 = [] diff --git a/core/infrastructure/plumbing.py b/core/infrastructure/plumbing.py index 359e7cb..e55e57a 100644 --- a/core/infrastructure/plumbing.py +++ b/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): diff --git a/core/models/Result.py b/core/models/Result.py index 66b6486..73d1aa3 100644 --- a/core/models/Result.py +++ b/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" diff --git a/core/models/manage/server_reply_tools.py b/core/models/manage/server_reply_tools.py index d314423..ca067e7 100644 --- a/core/models/manage/server_reply_tools.py +++ b/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: diff --git a/core/services/payment_phase/payment_state.py b/core/services/payment_phase/payment_state.py index bd697b4..97a866a 100644 --- a/core/services/payment_phase/payment_state.py +++ b/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): - logger.info("Starting wipe of pre-existing data") + def wipe_data_on_disk(self): + logger.info("Starting wipe of pre-existing data") - if not os.path.exists(self._data_file): - logger.info("File does NOT exist already") - return + if not os.path.exists(self._data_file): + logger.info("File does NOT exist already") + 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(): diff --git a/core/services/payment_phase/ticket_cycle.py b/core/services/payment_phase/ticket_cycle.py new file mode 100644 index 0000000..8ce3d7a --- /dev/null +++ b/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.") diff --git a/core/services/subscriptions/subscriptions_by_protocol.py b/core/services/subscriptions/subscriptions_by_protocol.py index 99bf2eb..fc8546f 100644 --- a/core/services/subscriptions/subscriptions_by_protocol.py +++ b/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: - payment_cycle.wipe() + if not payment_cycle: + 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() - \ No newline at end of file + return setup \ No newline at end of file diff --git a/core/utils/type_checker.py b/core/utils/type_checker.py index 7af5be8..1740f6b 100644 --- a/core/utils/type_checker.py +++ b/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 typing import get_origin, get_args, Union +from functools import wraps, partial +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] - - # Handle Type[T]: value must be a subclass of T - if origin is type: - 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__}" + if inner_type is Any: + return True, None + + if origin is type: + try: + 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() - # Handles both typing.Union and the Python 3.10+ | operator (UnionType) + + # 2. Handle Annotated[T, ...] (peel the wrapper to get T) + 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 not isinstance(value, check_type): + if is_callable_hint and not is_union: + # 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: