"""Broker CSV import: parse SelfWealth/CommSec exports into candidate trades.

Neither broker offers an API, so the flow is: download the CSV → upload it on
/imports-exports → review the parsed preview
(duplicates and unknown tickers are flagged) → commit. Column layouts are
format files: shipped in app/formats/brokers, or written by the in-app
mapper (Map a new broker CSV), so a changed export format is mapped again.

Parsing is strict on purpose: an unrecognised header or row fails loudly with
the offending text rather than guessing — first runs against a new broker
export are expected to need a mapping tweak.
"""

from __future__ import annotations

import csv
import datetime as dt
import io
import re
from dataclasses import dataclass, field
from decimal import Decimal
from itertools import accumulate

from sqlalchemy import select
from sqlalchemy.orm import Session

from . import money, queries
from .markettime import to_market
from .models import (
    CorporateAction,
    Instrument,
    Trade,
    exchange_problem,
    ticker_problem,
    trade_order,
)
from .pricefeed import MARKETS
from .settings import BrokerFormat, ImportSettings
from .tenancy import owned

# CommSec transactions export: Date,Reference,Details,Debit($),Credit($),Balance($)
# with trade rows like "B 50 ALPHA @ 94.500000" / "S 174 ACME @ 8.550000".
COMMSEC_DETAILS = re.compile(
    r"^\s*(?P<action>[BS])\s+(?P<units>[\d,]+)\s+(?P<ticker>[A-Z0-9]{2,6})\s+@\s+(?P<price>[\d,.]+)\s*$"
)


# Broker exports very commonly pad a date with a zero time —
# "2024-02-13 00:00:00" — so a date-only format has to tolerate one.
# The suffixes are tried in order after the bare format, longest first so
# "%H:%M:%S" is not matched as "%H:%M" with seconds left over.
TIME_SUFFIXES = (" %H:%M:%S", " %H:%M", "T%H:%M:%S", "T%H:%M")


# A strptime pattern as a person reads it, for saying which a date missed.
_WRITTEN = {"%d": "DD", "%m": "MM", "%Y": "YYYY", "%y": "YY", "%b": "Mon", "%B": "Month",
            "%H": "hh", "%I": "hh", "%M": "mm", "%S": "ss", "%p": "AM/PM"}


def written_as(date_format: str) -> str:
    """`%d/%m/%Y` as `DD/MM/YYYY`."""
    return re.sub(r"%[a-zA-Z]", lambda m: _WRITTEN.get(m.group(0), m.group(0)), date_format)


def strptime_any(text: str, date_format: str) -> dt.datetime:
    """`date_format`, or it with a time on the end."""
    try:
        return dt.datetime.strptime(text, date_format)
    except ValueError:
        for suffix in TIME_SUFFIXES:
            try:
                return dt.datetime.strptime(text, date_format + suffix)
            except ValueError:
                continue
        raise


def in_trading_hours(clock: dt.time | None, exchange: str) -> tuple[dt.time | None, bool]:
    """A trade time inside the exchange's session, and whether it had to move.

    Before the open becomes the open, after the close becomes the close. A time
    outside the session is not a trading time: it is a padded date ("00:00:00",
    which is most exports), a settlement stamp, or a clock in another timezone.

    Storing it as given would misorder the day. `time` feeds one thing —
    `models.trade_order`, which sequences same-day trades for FIFO — so a
    midnight stamp sorts an imported trade ahead of every hand-entered one on
    that date, and picks a different parcel for a same-day sale.

    The open comes back as None, which means exactly the same thing to
    `trade_order` and keeps a column of "10:00" out of the interface.

    Markets that never close (crypto) have no session to be outside of, so
    their times are taken as given.
    """
    if clock is None:
        return None, False
    market = MARKETS.get(exchange.upper())
    if market is None or market.open is None or market.close is None:
        return clock, False
    if clock < market.open:
        return None, True
    if clock > market.close:
        return market.close, True
    return (None, False) if clock == market.open else (clock, False)


