detailtrialbalance/trial_balance_parser.py
2026-07-01 16:04:18 -04:00

351 lines
No EOL
10 KiB
Python

import re
import io
import pandas as pd
import pdfplumber
import openpyxl
from openpyxl.styles import Font
from openpyxl.worksheet.table import Table, TableStyleInfo
from openpyxl.utils import get_column_letter
money_re = re.compile(r"-?\d{1,3}(?:,\d{3})*(?:\.\d{2})")
acct_re = re.compile(r"^(?P<account>\d{2}\.\d{5}\.\d{5})\s+-\s+(?P<fund>\d{2})\s+(?P<name>.+)$")
journal_re = re.compile(
r"^(?P<journal>[A-Z][A-Z0-9/]*)\s+"
r"(?P<date>\d{2}/\d{2}/\d{2})(?P<bch>\d{1,2})?\s+"
r"(?P<entry>\d+(?:-\d+)?)\b"
)
def parse_money(s):
if not s:
return None
m = money_re.search(s)
return float(m.group(0).replace(",", "")) if m else None
def is_page_header_noise(s):
s = " ".join(s.split()).upper()
return (
("FISCAL CALENDAR" in s or "FIRST CALENDAR" in s)
and ("ROUND MONEY" in s or "TRIAL" in s)
)
#change here
def clean_description(desc):
if not desc or is_page_header_noise(desc):
return ""
patterns = [
r"FISCAL\s+CALENDAR\s+[A-Z]{3,6}\s*,?\s*[A-Z]{3}\s+20\d{2}",
r"FIRST\s+CALENDAR\s+[A-Z]{3,6}\s*,?\s*[A-Z]{3}\s+20\d{2}",
r"ROUND\s+MONEY\s*:\s*\d+\.\d+",
r"\bTRIAL\b",
]
for p in patterns:
desc = re.sub(p, "", desc, flags=re.IGNORECASE)
return " ".join(desc.split())
def parse_trial_balance_pdf(pdf_file):
"""
Parses a MEDITECH detail trial balance PDF.
Accepts either:
- a file path string
- a Streamlit uploaded file
- a file-like object
Returns:
- df_txn
- df_acct
- title
"""
if hasattr(pdf_file, "seek"):
pdf_file.seek(0)
transactions = []
acct_summaries = []
report_title_parts = {
"fiscal": None,
"period": None,
"trial": None,
}
with pdfplumber.open(pdf_file) as pdf:
for page in pdf.pages:
text = page.extract_text(layout=True) or ""
lines = text.splitlines()
for ln in lines:
s = ln.strip()
if report_title_parts["fiscal"] is None and "Fiscal Calendar" in s:
report_title_parts["fiscal"] = " ".join(s.split())
if report_title_parts["period"] is None and re.search(r"\b[A-Z]{3}\s+20\d{2}\b", s):
report_title_parts["period"] = " ".join(s.split())
if report_title_parts["trial"] is None and s.upper() == "TRIAL":
report_title_parts["trial"] = "TRIAL"
txn_header = None
acct_header = None
for i, ln in enumerate(lines):
if "JOURNAL" in ln and "DESCRIPTION" in ln and "DEBITS" in ln:
txn_header = i
if "ACCOUNT" in ln and "OPEN" in ln and "CLOSE" in ln:
acct_header = i
deb_pos = cred_pos = desc_pos = None
open_pos = adeb_pos = acredit_pos = change_pos = close_pos = None
if txn_header is not None:
h = lines[txn_header]
deb_pos = h.find("DEBITS")
cred_pos = h.find("CREDITS", deb_pos)
desc_pos = h.find("DESCRIPTION", cred_pos)
if acct_header is not None:
h = lines[acct_header]
open_pos = h.find("OPEN")
adeb_pos = h.find("DEBITS", open_pos)
acredit_pos = h.find("CREDITS", adeb_pos)
change_pos = h.find("CHANGE", acredit_pos)
close_pos = h.find("CLOSE", change_pos)
current_account = None
current_account_name = None
pending_totals = False
last_txn = None
i = 0
while i < len(lines):
ln = lines[i]
s = ln.strip()
if not s or is_page_header_noise(s):
i += 1
continue
m = acct_re.match(s)
if m:
current_account = m.group("account")
current_account_name = m.group("name").strip()
pending_totals = True
last_txn = None
i += 1
continue
if pending_totals and money_re.search(s):
nums = [float(x.replace(",", "")) for x in money_re.findall(s)]
nums += [None] * 5
acct_summaries.append(
{
"ACCOUNT_NAME": current_account_name,
"ACCOUNT": current_account,
"OPEN": nums[0],
"TOTAL_DEBITS": nums[1],
"TOTAL_CREDITS": nums[2],
"NET_CHANGE": nums[3],
"CLOSE": nums[4],
}
)
pending_totals = False
i += 1
continue
m = journal_re.match(s)
if m and current_account:
journal = m.group("journal")
date = m.group("date")
bch = int(m.group("bch")) if m.group("bch") else None
entry = m.group("entry")
debit = None
credit = None
desc = ""
if deb_pos is not None and cred_pos is not None:
debit = parse_money(ln[deb_pos:cred_pos])
credit = parse_money(ln[cred_pos:desc_pos])
desc = ln[desc_pos:].strip()
else:
after = s[m.end():].strip()
nums = money_re.findall(after)
if nums:
debit = float(nums[0].replace(",", ""))
desc = money_re.sub("", after).strip()
row = {
"ACCOUNT_NAME": current_account_name,
"ACCOUNT": current_account,
"JOURNAL": journal,
"DATE": date,
"BCH": bch,
"ENTRY": entry,
"DEBITS": debit,
"CREDITS": credit,
"DESCRIPTION": clean_description(desc),
}
transactions.append(row)
last_txn = row
i += 1
continue
if last_txn:
add = clean_description(s)
if add:
last_txn["DESCRIPTION"] += " " + add
i += 1
continue
i += 1
title = ", ".join(v for v in report_title_parts.values() if v)
df_txn = pd.DataFrame(transactions)
df_acct = pd.DataFrame(acct_summaries)
return df_txn, df_acct, title
def build_excel_file(df_txn, df_acct, title):
"""
Builds a formatted Excel workbook in memory.
Returns:
- bytes suitable for Streamlit download_button
"""
import io
import pandas as pd
import openpyxl
from openpyxl.styles import Font
from openpyxl.worksheet.table import Table, TableStyleInfo
from openpyxl.utils import get_column_letter
txn_columns = [
"ACCOUNT_NAME",
"ACCOUNT",
"JOURNAL",
"DATE",
"BCH",
"ENTRY",
"DEBITS",
"CREDITS",
"DESCRIPTION",
]
acct_columns = [
"ACCOUNT_NAME",
"ACCOUNT",
"OPEN",
"TOTAL_DEBITS",
"TOTAL_CREDITS",
"NET_CHANGE",
"CLOSE",
]
if df_txn is None or df_txn.empty:
df_txn = pd.DataFrame(columns=txn_columns)
else:
df_txn = df_txn.reindex(columns=txn_columns)
if df_acct is None or df_acct.empty:
df_acct = pd.DataFrame(columns=acct_columns)
else:
df_acct = df_acct.reindex(columns=acct_columns)
output = io.BytesIO()
with pd.ExcelWriter(output, engine="openpyxl") as writer:
df_txn.to_excel(writer, sheet_name="Transactions", index=False, startrow=2)
df_acct.to_excel(writer, sheet_name="Account Summary", index=False, startrow=2)
wb = writer.book
for sheet_name in ["Transactions", "Account Summary"]:
ws = wb[sheet_name]
ws["A1"] = title or "MEDITECH Detail Trial Balance"
ws.merge_cells(
start_row=1,
start_column=1,
end_row=1,
end_column=ws.max_column,
)
ws["A1"].font = Font(bold=True)
ws.freeze_panes = "A4"
header_row = 3
# Only add Excel table if there is at least one data row
if ws.max_row > header_row:
end_col = get_column_letter(ws.max_column)
table_ref = f"A{header_row}:{end_col}{ws.max_row}"
table_name = sheet_name.replace(" ", "") + "Table"
tab = Table(displayName=table_name, ref=table_ref)
style = TableStyleInfo(
name="TableStyleMedium9",
showFirstColumn=False,
showLastColumn=False,
showRowStripes=True,
showColumnStripes=False,
)
tab.tableStyleInfo = style
ws.add_table(tab)
for col in ws.columns:
max_length = 0
col_letter = get_column_letter(col[0].column)
for cell in col:
if cell.value:
max_length = max(max_length, len(str(cell.value)))
ws.column_dimensions[col_letter].width = min(max_length + 2, 60)
output.seek(0)
return output.getvalue()
def parse_trial_balance_to_excel(pdf_file):
"""
Main function for Streamlit or command-line use.
Returns:
- excel_bytes
- df_txn
- df_acct
- title
"""
df_txn, df_acct, title = parse_trial_balance_pdf(pdf_file)
excel_bytes = build_excel_file(df_txn, df_acct, title)
return excel_bytes, df_txn, df_acct, title