Skip to content

Commit

Permalink
improves QIF split_category handling
Browse files Browse the repository at this point in the history
  • Loading branch information
WolfgangFahl committed Oct 12, 2024
1 parent f855793 commit d37141b
Show file tree
Hide file tree
Showing 5 changed files with 166 additions and 129 deletions.
17 changes: 9 additions & 8 deletions nomina/ledger.py
Original file line number Diff line number Diff line change
Expand Up @@ -153,16 +153,17 @@ def filter(
filtered_transactions = {}

for transaction_id, transaction in self.transactions.items():
transaction_date = transaction.isodate.split()[
0
] # Extract 'YYYY-MM-DD' part
if transaction.isodate is not None:
transaction_date = transaction.isodate.split()[
0
] # Extract 'YYYY-MM-DD' part

in_range = (not start_date or transaction_date >= start_date) and (
not end_date or transaction_date <= end_date
)
in_range = (not start_date or transaction_date >= start_date) and (
not end_date or transaction_date <= end_date
)

if in_range:
filtered_transactions[transaction_id] = transaction
if in_range:
filtered_transactions[transaction_id] = transaction

filtered_book = deepcopy(self)
filtered_book.transactions = filtered_transactions
Expand Down
9 changes: 7 additions & 2 deletions nomina/nomina_beancount.py
Original file line number Diff line number Diff line change
Expand Up @@ -147,8 +147,13 @@ def create_transaction(
meta = data.new_metadata("", 0, metadata)
txn_postings = []
for account, amount_value, currency in postings:
amt = amount.Amount(amount.D(str(amount_value)), currency)
txn_postings.append(data.Posting(account, amt, None, None, None, None))
if amount_value is None:
pass
else:
amount_str=str(amount_value)
amount_d=amount.D(amount_str)
amt = amount.Amount(amount_d, currency)
txn_postings.append(data.Posting(account, amt, None, None, None, None))

tx = data.Transaction(
meta,
Expand Down
101 changes: 55 additions & 46 deletions nomina/qif.py
Original file line number Diff line number Diff line change
Expand Up @@ -21,58 +21,68 @@
from nomina.date_utils import DateUtils
from nomina.stats import Stats

@dataclass
class ParseRecord:
"""
generic parse record to keep track of lines and errors
"""
start_line: int = 0
end_line: int = 0
errors: Dict[str, Exception] = field(default_factory=dict)

@dataclass
class SplitTarget:
target: str
transfer_account: Optional[str] = field(init=False, default=None)
account: Optional[str] = field(init=False, default=None)
subaccount: Optional[str] = field(init=False, default=None)
account_category: Optional[str] = field(init=False, default=None)
category: Optional[str] = field(init=False, default=None)
account_action: Optional[str] = field(init=False, default=None)
class SplitCategory:
"""
a Quicken Interchange Format (QIF) Split target
"""
markup: str # the original QIF markup for the split category
# parts of split
category:Optional[str]=None
account: Optional[str]=None
split_class: Optional[str]=None
# flags
has_pipe: bool=False
has_slash: bool=False

def __post_init__(self):
if "[" in self.target and "]" in self.target:
match = re.match(r"\[([^\]]+)\]", self.target)
if match:
self.transfer_account = match.group(1)
elif "|" in self.target:
match = re.match(r"([^|]+)\|x", self.target)
if match:
self.account_action = match.group(1)
elif "/" in self.target:
match = re.match(r"([^/]+)/([^/]+)", self.target)
if match:
self.account_category = match.group(1)
self.category = match.group(2)
elif ":" in self.target:
match = re.match(r"([^:]+):([^:]+)", self.target)
if match:
self.account = match.group(1)
self.subaccount = match.group(2)
else:
self.account = self.target
"""
parse my target string
"""
self.has_pipe="|" in self.markup
self.has_slash="/" in self.markup
# qif holds the markup which still needs processing
qif=self.markup

def __repr__(self):
attrs = ", ".join(
f"{k}={v!r}" for k, v in self.__dict__.items() if v is not None
)
return f"SplitTarget({attrs})"
pattern = r'\[(?P<account_name>[^\]]+)\]'

# Search for the pattern in the split_category string
match = re.search(pattern, self.markup)

@dataclass
class ParseRecord:
start_line: int = 0
end_line: int = 0
errors: Dict[str, Exception] = field(default_factory=dict)
if match:
# Extract the account name from the named group
self.account = match.group('account_name')
qif=qif.replace(f"[{self.account}]","")
else:
self.account = None

if self.has_pipe:
qif=qif.replace("|","")

if self.has_slash:
# Split by the first slash to separate category and class
parts = qif.split("/", 1)
if len(parts) > 1:
# If there's a class after the slash, set it
self.split_class = parts[1]
qif=parts[0]
if qif:
self.category=qif


@lod_storable
class ErrorRecord(ParseRecord):
line: Optional[str] = None


@lod_storable
class Category(ParseRecord):
"""
Expand All @@ -82,7 +92,6 @@ class Category(ParseRecord):
name: Optional[str] = None
description: str = ""


@lod_storable
class QifClass(ParseRecord):
"""
Expand All @@ -102,13 +111,11 @@ class Account(ParseRecord):
currency: str = "EUR" # Default to EUR
parent_account_id: Optional[str] = None


@lod_storable
class Transaction(ParseRecord):
"""
a single transaction
"""

isodate: Optional[str] = None
amount: Optional[str] = None
payee: Optional[str] = None
Expand All @@ -117,9 +124,9 @@ class Transaction(ParseRecord):
number: Optional[str] = None
cleared: Optional[str] = None
address: Optional[str] = None
split_category: List[str] = field(default_factory=list)
split_memo: List[str] = field(default_factory=list)
split_amount: List[str] = field(default_factory=list)
split_categories: List[SplitCategory] = field(default_factory=list)
split_memos: List[str] = field(default_factory=list)
split_amounts: List[str] = field(default_factory=list)
account: Optional[Account] = None
qif_class: Optional[QifClass] = None
category: Optional[Category] = None
Expand Down Expand Up @@ -147,7 +154,7 @@ def normalize(self):
self.errors["amount"] = ex

self.split_amounts_float = []
for i, amount in enumerate(self.split_amount):
for i, amount in enumerate(self.split_amounts):
try:
self.split_amounts_float.append(self.parse_amount(amount))
except Exception as ex:
Expand Down Expand Up @@ -296,6 +303,8 @@ def parse(self, lines: List[str], verbose: bool = False, debug: bool = False):
key = self.field_names.get(first)
value = line[1:].strip()
if key in ["split_category", "split_memo", "split_amount"]:
if key=="split_category":
value=SplitCategory(value)
if key not in current_record:
current_record[key] = []
current_record[key].append(value)
Expand Down
Loading

0 comments on commit d37141b

Please sign in to comment.