bootstrap

sync/excel_sync.py

@@ -0,0 +1,108 @@
+from pathlib import Path
+from uuid import UUID
+
+from openpyxl import Workbook, load_workbook
+
+from cashflow_model import Account, Asset, FinancialModel, Liability, RecurringCashflow, Transaction
+
+
+class SyncError(Exception):
+    pass
+
+
+_SHEET_CONFIG = {
+    "Accounts": {
+        "fields": ["id", "name", "currency", "balance"],
+        "cls": Account,
+    },
+    "Transactions": {
+        "fields": ["id", "date", "account", "category", "amount", "description"],
+        "cls": Transaction,
+    },
+    "Recurring": {
+        "fields": ["id", "start_date", "end_date", "frequency", "amount", "category"],
+        "cls": RecurringCashflow,
+    },
+    "Assets": {
+        "fields": ["id", "name", "value", "growth_rate"],
+        "cls": Asset,
+    },
+    "Liabilities": {
+        "fields": ["id", "name", "balance", "interest", "payment"],
+        "cls": Liability,
+    },
+}
+
+
+class ExcelSync:
+    def import_model(self, path: str | Path) -> FinancialModel:
+        path = Path(path)
+        if not path.exists():
+            raise SyncError(f"File not found: {path}")
+
+        wb = load_workbook(path, read_only=True, data_only=True)
+        model = FinancialModel()
+
+        for sheet_name, config in _SHEET_CONFIG.items():
+            if sheet_name not in wb.sheetnames:
+                continue
+            ws = wb[sheet_name]
+            rows = list(ws.iter_rows(values_only=True))
+            if len(rows) < 2:
+                continue
+
+            headers = [str(h).strip().lower() if h else "" for h in rows[0]]
+            for row in rows[1:]:
+                if not any(v is not None for v in row):
+                    continue
+                data = {}
+                for i, header in enumerate(headers):
+                    val = row[i] if i < len(row) else None
+                    if val is not None:
+                        data[header] = str(val) if not isinstance(val, (int, float)) else val
+                self._add_to_model(model, sheet_name, data)
+
+        wb.close()
+        return model
+
+    def export_model(self, model: FinancialModel, path: str | Path) -> None:
+        path = Path(path)
+        wb = Workbook()
+        wb.remove(wb.active)
+
+        collections = {
+            "Accounts": model.accounts,
+            "Transactions": model.transactions,
+            "Recurring": model.recurring,
+            "Assets": model.assets,
+            "Liabilities": model.liabilities,
+        }
+
+        for sheet_name, items in collections.items():
+            config = _SHEET_CONFIG[sheet_name]
+            ws = wb.create_sheet(title=sheet_name)
+            ws.append(config["fields"])
+            for item in items:
+                row = [
+                    str(getattr(item, f)) if isinstance(getattr(item, f), UUID)
+                    else getattr(item, f)
+                    for f in config["fields"]
+                ]
+                ws.append(row)
+
+        wb.save(path)
+
+    def _add_to_model(self, model: FinancialModel, sheet_name: str, data: dict) -> None:
+        try:
+            if sheet_name == "Accounts":
+                model.accounts.append(Account.from_dict(data))
+            elif sheet_name == "Transactions":
+                model.transactions.append(Transaction.from_dict(data))
+            elif sheet_name == "Recurring":
+                model.recurring.append(RecurringCashflow.from_dict(data))
+            elif sheet_name == "Assets":
+                model.assets.append(Asset.from_dict(data))
+            elif sheet_name == "Liabilities":
+                model.liabilities.append(Liability.from_dict(data))
+        except Exception as e:
+            raise SyncError(f"Failed to parse row in {sheet_name}: {e}") from e