commit bbe0ee3a4b97422110c7c58e5ad748903914fca5
parent 6821a5bf3750f24f1013a88ca9e3fcbd26cd1772
Author: lash <dev@holbrook.no>
Date: Tue, 14 Jul 2026 14:44:36 -0600
Merge branch 'carlos/gen-cleanup' into lash/wallet-only-in-store
Diffstat:
11 files changed, 110 insertions(+), 303 deletions(-)
diff --git a/dummy/usawa/gui/__init__.py b/dummy/usawa/gui/__init__.py
@@ -2,13 +2,10 @@ import logging
import signal
import gi
-from usawa.runnable.wallet import setup_wallet
-
gi.require_version("Gtk", "4.0")
gi.require_version("Adw", "1")
-from gi.repository import Gtk, Adw, Gio
+from gi.repository import Adw, Gio
from .main_window import UsawaMainWindow
-from usawa.config import load_config
logg = logging.getLogger("gui.app")
@@ -27,7 +24,6 @@ class Usawa(Adw.Application):
signal.signal(signal.SIGINT, self._handle_sigint)
def on_activate(self, app):
- #self.win = UsawaMainWindow(application=app, ledger_path=self.ledger_file, account_list=self.account_list)
self.win = UsawaMainWindow(app, self.ctx)
self.win.present()
@@ -36,37 +32,5 @@ class Usawa(Adw.Application):
self.quit()
def do_command_line(self, cli):
-# args = cli.get_arguments()
-#
-# config_dir = None
-# filtered_args = []
-# i = 1
-# while i < len(args):
-# if args[i] == "-c":
-# if i + 1 < len(args):
-# config_dir = args[i + 1]
-# i += 2
-# else:
-# logg.error("-c flag requires a directory argument")
-# return 1
-# else:
-# filtered_args.append(args[i])
-# i += 1
-# self.cfg = load_config(config_dir=config_dir)
-# logg.debug('config-dump:\n{}'.format(self.cfg))
-#
-# if len(filtered_args) == 0:
-# logg.error("missing ledger file argument")
-# return 1
-#
-# #if filtered_args[0] == "setup-wallet":
-# # wallet_dir = filtered_args[1] if len(filtered_args) > 1 else None
-# # setup_wallet(self.cfg, wallet_dir=wallet_dir)
-# # return 0
-#
-# self.ledger_file = filtered_args[0]
-# for k in self.cfg.all():
-# logg.debug("config {} => {}".format(k, self.cfg.get(k)))
-# self.activate()
self.activate()
return 0
diff --git a/dummy/usawa/gui/components/wallet_setup.py b/dummy/usawa/gui/components/wallet_setup.py
@@ -4,11 +4,8 @@ from gi.repository import Gtk, Adw, GLib, Gio
import threading
import logging
import threading
-
-from ..core.state_manager import StateManager
from usawa.crypto import DemoWallet
-
logg = logging.getLogger("gui.wallet_setup_view")
@@ -156,7 +153,6 @@ class ImportWalletDialog(Adw.Dialog):
toast.set_timeout(3)
self.parent.toast_overlay.add_toast(toast)
self.close()
- StateManager.set("wallet_path", self.privatekey_path)
self.parent._init_with_wallet(wallet, self.passphrase)
def _on_error(self):
diff --git a/dummy/usawa/gui/controllers/entry_controller.py b/dummy/usawa/gui/controllers/entry_controller.py
@@ -9,16 +9,17 @@ logg = logging.getLogger("gui.entry_controller")
class EntryController:
- """Handles entry creation logic"""
- def __init__(self, entry_service: EntryService):
+ def __init__(self, ctx, entry_service: EntryService):
self.entry_service = entry_service
self._entry_created_listeners = []
+ self.ctx = ctx
def collect_entry_data(self, view) -> Optional[LedgerEntry]:
- """Collect data from the view and create an entry"""
tx_date_str = view.date_entry.get_text().strip()
tx_time_str = view.time_entry.get_text().strip()
+ unit = self.ctx.store.ledger.uidx.base
+
try:
if tx_time_str:
for fmt in ("%Y-%m-%d %H:%M:%S", "%Y-%m-%d %H:%M"):
@@ -41,11 +42,11 @@ class EntryController:
external_reference=view.ref_entry.get_text().strip() or None,
description=view.desc_entry.get_text().strip() or None,
amount=float(view.amount_entry.get_text() or "0"),
- source_unit="BTC",
+ source_unit=unit,
tx_date=tx_date,
source_type=view.get_source_type(),
source_path=view.source_path_entry.get_text().strip(),
- dest_unit="BTC",
+ dest_unit=unit,
dest_type=view.get_dest_type(),
dest_path=view.dest_path_entry.get_text().strip(),
)
diff --git a/dummy/usawa/gui/core/entry_service.py b/dummy/usawa/gui/core/entry_service.py
@@ -9,56 +9,46 @@ logg = logging.getLogger("core.entry_service")
class EntryService:
- """Business logic for ledger entries"""
def __init__(self, repository: LedgerRepository):
self.repository = repository
def save_entry(self, entry: LedgerEntry) -> tuple[bool, str]:
- try:
- entry.date_registered = datetime.now()
- entry.transaction_ref = self._generate_transaction_ref()
+ entry.date_registered = datetime.now()
+ entry.transaction_ref = str(uuid.uuid4())
- is_valid, error_msg = entry.validate()
- if not is_valid:
- logg.error(f"Entry validation failed: {error_msg}")
- return False, error_msg
+ is_valid, error_msg = entry.validate()
+ if not is_valid:
+ logg.error(f"Entry validation failed: {error_msg}")
+ return False, error_msg
+ try:
self.repository.save(entry)
-
- logg.info(f"Entry saved successfully")
- return True, ""
-
- except FileExistsError as e:
- error_msg = (
- "Some file information for this entry is already recorded in the ledger"
+ except (FileExistsError, ValueError, IOError, Exception) as e:
+ error_msg = self._error_message_for(e)
+ logg.error(
+ f"Save failed: {e}",
+ exc_info=not isinstance(e, (FileExistsError, ValueError, IOError)),
)
return False, error_msg
- except ValueError as e:
- error_msg = f"Invalid entry data: {str(e)}"
- logg.error(f"Validation error: {e}")
- return False, error_msg
-
- except IOError as e:
- error_msg = f"File error: {str(e)}"
- logg.error(f"File operation failed: {e}")
- return False, error_msg
-
- except Exception as e:
- error_msg = f"Failed to save entry: {str(e)}"
- logg.error(f"Unexpected error: {e}", exc_info=True)
- return False, error_msg
+ logg.info("Entry saved successfully")
+ return True, ""
-# def save_wallet(self, wallet, passphrase):
-# return self.repository.save_wallet(wallet=wallet, passphrase=passphrase)
+ def _error_message_for(self, e: Exception) -> str:
+ if isinstance(e, FileExistsError):
+ return (
+ "Some file information for this entry is already recorded in the ledger"
+ )
+ if isinstance(e, ValueError):
+ return f"Invalid entry data: {e}"
+ if isinstance(e, IOError):
+ return f"File error: {e}"
+ return f"Failed to save entry: {e}"
def get_all_entries(self):
return self.repository.get_all_entries()
- def _generate_transaction_ref(self) -> str:
- return str(uuid.uuid4())
-
def get_asset_bytes(self, digest: bytes) -> bytes:
return self.repository.get_asset_bytes(digest=digest)
diff --git a/dummy/usawa/gui/core/state_manager.py b/dummy/usawa/gui/core/state_manager.py
@@ -1,28 +0,0 @@
-import json
-from pathlib import Path
-
-STATE_FILE = Path.home() / ".local" / "share" / "usawa" / "state.json"
-
-
-class StateManager:
-
- @staticmethod
- def get_state() -> dict:
- if STATE_FILE.exists():
- return json.loads(STATE_FILE.read_text())
- return {}
-
- @staticmethod
- def save_state(data: dict):
- STATE_FILE.parent.mkdir(parents=True, exist_ok=True)
- STATE_FILE.write_text(json.dumps(data, indent=2))
-
- @staticmethod
- def get(key: str, default=None):
- return StateManager.get_state().get(key, default)
-
- @staticmethod
- def set(key: str, value):
- state = StateManager.get_state()
- state[key] = value
- StateManager.save_state(state)
diff --git a/dummy/usawa/gui/main_window.py b/dummy/usawa/gui/main_window.py
@@ -1,24 +1,15 @@
import logging
-from pathlib import Path
-from .core.state_manager import StateManager
from usawa.crypto import DemoWallet
from usawa.gui.components.passphrase_dialog import (
PASSPHRASE_DIALOG_CSS,
PassphraseDialog,
)
-from usawa.gui.components.wallet_setup import ImportWalletDialog
-from usawa.ledger import Ledger
-from usawa.service import UnixClient
from .core.entry_service import EntryService
from usawa.storage.ledger_repository import LedgerRepository
from gi.repository import Adw, Gtk, Gio, GLib, Gdk
-from usawa import load
from usawa.gui.controllers.entry_controller import EntryController
from usawa.gui.views.entry_list_view import EntryListView
from datetime import datetime
-from usawa.store import LedgerStore
-from whee.valkey import ValkeyStore
-from whee.fs import FsStore
import usawa.error
logg = logging.getLogger("gui.mainwindow")
@@ -110,12 +101,12 @@ class UsawaMainWindow(Adw.ApplicationWindow):
page = Adw.NavigationPage(title="Ledger Entries", tag="entry-list")
entries = []
self.entry_list_view = EntryListView(
+ ctx=self.ctx,
nav_view=self.nav_view,
entry_controller=self.entry_controller,
entries=entries,
refresh_callback=self.refresh_entries,
toast_overlay=self.toast_overlay,
- #account_list=self.account_list,
)
self.entry_list_view._load_entries()
page.set_child(self.entry_list_view)
@@ -132,7 +123,7 @@ class UsawaMainWindow(Adw.ApplicationWindow):
self._init_with_wallet()
except usawa.error.VerifyError:
dialog = PassphraseDialog(
- #store=self.ctx.keystore,
+ # store=self.ctx.keystore,
ctx=self.ctx,
wallet_class=DemoWallet,
on_success=self._init_with_wallet,
@@ -140,10 +131,12 @@ class UsawaMainWindow(Adw.ApplicationWindow):
)
dialog.present(self)
- def _init_with_wallet(self, wallet):
+ def _init_with_wallet(self):
repository = LedgerRepository(self.ctx)
entry_service = EntryService(repository=repository)
- self.entry_controller = EntryController(entry_service=entry_service)
+ self.entry_controller = EntryController(
+ ctx=self.ctx, entry_service=entry_service
+ )
self.entry_controller.add_entry_created_listener(self.refresh_entries)
entry_list_page = self._create_entry_list_page()
diff --git a/dummy/usawa/gui/views/create_entry_view.py b/dummy/usawa/gui/views/create_entry_view.py
@@ -7,22 +7,19 @@ import mimetypes
logg = logging.getLogger("gui.create_entry_view")
-def create_entry_page(nav_view, controller, account_list=None):
- """Create a new entry page"""
+def create_entry_page(ctx, nav_view, controller, account_list=None):
page = Adw.NavigationPage(title="Create New Entry", tag="create-entry")
- view = CreateEntryView(nav_view, controller, account_list=account_list)
+ view = CreateEntryView(ctx, nav_view, controller, account_list=account_list)
page.set_child(view)
-
return page
class CreateEntryView(Gtk.Box):
- """Create entry view - UI ONLY"""
- def __init__(self, nav_view, controller, account_list=None):
+ def __init__(self, ctx, nav_view, controller, account_list=None):
super().__init__(orientation=Gtk.Orientation.VERTICAL, spacing=0)
-
+ self.ctx = ctx
self.nav_view = nav_view
self.controller = controller
self.account_list = account_list
@@ -31,7 +28,6 @@ class CreateEntryView(Gtk.Box):
self._build_ui(nav_view)
def _build_ui(self, nav_view):
- """Build the UI"""
header = self._create_header(nav_view=nav_view)
self.append(header)
@@ -65,7 +61,6 @@ class CreateEntryView(Gtk.Box):
self.append(action_bar)
def _create_header(self, nav_view):
- """Create the header with back button and serial number"""
header_box = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=12)
header_box.set_margin_top(12)
header_box.set_margin_bottom(12)
@@ -95,7 +90,6 @@ class CreateEntryView(Gtk.Box):
return header_box
def _create_basic_section(self):
- """Create basic details section - UI ONLY"""
section_box = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=12)
header = Gtk.Label(label="BASIC DETAILS")
@@ -303,9 +297,9 @@ class CreateEntryView(Gtk.Box):
unit_label.add_css_class("caption")
fields.append(unit_label)
+ unit = self.ctx.store.ledger.uidx.base
units = Gtk.StringList()
- units.append("Bitcoin(BTC)")
-
+ units.append(unit)
unit_dropdown = Gtk.DropDown(model=units)
unit_dropdown.set_selected(0)
fields.append(unit_dropdown)
@@ -439,7 +433,6 @@ class CreateEntryView(Gtk.Box):
)
def _on_files_selected(self, dialog, result):
- """Handle multiple file selection"""
try:
files = dialog.open_multiple_finish(result)
diff --git a/dummy/usawa/gui/views/entry_list_view.py b/dummy/usawa/gui/views/entry_list_view.py
@@ -10,10 +10,10 @@ logg = logging.getLogger("gui.entrylist")
class EntryListView(Gtk.Box):
- """The entry list view with filters, table, and FAB"""
def __init__(
self,
+ ctx,
nav_view,
entry_controller,
toast_overlay,
@@ -31,6 +31,7 @@ class EntryListView(Gtk.Box):
self.toast_overlay = toast_overlay
self.account_list = account_list
self.active_filter = None
+ self.ctx = ctx
overlay = Gtk.Overlay()
self.append(overlay)
@@ -308,15 +309,17 @@ class EntryListView(Gtk.Box):
dialog.destroy()
def on_sort_changed(self, button, sort_type):
- """Handle sort option changes"""
if button.get_active():
logg.info(f"Sort changed to: {sort_type}")
self._sort_and_reload(sort_type)
def on_fab_clicked(self, button):
- logg.info("FAB clicked - opening create entry window")
-
- create_page = create_entry_page(self.nav_view, self.entry_controller, account_list=self.account_list)
+ create_page = create_entry_page(
+ self.ctx,
+ self.nav_view,
+ self.entry_controller,
+ account_list=self.account_list,
+ )
self.nav_view.push(create_page)
def refresh_data(self):
@@ -372,7 +375,6 @@ class EntryListView(Gtk.Box):
return False
def _create_table_section(self):
- """Create the entry list table with all columns"""
table_box = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=0)
self.entry_store = Gio.ListStore.new(EntryItem)
diff --git a/dummy/usawa/storage/entry_mapper.py b/dummy/usawa/storage/entry_mapper.py
@@ -4,7 +4,6 @@ from datetime import datetime, date
from usawa.entry import Entry, EntryPart
from usawa.ledger import Ledger
from usawa.account import Account
-from usawa.unit import UnitIndex
from ..gui.core.models import LedgerEntry
from usawa import Entry, EntryPart
@@ -15,20 +14,7 @@ class EntryMapper:
"""Maps between domain model (LedgerEntry) and storage model (Entry)"""
@staticmethod
- def to_entry(domain: LedgerEntry, ledger, unitindex=UnitIndex("BTC")):
- """
- Convert LedgerEntry (domain) to Entry (storage)
-
- :param domain: Domain model entry
- :type domain: LedgerEntry
- :param ledger: The ledger object (for parent digest)
- :type ledger: usawa.Ledger
- :param unitindex: UnitIndex for validation
- :type unitindex: UnitIndex
- :return: Storage model entry
- :rtype: Entry
- """
-
+ def to_entry(ctx, domain: LedgerEntry, ledger):
if domain.tx_date is None:
raise ValueError("Transaction date is required for storage")
@@ -45,33 +31,26 @@ class EntryMapper:
parent=parent,
description=domain.description or "",
ref=ref,
- unitindex=unitindex,
+ unitindex=ctx.uidx,
)
- source_amount = unitindex.from_floatstring(
- unitindex.default_unit, str(domain.amount)
- )
+ source_amount = ctx.uidx.from_floatstring(ctx.uidx.base, str(domain.amount))
dest_amount = -source_amount
- account = Account(unitindex.default_unit, domain.source_type.lower(), domain.source_path)
+ account = Account(ctx.uidx.base, domain.source_type.lower(), domain.source_path)
source_part = EntryPart(
- #unitindex.default_unit,
- #domain.source_type.lower(),
- #domain.source_path,
account,
source_amount,
debit=True,
)
- #entry.add_part(source_part, debit=True)
entry.add_part(source_part)
- account = Account(unitindex.default_unit, domain.dest_type.lower(), domain.dest_path)
+ account = Account(ctx.uidx.base, domain.dest_type.lower(), domain.dest_path)
dest_part = EntryPart(
account,
dest_amount,
debit=False,
)
- #entry.add_part(dest_part, debit=False)
entry.add_part(dest_part)
return entry
diff --git a/dummy/usawa/storage/ledger_repository.py b/dummy/usawa/storage/ledger_repository.py
@@ -1,121 +1,58 @@
import logging
from typing import List
from usawa.gui.core.models import LedgerEntry
-from usawa.storage.file_utils import path_from_uri
from usawa.storage.xml_utils import (
_write_xml_to_file,
)
from usawa.asset import Asset
-from usawa.crypto import ACL, DemoWallet, Wallet
+from usawa.crypto import ACL, Wallet
from usawa.ledger import Ledger
-from usawa.resolve.fs import FSResolver
-from usawa.service import UnixClient
from usawa.store import LedgerStore
from .entry_mapper import EntryMapper
-from whee.valkey import ValkeyStore
-from usawa import Ledger, DemoWallet, load
+from usawa import Ledger, load
from pathlib import Path
import mimetypes
import lxml.etree as ET
from pathlib import Path
import lxml.etree as ET
-
logg = logging.getLogger("storage.ledger_repository")
class LedgerRepository:
- """Repository that wraps LedgerStore and handles mapping"""
def __init__(
self,
ctx,
):
- """
- Initialize the LedgerRepository.
-
- :param ledger_path: Path to the ledger definition file to import.
- :type ledger_path: str | None
-
- :param unix_client: Unix socket client used to communicate with the storage service.
- :type unix_client: usawa.UnixClient
-
- """
+ self.ctx = ctx
self.resolver = ctx.resolver
self.wallet = ctx.wallet
self.store = ctx.store
self.db = ctx.db
self.ledger = ctx.ledger
- self.ledger_path = ctx.get('ledger_path')
-
+ self.ledger_path = ctx.get("ledger_path")
def _init_store(self, write=False) -> tuple[LedgerStore, Ledger, Wallet]:
ledger_tree = load(self.ledger_path)
ledger = Ledger.from_tree(ledger_tree)
- if write:
- logg.info("init store for write")
- self.store = LedgerStore(self.db, ledger)
- else:
- logg.info("init store for read")
- self.store = LedgerStore(self.db, ledger)
-
- logg.info("wallet ready, pubkey: %s...", self.wallet.pubkey().hex()[:16])
+ self.store = LedgerStore(self.db, ledger)
ledger.set_wallet(self.wallet)
ledger.acl = ACL.from_wallet(self.wallet)
self.store.load(acl=ledger.acl)
return self.store, ledger, self.wallet
-
- #def _init_store(self, write=False) -> tuple[LedgerStore, Ledger, Wallet]:
- # return self.store, self.ledger, self.wallet
-
-
def save(self, domain_entry: LedgerEntry) -> None:
- """
- Save a domain entry to storage
-
- :param domain_entry: Entry to save
- :type domain_entry: LedgerEntry
- :raises ValueError: If validation fails
- :raises FileExistsError: If attachment already exists in the store
- :raises IOError: If file operations fail
- :raises Exception: For other storage errors
- """
try:
store, ledger, wallet = self._init_store(write=True)
- entry = EntryMapper.to_entry(domain_entry, ledger=ledger)
+ entry = EntryMapper.to_entry(self.ctx, domain_entry, ledger=ledger)
entry.sign(wallet)
- logg.debug(
- "Mapped entry - Serial: %s, Parent: %s, Attachments: %s",
- entry.serial,
- entry.parent.hex(),
- entry.attachment,
- )
-
for attachment in domain_entry.attachments:
- try:
- info = self.get_file_info(attachment)
- asset = Asset.from_file(
- attachment,
- slug=info["slug"],
- description=info["description"],
- mimetype=info["mimetype"],
- )
- store.add_asset(asset)
- entry.attach(asset)
-
- with open(attachment, "rb") as f:
- data = f.read()
- self.resolver.put(asset.get_digest(binary=True), data)
-
- except FileNotFoundError as e:
- raise IOError(f"Attachment file not found: {attachment}") from e
- except PermissionError as e:
- raise IOError(f"Cannot read attachment file: {attachment}") from e
+ self._attach_file(store, entry, attachment)
store.add_entry(entry, update_ledger=True)
@@ -123,19 +60,30 @@ class LedgerRepository:
ledger.sign()
logg.info(f"Successfully saved entry #{entry.serial}")
- except FileExistsError as e:
- logg.debug(f"Entry fileinfo already exists: {e}")
- raise
- except ValueError as e:
- logg.debug(f"Validation error: {e}")
- raise
- except IOError as e:
- logg.debug(f"File operation failed: {e}")
- raise
except Exception as e:
logg.debug(f"Failed to save entry: {e}", exc_info=True)
raise
+ def _attach_file(self, store, entry, attachment) -> None:
+ try:
+ info = self.get_file_info(attachment)
+ asset = Asset.from_file(
+ attachment,
+ slug=info["slug"],
+ description=info["description"],
+ mimetype=info["mimetype"],
+ )
+ store.add_asset(asset)
+ entry.attach(asset)
+
+ with open(attachment, "rb") as f:
+ data = f.read()
+ self.resolver.put(asset.get_digest(binary=True), data)
+
+ except FileNotFoundError as e:
+ raise IOError(f"Attachment file not found: {attachment}") from e
+ except PermissionError as e:
+ raise IOError(f"Cannot read attachment file: {attachment}") from e
def get_all_entries(self) -> List[LedgerEntry]:
"""Get all entries"""
@@ -178,65 +126,53 @@ class LedgerRepository:
def export_all_entries_to_xml(self, output_path: str) -> tuple[bool, str]:
try:
- _, ledger, _ = self._init_store()
- for k in ledger.entries:
- self.resolver.put_entry(ledger.entries[k], lookup="sha512")
+ for entry in self.ledger.entries.values():
+ self.resolver.put_entry(entry, lookup="sha512")
+
+ tree = self.ledger.to_tree()
- tree = ledger.to_tree()
output_file = Path(output_path)
output_file.parent.mkdir(parents=True, exist_ok=True)
xml_string = ET.tostring(
- tree, encoding="utf-8", xml_declaration=True, pretty_print=True
+ tree,
+ encoding="utf-8",
+ xml_declaration=True,
+ pretty_print=True,
)
with open(output_file, "wb") as f:
f.write(xml_string)
- logg.info(f"Exported ledger to {output_path}")
+ logg.info("Exported ledger to %s", output_path)
return True, ""
- except PermissionError as e:
- error_msg = "Permission denied. Cannot write to the specified location."
- logg.debug(f"Permission error: {e}")
- return False, error_msg
-
- except IOError as e:
- error_msg = f"Failed to write file: {str(e)}"
- logg.debug(f"I/O error: {e}")
- return False, error_msg
-
except Exception as e:
- error_msg = f"Failed to export ledger: {str(e)}"
- logg.debug("Unexpected error during export")
- return False, error_msg
+ return self._handle_export_error(e)
def export_entry_to_xml(self, serial: int, output_path: str) -> tuple[bool, str]:
try:
storage_entry = self.store.ledger.entries.get(serial)
- if not storage_entry:
+ if storage_entry is None:
return False, f"Entry #{serial} not found"
- _, ledger, _ = self._init_store()
- try:
- _write_xml_to_file(storage_entry.to_string(), output_path)
- except Exception as e:
- logg.debug(
- "Failed to write entry #%d to file %s: %s", serial, output_path, e
- )
-
- ledger.truncate()
- logg.debug(
- "Ledger entries after truncate: %d", len(self.store.ledger.entries)
- )
- self.resolver.put_entry(entry=storage_entry, lookup="sha512")
+ self.resolver.put_entry(storage_entry, lookup="sha512")
+ _write_xml_to_file(storage_entry.to_string(), output_path)
- logg.info(f"Successfully exported entry #{serial} -> {output_path}")
+ logg.info("Successfully exported entry #%d", serial)
return True, ""
- except PermissionError:
- return False, "Permission denied"
- except IOError as e:
- return False, str(e)
except Exception as e:
- return False, str(e)
+ return self._handle_export_error(e)
+
+ def _handle_export_error(self, e: Exception) -> tuple[bool, str]:
+ if isinstance(e, PermissionError):
+ logg.exception("Permission denied during export")
+ return False, "Permission denied. Cannot write to the specified location."
+
+ if isinstance(e, IOError):
+ logg.exception("I/O error during export")
+ return False, f"Failed to write file: {e}"
+
+ logg.exception("Unexpected error during export")
+ return False, f"Failed to export: {e}"
diff --git a/dummy/usawa/storage/xml_utils.py b/dummy/usawa/storage/xml_utils.py
@@ -1,19 +0,0 @@
-import logging
-from pathlib import Path
-
-from lxml import etree as ET
-
-logg = logging.getLogger("usawa.xml_utils")
-
-
-def _write_xml_to_file(xml_string, output_path: str) -> None:
-
- output_file = Path(output_path)
- output_file.parent.mkdir(parents=True, exist_ok=True)
-
- with open(output_file, "wb") as f:
- if isinstance(xml_string, str):
- xml_string = xml_string.encode("utf-8")
- f.write(xml_string)
-
- logg.debug("Wrote XML to %s (%d bytes)", output_path, len(xml_string))