def parse_when(value: str, date_format: str) -> tuple[dt.date, dt.time | None]:
    """The date, and the time if the cell carried a real one.

    Midnight is no time at all: a broker writing "2024-02-13 00:00:00" is
    padding a date, not claiming a trade at midnight. It is separated from a
    stated out-of-hours time here — see `in_trading_hours` — so that a
    date-only export does not report every one of its rows as adjusted.
    """
    stamp = strptime_any(value.strip(), date_format)  # raises for the caller
    clock = stamp.time()
    return stamp.date(), None if clock == dt.time(0) else clock


@dataclass
class CandidateTrade:
    date: dt.date
    ticker: str
    type: str  # buy / sell
    quantity: Decimal
    unit_price: Decimal
    brokerage: Decimal
    currency: str
    exchange: str
    time: dt.time | None = None  # only when the export carried a real one
    status: str = "new"  # new / duplicate / unknown-instrument / refused
    detail: str = ""

    def as_dict(self) -> dict:
        return {
            "date": self.date.isoformat(),
            "time": self.time.isoformat() if self.time else None,
            "ticker": self.ticker,
            "type": self.type,
            "quantity": str(self.quantity),
            "unit_price": str(self.unit_price),
            "brokerage": str(self.brokerage),
            "currency": self.currency,
            "exchange": self.exchange,
            "status": self.status,
            "detail": self.detail,
        }

    @classmethod
    def from_dict(cls, d: dict) -> CandidateTrade:
        return cls(
            date=dt.date.fromisoformat(d["date"]),
            time=dt.time.fromisoformat(d["time"]) if d.get("time") else None,
            ticker=d["ticker"],
            type=d["type"],
            quantity=Decimal(d["quantity"]),
            unit_price=Decimal(d["unit_price"]),
            brokerage=Decimal(d["brokerage"]),
            currency=d["currency"],
            exchange=d["exchange"],
            status=d["status"],
            detail=d["detail"],
        )


@dataclass
class ParseResult:
    candidates: list[CandidateTrade] = field(default_factory=list)
    skipped: list[str] = field(default_factory=list)  # non-trade rows (cash etc.)
    errors: list[str] = field(default_factory=list)
    adjusted: int = 0  # times moved into trading hours; see `in_trading_hours`


def _num(raw: str, column, name: str) -> Decimal:
    return money.parse(str(raw).replace(",", "").replace("$", ""), column, name)


def _check_trade_figures(units: Decimal, price: Decimal, brokerage: Decimal) -> None:
    """The trade form's rules, so a file cannot stage what the form would refuse.

    Zero is a real price (a bonus issue, a demerger allocation). A negative
    quantity is refused rather than flipped: whether it means a sell is the
    export's to say, and none of the shipped formats do.
    """
    if units <= 0:
        raise money.FigureError(f"Units: {units} has to be more than zero")
    if price < 0:
        raise money.FigureError(f"Price: {price} can't be negative")
    if brokerage < 0:
        raise money.FigureError(f"Brokerage: {brokerage} can't be negative")


def parse_csv(text: str, broker: str, fmt: BrokerFormat) -> ParseResult:
    """Every trade in an export, with an account of what was skipped.

    A field past the csv module's size limit (128 KiB: a file that is not a
    CSV at all, or one runaway quote) raises inside the reader, and was a
    server error rather than a sentence about the file."""
    try:
        return _parse_csv(text, broker, fmt)
    except csv.Error:
        return ParseResult(errors=["The file could not be read as a CSV."])


