voiceledger/ledger/database.py

"""SQLite persistence layer for VoiceLedger transactions."""

from __future__ import annotations

import sqlite3
import tempfile
from datetime import datetime
from pathlib import Path
from typing import Any

import pandas as pd

from voiceledger.config import get_database_path
from voiceledger.ledger.corrections import initialize_correction_log_table
from voiceledger.ledger.customers import add_credit, initialize_customers_table, record_payment
from voiceledger.ledger.inventory import add_stock, initialize_inventory_table, remove_stock
from voiceledger.ledger.settings import initialize_business_settings_table
from voiceledger.parser.schema import Transaction


SCHEMA_SQL = """
CREATE TABLE IF NOT EXISTS transactions (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    transaction_type TEXT NOT NULL,
    item TEXT,
    quantity REAL,
    unit_price REAL,
    amount REAL,
    customer TEXT,
    payment_status TEXT NOT NULL,
    notes TEXT NOT NULL,
    confidence REAL NOT NULL,
    created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
);
"""

COLUMNS = [
    "id",
    "transaction_type",
    "item",
    "quantity",
    "unit_price",
    "amount",
    "customer",
    "payment_status",
    "notes",
    "confidence",
    "created_at",
]


def initialize_database(db_path: str | Path | None = None) -> Path:
    """Create the SQLite database and transactions table if needed."""
    path = _resolve_db_path(db_path)
    path.parent.mkdir(parents=True, exist_ok=True)

    with sqlite3.connect(path) as connection:
        connection.execute(SCHEMA_SQL)
        connection.commit()

    initialize_customers_table(path)
    initialize_inventory_table(path)
    initialize_business_settings_table(path)
    initialize_correction_log_table(path)
    return path


def add_transaction(transaction: Transaction, db_path: str | Path | None = None) -> int:
    """Insert a transaction and return its database id."""
    path = initialize_database(db_path)
    payload = transaction.model_dump()

    with sqlite3.connect(path) as connection:
        cursor = connection.execute(
            """
            INSERT INTO transactions (
                transaction_type,
                item,
                quantity,
                unit_price,
                amount,
                customer,
                payment_status,
                notes,
                confidence,
                created_at
            )
            VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
            """,
            (
                payload["transaction_type"],
                payload["item"],
                payload["quantity"],
                payload["unit_price"],
                payload["amount"],
                payload["customer"],
                payload["payment_status"],
                payload["notes"],
                payload["confidence"],
                datetime.now().isoformat(sep=" ", timespec="seconds"),
            ),
        )
        connection.commit()
        transaction_id = int(cursor.lastrowid)

    _apply_customer_balance_update(transaction, path)
    _apply_inventory_update(transaction, path)
    return transaction_id


def get_transactions(db_path: str | Path | None = None) -> pd.DataFrame:
    """Return all saved transactions as a Pandas DataFrame."""
    path = initialize_database(db_path)

    with sqlite3.connect(path) as connection:
        rows = connection.execute(
            """
            SELECT
                id,
                transaction_type,
                item,
                quantity,
                unit_price,
                amount,
                customer,
                payment_status,
                notes,
                confidence,
                created_at
            FROM transactions
            ORDER BY id DESC
            """
        ).fetchall()

    records: list[dict[str, Any]] = [dict(zip(COLUMNS, row, strict=True)) for row in rows]
    return pd.DataFrame.from_records(records, columns=COLUMNS)


def get_transaction(transaction_id: int, db_path: str | Path | None = None) -> Transaction | None:
    """Return one transaction by id, or None when it does not exist."""
    path = initialize_database(db_path)

    with sqlite3.connect(path) as connection:
        row = connection.execute(
            """
            SELECT
                transaction_type,
                item,
                quantity,
                unit_price,
                amount,
                customer,
                payment_status,
                notes,
                confidence
            FROM transactions
            WHERE id = ?
            """,
            (int(transaction_id),),
        ).fetchone()

    if row is None:
        return None

    payload = dict(zip(COLUMNS[1:-1], row, strict=True))
    return Transaction.model_validate(payload)


