mirror of
https://github.com/escalante29/WealthySmart.git
synced 2026-07-17 17:48:46 +02:00
82a7bfc4c5
read_pdf_upload() bounds reads at 10 MB and requires the %PDF magic before anything reaches pdftotext (SEC-06); rejections land in the per-file errors list with the filename. Parsing now runs via asyncio.to_thread so a slow PDF no longer blocks the event loop for up to 30s (BE-14). Review notes: temp files were already context-managed (BE-09 overstated) and the pension batch already commits atomically (BE-10 overstated). 69 tests green. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
290 lines
9.1 KiB
Python
290 lines
9.1 KiB
Python
from datetime import datetime
|
|
from dateutil.relativedelta import relativedelta
|
|
from typing import Optional
|
|
|
|
import asyncio
|
|
|
|
from fastapi import APIRouter, Depends, Query, UploadFile
|
|
from pydantic import BaseModel
|
|
from sqlmodel import Session, select
|
|
|
|
from app.auth import get_current_user
|
|
from app.uploads import read_pdf_upload
|
|
from app.db import get_session
|
|
from app.models.models import (
|
|
Category,
|
|
Currency,
|
|
MunicipalReceipt,
|
|
MunicipalReceiptRead,
|
|
Transaction,
|
|
TransactionSource,
|
|
TransactionType,
|
|
WaterMeterReading,
|
|
WaterMeterReadingRead,
|
|
)
|
|
from app.services.municipal_receipt_pdf import extract_municipal_receipt
|
|
|
|
router = APIRouter(prefix="/municipal-receipts", tags=["municipal-receipts"])
|
|
|
|
|
|
# --- Response models ---
|
|
|
|
|
|
class MunicipalReceiptDetailRead(MunicipalReceiptRead):
|
|
water_readings: list[WaterMeterReadingRead] = []
|
|
|
|
|
|
class MunicipalReceiptUploadResult(BaseModel):
|
|
imported: int
|
|
updated: int
|
|
errors: list[str]
|
|
receipt: Optional[MunicipalReceiptRead] = None
|
|
|
|
|
|
# --- Helpers ---
|
|
|
|
|
|
def _auto_categorize(merchant: str, session: Session) -> Optional[int]:
|
|
categories = session.exec(select(Category)).all()
|
|
merchant_lower = merchant.lower()
|
|
for cat in categories:
|
|
if cat.auto_match_patterns:
|
|
patterns = [p.strip().lower() for p in cat.auto_match_patterns.split(",")]
|
|
if any(p in merchant_lower for p in patterns if p):
|
|
return cat.id
|
|
return None
|
|
|
|
|
|
def _upsert_receipt(
|
|
session: Session, data: dict, filename: str
|
|
) -> tuple[MunicipalReceipt, bool]:
|
|
"""Insert or update a municipal receipt. Returns (row, is_new)."""
|
|
r = data["receipt"]
|
|
totals = data["totals"]
|
|
receipt_date_str = r["date"]
|
|
# The receipt is issued in month N but covers month N-1
|
|
receipt_dt = datetime.strptime(receipt_date_str, "%Y-%m-%d").date()
|
|
billing_month = receipt_dt - relativedelta(months=1)
|
|
period = billing_month.strftime("%Y-%m")
|
|
|
|
existing = session.exec(
|
|
select(MunicipalReceipt).where(
|
|
MunicipalReceipt.account == r["account"],
|
|
MunicipalReceipt.period == period,
|
|
)
|
|
).first()
|
|
|
|
charges = [
|
|
{"detail": c["detail"], "amount": c.get("amount", 0)}
|
|
for c in data.get("charges", [])
|
|
]
|
|
|
|
fields = dict(
|
|
receipt_date=datetime.strptime(receipt_date_str, "%Y-%m-%d").date(),
|
|
due_date=datetime.strptime(r["due_date"], "%Y-%m-%d").date(),
|
|
period=period,
|
|
account=r["account"],
|
|
finca=r.get("finca", ""),
|
|
holder_name=r.get("account_holder", {}).get("name", ""),
|
|
holder_cedula=r.get("account_holder", {}).get("cedula", ""),
|
|
holder_address=r.get("account_holder", {}).get("address", ""),
|
|
subtotal=totals.get("subtotal", 0),
|
|
interests=totals.get("interests", 0),
|
|
iva=totals.get("iva", 0),
|
|
total=totals.get("total", 0),
|
|
raw_charges=charges,
|
|
source_filename=filename,
|
|
)
|
|
|
|
if existing:
|
|
for k, v in fields.items():
|
|
setattr(existing, k, v)
|
|
session.add(existing)
|
|
# Delete old water readings for this receipt
|
|
old_readings = session.exec(
|
|
select(WaterMeterReading).where(
|
|
WaterMeterReading.receipt_id == existing.id
|
|
)
|
|
).all()
|
|
for rd in old_readings:
|
|
session.delete(rd)
|
|
session.flush()
|
|
return existing, False
|
|
|
|
row = MunicipalReceipt(**fields)
|
|
session.add(row)
|
|
session.flush()
|
|
return row, True
|
|
|
|
|
|
def _insert_water_readings(
|
|
session: Session, receipt: MunicipalReceipt, data: dict
|
|
) -> None:
|
|
"""Insert water meter readings (current + historical) for a receipt."""
|
|
# Current period readings
|
|
for wm in data.get("water_meters", []):
|
|
reading = WaterMeterReading(
|
|
receipt_id=receipt.id,
|
|
meter_id=str(wm["meter_id"]),
|
|
period=wm["period"],
|
|
reading_previous=wm.get("reading_previous", 0),
|
|
reading_current=wm.get("reading_current", 0),
|
|
consumption_m3=wm.get("consumption_m3", 0),
|
|
agua_potable=wm.get("agua_potable", 0),
|
|
serv_ambientales=wm.get("serv_ambientales", 0),
|
|
alcant_sanitario=wm.get("alcant_sanitario", 0),
|
|
iva=wm.get("iva", 0),
|
|
is_historical=False,
|
|
)
|
|
session.add(reading)
|
|
|
|
# Historical consumption entries
|
|
for hc in data.get("historical_consumption", []):
|
|
period = hc["period"]
|
|
meter_id = str(hc["meter_id"])
|
|
# Upsert: check if this historical entry already exists
|
|
existing = session.exec(
|
|
select(WaterMeterReading).where(
|
|
WaterMeterReading.meter_id == meter_id,
|
|
WaterMeterReading.period == period,
|
|
WaterMeterReading.is_historical == True, # noqa: E712
|
|
)
|
|
).first()
|
|
if existing:
|
|
existing.consumption_m3 = hc.get("consumption_m3", 0)
|
|
session.add(existing)
|
|
else:
|
|
session.add(
|
|
WaterMeterReading(
|
|
receipt_id=receipt.id,
|
|
meter_id=meter_id,
|
|
period=period,
|
|
consumption_m3=hc.get("consumption_m3", 0),
|
|
is_historical=True,
|
|
)
|
|
)
|
|
|
|
|
|
def _ensure_transaction(
|
|
session: Session, receipt: MunicipalReceipt
|
|
) -> None:
|
|
"""Create a budget Transaction for this receipt if one doesn't exist."""
|
|
reference = f"municipal-{receipt.account}-{receipt.period}"
|
|
existing = session.exec(
|
|
select(Transaction).where(Transaction.reference == reference)
|
|
).first()
|
|
if existing:
|
|
# Update amount in case receipt was re-uploaded with corrections
|
|
existing.amount = receipt.total
|
|
session.add(existing)
|
|
return
|
|
|
|
category_id = _auto_categorize("municipalidad", session)
|
|
tx = Transaction(
|
|
amount=receipt.total,
|
|
currency=Currency.CRC,
|
|
merchant="Municipalidad de Belén",
|
|
date=datetime.combine(receipt.receipt_date, datetime.min.time()),
|
|
transaction_type=TransactionType.COMPRA,
|
|
source=TransactionSource.TRANSFER,
|
|
reference=reference,
|
|
category_id=category_id,
|
|
notes=f"Recibo municipal {receipt.period}",
|
|
)
|
|
session.add(tx)
|
|
|
|
|
|
# --- Endpoints ---
|
|
|
|
|
|
@router.post("/upload", response_model=MunicipalReceiptUploadResult)
|
|
async def upload_municipal_receipt(
|
|
file: UploadFile,
|
|
session: Session = Depends(get_session),
|
|
_user: str = Depends(get_current_user),
|
|
):
|
|
filename = file.filename or "unknown.pdf"
|
|
errors: list[str] = []
|
|
|
|
try:
|
|
pdf_bytes = await read_pdf_upload(file)
|
|
# pdftotext runs as a subprocess; keep the event loop free
|
|
data = await asyncio.to_thread(extract_municipal_receipt, pdf_bytes, filename)
|
|
except ValueError as e:
|
|
return MunicipalReceiptUploadResult(imported=0, updated=0, errors=[str(e)])
|
|
except Exception as e:
|
|
return MunicipalReceiptUploadResult(
|
|
imported=0, updated=0, errors=[f"{filename}: {e}"]
|
|
)
|
|
|
|
receipt, is_new = _upsert_receipt(session, data, filename)
|
|
_insert_water_readings(session, receipt, data)
|
|
_ensure_transaction(session, receipt)
|
|
|
|
session.commit()
|
|
session.refresh(receipt)
|
|
|
|
return MunicipalReceiptUploadResult(
|
|
imported=1 if is_new else 0,
|
|
updated=0 if is_new else 1,
|
|
errors=errors,
|
|
receipt=MunicipalReceiptRead.model_validate(receipt),
|
|
)
|
|
|
|
|
|
@router.get("/", response_model=list[MunicipalReceiptRead])
|
|
def list_receipts(
|
|
session: Session = Depends(get_session),
|
|
_user: str = Depends(get_current_user),
|
|
):
|
|
rows = session.exec(
|
|
select(MunicipalReceipt).order_by(
|
|
MunicipalReceipt.receipt_date.desc() # type: ignore[union-attr]
|
|
)
|
|
).all()
|
|
return rows
|
|
|
|
|
|
@router.get("/water-consumption", response_model=list[WaterMeterReadingRead])
|
|
def get_water_consumption(
|
|
months: int = Query(default=24, ge=1, le=120),
|
|
session: Session = Depends(get_session),
|
|
_user: str = Depends(get_current_user),
|
|
):
|
|
rows = session.exec(
|
|
select(WaterMeterReading)
|
|
.where(WaterMeterReading.is_historical == False) # noqa: E712
|
|
.order_by(
|
|
WaterMeterReading.period.asc(), # type: ignore[union-attr]
|
|
WaterMeterReading.meter_id.asc(), # type: ignore[union-attr]
|
|
)
|
|
.limit(months * 3) # up to 3 meters per month
|
|
).all()
|
|
return rows
|
|
|
|
|
|
@router.get("/{receipt_id}", response_model=MunicipalReceiptDetailRead)
|
|
def get_receipt_detail(
|
|
receipt_id: int,
|
|
session: Session = Depends(get_session),
|
|
_user: str = Depends(get_current_user),
|
|
):
|
|
receipt = session.get(MunicipalReceipt, receipt_id)
|
|
if not receipt:
|
|
from fastapi import HTTPException
|
|
|
|
raise HTTPException(status_code=404, detail="Receipt not found")
|
|
|
|
readings = session.exec(
|
|
select(WaterMeterReading).where(
|
|
WaterMeterReading.receipt_id == receipt_id
|
|
)
|
|
).all()
|
|
|
|
return MunicipalReceiptDetailRead(
|
|
**MunicipalReceiptRead.model_validate(receipt).model_dump(),
|
|
water_readings=[
|
|
WaterMeterReadingRead.model_validate(r) for r in readings
|
|
],
|
|
)
|