gizmoduck/ingestion/services.py

"""Audited review lifecycle. No financial rows exist until an explicit commit."""
from datetime import date
from functools import wraps
import hashlib
import re
from uuid import UUID, uuid5, NAMESPACE_URL

from django.core.exceptions import ValidationError
from django.db import IntegrityError, connection, transaction
from django.db.models import Q

from finance.models import Account, Payee, Category, CardStatement, InstallmentPlan, Transaction, SourceRecord
from finance.services import Actor, DomainError
from finance.services.base import audit, identity, save
from finance.services.posting import record
from ingestion.models import ImportBatch, ImportItem
from ingestion.parsers import parse_document, ParseError
from ingestion.normalization import normalize
from ingestion.duplicates import candidate_signature, transaction_signature
from ingestion.validation import validate


REASONS = {"USER_EXCLUDED", "MALFORMED", "UNRECOGNIZED", "ALREADY_IMPORTED", "SECTION_EXCLUDED"}
REFERENCE_FIELDS = {"payee_id": ("payees", Payee), "category_id": ("categories", Category),
                    "statement_id": ("statements", CardStatement), "installment_plan_id": ("plans", InstallmentPlan)}


def operation(name):
    def decorate(function):
        @wraps(function)
        def wrapped(*args, actor, **kwargs):
            if not isinstance(actor, Actor) or actor.kind not in {"HUMAN", "APPLICATION", "AGENT"} or not isinstance(actor.identifier, str) or not actor.identifier.strip() or len(actor.identifier) > 200:
                raise DomainError("invalid_actor")
            if connection.in_atomic_block:
                raise DomainError("transaction_boundary")
            try:
                with transaction.atomic(durable=True):
                    batch, fields, emit = function(*args, actor=actor, **kwargs)
                    if emit:
                        audit(actor, name, "SUCCESS", batch.pk, fields)
                    return batch
            except (DomainError, ParseError, ValidationError, IntegrityError) as error:
                code = getattr(error, "code", None)
                if not isinstance(code, str):
                    code = "CONFLICT" if isinstance(error, IntegrityError) else "VALIDATION_ERROR"
                with transaction.atomic(durable=True):
                    audit(actor, name, "FAILURE", error_code=code)
                raise DomainError(code) from None
        return wrapped
    return decorate


def _batch(batch_id, revision=None, *, preview=True):
    try:
        batch = ImportBatch.objects.select_for_update().get(pk=identity(batch_id))
    except ImportBatch.DoesNotExist:
        raise DomainError("BATCH_NOT_FOUND") from None
    if revision is not None and (type(revision) is not int or revision != batch.revision):
        raise DomainError("STALE_REVISION")
    if preview and batch.state != "PREVIEW":
        raise DomainError("BATCH_NOT_PREVIEW")
    return batch


def _rows(query, locking):
    if locking:
        query = query.select_for_update(of=("self",))
    return {str(obj.pk): obj for obj in query.order_by("pk")}


def _load(batch, items, locking):
    account_ids = [value for value in batch.section_mapping.values() if value]
    accounts = _rows(Account.objects.filter(pk__in=account_ids), locking)
    refs = {}
    for field, (key, model) in REFERENCE_FIELDS.items():
        ids = [item.decision[field] for item in items if item.decision.get(field)]
        refs[key] = _rows(model.objects.filter(pk__in=ids), locking)
    dates = set()
    for item in items:
        try:
            dates.add(date.fromisoformat(item.candidate.get("posting_date")))
        except (TypeError, ValueError):
            pass
    target_ids = [item.decision[key] for item in items for key in ("transaction_id", "refund_of_id") if item.decision.get(key)]
    query = Transaction.objects.filter(Q(account_id__in=account_ids, posting_date__in=dates) | Q(pk__in=target_ids))
    refs["transactions"] = _rows(query.select_related("account", "installment_plan"), locking)
    sources = {row.item_locator: row for row in SourceRecord.objects.filter(document_hash=batch.document_hash).select_related("transaction")}
    return accounts, refs, sources


def _fingerprint(obj):
    # IDs and modification timestamps detect reference annotations/archive changes.
    return [str(obj.pk), obj.updated_at.isoformat()]


