109 lines
3.6 KiB
Python
109 lines
3.6 KiB
Python
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
|