diff --git a/core/controllers/tickets/TicketPrepController.py b/core/controllers/tickets/TicketPrepController.py index ddcd2c7..e2a82e1 100644 --- a/core/controllers/tickets/TicketPrepController.py +++ b/core/controllers/tickets/TicketPrepController.py @@ -5,6 +5,8 @@ if TYPE_CHECKING: from core.essentials.observers.ConnectionObserver import ConnectionObserver from core.observers.TicketObserver import TicketObserver +from core.services.payment_phase.payment_state import TicketPaymentData +# from core.services.prepare_tickets.get_pub_key import get_and_save_pub_key from core.utils.type_checker import type_checker from core.services.prepare_tickets.ticket_prep_orchestrator import ticket_prep_orchestrator from core.services.prepare_tickets.make_sure_pub_key_exists import make_sure_pub_key_exists @@ -44,33 +46,47 @@ def prepare_tickets( billing_code: Optional[int] = None ) -> Result: + logger.info("prepare_tickets has begun") + # these two are for preparation of a single ticket to a profile. # can't have one blank without the other. (profile or ticket) if bool(which_ticket) != bool(billing_code): - return {"valid": False, "error_code": "failed_input"} + return Result(valid=False, error_type=ResultError.INVALID_INPUT, message="which ticket must match the billing code") + # return {"valid": False, "error_code": "failed_input"} # make sure it's a number: - if not isinstance(how_many_profiles, int): + if not how_many_profiles or not isinstance(how_many_profiles, int): if ticket_observer is not None: ticket_observer.notify("failed_input", None) - return {"valid": False, "error_code": "failed_input"} + return Result(valid=False, error_type=ResultError.INVALID_INPUT, message=f"The profile quantity must be a number, and not type({how_many_profiles}) with {how_many_profiles}") + # return {"valid": False, "error_code": "failed_input"} # allow single prep: - if how_many_profiles != 1: - # but if it's not a single ticket, then make sure that "how_many_profiles" is actually the number of profiles ordered - # (which is based on locally saved data from the previous step): - how_many_ordered = get_how_many_profiles_were_ordered() - if how_many_profiles != how_many_ordered: - if ticket_observer is not None: - ticket_observer.notify("failed_input", None) - return {"valid": False, "error_code": "failed_input"} + # if how_many_profiles != 1: + # # but if it's not a single ticket, then make sure that "how_many_profiles" is actually the number of profiles ordered + # # (which is based on locally saved data from the previous step): + # how_many_ordered = get_how_many_profiles_were_ordered() + # if how_many_profiles != how_many_ordered: + # if ticket_observer is not None: + # ticket_observer.notify("failed_input", None) + # return Result(valid=False, error_type=ResultError.INVALID_INPUT, message=f"The profile quantity must be a number, and not type({how_many_profiles}) with {how_many_profiles}") + # return {"valid": False, "error_code": "failed_input"} # make sure this guy has a public key to verify against: - does_he_have_public_key = make_sure_pub_key_exists(connection_observer) - if does_he_have_public_key == False: - if ticket_observer is not None: - ticket_observer.notify("failed_input", None) - return {"valid": False, "error_code": "failed_input"} + + logger.info("Checking public key..") + billing_data = TicketPaymentData() + which_key = billing_data.which_key + if not which_key: + error_msg = "You lack knowing the public key for this ticket plan." + logger.error(error_msg) + return Result(valid=False, error_type=ResultError.INVALID_INPUT, message=error_msg) + + # got_key = get_and_save_pub_key(which_key) + # if not got_key.valid: + # logger.error(got_key.message) + # return got_key + # logger.info("Got public key..") notification = "Preparing Cryptography Locally" if ticket_observer is not None: diff --git a/core/infrastructure/plumbing.py b/core/infrastructure/plumbing.py index e55e57a..bd80639 100644 --- a/core/infrastructure/plumbing.py +++ b/core/infrastructure/plumbing.py @@ -90,16 +90,18 @@ class ThreadPlumbing: return # does this have an auto-navigate page? - if self.navigator and self.auto_navigate_dict: - self.check_auto_navigation(result) + # if self.navigator and self.auto_navigate_dict: + # logger.info("checking auto-navigate..") + # self.check_auto_navigation(result) # If the result sender explicitly asked us to navigate to a page with a ui callback, then bypass the popup: - if result.error_type == ResultError.NAVIGATE_TO: - if self.ui_caller: - self.ui_caller(result) + if result.error_type == ResultError.NAVIGATE_TO and self.ui_caller: + logger.info("Plumbing was asked to NAVIGATE_TO so skipping to the ui_caller..") + self.ui_caller(result) # do we need to decide on an action? choice = bool(result.function_choice) + logger.info(f"Do we have to make a choice? {choice}") # if there's a message or a choice to make: if result.message or choice: @@ -113,13 +115,15 @@ class ThreadPlumbing: end_app=False ) # framework neutral show method + logger.info("Showing a popup..") popup.show() + logger.info("Showed the error popup.") # otherwise return it as is: - else: - if self.ui_caller: - self.ui_caller(result) - + if self.ui_caller: + logger.info("Returning from plumbing..") + self.ui_caller(result) + logger.error("No ui caller!!!") def _on_popup_answer(self, user_choice: bool, passed_result: Result): """User responded to popup""" diff --git a/core/models/Result.py b/core/models/Result.py index 23cbad3..a4c82f0 100644 --- a/core/models/Result.py +++ b/core/models/Result.py @@ -21,6 +21,7 @@ class ResultError(Enum): SUBSCRIPTION = "subscription" EXPIRED_INVOICE = "expired_invoice" # meant for invoice UNPAID = "unpaid" + PAID = "paid" # this is in-case it was expected to be unpaid EXPIRED_SUB = "expired_sub" ACTIVE_BILLING = "active_billing" BILLING_CONFLICT = "billing_conflict" diff --git a/core/services/payment_phase/check_payment_loop.py b/core/services/payment_phase/check_payment_loop.py index b28d4d0..90c7ff0 100644 --- a/core/services/payment_phase/check_payment_loop.py +++ b/core/services/payment_phase/check_payment_loop.py @@ -1,23 +1,175 @@ +# inside the class: from core.infrastructure.thread_dispatch import MainThreadDispatcher -from core.services.payment_phase.payment_update_tools import check_paid, mark_as_paid -from core.controllers.tickets.TicketPayController import check_if_ticket_paid - +# from core.services.payment_phase.payment_update_tools import check_paid, mark_as_paid from core.observers.ConnectionObserver import ConnectionObserver from core.models.enum_models import PaymentType -from core.models.Result import Result, ResultError -# from infrastructure.observer_service import notify from core.errors.logger import logger from core.models.Result import Result, ResultError -from threading import Thread +# from threading import Thread from typing import Callable import threading import time - MAX_FAILED_CONNECTIONS = 8 +# Top level loose functions: + +from core.services.payment_phase.conclude import delete_individual_cycle +from core.models.manage.server_reply_tools import api_call_and_extraction, ApiCall +from core.controllers.ProfileController import ProfileController +from core.services.payment_phase.payment_state import OverallPaymentCycle +from core.models.enum_models import PaymentCategory +# from core.observers.ConnectionObserver import ConnectionObserver +from core.Constants import Constants +from core.errors.error_shortcuts import not_supported +from core.observers.Notifications import Notifications, EventType + +from typing import Optional + +def _mark_one_profile_as_paid(profile_id: int, payment_cycle: OverallPaymentCycle): + """Save to disk that an individual profile is paid.""" + + #------------------------- + # UPDATE SAVED PROFILE DATA + #------------------------- + logger.info(f"Marking profile_id {str(profile_id)} as paid..") + + # Get real profile object: + profile = ProfileController.get(profile_id) + if not profile: + error_msg = f"Invalid profile id of {profile_id}" + logger.error(error_msg) + return Result(valid=False, message=error_msg, error_type=ResultError.INVALID_INPUT) + + profile.subscription.billing_pending = False + ProfileController.update(profile) + + #------------------------- + # DELETE BILLING FOR THAT PROFILE + #------------------------- + return delete_individual_cycle(payment_cycle=payment_cycle, inserted_profile_id=profile_id) + + +def mark_as_paid(profile_id: Optional[int] = None) -> Result: + """Conclude the payment phase for BOTH payment types.""" + + logger.info("Marking as paid..") + + payment_cycle = OverallPaymentCycle() + + if not payment_cycle: + error_msg = "There is no current billing cycle." + return Result(valid=False, error_type=ResultError.INVALID_INPUT, message=error_msg) + + billing_data = payment_cycle.billing_data + + if not billing_data: + error_msg = "There is no current billing data." + return Result(valid=False, error_type=ResultError.INVALID_INPUT, message=error_msg) + + category = payment_cycle.category + + if category == PaymentCategory.INDIVIDUAL: + if not profile_id: + profile_id = billing_data.instance_id + result = _mark_one_profile_as_paid(profile_id=profile_id, payment_cycle=payment_cycle) + if not result.valid: + return result + + elif category == PaymentCategory.TICKET: + # saving global ticket data: + logger.info("Marking the global tickets as paid") + # billing_data.billing_pending = False + billing_data.paid = True + billing_data.save() + else: + return not_supported() + + # finally: + logger.info("wiping the global payment cycle") + payment_cycle.wipe() + notification = "Paid!" + Notifications.emit(EventType.STATUS_UPDATE, notification) + return Result(valid=True) + + +def check_paid(billing_code: str, connection_observer: ConnectionObserver, get_expired_date: Optional[bool] = False) -> Result: + """ + Purpose: + Evaluates Payment Status for EITHER payment category + + Requires: + Valid OverallPaymentCycle + billing_code is provided by the timer for BOTH types. + """ + + payment_cycle = OverallPaymentCycle() + category = payment_cycle.category + + logger.info(f"Checking paid with category of {category}") + + # Get endpoint: + if category == PaymentCategory.INDIVIDUAL: + url = f'{Constants.SP_BASE_URL}/v2/check_paid' + elif category == PaymentCategory.TICKET: + url = f'{Constants.TICKET_API_BASE_URL}/check_ticket_paid' + else: + return Result(valid=False, error_type=ResultError.INVALID_INPUT, message="This is not a supported payment category to check") + + submit_data = { + "billing_code": billing_code + } + + #----------------- + # Just get paid or not + #----------------- + if not get_expired_date: + + result = api_call_and_extraction( + desired_type="payment_status", + method=ApiCall.POST, + url=url, + connection_observer=connection_observer, + payload=submit_data + ) + + if not result.valid: + return result + + payment_status = result.data + + # paid + if payment_status: + return Result(valid=True, message="paid") + else: + return Result(valid=False, error_type=ResultError.UNPAID, message="The API call went through, but this subscription is not yet paid.") + + + #----------------- + # Find out if Paid & Get expiration: + #----------------- + # Get entire ServerReply + server_reply = api_call_and_extraction( + desired_type=None, + method=ApiCall.POST, + url=url, + connection_observer=connection_observer, + payload=submit_data, + return_server_reply=True # get server reply + ) + + if not server_reply.valid: + return server_reply + + payment_status = server_reply.payment_status + expires_at = server_reply.expires_at + + if payment_status and expires_at: + return Result(valid=True, data=expires_at, message="Paid with an expiration") + + return Result(valid=False, error_type=ResultError.UNPAID, message="The API call went through, but this subscription is not yet paid.") class CheckPaidTimer: @@ -91,14 +243,6 @@ class CheckPaidTimer: # in the background thread: print("checking if paid..") result = check_paid(self.billing_code, self.connection_observer) - - - # if self.payment_type == PaymentType.CRYPTO: - # result = check_paid(self.billing_code, self.connection_observer) - # elif self.payment_type == PaymentType.TICKET: - # result = check_if_ticket_paid(self.billing_code, self.connection_observer) - # else: - # result = Result(valid=False, error_type=ResultError.INVALID_INPUT, message="Invalid payment type chosen to check.") print("checked") # Bring result to main thread: diff --git a/core/services/payment_phase/conclude.py b/core/services/payment_phase/conclude.py new file mode 100644 index 0000000..6c3ecf2 --- /dev/null +++ b/core/services/payment_phase/conclude.py @@ -0,0 +1,32 @@ +from core.services.payment_phase.payment_state import OverallPaymentCycle, PaymentData +from core.models.Result import Result, ResultError +from core.errors.logger import logger + + +def return_wrong_profile(inserted_profile_id: int, cycle_profile_id: int) -> Result: + """Provides an error message for a profile delete mismatch.""" + 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) + + +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: + return return_wrong_profile(inserted_profile_id=inserted_profile_id, cycle_profile_id=cycle_profile_id) + + payment_cycle.wipe() + + individual_cycle = PaymentData(instance_id=inserted_profile_id) + individual_cycle.wipe() + return Result(valid=True) + diff --git a/core/services/payment_phase/payment_update_tools.py b/core/services/payment_phase/payment_update_tools.py index 2c87e87..fa34d36 100644 --- a/core/services/payment_phase/payment_update_tools.py +++ b/core/services/payment_phase/payment_update_tools.py @@ -1,3 +1,4 @@ +from core.services.payment_phase.conclude import delete_individual_cycle from core.models.manage.server_reply_tools import api_call_and_extraction, ApiCall from core.controllers.ProfileController import ProfileController from core.services.payment_phase.payment_state import OverallPaymentCycle @@ -11,9 +12,12 @@ from core.observers.Notifications import Notifications, EventType from typing import Optional -def _mark_profile_as_paid(profile_id: int): +def _mark_one_profile_as_paid(profile_id: int, payment_cycle: OverallPaymentCycle): """Save to disk that an individual profile is paid.""" + #------------------------- + # UPDATE SAVED PROFILE DATA + #------------------------- logger.info(f"Marking profile_id {str(profile_id)} as paid..") # Get real profile object: @@ -25,12 +29,15 @@ def _mark_profile_as_paid(profile_id: int): profile.subscription.billing_pending = False ProfileController.update(profile) - return Result(valid=True) + + #------------------------- + # DELETE BILLING FOR THAT PROFILE + #------------------------- + return delete_individual_cycle(payment_cycle=payment_cycle, inserted_profile_id=profile_id) - -def mark_as_paid() -> Result: - """Conclude the payment phase for ALL profiles.""" +def mark_as_paid(profile_id: Optional[int] = None) -> Result: + """Conclude the payment phase for BOTH payment types.""" logger.info("Marking as paid..") @@ -40,7 +47,7 @@ def mark_as_paid() -> Result: error_msg = "There is no current billing cycle." return Result(valid=False, error_type=ResultError.INVALID_INPUT, message=error_msg) - billing_data = payment_cycle.instance + billing_data = payment_cycle.billing_data if not billing_data: error_msg = "There is no current billing data." @@ -48,9 +55,8 @@ def mark_as_paid() -> Result: category = payment_cycle.category - if category == PaymentCategory.INDIVIDUAL: - profile_id = billing_data.instance_id - result = _mark_profile_as_paid(profile_id) + if category == PaymentCategory.INDIVIDUAL and profile_id: + result = _mark_one_profile_as_paid(profile_id=profile_id, payment_cycle=payment_cycle) if not result.valid: return result diff --git a/core/services/subscriptions/subscriptions_by_protocol.py b/core/services/subscriptions/subscriptions_by_protocol.py index 1408425..0dadc69 100644 --- a/core/services/subscriptions/subscriptions_by_protocol.py +++ b/core/services/subscriptions/subscriptions_by_protocol.py @@ -1,5 +1,6 @@ -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.services.payment_phase.conclude import delete_individual_cycle +from core.services.payment_phase.check_payment_loop import CheckPaidTimer, mark_as_paid +from core.models.manage.server_reply_tools import api_call_and_extraction, ApiCall, translate_server_failure from core.controllers.ProfileController import ProfileController from core.controllers.ConfigurationController import ConfigurationController from core.models.BaseProfile import BaseProfile as Profile @@ -8,7 +9,7 @@ from core.models.enum_models import ProtocolChoice, PaymentCategory from core.observers.ConnectionObserver import ConnectionObserver from core.models.pydantic_models.type_checker import validate_model from core.observers.Notifications import Notifications, EventType -from core.models.pydantic_models.ServerReply import SubscriptionInvoice +from core.models.pydantic_models.ServerReply import SubscriptionInvoice, ServerReply from core.models.Subscription import Subscription from core.utils.type_checker import type_checker from core.Constants import Constants @@ -27,6 +28,18 @@ def invalid_server_format() -> Result: return Result(valid=False, error_type=ResultError.INVALID_API_REPLY, message=error_msg) +def _already_paid() -> Result: + """Wrap the marking of payment in a failed Result, because an invoice was expected.""" + error_msg = "This invoice is already paid. Marking it as so" + logger.info(error_msg) + marked_paid = mark_as_paid() + if marked_paid.valid: + return Result(valid=False, error_type=ResultError.PAID, message=error_msg) + else: + logger.error(marked_paid.message) + return marked_paid + + def _api_call_for_invoice(url: str, payload: dict, connection_observer: ConnectionObserver) -> Result: """ Cross-Category Helper function. @@ -42,21 +55,36 @@ def _api_call_for_invoice(url: str, payload: dict, connection_observer: Connecti notification = "Sending to API.." Notifications.emit(EventType.STATUS_UPDATE, notification) - result = api_call_and_extraction( - desired_type="data", + server_reply = api_call_and_extraction( + desired_type=None, method=ApiCall.POST, url=url, connection_observer=connection_observer, - payload=payload + payload=payload, + return_server_reply=True ) - if not result.valid: - return result + # this is a connection error or invalid format itself: + if isinstance(server_reply, Result) or not isinstance(server_reply, ServerReply): + return server_reply #----------------------------- - # EVALUATE SERVER'S ACTUAL DATA + # EVALUATE IF PAID ALREADY #----------------------------- - payload_data = result.data + paid = server_reply.payment_status + if paid: + return _already_paid() + + #----------------------------- + # TRANSLATE OTHER FAILURES: + #----------------------------- + if not server_reply.valid: + return translate_server_failure(server_reply=server_reply) + + #----------------------------- + # EVALUATE INVOICE + #----------------------------- + payload_data = server_reply.data logger.info("Validating the final invoice") final_invoice = validate_model(SubscriptionInvoice, payload_data) @@ -115,7 +143,7 @@ def evaluate_repeat_cycle(payment_cycle: OverallPaymentCycle, profile_id: int) - Notifications.emit(EventType.STATUS_UPDATE, notification) return Result(valid=True, data=payment_cycle.billing_data) else: - error_msg = f"You activated profile {profile_id}, but profile {payment_cycle.billing_data.instance_id} has active billing, stop that?" + error_msg = f"You activated profile {profile_id}, but another profile has active billing, stop that?" logger.error(error_msg) func_choice = partial(wipe_billing_cycle) return Result(valid=False, function_choice=func_choice, error_type=ResultError.BILLING_CONFLICT, message=error_msg) @@ -249,26 +277,35 @@ def return_pre_existing_invoice(billing_code: str, connection_observer: Connecti result = _api_call_for_invoice(url=url, payload=payload, connection_observer=connection_observer) + #----------------------------- + # SUCCESS + #----------------------------- if result.valid: notification = "Got Pre-Existing Invoice" Notifications.emit(EventType.STATUS_UPDATE, notification) + return result + + #----------------------------- + # NO INVOICE BECAUSE ALREADY PAID + #----------------------------- + if result.error_type == ResultError.PAID: + return result #----------------------------- # EXPIRED? CLARIFY ON WHAT'S EXPIRED (TICKETS VS INDIVIDUAL) #----------------------------- - if not result.valid and result.error_type == ResultError.EXPIRED_INVOICE: + if result.error_type == ResultError.EXPIRED_INVOICE: + # INDIVIDUAL + if category == PaymentCategory.INDIVIDUAL: + result.message = "This invoice is expired. Do you want to wipe it and start over?" + return result + # 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 @@ -484,7 +521,7 @@ def wipe_billing_cycle_IF_its_this_profile(profile_id) -> Result: if payment_cycle.billing_code != profile_id: return Result(valid=True) - return wipe_billing_cycle(inserted_profile_id=profile.id) + return wipe_billing_cycle(inserted_profile_id=profile_id) @@ -520,27 +557,27 @@ def wipe_billing_cycle(inserted_profile_id: Optional[int] = None) -> Result: 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: +# 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) +# 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 +# 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) +# 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() +# payment_cycle.wipe() - individual_cycle = PaymentData(instance_id=inserted_profile_id) - individual_cycle.wipe() - return Result(valid=True) +# individual_cycle = PaymentData(instance_id=inserted_profile_id) +# individual_cycle.wipe() +# return Result(valid=True) @type_checker diff --git a/core/utils/basic_operations/file_tools.py b/core/utils/basic_operations/file_tools.py index f769242..e1a7d08 100644 --- a/core/utils/basic_operations/file_tools.py +++ b/core/utils/basic_operations/file_tools.py @@ -1,18 +1,84 @@ from core.errors.logger import logger from pathlib import Path +from typing import Tuple -def delete_file(file_path: str | Path) -> bool: - """ - Deletes a file at the specified path. - """ +def write_to_file(data: str, filepath: str) -> bool: try: - # Path(file_path) handles both string and Path object inputs - # missing_ok=True prevents FileNotFoundError from being raised - Path(file_path).unlink(missing_ok=True) + file_path = Path(filepath) + file_path.parent.mkdir(parents=True, exist_ok=True) + file_path.write_text(data) return True - except (PermissionError, OSError) as e: - # Log the error for debugging purposes - logging.error(f"Failed to delete {file_path}: {e}") + + except PermissionError: + error_msg = f"There was an error with saving/writing the file. You don't have permission to modify this path: {filepath}" + logger.error(error_msg) + return False + except UnicodeDecodeError: + logger.error("File encoding is not UTF-8") + return False + except Exception as e: + logger.error(f"Unexpected error: {e}") return False + +def read_a_file(path_of_file: str) -> str | bool: + file_path = Path(path_of_file) + + try: + with open(file_path, "r") as file: + content = file.read() + return content + + except FileNotFoundError: + logger.error(f"File does not exist: {path_of_file}") + return False + except PermissionError: + logger.error("You don't have permission to read this file") + return False + except UnicodeDecodeError: + logger.error("File encoding is not UTF-8") + return False + except Exception as e: + logger.error(f"Unexpected error: {e}") + return False + + +def delete_file(file_path: str | Path) -> Tuple[bool, str]: + """ + Robustly delete a file with comprehensive error handling. + + Returns: + Tuple of (success: bool, message: str) + """ + try: + # Convert to Path object for better handling + path = Path(file_path) + + # Check if file exists + if not path.exists(): + error_msg = f"File does not exist: {file_path}" + logger.info(error_msg) + return True, error_msg + + # Verify it's actually a file (not a directory) + if not path.is_file(): + error_msg = f"Already deleted! Path is not a file (may be a directory): {file_path}" + return True, error_msg + + # Attempt deletion + path.unlink() + return True, f"File successfully deleted: {file_path}" + + except PermissionError: + return False, f"Permission denied: No permission to delete {file_path}" + except FileNotFoundError: + error_msg = f"File not found: {file_path}" + logger.info(f"File not found: {file_path}") + return True, error_msg + except IsADirectoryError: + return False, f"Path is a directory, not a file: {file_path}" + except OSError as e: + return False, f"OS error while deleting file: {e}" + except Exception as e: + return False, f"Unexpected error while deleting file: {type(e).__name__}: {e}" diff --git a/core/utils/get_data.py b/core/utils/get_data.py index b21c0e7..06430e5 100644 --- a/core/utils/get_data.py +++ b/core/utils/get_data.py @@ -1,8 +1,9 @@ +from core.utils.basic_operations.file_tools import read_a_file + from core.Constants import Constants import os import json import ast -from core.errors.exceptions import * from core.errors.logger import logger @@ -32,6 +33,27 @@ def get_data(which_ticket, name_of_file_and_folder): return False +def read_and_convert_with_eval(where_is_the_data: str) -> dict: + """ + Purpose: + Opens a file and uses literal_eval to convert it. + Why: + The py_ecc library format gives data in a weird format that traditional JSON tools struggle with. + Called by: + get_data (same module) + get_pub_key's get_local_copy_of_public_key + """ + raw_string_content = read_a_file(where_is_the_data) + + if not raw_string_content: + return False + + # Convert string to Python dict + python_dict = ast.literal_eval(raw_string_content) + return python_dict + + + def get_raw_string(which_ticket, name_of_file_and_folder): data_folder = Constants.HV_TICKETING_DATA_HOME folder_path = f"{data_folder}/{name_of_file_and_folder}"