def _evaluate(batch, items, *, locking=False):
    accounts, refs, sources = _load(batch, items, locking)
    keys = {}
    existing = {}
    for row in refs["transactions"].values():
        existing.setdefault(transaction_signature(row), []).append(str(row.pk))
    for item in items:
        account = accounts.get(batch.section_mapping.get(item.candidate["section"]))
        if account and not item.candidate.get("errors"):
            try:
                keys[item.locator] = candidate_signature(item.candidate, account, item.decision)
            except (KeyError, TypeError, ValueError):
                pass
    results = {}
    plan_definitions = {}
    plan_memberships = set()
    for item in items:
        candidate, decision = item.candidate, item.decision
        mapped = batch.section_mapping.get(candidate["section"])
        account = accounts.get(mapped)
        issues = validate(candidate, decision, account, refs, batch.profile) if mapped else ["SECTION_EXCLUDED"]
        key = keys.get(item.locator)
        source = sources.get(item.locator)
        matches = sorted(existing.get(key, [])) if key is not None else []
        within = sorted(locator for locator, other in keys.items() if locator != item.locator and other == key) if key is not None else []
        duplicate = {"existing": matches, "within": within,
                     "source": str(source.transaction_id) if source else None,
                     "source_account": str(source.transaction.account_id) if source else None,
                     "status": "ALREADY_IMPORTED" if source else None}
        if source:
            issues.append("ALREADY_IMPORTED" if str(source.transaction.account_id) == mapped else "SOURCE_ACCOUNT_CONFLICT")
        if decision.get("new_plan") and decision.get("action") == "CREATE":
            name = decision["new_plan"]
            definition = (mapped, candidate.get("installment_total"))
            if name in plan_definitions and plan_definitions[name] != definition:
                issues.append("NEW_PLAN_CONFLICT")
            plan_definitions[name] = definition
        if decision.get("action") == "CREATE" and candidate.get("installment_number"):
            plan_key = decision.get("installment_plan_id") or decision.get("new_plan")
            membership = (plan_key, candidate["installment_number"])
            if plan_key and membership in plan_memberships:
                issues.append("DUPLICATE_INSTALLMENT")
            plan_memberships.add(membership)
        results[item.locator] = {"issues": sorted(set(issues)), "duplicates": duplicate}
    reference_snapshot = {name: [_fingerprint(obj) for obj in table.values()] for name, table in refs.items() if name != "transactions"}
    relevant_ids = {tid for result in results.values() for tid in result["duplicates"]["existing"]}
    relevant_ids.update(item.decision[key] for item in items for key in ("transaction_id", "refund_of_id") if item.decision.get(key))
    reference_snapshot["transactions"] = [_fingerprint(obj) for key, obj in refs["transactions"].items() if key in relevant_ids]
    return {"accounts": [_fingerprint(obj) for obj in accounts.values()], "references": reference_snapshot, "items": results}


def _persist(batch, items, snapshot):
    for item in items:
        current = snapshot["items"][item.locator]
        item.issues, item.duplicates = current["issues"], current["duplicates"]
    ImportItem.objects.bulk_update(items, ["issues", "duplicates", "decision"])
    batch.snapshot = snapshot
    batch.summary = {**batch.summary, "rows": len(items), "error_rows": sum(bool(item.issues) for item in items),
                     "decided_rows": sum(bool(item.decision) for item in items)}
    batch.save()


def _current(batch, items):
    snapshot = _evaluate(batch, items, locking=True)
    if snapshot != batch.snapshot:
        raise DomainError("STALE_PREVIEW")
    return snapshot