def _parse_csv(text: str, broker: str, fmt: BrokerFormat) -> ParseResult:
    # `restval`: a row shorter than the header (an export's one-line footer)
    # reads as blanks, and is skipped by line like any other, instead of
    # handing None to the first `.strip()`.
    reader = csv.DictReader(io.StringIO(text), restval="")
    if reader.fieldnames is None:
        return ParseResult(errors=["The file is empty."])
    # "Date, Code" names its columns with the spaces. The rows are read by
    # the names the header check and the broker designer use: without them.
    reader.fieldnames = headers = [h.strip() for h in reader.fieldnames]
    result = ParseResult()

    if fmt.kind == "mapped":
        missing = [v for v in fmt.columns.values() if v not in headers]
        if missing:
            result.errors.append(
                f"This file has no {', '.join(missing)} column{'s' if len(missing) > 1 else ''}, "
                f"which the {broker} format reads. Pick the broker it came from, or map its "
                "columns with Map a new broker CSV.")
            return result
        col = fmt.columns
        for i, row in enumerate(reader, start=2):
            try:
                action = trade_type(row[col["action"]], fmt.actions)
                if action is None:
                    result.skipped.append(f"line {i}: action {row[col['action']]!r}")
                    continue
                # A DRP allotment carries units and no price: the registry
                # bought them out of the distribution, and the cost base is in
                # the statement rather than this export. Booking it at zero
                # would understate the cost base of every parcel and overstate
                # the gain on sale — decisions.md #100.
                if action == "drp" and not str(row.get(col["price"], "")).strip():
                    result.skipped.append(
                        f"line {i}: DRP allotment with no price — import the "
                        f"dividend statement, which carries the cost base")
                    continue
                written = row[col["date"]]
                try:
                    when, clock = parse_when(written, fmt.date_format)
                except ValueError:
                    result.errors.append(f"line {i}: {written.strip()!r} is not a date "
                                         f"written as {written_as(fmt.date_format)}")
                    continue
                if clock and fmt.times_zone:
                    when, clock = to_market(clock, when, fmt.times_zone, fmt.exchange)
                clock, moved = in_trading_hours(clock, fmt.exchange)
                result.adjusted += moved
                units = _num(row[col["units"]], Trade.quantity, "Units")
                price = _num(row[col["price"]], Trade.unit_price, "Price")
                brokerage = _num(row.get(col.get("brokerage", ""), "0") or "0",
                                 Trade.brokerage, "Brokerage")
                _check_trade_figures(units, price, brokerage)
                result.candidates.append(
                    CandidateTrade(
                        date=when,
                        time=clock,
                        ticker=row[col["ticker"]].strip().upper(),
                        type=action,
                        quantity=units,
                        unit_price=price,
                        brokerage=brokerage,
                        currency=fmt.currency,
                        exchange=fmt.exchange,
                    )
                )
            except (money.FigureError, KeyError, ValueError) as exc:
                result.errors.append(f"line {i}: {exc}")
        return result

    if fmt.kind == "commsec_transactions":
        if "Details" not in headers or "Debit($)" not in headers:
            result.errors.append(
                "This is not CommSec's transactions export header, which reads Date, "
                "Reference, Details, Debit($), Credit($), Balance($). Pick the broker the "
                "file came from.")
            return result
        for i, row in enumerate(reader, start=2):
            details = (row.get("Details") or "").strip()
            m = COMMSEC_DETAILS.match(details)
            if not m:
                result.skipped.append(f"line {i}: {details!r}")
                continue
            try:
                qty = _num(m.group("units"), Trade.quantity, "Units")
                price = _num(m.group("price"), Trade.unit_price, "Price")
                _check_trade_figures(qty, price, Decimal(0))
                is_buy = m.group("action") == "B"
                # Brokerage isn't a column: recover it from the cash movement.
                # buy: debit = qty*price + brokerage; sell: credit = qty*price - brokerage.
                gross = qty * price
                try:
                    when, clock = parse_when(row["Date"], fmt.date_format)
                except ValueError:
                    result.errors.append(f"line {i}: {row['Date'].strip()!r} is not a date "
                                         f"written as {written_as(fmt.date_format)}")
                    continue
                clock, moved = in_trading_hours(clock, fmt.exchange)
                result.adjusted += moved
                cash = (_num(row.get("Debit($)") or "0", Trade.quantity, "Debit")
                        if is_buy else
                        _num(row.get("Credit($)") or "0", Trade.quantity, "Credit"))
                brokerage = (cash - gross) if is_buy else (gross - cash)
                if brokerage < 0 or brokerage > max(Decimal("100"), gross / 10):
                    result.errors.append(
                        f"line {i}: implausible brokerage {brokerage} from {details!r} "
                        f"(cash {cash}, gross {gross}) — check the export"
                    )
                    continue
                result.candidates.append(
                    CandidateTrade(
                        date=when,
                        time=clock,
                        ticker=m.group("ticker"),
                        type="buy" if is_buy else "sell",
                        quantity=qty,
                        unit_price=price,
                        brokerage=brokerage.quantize(Decimal("0.01")),
                        currency=fmt.currency,
                        exchange=fmt.exchange,
                    )
                )
            except (money.FigureError, ValueError) as exc:
                result.errors.append(f"line {i}: {exc}")
        return result

    result.errors.append(f"unknown broker format kind {fmt.kind!r}")
    return result


