Spaces:
Running
Running
File size: 4,126 Bytes
0ebef7d | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 | """POST /convert — PDF in, reconciled statement data out (JSON or .xlsx).
Stateless: the uploaded PDF is read into memory, processed in an isolated temp
dir, and deleted before the response is built. Nothing is persisted.
"""
from __future__ import annotations
import io
import logging
from fastapi import APIRouter, File, Header, HTTPException, Query, UploadFile
from fastapi.responses import JSONResponse, StreamingResponse
from openpyxl import Workbook
from app.core.config import API_KEY, MAX_UPLOAD_BYTES
from app.core.files import temp_pdf
from app.engine.converter import BankStatementConverter, ExtractedStatement
router = APIRouter()
logger = logging.getLogger("ebs.convert")
XLSX_MEDIA = "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet"
def _is_pdf(file: UploadFile) -> bool:
if (file.content_type or "").lower() in {"application/pdf", "application/x-pdf"}:
return True
return (file.filename or "").lower().endswith(".pdf")
def _statement_dict(st: ExtractedStatement) -> dict:
return {
"account": st.account,
"accountHolder": st.account_holder,
"period": st.period,
"openingBalance": st.opening_balance,
"closingBalance": st.closing_balance,
"reconciles": st.reconciles,
"reconcileStatus": st.reconcile_status,
"runningBalanceOk": st.running_balance_ok,
"statedSummaryOk": st.stated_summary_ok,
"reconciliationDetail": st.reconciliation_detail,
"transactions": [
{
"date": t.date,
"description": t.description,
"debit": t.debit,
"credit": t.credit,
"balance": t.balance,
}
for t in st.transactions
],
}
def _xlsx_bytes(statements: list[ExtractedStatement]) -> io.BytesIO:
wb = Workbook()
wb.remove(wb.active)
for n, st in enumerate(statements, 1):
ws = wb.create_sheet(f"Account{n}")
ws.append(["Date", "Description", "Debit", "Credit", "Balance"])
for t in st.transactions:
ws.append([t.date, t.description, t.debit, t.credit, t.balance])
buf = io.BytesIO()
wb.save(buf)
buf.seek(0)
return buf
@router.post("/convert")
async def convert(
file: UploadFile = File(...),
format: str = Query("json", pattern="^(json|xlsx)$"),
x_api_key: str | None = Header(default=None),
):
if API_KEY and x_api_key != API_KEY:
raise HTTPException(status_code=401, detail="Unauthorized.")
if not _is_pdf(file):
raise HTTPException(status_code=415, detail="Only PDF files are accepted.")
data = await file.read()
if not data:
raise HTTPException(status_code=400, detail="Empty file.")
if len(data) > MAX_UPLOAD_BYTES:
mb = MAX_UPLOAD_BYTES // (1024 * 1024)
raise HTTPException(status_code=413, detail=f"File too large (max {mb} MB).")
try:
with temp_pdf(data) as pdf_path:
statements = BankStatementConverter(pdf_path).extract()
except Exception:
# Bad user input (corrupt / password-protected / not a real PDF) must
# surface as 4xx, never a 500. Logged for observability.
logger.exception("extraction failed for upload %r", file.filename)
raise HTTPException(
status_code=422,
detail="Couldn't read this PDF. It may be corrupt, password-protected, or not a valid PDF.",
)
if not statements or all(not s.transactions for s in statements):
raise HTTPException(
status_code=422,
detail="Couldn't read any transactions. If this is a scanned image, a clearer scan may help.",
)
if format == "xlsx":
return StreamingResponse(
_xlsx_bytes(statements),
media_type=XLSX_MEDIA,
headers={"Content-Disposition": 'attachment; filename="statement.xlsx"'},
)
return JSONResponse(
{
"count": sum(len(s.transactions) for s in statements),
"statements": [_statement_dict(s) for s in statements],
}
)
|