@operation("import_preview")
def create_preview(*, data, profile, section_mapping, context=None, actor):
    parsed = parse_document(data, profile, context or {})
    candidates = [candidate.to_dict() for candidate in normalize(parsed)]
    if not isinstance(section_mapping, dict) or any(not isinstance(key, str) for key in section_mapping):
        raise DomainError("INVALID_SECTION_MAPPING")
    sections = {item["section"] for item in candidates}
    if set(section_mapping) != sections:
        raise DomainError("SECTION_MAPPING_REQUIRED")
    mapping = {key: str(identity(value)) if value is not None else None for key, value in section_mapping.items()}
    batch = ImportBatch.objects.create(document_hash=hashlib.sha256(data).hexdigest(),
        source_format="BANK_CSV" if profile == "BANK_CSV_V1" else "OUROCARD_TXT",
        profile=profile, encoding=parsed.encoding, context=context or {}, section_mapping=mapping,
        summary={"parser_skipped": dict(parsed.skipped)}, actor_kind=actor.kind, actor_identifier=actor.identifier)
    items = [ImportItem(batch=batch, locator=candidate["locator"], candidate=candidate) for candidate in candidates]
    ImportItem.objects.bulk_create(items)
    _persist(batch, items, _evaluate(batch, items, locking=True))
    return batch, ["preview"], True


def export_review(*, batch_id):
    # Keep exported revision and item decisions consistent with concurrent editing.
    with transaction.atomic():
        batch = _batch(batch_id, preview=False)
        items = list(batch.items.order_by("created_at", "locator"))
        return {"batch_id": str(batch.pk), "revision": batch.revision, "state": batch.state,
                "profile": batch.profile, "encoding": batch.encoding, "summary": batch.summary,
                "items": [{"locator": row.locator, "candidate_id": str(uuid5(NAMESPACE_URL, batch.document_hash + ":" + row.locator)),
                           "candidate": row.candidate, "issues": row.issues, "duplicates": row.duplicates,
                           "decision": row.decision} for row in items],
                "decisions": [{"locator": row.locator, **row.decision} for row in items]}


def _decisions(value, locators):
    if not isinstance(value, list):
        raise DomainError("INVALID_DECISIONS")
    result = {}
    for supplied in value:
        if not isinstance(supplied, dict):
            raise DomainError("INVALID_DECISION")
        row = dict(supplied)
        locator = row.pop("locator", None)
        if not isinstance(locator, str) or locator not in locators or locator in result:
            raise DomainError("INVALID_LOCATOR")
        action = row.get("action")
        allowed = {"action"}
        if action == "CREATE":
            allowed |= {"acknowledge_duplicate", "kind", "payee_id", "category_id", "statement_id", "refund_of_id", "installment_plan_id", "new_plan"}
        elif action == "MATCH":
            allowed.add("transaction_id")
        elif action == "SKIP":
            allowed.add("reason")
        else:
            raise DomainError("INVALID_ACTION")
        if set(row) - allowed:
            raise DomainError("UNKNOWN_DECISION_FIELD")
        if "acknowledge_duplicate" in row and type(row["acknowledge_duplicate"]) is not bool:
            raise DomainError("INVALID_ACKNOWLEDGEMENT")
        if action == "SKIP" and row.get("reason") not in REASONS:
            raise DomainError("SKIP_REASON_REQUIRED")
        if action == "MATCH" and not row.get("transaction_id"):
            raise DomainError("MATCH_TARGET_REQUIRED")
        for field in set(row) & {"transaction_id", "payee_id", "category_id", "statement_id", "refund_of_id", "installment_plan_id"}:
            if row[field] is not None:
                row[field] = str(identity(row[field]))
        if "new_plan" in row and (not isinstance(row["new_plan"], str) or not re.fullmatch(r"[A-Za-z0-9_-]{1,64}", row["new_plan"])):
            raise DomainError("INVALID_NEW_PLAN")
        if "kind" in row and row["kind"] not in {"INCOME", "EXPENSE", "REFUND", "TRANSFER"}:
            raise DomainError("INVALID_KIND")
        result[locator] = row
    return result


@operation("import_decisions")
def apply_decisions(*, batch_id, revision, decisions, actor):
    batch = _batch(batch_id, revision)
    items = list(batch.items.order_by("locator"))
    _current(batch, items)
    values = _decisions(decisions, {item.locator for item in items})
    for item in items:
        item.decision = values.get(item.locator, {})
    batch.revision += 1
    _persist(batch, items, _evaluate(batch, items, locking=True))
    return batch, ["decisions", "revision"], True