BUILTIN_ACTIONS = {"buy": "buy", "b": "buy", "sell": "sell", "s": "sell"}


def trade_type(raw: str, mapped: dict[str, str] | None = None) -> str | None:
    """This broker's word for a trade type, as one of ours, or None to skip.

    The format's own vocabulary wins, matched without case, so a broker calling
    a DRP allotment "In" needs no code. Everything else falls back to the words
    every broker agrees on.
    """
    word = raw.strip()
    for source, target in (mapped or {}).items():
        if source.strip().lower() == word.lower():
            return target if target in ("buy", "sell", "drp") else None
    return BUILTIN_ACTIONS.get(word.lower())


def unresolved(candidates: list[CandidateTrade]) -> list[tuple[str, str, int]]:
    """The distinct (ticker, exchange) pairs that would be created, with counts.

    Grouped rather than per row: a file with ninety trades in six instruments
    asks six questions, not ninety.
    """
    seen: dict[tuple[str, str], int] = {}
    for c in candidates:
        if c.status == "unknown-instrument":
            seen[(c.ticker, c.exchange)] = seen.get((c.ticker, c.exchange), 0) + 1
    return [(t, e, n) for (t, e), n in sorted(seen.items())]


def relabel(candidates: list[CandidateTrade], moves: dict[tuple[str, str], tuple[str, str]]) -> int:
    """Apply corrected tickers and exchanges to every row that carried the old.

    Returns how many rows moved. A correction is per instrument, so fixing
    ALPHA on the wrong exchange fixes every trade of it at once.
    """
    changed = 0
    for c in candidates:
        new = moves.get((c.ticker, c.exchange))
        if new and new != (c.ticker, c.exchange):
            c.ticker, c.exchange = new
            changed += 1
    return changed


def _own(session: Session, ticker: str, exchange: str, own_ids: set[int]) -> Instrument | None:
    """This portfolio's instrument for a row, never the catalogue's. Whether
    some other portfolio holds a ticker is not this one's to learn (#139), so a
    row for one is new here, exactly as for a ticker nobody holds."""
    return session.scalars(
        select(Instrument).where(Instrument.ticker == ticker, Instrument.exchange == exchange,
                                 Instrument.id.in_(own_ids))
    ).first()


def annotate(session: Session, candidates: list[CandidateTrade], imports: ImportSettings) -> None:
    """Mark each candidate as new / duplicate / unknown-instrument.

    Re-entrant: status and detail are reset first, because the resolve step
    calls this again after a correction and a stale "unknown-instrument" would
    keep refusing a ticker that now resolves.

    A recorded trade answers for one row at most, so a second identical fill
    on the day is new. Rows with a time take the trade at that time first; a
    trade typed without one sits at market open, so time is never the only
    reason a row is new.
    """
    matching: list[tuple[CandidateTrade, list[Trade]]] = []
    own_ids = queries.portfolio_instrument_ids(session)
    for c in candidates:
        c.status, c.detail = "new", ""
        inst = _own(session, c.ticker, c.exchange, own_ids)
        if inst is None:
            c.status = "unknown-instrument"
            # Refused when new instruments are off: the note under the table
            # says what to do, so the row says only what is true of it.
            c.detail = "will be created" if imports.allow_new_instruments else "not in this portfolio"
            continue
        matching.append((c, list(session.scalars(
            select(Trade).where(
                Trade.instrument_id == inst.id,
                Trade.date == c.date,
                Trade.type == c.type,
                Trade.quantity == c.quantity,
                Trade.unit_price == c.unit_price,
            ).order_by(Trade.id)
        ))))
    claimed: set[int] = set()
    for same_time in (True, False):
        for c, trades in matching:
            if c.status == "duplicate":
                continue
            dupe = next((t for t in trades if t.id not in claimed
                         and (not same_time or t.time == c.time)), None)
            if dupe is not None:
                claimed.add(dupe.id)
                c.status = "duplicate"
                c.detail = f"matches trade #{dupe.id}"
    _refuse_shortfalls(session, candidates)


