app/statements.py
Rama main · réplica de lectura del repositorio original.
"""Dividend-statement PDF reader.
Parses registry statements with pdfplumber, entirely locally. Extraction is
best-effort by design: the preview form shows every parsed field for correction
before anything is committed, so an unrecognised layout costs a minute of
typing rather than a bad row.
Layouts are data, not code — the templates live in `app/formats/statements/`
and are read by `docformats.py`, so adding a registry needs no Python
(decisions.md #64).
Captured per statement: instrument, payment date, net cash, franked amount and
franking credits, and DRP units and price where present — committed as a
Dividend, plus a linked DRP Trade when units were allotted.
"""
from __future__ import annotations
import datetime as dt
import io
import re
from collections import Counter
from dataclasses import dataclass, field
from pathlib import Path
from sqlalchemy import select
from sqlalchemy.orm import Session
from . import docformats, money, ocr, queries
from .models import Dividend, Instrument
@dataclass
class StatementRow:
"""One fund's line on a combined DRP advice."""
ticker: str
cash_amount: str # the distribution for the period ("Reinvestment Amount")
drp_price: str
drp_units: str # "0" = nothing allotted, cash carried forward
units_held: str # per the statement — sanity-check against the DB
db_units: str | None = None
carry_note: str = ""
status: str = "new" # new / duplicate / unknown-instrument
# A fund held here on more than one exchange: which one the advice means,
# it does not say, so the row offers each, as (form value, label).
choices: list[tuple[str, str]] = field(default_factory=list)
# Per row, so a date the advice did not give can be typed for each fund.
payment_date: str = ""
listing: str = "" # the one of `choices` picked, as TICKER:EXCHANGE
ticked: bool = True
problem: str = "" # why saving it was refused
@dataclass
class ParsedStatement:
text_sample: str
# Which template read this document, and its full extracted text. Both are
# for the person looking at a field that came back empty: without them the
# only feedback is a blank box, and there is no way to tell a layout this
# app has never seen from a label it looked for and did not find.
template_key: str | None = None
template_name: str | None = None
full_text: str = ""
ticker: str | None = None
payment_date: dt.date | None = None
net_amount: str | None = None
franked_amount: str | None = None
franking_credits: str | None = None
drp_units: str | None = None
drp_price: str | None = None
# A trust's period end, typed in: no statement template reads one yet.
period_end: str | None = None
# Foreign tax withheld, typed in, for the same reason.
withholding_tax: str | None = None
rows: list[StatementRow] | None = None # combined multi-fund advice
# This document was read from pixels, not from a text layer. Carried all
# the way to the preview because OCR misreads digits and these are money —
# the person checking the form needs to know to check it harder.
used_ocr: bool = False
# Why nothing came out, when nothing came out. "Statement not recognised"
# is true and useless — it sends the reader to their templates when the
# answer is that the PDF has no text in it.
problem: str = ""
# A text layer this short is page furniture — a producer string, a page number,
# a stray header — not a statement. The number sits above what a scan wrapper
# typically carries and well below the shortest real advice; there is a
# parametrised test pinning both edges.
SCANNED_UNDER_CHARS = 60
def looks_scanned(text: str) -> bool:
"""Whether this PDF's text layer is worth nothing.
Not `if not text`: a scan is usually wrapped in a PDF that carries a few
characters of metadata, so "not empty" is a long way from "has the
statement in it".
"""
return len(" ".join(text.split())) < SCANNED_UNDER_CHARS
@dataclass
class Extracted:
text: str
used_ocr: bool = False
problem: str = ""
# A dividend statement is a page or two, a combined advice a handful. Past this
# the upload is REFUSED, not truncated: reading every page ties up a worker,
# and reading only the first ones would drop rows without saying so (#46).
MAX_PAGES = 30
class TooManyPages(Exception):
def __init__(self, pages: int):
super().__init__(pages)
self.pages = pages
class NotAPdf(Exception):
"""The upload is not a PDF pdfplumber can open."""
def _pdf_text(data: bytes) -> str:
# Imported here, not at module scope: uploading a statement is a rare,
# deliberate act, and this module is imported at startup by main.py.
# pdfplumber costs ~8 MiB in the container — small next to yfinance, but it
# is the same principle, and this is the only function that touches it.
import pdfplumber
from pdfplumber.utils.exceptions import PdfminerException
try:
pdf = pdfplumber.open(io.BytesIO(data))
except PdfminerException:
raise NotAPdf from None
with pdf:
# Counted before any page is read: the point is not doing the work.
if len(pdf.pages) > MAX_PAGES:
raise TooManyPages(len(pdf.pages))
return "\n".join(page.extract_text() or "" for page in pdf.pages)
def extract(data: bytes, *, ocr_enabled: bool = True) -> Extracted:
"""The document's text, from its text layer or — failing that — from OCR.
OCR is the fallback and never the first choice: running it on a PDF that
already has a text layer is slower and strictly worse, because it
re-derives from pixels what the file already states exactly.
Never raises. Every way this can fail produces a sentence for the person
who uploaded the file, because all of them are things about *their file*
rather than faults in the app.
"""
try:
text = _pdf_text(data)
except NotAPdf:
return Extracted(text="", problem=(
"This file is not a PDF, or not one that can be opened. Upload the "
"statement as the PDF your registry sent."))
except TooManyPages as exc:
return Extracted(text="", problem=(
f"This PDF has {exc.pages} pages; statements are read up to "
f"{MAX_PAGES}. Split it and upload the pages you need."))
if not looks_scanned(text):
return Extracted(text=text)
if not ocr_enabled:
return Extracted(text="", problem=(
"This PDF has no text layer, so it is probably a scan. Reading "
"scans is turned off on this install (imports.ocr.enabled)."))
if not ocr.available():
return Extracted(text="", problem=(
"This PDF has no text layer, so it is probably a scan. Reading one "
"needs OCR, and the tesseract program is not installed here. "
"Install tesseract-ocr, or enter the figures by hand below."))
try:
scanned = ocr.read_pdf(data)
except ocr.OcrFailed as exc:
return Extracted(text="", problem=(
f"This PDF has no text layer and OCR could not read it: {exc}"))
# A DIFFERENT question from `looks_scanned`, deliberately. That one asks
# whether a text layer is worth using when an exact one might exist; this
# one asks whether OCR found anything at all, and there is no better
# alternative to fall back to. Holding OCR output to the same length bar
# would throw away a terse but perfectly good read.
if not scanned.strip():
return Extracted(text="", problem=(
"This PDF has no text layer, and OCR found no words in it. Check "
"it is the right file, and that the scan is the right way up."))
return Extracted(text=scanned, used_ocr=True)
def extract_text(data: bytes) -> str:
"""The text alone. Kept because it is the simple question, and because the
statement designer and several tests ask exactly it."""
return extract(data).text
def own_instruments(session: Session) -> list[Instrument]:
"""This portfolio's instruments, in ticker order. A statement is for
something it holds or held (decisions.md #139): another portfolio's is not
it, nor the same ticker on another exchange, and the catalogue's own order
is no order at all."""
ids = queries.portfolio_instrument_ids(session)
return list(session.scalars(select(Instrument).where(Instrument.id.in_(ids))
.order_by(Instrument.ticker, Instrument.id)))
def choices(session: Session) -> list[tuple[str, str]]:
"""This portfolio's instruments as a form offers them, as (value, label):
the ticker, or each listing of one held on two exchanges, as
TICKER:EXCHANGE — the form `queries.held_as` reads."""
own = own_instruments(session)
twice = {t for t, n in Counter(i.ticker for i in own).items() if n > 1}
return [_choice(i) if i.ticker in twice else (i.ticker, i.ticker) for i in own]
def _choice(inst: Instrument) -> tuple[str, str]:
return f"{inst.ticker}:{inst.exchange}", f"{inst.ticker} ({inst.exchange})"
def _rows_from(
values: dict, payment_date: dt.date | None, session: Session
) -> list[StatementRow] | None:
"""Turn a template's extracted rows into the multi-fund preview.
Only templates that declare a `rows:` shape produce these; a single-payment
statement returns None and the caller falls back to the one-off form.
"""
matches = values.get("rows") or []
if not matches:
return None
rows = []
for row_values in matches:
brought = row_values["brought_forward"]
carried = row_values["carried_forward"]
rows.append(StatementRow(
ticker=row_values["ticker"],
cash_amount=row_values["amount"].replace(",", ""),
drp_price=row_values["drp_price"],
drp_units=row_values["allotted"],
units_held=row_values["units_held"].replace(",", ""),
carry_note=f"DRP carry: brought fwd {brought}, carried fwd {carried}",
payment_date=payment_date.isoformat() if payment_date else "",
))
annotate_rows(session, rows)
return rows
def annotate_rows(session: Session, rows: list[StatementRow]) -> None:
"""Each row against this portfolio's holdings: the listings to choose
from, the units held here, and whether it is recorded already. Asked
again when the rows come back from the form, so they are text."""
from .models import Trade
own: dict[str, list[Instrument]] = {}
for inst in own_instruments(session):
own.setdefault(inst.ticker, []).append(inst)
for row in rows:
row.status, row.choices, row.db_units = "new", [], None
listings = own.get(row.ticker, [])
if not listings:
row.status = "unknown-instrument"
continue
if len(listings) > 1:
row.choices = [_choice(i) for i in listings]
continue
inst = listings[0]
trades = session.scalars(select(Trade).where(Trade.instrument_id == inst.id)).all()
units = sum(t.quantity for t in trades if t.type in ("buy", "drp")) - sum(
t.quantity for t in trades if t.type == "sell"
)
row.db_units = f"{units.normalize():f}" if trades else "0"
try:
date = dt.date.fromisoformat(row.payment_date)
# Read as the save reads it. A figure no column can hold is never
# compared with one: Postgres refuses 1E+999999 outright, and one
# such row took the whole advice down.
amount = money.parse(row.cash_amount.strip().replace(",", "").replace("$", ""),
Dividend.cash_amount)
except ValueError:
continue
if duplicate_of(session, inst, date, amount):
row.status = "duplicate"
def parse_statement(data: bytes, session: Session,
templates_dir: Path | None = None,
*, ocr_enabled: bool = True,
wanted_template: str | None = None) -> ParsedStatement:
read = extract(data, ocr_enabled=ocr_enabled)
text = read.text
parsed = ParsedStatement(text_sample="\n".join(text.splitlines()[:25]),
full_text=text, used_ocr=read.used_ocr,
problem=read.problem)
# Instrument: match against what we track — ticker first, then name words.
instruments = own_instruments(session)
upper = f" {text.upper()} "
for inst in instruments:
if re.search(rf"\b{re.escape(inst.ticker)}\b", upper):
parsed.ticker = inst.ticker
break
if parsed.ticker is None:
for inst in instruments:
if inst.name and inst.name.upper() in upper:
parsed.ticker = inst.ticker
break
# `wanted_template` is somebody choosing on the upload form; guessing is the
# fallback. Without the choice, an installed template with no marker could
# never beat the generic one, so it silently never ran.
template = docformats.pick(
docformats.load_statement_templates(templates_dir), text,
wanted=wanted_template)
if template is None: # no templates installed at all
return parsed
parsed.template_key = template.key
parsed.template_name = template.name
values = template.extract(text)
parsed.payment_date = values.get("payment_date")
parsed.rows = _rows_from(values, parsed.payment_date, session)
if parsed.rows:
return parsed # combined advice: per-row preview supersedes the single form
# The preview form is text inputs, so values are handed over as the strings
# they will be edited as. `_text` keeps the trailing zeros a Decimal would
# otherwise drop — "104.70" must not become "104.7" in a money field.
parsed.net_amount = _text(values.get("net_amount"))
parsed.franked_amount = _text(values.get("franked_amount"))
parsed.franking_credits = _text(values.get("franking_credits"))
parsed.drp_units = _text(values.get("drp_units"))
parsed.drp_price = _text(values.get("drp_price"))
return parsed
def _text(value) -> str | None:
return None if value is None else str(value)
def duplicate_of(session: Session, instrument: Instrument, date: dt.date, amount) -> Dividend | None:
return session.scalars(
select(Dividend).where(
Dividend.instrument_id == instrument.id,
Dividend.date == date,
Dividend.cash_amount == amount,
)
).first()