diff --git a/app/apps/import_app/schemas/v1.py b/app/apps/import_app/schemas/v1.py index 5b735d5..ef79c19 100644 --- a/app/apps/import_app/schemas/v1.py +++ b/app/apps/import_app/schemas/v1.py @@ -106,6 +106,17 @@ class ExcelImportSettings(BaseModel): sheets: list[str] | str = "*" +class QIFImportSettings(BaseModel): + skip_errors: bool = Field( + default=False, + description="If True, errors during import will be logged and skipped", + ) + file_type: Literal["qif"] = "qif" + importing: Literal["transactions"] = "transactions" + encoding: str = Field(default="utf-8", description="File encoding") + date_format: str = Field(..., description="Date format (e.g. %d/%m/%Y)") + + class ColumnMapping(BaseModel): source: Optional[str] | Optional[list[str]] = Field( default=None, @@ -342,7 +353,7 @@ class CurrencyExchangeMapping(ColumnMapping): class ImportProfileSchema(BaseModel): - settings: CSVImportSettings | ExcelImportSettings + settings: CSVImportSettings | ExcelImportSettings | QIFImportSettings mapping: Dict[ str, TransactionAccountMapping diff --git a/app/apps/import_app/services/v1.py b/app/apps/import_app/services/v1.py index fbebc7b..19ad85d 100644 --- a/app/apps/import_app/services/v1.py +++ b/app/apps/import_app/services/v1.py @@ -3,6 +3,8 @@ import hashlib import logging import os import re +import zipfile +from django.db import transaction from datetime import datetime, date from decimal import Decimal, InvalidOperation from typing import Dict, Any, Literal, Union @@ -845,6 +847,219 @@ class ImportService: f"Invalid {self.settings.file_type.upper()} file format: {str(e)}" ) + def _parse_and_import_qif(self, content_lines: list[str], filename: str) -> None: + # Infer account from filename (remove extension) + account_name = os.path.splitext(os.path.basename(filename))[0] + + current_transaction = {} + raw_lines_buffer = [] + + account = Account.objects.filter(name=account_name).first() + if not account: + raise ValueError(f"Account '{account_name}' not found.") + + row_number = 0 + for line in content_lines: + row_number += 1 + line = line.strip() + if not line: + continue + + raw_lines_buffer.append(line) + + if line == "^": + if current_transaction: + # Deduplication using hash of raw lines + raw_content = "".join(raw_lines_buffer) + internal_id = hashlib.sha256( + raw_content.encode("utf-8") + ).hexdigest() + + # Reset buffer for next transaction + raw_lines_buffer = [] + + try: + with transaction.atomic(): + if Transaction.objects.filter( + internal_id=internal_id + ).exists(): + self._increment_totals("skipped", 1) + self._log( + "info", + f"Skipped duplicate transaction from {filename}", + ) + current_transaction = {} + continue + + # Handle Account + if account: + current_transaction["account"] = account + else: + acc = Account.objects.filter(name=account_name).first() + if acc: + current_transaction["account"] = acc + else: + raise ValueError( + f"Account '{account_name}' not found." + ) + + current_transaction["internal_id"] = internal_id + + # Handle Description/Memo mapping + if "memo" in current_transaction: + current_transaction["description"] = ( + current_transaction.pop("memo") + ) + + # Handle Payee mapping + entities = [] + if "payee" in current_transaction: + payee_name = current_transaction.pop("payee") + # "Treat the payee (P) as the entity. Use existing or create" + entity, _ = TransactionEntity.objects.get_or_create( + name=payee_name + ) + entities.append(entity) + + # Handle Label/Category + category = None + tags = [] + if "label" in current_transaction: + label = current_transaction.pop("label") + if label.startswith("[") and label.endswith("]"): + # Transfer: set label as description, ignore category/tags + clean_label = label[1:-1] + current_transaction["description"] = clean_label + else: + parts = label.split(":") + if parts: + cat_name = parts[0].strip() + if cat_name: + category, _ = ( + TransactionCategory.objects.get_or_create( + name=cat_name + ) + ) + + if len(parts) > 1: + for tag_name in parts[1:]: + tag_name = tag_name.strip() + if tag_name: + tag, _ = ( + TransactionTag.objects.get_or_create( + name=tag_name + ) + ) + tags.append(tag) + + current_transaction["category"] = category + + # Create transaction + new_trans = Transaction.objects.create( + **current_transaction + ) + if entities: + new_trans.entities.set(entities) + if tags: + new_trans.tags.set(tags) + + self.import_run.transactions.add(new_trans) + self._increment_totals("successful", 1) + + except Exception as e: + if not self.settings.skip_errors: + raise e + self._log( + "warning", + f"Error processing transaction in {filename}: {str(e)}", + ) + self._increment_totals("failed", 1) + + # Reset for next transaction + current_transaction = {} + else: + # Empty transaction record (orphaned ^) + raw_lines_buffer = [] + pass + self._increment_totals("processed", 1) + continue + + if line.startswith("!"): + continue + + code = line[0] + value = line[1:] + + if code == "D": + try: + current_transaction["date"] = datetime.strptime( + value, self.settings.date_format + ).date() + except ValueError: + self._log( + "warning", + f"Could not parse date '{value}' using format '{self.settings.date_format}' in {filename}", + ) + if not self.settings.skip_errors: + raise ValueError(f"Invalid date format '{value}'") + + elif code == "T": + try: + cleaned_value = value.replace(",", "") + amount = Decimal(cleaned_value) + if amount < 0: + current_transaction["type"] = Transaction.Type.EXPENSE + current_transaction["amount"] = abs(amount) + else: + current_transaction["type"] = Transaction.Type.INCOME + current_transaction["amount"] = amount + except InvalidOperation: + self._log( + "warning", f"Could not parse amount '{value}' in {filename}" + ) + if not self.settings.skip_errors: + raise ValueError(f"Invalid amount format '{value}'") + + elif code == "P": + current_transaction["payee"] = value + elif code == "M": + current_transaction["memo"] = value + elif code == "L": + current_transaction["label"] = value + elif code == "N": + pass + + def _process_qif(self, file_path): + def process_logic(): + if zipfile.is_zipfile(file_path): + try: + with zipfile.ZipFile(file_path, "r") as zf: + for filename in zf.namelist(): + if filename.lower().endswith( + ".qif" + ) and not filename.startswith("__MACOSX"): + self._log( + "info", f"Processing QIF from ZIP: {filename}" + ) + with zf.open(filename) as f: + content = f.read().decode(self.settings.encoding) + self._parse_and_import_qif( + content.splitlines(), filename + ) + except Exception as e: + raise ValueError(f"Error processing ZIP file: {str(e)}") + else: + with open(file_path, "r", encoding=self.settings.encoding) as f: + self._parse_and_import_qif( + f.readlines(), os.path.basename(file_path) + ) + + if not self.settings.skip_errors: + with transaction.atomic(): + process_logic() + else: + process_logic() + def _validate_file_path(self, file_path: str) -> str: """ Validates that the file path is within the allowed temporary directory. @@ -871,6 +1086,8 @@ class ImportService: self._process_csv(file_path) elif isinstance(self.settings, version_1.ExcelImportSettings): self._process_excel(file_path) + elif isinstance(self.settings, version_1.QIFImportSettings): + self._process_qif(file_path) self._update_status("FINISHED") self._log( diff --git a/app/apps/import_app/tests/test_qif_import.py b/app/apps/import_app/tests/test_qif_import.py new file mode 100644 index 0000000..a79662b --- /dev/null +++ b/app/apps/import_app/tests/test_qif_import.py @@ -0,0 +1,259 @@ +from decimal import Decimal +import os +import shutil +from django.test import TestCase +from django.contrib.auth import get_user_model +from apps.accounts.models import Account, AccountGroup +from apps.currencies.models import Currency +from apps.common.middleware.thread_local import write_current_user, delete_current_user +from apps.import_app.models import ImportProfile, ImportRun +from apps.import_app.services.v1 import ImportService +from apps.transactions.models import ( + Transaction, +) + + +class QIFImportTests(TestCase): + def setUp(self): + # Patch TEMP_DIR for testing + self.original_temp_dir = ImportService.TEMP_DIR + self.test_dir = os.path.abspath("temp_test_import") + ImportService.TEMP_DIR = self.test_dir + os.makedirs(self.test_dir, exist_ok=True) + + # Create user and set context + User = get_user_model() + self.user = User.objects.create_user( + email="test@example.com", password="password" + ) + write_current_user(self.user) + + self.currency = Currency.objects.create( + code="BRL", name="Real", decimal_places=2, prefix="R$ " + ) + self.group = AccountGroup.objects.create(name="Test Group", owner=self.user) + self.account = Account.objects.create( + name="bradesco-checking", + group=self.group, + currency=self.currency, + owner=self.user, + ) + + def tearDown(self): + delete_current_user() + ImportService.TEMP_DIR = self.original_temp_dir + if os.path.exists(self.test_dir): + shutil.rmtree(self.test_dir) + + def test_import_single_qif_valid_mapping(self): + content = """!Type:Bank +D04/01/2015 +T8069.46 +PMy Payee -> Entity +MNote -> Desc +LOld Cat:New Tag +^ +D05/01/2015 +T-100.00 +PSupermarket +MWeekly shopping +L[Transfer] +^ +""" + filename = "bradesco-checking.qif" + file_path = os.path.join(self.test_dir, filename) + with open(file_path, "w", encoding="utf-8") as f: + f.write(content) + + yaml_config = """ +settings: + file_type: qif + importing: transactions + date_format: "%d/%m/%Y" +mapping: {} +""" + profile = ImportProfile.objects.create( + name="QIF Profile", + yaml_config=yaml_config, + version=ImportProfile.Versions.VERSION_1, + ) + run = ImportRun.objects.create(profile=profile, file_name=filename) + service = ImportService(run) + + service.process_file(file_path) + + self.assertEqual(Transaction.objects.count(), 2) + + # Transaction 1: Income, Category+Tag + t1 = Transaction.objects.get(description="Note -> Desc") + self.assertEqual(t1.amount, Decimal("8069.46")) + self.assertEqual(t1.type, Transaction.Type.INCOME) + self.assertEqual(t1.category.name, "Old Cat") + self.assertTrue(t1.tags.filter(name="New Tag").exists()) + self.assertTrue(t1.entities.filter(name="My Payee -> Entity").exists()) + self.assertEqual(t1.account, self.account) + + # Transaction 2: Expense, Transfer ([Transfer] -> Description) + t2 = Transaction.objects.get(description="Transfer") + self.assertEqual(t2.amount, Decimal("100.00")) + self.assertEqual(t2.type, Transaction.Type.EXPENSE) + self.assertIsNone(t2.category) + self.assertFalse(t2.tags.exists()) + self.assertTrue(t2.entities.filter(name="Supermarket").exists()) + self.assertEqual(t2.description, "Transfer") + + def test_import_deduplication_hash(self): + # Same content twice. Should result in only 1 transaction due to hash deduplication. + content = """!Type:Bank +D04/01/2015 +T100.00 +POK +^ +""" + filename = "bradesco-checking.qif" + file_path = os.path.join(self.test_dir, filename) + with open(file_path, "w", encoding="utf-8") as f: + f.write(content) + + yaml_config = """ +settings: + file_type: qif + importing: transactions + date_format: "%d/%m/%Y" +mapping: {} +""" + profile = ImportProfile.objects.create( + name="QIF Profile", + yaml_config=yaml_config, + version=ImportProfile.Versions.VERSION_1, + ) + run = ImportRun.objects.create(profile=profile, file_name=filename) + service = ImportService(run) + + # First run + service.process_file(file_path) + self.assertEqual(Transaction.objects.count(), 1) + + # Service deletes file after processing, so recreate it for second run + with open(file_path, "w", encoding="utf-8") as f: + f.write(content) + + # Second run - Duplicate content + service.process_file(file_path) + self.assertEqual(Transaction.objects.count(), 1) + + def test_import_strict_error_rollback(self): + # atomic check. + # Transaction 1 valid, Transaction 2 invalid date. + content = """!Type:Bank +D04/01/2015 +T100.00 +POK +^ +DINVALID +T100.00 +PBad +^ +""" + filename = "bradesco-checking.qif" + file_path = os.path.join(self.test_dir, filename) + with open(file_path, "w", encoding="utf-8") as f: + f.write(content) + + yaml_config = """ +settings: + file_type: qif + importing: transactions + date_format: "%d/%m/%Y" + skip_errors: false +mapping: {} +""" + profile = ImportProfile.objects.create( + name="QIF Profile", + yaml_config=yaml_config, + version=ImportProfile.Versions.VERSION_1, + ) + run = ImportRun.objects.create(profile=profile, file_name=filename) + service = ImportService(run) + + with self.assertRaises(Exception) as cm: + service.process_file(file_path) + self.assertEqual(str(cm.exception), "Import failed") + + # Should be 0 transactions because of atomic rollback + self.assertEqual(Transaction.objects.count(), 0) + + def test_import_missing_account(self): + # File with account name that doesn't exist + content = """!Type:Bank +D04/01/2015 +T100.00 +POK +^ +""" + filename = "missing-account.qif" + file_path = os.path.join(self.test_dir, filename) + with open(file_path, "w", encoding="utf-8") as f: + f.write(content) + + yaml_config = """ +settings: + file_type: qif + importing: transactions + date_format: "%d/%m/%Y" +mapping: {} +""" + profile = ImportProfile.objects.create( + name="QIF Profile", + yaml_config=yaml_config, + version=ImportProfile.Versions.VERSION_1, + ) + run = ImportRun.objects.create(profile=profile, file_name=filename) + service = ImportService(run) + + # Should fail because account doesn't exist + with self.assertRaises(Exception) as cm: + service.process_file(file_path) + self.assertEqual(str(cm.exception), "Import failed") + + def test_import_skip_errors(self): + # skip_errors: true. + # Transaction 1 valid, Transaction 2 invalid date. + content = """!Type:Bank +D04/01/2015 +T100.00 +POK +^ +DINVALID +T100.00 +PBad +^ +""" + filename = "bradesco-checking.qif" + file_path = os.path.join(self.test_dir, filename) + with open(file_path, "w", encoding="utf-8") as f: + f.write(content) + + yaml_config = """ +settings: + file_type: qif + importing: transactions + date_format: "%d/%m/%Y" + skip_errors: true +mapping: {} +""" + profile = ImportProfile.objects.create( + name="QIF Profile", + yaml_config=yaml_config, + version=ImportProfile.Versions.VERSION_1, + ) + run = ImportRun.objects.create(profile=profile, file_name=filename) + service = ImportService(run) + + service.process_file(file_path) + + # Should be 1 transaction (valid one) + self.assertEqual(Transaction.objects.count(), 1) + self.assertEqual( + Transaction.objects.first().description, "" + ) # empty desc if no memo diff --git a/app/import_presets/qif_standard/config.yml b/app/import_presets/qif_standard/config.yml new file mode 100644 index 0000000..91ba715 --- /dev/null +++ b/app/import_presets/qif_standard/config.yml @@ -0,0 +1,9 @@ +settings: + file_type: qif + importing: transactions + date_format: "%d/%m/%Y" + skip_errors: true + +mapping: {} + +deduplicate: [] diff --git a/app/import_presets/qif_standard/manifest.json b/app/import_presets/qif_standard/manifest.json new file mode 100644 index 0000000..29f8656 --- /dev/null +++ b/app/import_presets/qif_standard/manifest.json @@ -0,0 +1,7 @@ +{ + "author": "eitchtee", + "description": "Standard QIF Import. Mapping is automatic.", + "schema_version": 1, + "name": "Standard QIF", + "message": "Account is inferred from filename (e.g., 'Checking.qif' -> Account 'Checking').\nYou might need to change the date format to match the date format on your file." +} \ No newline at end of file