def update_transaction(
    transaction_id: int,
    transaction: Transaction,
    db_path: str | Path | None = None,
) -> bool:
    """Update a transaction and rebuild derived balances when found."""
    path = initialize_database(db_path)
    payload = transaction.model_dump()

    with sqlite3.connect(path) as connection:
        cursor = connection.execute(
            """
            UPDATE transactions
            SET
                transaction_type = ?,
                item = ?,
                quantity = ?,
                unit_price = ?,
                amount = ?,
                customer = ?,
                payment_status = ?,
                notes = ?,
                confidence = ?
            WHERE id = ?
            """,
            (
                payload["transaction_type"],
                payload["item"],
                payload["quantity"],
                payload["unit_price"],
                payload["amount"],
                payload["customer"],
                payload["payment_status"],
                payload["notes"],
                payload["confidence"],
                int(transaction_id),
            ),
        )
        connection.commit()
        updated = cursor.rowcount > 0

    if updated:
        rebuild_derived_tables(path)
    return updated


def delete_transaction(transaction_id: int, db_path: str | Path | None = None) -> bool:
    """Delete a transaction and rebuild derived balances when found."""
    path = initialize_database(db_path)

    with sqlite3.connect(path) as connection:
        cursor = connection.execute(
            "DELETE FROM transactions WHERE id = ?",
            (int(transaction_id),),
        )
        connection.commit()
        deleted = cursor.rowcount > 0

    if deleted:
        rebuild_derived_tables(path)
    return deleted


def export_transactions_csv(
    db_path: str | Path | None = None,
    export_path: str | Path | None = None,
) -> Path:
    """Export all transactions to a CSV file and return the file path."""
    ledger = get_transactions(db_path)
    if export_path is None:
        export_path = Path(tempfile.gettempdir()) / "voiceledger_transactions.csv"
    path = Path(export_path).expanduser()
    path.parent.mkdir(parents=True, exist_ok=True)
    ledger.to_csv(path, index=False, columns=COLUMNS)
    return path


def rebuild_derived_tables(db_path: str | Path | None = None) -> None:
    """Rebuild customer balances and inventory from saved transactions."""
    path = initialize_database(db_path)

    with sqlite3.connect(path) as connection:
        connection.execute("DELETE FROM customers")
        connection.execute("DELETE FROM inventory")
        rows = connection.execute(
            """
            SELECT
                transaction_type,
                item,
                quantity,
                unit_price,
                amount,
                customer,
                payment_status,
                notes,
                confidence
            FROM transactions
            ORDER BY id ASC
            """
        ).fetchall()
        connection.commit()

    for row in rows:
        transaction = Transaction.model_validate(dict(zip(COLUMNS[1:-1], row, strict=True)))
        _apply_customer_balance_update(transaction, path)
        _apply_inventory_update(transaction, path)


def _resolve_db_path(db_path: str | Path | None) -> Path:
    """Resolve an explicit or configured database path."""
    if db_path is None:
        return get_database_path()
    return Path(db_path).expanduser()


def _apply_customer_balance_update(transaction: Transaction, db_path: Path) -> None:
    """Apply customer balance side effects for credit-related transactions."""
    if not transaction.customer or transaction.amount is None:
        return

    if transaction.transaction_type == "customer_credit":
        add_credit(transaction.customer, transaction.amount, db_path)
    elif transaction.transaction_type == "customer_payment":
        record_payment(transaction.customer, transaction.amount, db_path)


def _apply_inventory_update(transaction: Transaction, db_path: Path) -> None:
    """Apply inventory side effects for stock-related transactions."""
    if not transaction.item or transaction.quantity is None:
        return

    if transaction.transaction_type == "inventory_purchase":
        add_stock(transaction.item, transaction.quantity, db_path)
    elif transaction.transaction_type == "sale":
        remove_stock(transaction.item, transaction.quantity, db_path)