AtlatestRepositorytally

tally / tree / scriptsconvert-qbo-xlsx.py

1#!/usr/bin/env python3
2"""Convert a QBO Journal XLSX export to yearly hledger journal files.
3
4Usage:
5 python3 convert-qbo-xlsx.py <journal.xlsx> <output-dir>
6
7Reads the QBO "Journal" report export (one transaction = date row +
8continuation rows + totals row + blank row) and writes one .journal
9file per year to the output directory.
11Requires: openpyxl
12"""
14import sys
15import os
16from collections import defaultdict
17from decimal import Decimal, ROUND_HALF_UP
19try:
20 import openpyxl
21except ImportError:
22 print("Error: openpyxl required. Install with: pip install openpyxl", file=sys.stderr)
23 print(" or: guix shell python python-openpyxl -- python3 convert-qbo-xlsx.py ...", file=sys.stderr)
24 sys.exit(1)
27# ── Account Mapping ──────────────────────────────────────────────
28# Maps QBO account names to hledger account hierarchy.
30ACCOUNT_MAP = {
31 # Assets
32 "QuickBooks Checking Account": "assets:bank:checking",
33 "Payments to deposit": "assets:payments-to-deposit",
34 "Uncategorized Asset": "assets:uncategorized",
35 "Accounts Receivable (A/R)": "assets:accounts-receivable",
37 # Equity
38 "Owner draws": "equity:owner-draws",
39 "Owner investments": "equity:owner-investments",
41 # Income
42 "Services": "income:services",
43 "Services:Consulting Services": "income:services:consulting",
44 "Services:Viewer Memberships": "income:services:memberships",
45 "Uncategorized Income:Advertising Revenue": "income:advertising",
46 "Uncategorized Income:Interest Received": "income:interest",
47 "Uncategorized Income:Viewer Tips": "income:tips",
48 "Sales of Product Income": "income:product-sales",
49 "Other income:Interest earned": "income:interest",
51 # COGS
52 "Cost of goods sold": "expenses:cogs",
53 "Cost of goods sold:Merchandise Printing and Distribution": "expenses:cogs:merchandise",
55 # Expenses
56 "General business expenses": "expenses:general",
57 "General business expenses:Bank fees & service charges": "expenses:bank-fees",
58 "General business expenses:Cloud Hosting": "expenses:hosting",
59 "General business expenses:Analytics Services": "expenses:analytics",
60 "General business expenses:Online Store Hosting": "expenses:hosting:store",
61 "Office expenses:Phone Service": "expenses:phone",
62 "Office expenses:Software & apps": "expenses:software",
63 "Advertising & marketing": "expenses:marketing",
64 "Legal & accounting services": "expenses:legal-accounting",
65 "Contract labor": "expenses:contract-labor",
66 "Business licences": "expenses:licenses",
67 "Utilities:Internet & TV services": "expenses:internet",
68 "QuickBooks Payments Fees": "expenses:payment-processing",
72def map_account(qbo_name):
73 """Map a QBO account name to hledger hierarchy."""
74 mapped = ACCOUNT_MAP.get(qbo_name)
75 if mapped:
76 return mapped
77 # Fallback: sanitize as-is
78 return qbo_name.lower().replace(" ", "-").replace("&", "and")
81# ── Amount Formatting ────────────────────────────────────────────
83def fmt_amount(value):
84 """Format a number as $X,XXX.XX with proper sign handling."""
85 if value is None:
86 return None
87 d = Decimal(str(value)).quantize(Decimal("0.01"), rounding=ROUND_HALF_UP)
88 sign = "-" if d < 0 else ""
89 d = abs(d)
90 # Format with comma separators
91 int_part = int(d)
92 frac_part = d - int_part
93 int_str = f"{int_part:,}"
94 return f"{sign}${int_str}.{str(frac_part)[2:4].ljust(2, '0')}"
97# ── XLSX Parsing ─────────────────────────────────────────────────
99def parse_xlsx(path):
100 """Parse QBO Journal XLSX into a list of transaction dicts.
102 Each transaction dict has:
103 date: str (MM/DD/YYYY)
104 type: str
105 num: str or None
106 name: str or None
107 postings: list of {account, memo, debit, credit}
108 """
109 wb = openpyxl.load_workbook(path, read_only=True)
110 ws = wb.active
111 all_rows = list(ws.iter_rows(values_only=True))
112 wb.close()
114 transactions = []
115 i = 5 # skip header rows (0-indexed: rows 1-5)
117 while i < len(all_rows):
118 row = all_rows[i]
120 # Skip blank rows
121 if row[1] is None:
122 i += 1
123 continue
125 # Skip the final TOTAL row
126 if row[0] == "TOTAL":
127 break
129 # Skip footer/timestamp rows
130 if isinstance(row[0], str) and row[0] and not row[1]:
131 i += 1
132 continue
134 # Date row — start of a new transaction
135 date = row[1]
136 txn_type = row[2] or ""
137 num = row[3]
138 name = row[4] or ""
139 memo_first = row[5] or ""
140 account_first = row[6] or ""
141 debit_first = row[7]
142 credit_first = row[8]
144 # Clean up Num (openpyxl reads as float)
145 if num is not None:
146 num = str(int(num)) if isinstance(num, float) and num == int(num) else str(num)
148 postings = [{
149 "account": account_first,
150 "memo": memo_first,
151 "debit": debit_first,
152 "credit": credit_first,
153 }]
155 # Read continuation rows
156 i += 1
157 while i < len(all_rows):
158 row = all_rows[i]
160 # Blank row or totals row = end of transaction
161 if all(cell is None for cell in row):
162 i += 1
163 break
164 # Totals row: debit and credit both non-None and equal
165 if row[7] is not None and row[8] is not None and row[6] is None:
166 i += 1
167 # Skip the blank row after totals
168 if i < len(all_rows) and all(cell is None for cell in all_rows[i]):
169 i += 1
170 break
172 # Continuation posting row
173 if row[6] is not None:
174 postings.append({
175 "account": row[6] or "",
176 "memo": row[5] or "",
177 "debit": row[7],
178 "credit": row[8],
179 })
180 i += 1
182 transactions.append({
183 "date": date,
184 "type": txn_type,
185 "num": num,
186 "name": name,
187 "postings": postings,
188 })
190 return transactions
193# ── Journal Formatting ───────────────────────────────────────────
195def posting_amount(posting):
196 """Compute signed amount: debit positive, credit negative."""
197 if posting["debit"] is not None and posting["debit"] != 0:
198 return posting["debit"]
199 if posting["credit"] is not None and posting["credit"] != 0:
200 return -posting["credit"]
201 return 0
204def txn_description(txn):
205 """Build description from name and/or first posting memo."""
206 name = txn["name"].strip() if txn["name"] else ""
207 memo = txn["postings"][0]["memo"].strip() if txn["postings"] else ""
209 if name and memo:
210 return f"{name} | {memo}"
211 return name or memo or "Unknown"
214def format_ref_tag(txn):
215 """Build a ref tag for deduplication."""
216 t = txn["type"].replace(" ", "-").lower()
217 if txn["num"]:
218 return f"qbo:{t}:{txn['num']}"
219 # Use date + description hash for uniqueness
220 desc = txn_description(txn).replace(" ", "-").lower()[:40]
221 return f"qbo:{t}:{txn['date'].replace('/', '-')}:{desc}"
224def qbo_date_to_hledger(date_str):
225 """Convert MM/DD/YYYY to YYYY-MM-DD."""
226 parts = date_str.split("/")
227 if len(parts) == 3:
228 return f"{parts[2]}-{parts[0].zfill(2)}-{parts[1].zfill(2)}"
229 return date_str
232def format_transaction(txn):
233 """Format a transaction as hledger journal text."""
234 date = qbo_date_to_hledger(txn["date"])
235 desc = txn_description(txn)
236 code = f" ({txn['num']})" if txn["num"] else ""
237 ref = format_ref_tag(txn)
239 lines = [f"{date} !{code} {desc}"]
240 lines.append(f" ; ref: {ref}")
242 for posting in txn["postings"]:
243 account = map_account(posting["account"])
244 amount = posting_amount(posting)
245 amount_str = fmt_amount(amount)
246 lines.append(f" {account:<45s} {amount_str}")
248 return "\n".join(lines)
251# ── Yearly Splitting ─────────────────────────────────────────────
253def group_by_year(transactions):
254 """Group transactions by year from their date."""
255 by_year = defaultdict(list)
256 for txn in transactions:
257 year = txn["date"].split("/")[-1]
258 by_year[year].append(txn)
259 return dict(by_year)
262def write_yearly_journals(transactions, output_dir):
263 """Write one .journal file per year."""
264 by_year = group_by_year(transactions)
266 os.makedirs(output_dir, exist_ok=True)
268 for year in sorted(by_year.keys()):
269 txns = by_year[year]
270 path = os.path.join(output_dir, f"{year}.journal")
272 with open(path, "w") as f:
273 f.write(f"; QBO Journal export — {year}\n")
274 f.write(f"; Converted from QuickBooks Online on {txns[0]['date'] if txns else 'N/A'}\n")
275 f.write(f"; {len(txns)} transactions\n\n")
277 for i, txn in enumerate(txns):
278 f.write(format_transaction(txn))
279 if i < len(txns) - 1:
280 f.write("\n\n")
281 else:
282 f.write("\n")
284 print(f" {path}: {len(txns)} transactions")
286 # Write a master include file
287 master_path = os.path.join(output_dir, "qbo.journal")
288 with open(master_path, "w") as f:
289 f.write("; QBO imported data — master include file\n")
290 f.write("; Generated from QuickBooks Online Journal export\n\n")
291 for year in sorted(by_year.keys()):
292 f.write(f"include {year}.journal\n")
294 print(f" {master_path}: master include file")
297# ── Main ─────────────────────────────────────────────────────────
299def main():
300 if len(sys.argv) < 3:
301 print(f"Usage: {sys.argv[0]} <journal.xlsx> <output-dir>", file=sys.stderr)
302 sys.exit(1)
304 xlsx_path = sys.argv[1]
305 output_dir = sys.argv[2]
307 if not os.path.exists(xlsx_path):
308 print(f"Error: {xlsx_path} not found", file=sys.stderr)
309 sys.exit(1)
311 print(f"Reading {xlsx_path}...")
312 transactions = parse_xlsx(xlsx_path)
313 print(f"Parsed {len(transactions)} transactions.")
315 print(f"\nWriting yearly journals to {output_dir}/")
316 write_yearly_journals(transactions, output_dir)
317 print(f"\nDone! Validate with: hledger print -f {output_dir}/qbo.journal")
320if __name__ == "__main__":
321 main()