"""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()