@operation("import_refresh")
def refresh_preview(*, batch_id, revision, actor):
    batch = _batch(batch_id, revision)
    items = list(batch.items.order_by("locator"))
    for item in items:
        item.decision = {}
    batch.revision += 1
    _persist(batch, items, _evaluate(batch, items, locking=True))
    return batch, ["preview", "decisions", "revision"], True


@operation("import_cancel")
def cancel_preview(*, batch_id, revision, actor):
    batch = _batch(batch_id, revision)
    batch.state = "CANCELLED"
    batch.save()
    return batch, ["state"], True


def _check_decision(item):
    decision = item.decision
    if not decision:
        raise DomainError("DECISION_REQUIRED")
    if "SOURCE_ACCOUNT_CONFLICT" in item.issues:
        raise DomainError("SOURCE_ACCOUNT_CONFLICT")
    if decision["action"] == "SKIP":
        if item.duplicates.get("source") and decision["reason"] != "ALREADY_IMPORTED":
            raise DomainError("ALREADY_IMPORTED")
        if "SECTION_EXCLUDED" in item.issues and decision["reason"] != "SECTION_EXCLUDED":
            raise DomainError("SECTION_EXCLUDED")
        return
    if item.duplicates.get("source"):
        raise DomainError("ALREADY_IMPORTED")
    # MATCH validates the signature rather than requiring a new plan assignment.
    issues = [issue for issue in item.issues if not (decision["action"] == "MATCH" and issue == "INSTALLMENT_PLAN_REQUIRED")]
    if issues:
        raise DomainError("UNRESOLVED_ISSUES")
    if decision["action"] == "CREATE" and (item.duplicates["existing"] or item.duplicates["within"]) and not decision.get("acknowledge_duplicate"):
        raise DomainError("DUPLICATE_ACKNOWLEDGEMENT_REQUIRED")
    if decision["action"] == "MATCH" and decision["transaction_id"] not in item.duplicates["existing"]:
        raise DomainError("MATCH_SIGNATURE_MISMATCH")


@operation("import_commit")
def commit_preview(*, batch_id, revision, actor):
    batch = _batch(batch_id, revision, preview=False)
    if batch.state == "COMMITTED":
        return batch, [], False
    if batch.state != "PREVIEW":
        raise DomainError("BATCH_NOT_PREVIEW")
    items = list(batch.items.order_by("locator"))
    _current(batch, items)
    for item in items:
        _check_decision(item)
    plans = {}
    result = {"created": 0, "matched": 0, "skipped": 0, "transaction_ids": [], "parser_skipped": batch.summary.get("parser_skipped", {}), "error_rows": batch.summary.get("error_rows", 0)}
    for item in items:
        candidate, decision = item.candidate, item.decision
        if decision["action"] == "SKIP":
            result["skipped"] += 1
            continue
        if decision["action"] == "MATCH":
            row = Transaction.objects.get(pk=decision["transaction_id"])
            result["matched"] += 1
        else:
            account_id = batch.section_mapping[candidate["section"]]
            plan_id = decision.get("installment_plan_id")
            if decision.get("new_plan"):
                name = decision["new_plan"]
                if name not in plans:
                    plans[name] = save(InstallmentPlan(account_id=account_id, description=name, total_installments=candidate["installment_total"]))
                plan_id = plans[name].pk
            fields = {key: decision.get(key) for key in ("payee_id", "category_id", "statement_id", "refund_of_id")}
            row = record(account_id=account_id, amount=candidate["amount"], posting_date=candidate["posting_date"],
                         occurrence_date=candidate["occurrence_date"], kind=decision.get("kind", candidate["kind"]),
                         description=candidate["description"], installment_plan_id=plan_id,
                         installment_number=candidate.get("installment_number"), **fields)
            result["created"] += 1
        save(SourceRecord(transaction=row, document_hash=batch.document_hash, source_format=batch.source_format,
                          parser_version=batch.parser_version, item_locator=item.locator))
        result["transaction_ids"].append(str(row.pk))
    batch.state = "COMMITTED"
    batch.result = result
    batch.save()
    return batch, ["state", "transactions", "source_records", "plans"], True