def _refuse_shortfalls(session: Session, candidates: list[CandidateTrade]) -> None:
    """Refuse a row that would sell more than is held, as the trade form
    refuses one typed in. Each holding's rows are walked in date order against
    what is already recorded and the rows accepted before them; a refused row
    is left out of the walk, so the rest are judged without it.
    """
    by_holding: dict[tuple[str, str], list[CandidateTrade]] = {}
    for c in candidates:
        if c.status in ("new", "unknown-instrument"):
            by_holding.setdefault((c.ticker, c.exchange), []).append(c)
    own_ids = queries.portfolio_instrument_ids(session)
    for (ticker, exchange), rows in by_holding.items():
        inst = _own(session, ticker, exchange, own_ids)
        held = (list(session.scalars(select(Trade).where(Trade.instrument_id == inst.id)))
                if inst is not None else [])
        actions = (list(session.scalars(select(CorporateAction)
                                        .where(CorporateAction.instrument_id == inst.id)))
                   if inst is not None else [])
        found = (_shortfalls_walked(held, rows, actions)
                 if any(a.kind == "split" for a in actions) else _shortfalls(held, rows))
        for c, breach in found:
            c.status, c.detail = "refused", queries.breach_message(ticker, breach)


def _candidate(c: CandidateTrade) -> Trade:
    return Trade(date=c.date, time=c.time, type=c.type, quantity=c.quantity)


def _shortfalls_walked(held: list[Trade], rows: list[CandidateTrade],
                       actions: list[CorporateAction]) -> list[tuple[CandidateTrade, queries.Breach]]:
    """`_shortfalls`, for a holding with a split: each row walked as the trade
    form walks one. The running totals there cannot take a split, which
    rounds the holding where it falls (#146); such a holding's rows are few."""
    refused, accepted = [], []
    for c in sorted(rows, key=lambda c: trade_order(_candidate(c))):
        candidate = _candidate(c)
        breach = queries.balance_breach(held + accepted, candidate, actions=actions)
        if breach is None:
            accepted.append(candidate)
        else:
            refused.append((c, breach))
    return refused


def _shortfalls(held: list[Trade], rows: list[CandidateTrade]) -> list[tuple[CandidateTrade, queries.Breach]]:
    """Each row the trade form would refuse, with the breach it would name.

    The form walks the whole timeline for its one row; done for each row of a
    file, that took minutes over a 40,000-row export. So the timeline is
    walked once: a row is checked against the units held at its place, and
    against what the recorded trades after it need. Only recorded trades come
    after it: the rows accepted so far all sort before it, and the rest are
    judged later.
    """
    timeline = sorted(
        [(t, None) for t in held]
        + [(Trade(date=c.date, time=c.time, type=c.type, quantity=c.quantity), c) for c in rows],
        key=lambda pair: trade_order(pair[0]))
    change = [t.quantity if t.type in ("buy", "drp") else -t.quantity for t, _ in timeline]
    # The recorded trades alone: their running total at each place, and
    # `need`, the most those from a place on take from what is held there.
    recorded = list(accumulate(Decimal(0) if c else step for (_t, c), step in zip(timeline, change)))
    need = [Decimal(0)] * (len(timeline) + 1)
    for i in reversed(range(len(timeline))):
        need[i] = need[i + 1] if timeline[i][1] else max(Decimal(0), need[i + 1] - change[i])
    first_below = _first_below(recorded)

    refused, units, broken = [], Decimal(0), None
    for i, (t, c) in enumerate(timeline):
        after = units + change[i]
        if c is None:
            # Recorded trades the file never touched can already be short;
            # the form names the first, before any row of its own.
            if after < 0 and broken is None:
                broken = queries.Breach(trade=t, balance=after, held_before=units,
                                        is_candidate=False)
            units = after
        elif broken is not None:
            refused.append((c, broken))
        elif after < 0:
            refused.append((c, queries.Breach(trade=t, balance=after, held_before=units,
                                              is_candidate=True)))
        elif after < need[i + 1]:
            k = first_below(i + 1, recorded[i] - after)
            left = after + recorded[k] - recorded[i]
            refused.append((c, queries.Breach(trade=timeline[k][0], balance=left,
                                              held_before=left - change[k], is_candidate=False)))
        else:
            units = after
    return refused


