commit f8478853855bf8c2a8363dd0f8978f8f28456304
parent fc0eac13c7e88308a6de4e8c279b00be21ceee11
Author: Carlosokumu <carlosokumu254@gmail.com>
Date: Tue, 24 Mar 2026 20:40:38 +0300
chore: restructure core classes
Diffstat:
5 files changed, 0 insertions(+), 351 deletions(-)
diff --git a/dummy/usawa/core/__init__.py b/dummy/usawa/core/__init__.py
diff --git a/dummy/usawa/core/chain_manager.py b/dummy/usawa/core/chain_manager.py
@@ -1,157 +0,0 @@
-import os
-import logging
-from usawa import Ledger, load
-from usawa.crypto import ACL, DemoWallet
-from usawa.store import LedgerStore
-from whee.valkey import ValkeyStore
-import os
-import shutil
-import logging
-
-from usawa import Ledger, load
-
-logg = logging.getLogger("core.chain_manager")
-
-
-def default_chain_dir():
- """Return the conventional XDG data directory for usawa ledger files.
-
- Defaults to ~/.local/share/usawa/ledger/ unless XDG_DATA_HOME is set.
- Directory is created if it does not exist.
- """
- xdg_data = os.environ.get(
- 'XDG_DATA_HOME',
- os.path.join(os.path.expanduser('~'), '.local', 'share'),
- )
- path = os.path.join(xdg_data, 'usawa', 'ledger')
- os.makedirs(path, exist_ok=True)
- return path
-
-
-class LedgerChainManager:
-
- def __init__(self, genesis_path=None):
- """Initialise the chain manager.
-
- On first ever launch, supply genesis_path so it can be copied into
- the conventional directory as 0.xml. On subsequent launches, omit
- genesis_path — the manager will reconstruct the chain from disk.
-
- :param genesis_path: Path to the genesis XML file (optional after first run)
- :type genesis_path: str or None
- """
- self.basedir = default_chain_dir()
- logg.debug('chain dir: {}'.format(self.basedir))
-
- zero = os.path.join(self.basedir, '0.xml')
- if not os.path.exists(zero):
- if genesis_path is None:
- raise FileNotFoundError(
- 'no chain found in {} and no genesis_path provided'.format(self.basedir)
- )
- genesis_path = os.path.realpath(genesis_path)
- if not os.path.exists(genesis_path):
- raise FileNotFoundError('genesis file not found: {}'.format(genesis_path))
- shutil.copy(genesis_path, zero)
- logg.debug('copied genesis {} → {}'.format(genesis_path, zero))
-
- self.chain = self._reconstruct_chain()
- logg.debug('chain reconstructed with depth {}'.format(self.depth()))
-
-
- def _reconstruct_chain(self):
- """Rebuild the chain list by scanning sequential xml files on disk.
-
- Starts at 0.xml and stops at the first missing index.
- """
- chain = []
- index = 0
- while True:
- path = os.path.join(self.basedir, '{}.xml'.format(index))
- if not os.path.exists(path):
- break
- chain.append(path)
- index += 1
- if not chain:
- raise FileNotFoundError('no chain files found in: {}'.format(self.basedir))
- return chain
-
-
- def current(self):
- """Return the path of the most recent XML file in the chain."""
- return self.chain[-1]
-
-
- def derive_next(self):
- """Derive the next output file path based on the current chain length.
-
- genesis = index 0, first entry output = index 1, and so on.
- """
- next_index = len(self.chain)
- basename = '{}.xml'.format(next_index)
- return os.path.join(self.basedir, basename)
-
-
- def advance(self, written_path):
- """Call this after successfully writing an entry output file.
-
- Verifies the file exists before appending to the chain.
- """
- written_path = os.path.realpath(written_path)
- if not os.path.exists(written_path):
- raise FileNotFoundError(
- 'written file not found, cannot advance chain: {}'.format(written_path)
- )
- self.chain.append(written_path)
- logg.debug('chain advanced to: {}'.format(written_path))
-
-
- def load_current(self):
- """Load and return a fresh Ledger instance from the current chain tail."""
- ledger_path = self.current()
- ledger_tree = load(ledger_path)
- ledger = Ledger.from_tree(ledger_tree)
- logg.debug('loaded ledger from: {}'.format(ledger_path))
- return ledger
-
-
- def write_entry(self, entry, ledger, store, wallet):
- """Sign and write an entry, then advance the chain.
-
- Encapsulates the full write sequence:
- entry.sign → store.add_entry → ledger.truncate → ledger.sign → write file
- """
- next_path = self.derive_next()
-
- db = ValkeyStore('')
-
- store = LedgerStore(db, ledger)
- pk = store.get_key()
- wallet = DemoWallet(privatekey=pk)
- logg.debug("wallet pk: %s pubk: %s", wallet.privkey().hex(), wallet.pubkey().hex())
- ledger.set_wallet(wallet)
-
-
- ledger.acl = ACL.from_wallet(wallet)
- self.ledger = ledger
-
- entry.sign(wallet)
- store.add_entry(entry, update_ledger=True)
- ledger.truncate()
- ledger.sign()
-
- with open(next_path, 'wb') as f:
- f.write(ledger.to_string())
- logg.debug('entry written to: {}'.format(next_path))
-
- self.advance(next_path)
- return next_path
-
-
- def depth(self):
- """Return the number of files in the chain including genesis."""
- return len(self.chain)
-
-
- def __repr__(self):
- return 'LedgerChainManager(depth={}, current={})'.format(self.depth(), self.current())
-\ No newline at end of file
diff --git a/dummy/usawa/core/entry_service.py b/dummy/usawa/core/entry_service.py
@@ -1,70 +0,0 @@
-import logging
-from datetime import datetime
-import uuid
-
-from .models import LedgerEntry
-from usawa.storage.ledger_repository import LedgerRepository
-
-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.tx_date = datetime.now()
- entry.date_registered = datetime.now()
- entry.transaction_ref = self._generate_transaction_ref()
-
- is_valid, error_msg = entry.validate()
- if not is_valid:
- logg.error(f"Entry validation failed: {error_msg}")
- return False, error_msg
-
- 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"
- )
- 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
-
- def save_wallet(self, wallet, passphrase):
- return self.repository.save_wallet(wallet=wallet, passphrase=passphrase)
-
- 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)
-
- def export_all_entries_to_xml(self, output_path: str) -> tuple[bool, str]:
- return self.repository.export_all_entries_to_xml(output_path=output_path)
-
- def export_entry_to_xml(self, serial: int, output_path: str) -> tuple[bool, str]:
- return self.repository.export_entry_to_xml(serial, output_path)
diff --git a/dummy/usawa/core/models.py b/dummy/usawa/core/models.py
@@ -1,95 +0,0 @@
-from dataclasses import dataclass,field
-from typing import Optional,List, Union
-from datetime import datetime
-from pathlib import Path
-
-@dataclass
-class LedgerEntry:
- """Ledger entry data model"""
-
- # Basic details
- external_reference: Optional[str] = None
- description: Optional[str] = None
-
- # Transaction details
- amount: float = 0.0
- source_unit: str = ""
- source_type: str = ""
- source_path: str = "general"
- dest_unit: str = ""
- dest_type: str = ""
- dest_path: str = "general"
-
- # Attachments
- attachments: List[str] = field(default_factory=list)
-
- # Signers (public keys)
- signer_pubkeys: List[str] = field(default_factory=list)
-
- serial: Optional[int] = None
- tx_date: Optional[datetime] = None
- tx_reference: Optional[str] = None
- date_registered: Optional[datetime] = None
- parent_digest: Optional[str] = None
- unit_index: Optional[int] = None
-
- def validate(self) -> tuple[bool, str]:
- """Validate entry data"""
- if self.amount <= 0:
- return False, "Amount must be greater than 0"
-
- if not self.source_unit or not self.dest_unit:
- return False, "Unit/Currency is required for both source and destination"
-
- if not self.source_type or not self.dest_type:
- return False, "Account type is required for both source and destination"
-
- if not self.source_path or not self.dest_path:
- return False, "Account path is required for both source and destination"
-
- # Validate attachments
- if self.attachments:
- for filepath in self.attachments:
- if not Path(filepath).exists():
- return False, f"Attachment file not found: {filepath}"
-
- return True, ""
-
- def __repr__(self):
- return (
- f"LedgerEntry("
- f"external_reference={self.external_reference!r}, "
- f"description={self.description!r}, "
- f"serial={self.serial!r}, "
- f"parent_digest={self.parent_digest}, "
- f"amount={self.amount}, "
- f"source_unit={self.source_unit}, "
- f"tx_ref={self.tx_reference}, "
- f"source_type={self.source_type}, "
- f"dest_unit={self.dest_unit}, "
- f"dest_type={self.dest_type})"
- f"attachments={self.attachments!r})"
- )
-
- def add_attachment(self, filepath: Union[str, List[str]]):
- """
- Add one or more attachment file paths
- """
- if isinstance(filepath, str):
- if filepath not in self.attachments:
- self.attachments.append(filepath)
- elif isinstance(filepath, list):
- for path in filepath:
- if path not in self.attachments:
- self.attachments.append(path)
- else:
- raise TypeError(f"filepath must be str or List[str], got {type(filepath)}")
-
- def remove_attachment(self, filepath: str):
- """Remove an attachment file path"""
- if filepath in self.attachments:
- self.attachments.remove(filepath)
-
- def get_attachment_count(self) -> int:
- """Get number of attachments"""
- return len(self.attachments)
diff --git a/dummy/usawa/core/state_manager.py b/dummy/usawa/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)