def _first_below(values: list[Decimal]):
    """`find(start, limit)`: the first index from `start` whose value is below
    `limit`, in a handful of steps rather than a scan. `levels[k][i]` is the
    least of `values[i : i + 2**k]`, so whole runs that stay at or above the
    limit are stepped over, longest first."""
    levels = [values]
    while 2 ** len(levels) <= len(values):
        last, run = levels[-1], 2 ** (len(levels) - 1)
        levels.append([min(a, b) for a, b in zip(last, last[run:])])

    def find(start: int, limit: Decimal) -> int:
        i = start
        for k in reversed(range(len(levels))):
            if i < len(levels[k]) and levels[k][i] >= limit:
                i += 2 ** k
        return i

    return find


def commit(session: Session, candidates: list[CandidateTrade], imports: ImportSettings, source: str) -> dict[str, int]:
    counts = {"inserted": 0, "duplicates_skipped": 0, "instruments_created": 0}
    refused = [c for c in candidates if c.status == "refused"]
    if refused:
        raise ValueError(refused[0].detail)
    own_ids = queries.portfolio_instrument_ids(session)
    for c in candidates:
        if c.status == "duplicate":
            counts["duplicates_skipped"] += 1
            continue
        inst = _own(session, c.ticker, c.exchange, own_ids)
        if inst is None:
            if not imports.allow_new_instruments:
                raise ValueError(f"unknown instrument {c.ticker} ({c.exchange})")
            # A CSV is arbitrary input; a malformed ticker must not become a row.
            problem = ticker_problem(c.ticker) or exchange_problem(c.exchange)
            if problem:
                raise ValueError(f"{problem} (got {c.ticker!r} on {c.exchange!r})")
            # New to this portfolio, and only now looked for in the shared
            # catalogue (#125), which the preview could not tell from new.
            inst = session.scalars(
                select(Instrument).where(Instrument.ticker == c.ticker,
                                         Instrument.exchange == c.exchange)
            ).first()
        if inst is None:
            inst = Instrument(
                ticker=c.ticker,
                exchange=c.exchange,
                currency=c.currency,
                asset_class="share",
                drp=False,
                yahoo_symbol=f"{c.ticker}.AX" if c.exchange == "ASX" else c.ticker,
                note=f"created by {source} import",
            )
            session.add(inst)
            session.flush()
            counts["instruments_created"] += 1
        # A removed one comes back, as it does when a trade is typed in.
        inst.active = True
        session.add(
            owned(
                session,
                Trade(
                    instrument=inst,
                    date=c.date,
                    # Only when the export carried one; unset means MARKET_OPEN.
                    **({"time": c.time} if c.time else {}),
                    type=c.type,
                    quantity=c.quantity,
                    unit_price=c.unit_price,
                    brokerage=c.brokerage,
                    fx_rate=Decimal(1) if c.currency == "AUD" else None,  # feed backfills
                    note=f"imported from {source} CSV",
                ),
            )
        )
        counts["inserted"] += 1
    session.flush()  # the caller's scoped session owns the commit
    return counts