db.py

import os
import hashlib
from contextlib import contextmanager
from typing import Any, Dict, List, Optional, Sequence

import psycopg2
from psycopg2.extras import RealDictCursor

DATABASE_URL = os.environ.get("NEON_DATABASE_URL")
PRIVATE_TAX_ID_PREFIX = "__PRIVATE__:"

if not DATABASE_URL:
    raise RuntimeError(
        "Brak zmiennej NEON_DATABASE_URL. Ustaw sekret w Hugging Face lub "
        "ustaw zmienną środowiskową lokalnie."
    )



@contextmanager
def db_conn():
    conn = psycopg2.connect(DATABASE_URL, cursor_factory=RealDictCursor)
    try:
        yield conn
        conn.commit()
    except Exception:
        conn.rollback()
        raise
    finally:
        conn.close()


def fetch_one(query: str, params: Sequence[Any]) -> Optional[Dict[str, Any]]:
    with db_conn() as conn, conn.cursor() as cur:
        cur.execute(query, params)
        return cur.fetchone()


def fetch_all(query: str, params: Sequence[Any] = ()) -> List[Dict[str, Any]]:
    with db_conn() as conn, conn.cursor() as cur:
        cur.execute(query, params)
        return cur.fetchall()


def execute(query: str, params: Sequence[Any]) -> None:
    with db_conn() as conn, conn.cursor() as cur:
        cur.execute(query, params)


def create_account(login: str, email: str, password_hash: str) -> int:
    with db_conn() as conn, conn.cursor() as cur:
        cur.execute(
            """
            INSERT INTO accounts (login, password_hash)
            VALUES (%s, %s)
            RETURNING id
            """,
            (login, password_hash),
        )
        account_id = cur.fetchone()["id"]
        cur.execute(
            """
            INSERT INTO business_profiles (account_id, company_name, owner_name,
                                           address_line, postal_code, city, tax_id, bank_account)
            VALUES (%s, '', '', '', '', '', '', '')
            """,
            (account_id,),
        )
        return account_id


def update_business(account_id: int, data: Dict[str, str]) -> None:
    execute(
        """
        UPDATE business_profiles
        SET company_name = %s,
            owner_name = %s,
            address_line = %s,
            postal_code = %s,
            city = %s,
            tax_id = %s,
            bank_account = %s
        WHERE account_id = %s
        """,
        (
            data["company_name"],
            data["owner_name"],
            data["address_line"],
            data["postal_code"],
            data["city"],
            data["tax_id"],
            data["bank_account"],
            account_id,
        ),
    )



def fetch_business_logo(account_id: int) -> Optional[Dict[str, Optional[str]]]:
    row = fetch_one(
        """
        SELECT logo_mime_type, logo_data_base64
        FROM business_profiles
        WHERE account_id = %s
        """,
        (account_id,),
    )
    if not row:
        return None
    mime_type = row.get("logo_mime_type")
    data_base64 = row.get("logo_data_base64")
    if not mime_type or not data_base64:
        return None
    return {"mime_type": mime_type, "data": data_base64}


def update_business_logo(account_id: int, mime: Optional[str], data_base64: Optional[str]) -> None:
    execute(
        """
        UPDATE business_profiles
        SET logo_mime_type = %s,
            logo_data_base64 = %s
        WHERE account_id = %s
        """,
        (mime, data_base64, account_id),
    )

def upsert_client(account_id: int, payload: Dict[str, str]) -> int:
    tax_id = (payload.get("tax_id") or "").strip()
    if tax_id:
        stored_tax_id = tax_id
    else:
        identity = "|".join(
            [
                str(account_id),
                (payload.get("name") or "").strip().lower(),
                (payload.get("address_line") or "").strip().lower(),
                (payload.get("postal_code") or "").strip().lower(),
                (payload.get("city") or "").strip().lower(),
                (payload.get("phone") or "").strip().lower(),
            ]
        )
        stored_tax_id = f"{PRIVATE_TAX_ID_PREFIX}{hashlib.sha1(identity.encode('utf-8')).hexdigest()[:20]}"

    row = fetch_one(
        """
        SELECT id FROM clients
        WHERE account_id = %s AND tax_id = %s
        """,
        (account_id, stored_tax_id),
    )
    if row:
        client_id = row["id"]
        execute(
            """
            UPDATE clients
            SET name = %s,
                address_line = %s,
                postal_code = %s,
                city = %s,
                phone = %s
            WHERE id = %s
            """,
            (
                payload["name"],
                payload["address_line"],
                payload["postal_code"],
                payload["city"],
                payload.get("phone"),
                client_id,
            ),
        )
        return client_id

    with db_conn() as conn, conn.cursor() as cur:
        cur.execute(
            """
            INSERT INTO clients (account_id, name, address_line, postal_code, city, tax_id, phone)
            VALUES (%s, %s, %s, %s, %s, %s, %s)
            RETURNING id
            """,
            (
                account_id,
                payload["name"],
                payload["address_line"],
                payload["postal_code"],
                payload["city"],
                stored_tax_id,
                payload.get("phone"),
            ),
        )
        return cur.fetchone()["id"]


def search_clients(account_id: int, term: str, limit: int = 10) -> List[Dict[str, Any]]:
    query = (term or "").strip().lower()
    like = f"%{query}%"
    return fetch_all(
        """
        SELECT name,
               CASE WHEN tax_id LIKE %s THEN '' ELSE tax_id END AS tax_id,
               address_line, postal_code, city, phone
        FROM clients
        WHERE account_id = %s
          AND (
            %s = '' OR
            LOWER(COALESCE(name, '')) LIKE %s OR
            LOWER(COALESCE(tax_id, '')) LIKE %s
          )
        ORDER BY LOWER(COALESCE(name, tax_id, '')) ASC
        LIMIT %s
        """,
        (f"{PRIVATE_TAX_ID_PREFIX}%", account_id, query, like, like, limit),
    )


def insert_invoice(account_id: int, client_id: int, invoice: Dict[str, Any]) -> int:
    with db_conn() as conn, conn.cursor() as cur:
        cur.execute(
            """
            INSERT INTO invoices (account_id, client_id, invoice_number, issued_at,
                                  sale_date, payment_term_days, exemption_note,
                                  total_net, total_vat, total_gross)
            VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
            RETURNING id
            """,
            (
                account_id,
                client_id,
                invoice["invoice_id"],
                invoice["issued_at"],
                invoice["sale_date"],
                invoice.get("payment_term", 14),
                invoice.get("exemption_note"),
                invoice["totals"]["net"],
                invoice["totals"]["vat"],
                invoice["totals"]["gross"],
            ),
        )
        invoice_id = cur.fetchone()["id"]

        cur.executemany(
            """
            INSERT INTO invoice_items (invoice_id, line_no, name, quantity, unit,
                                       vat_code, vat_label, unit_price_net,
                                       unit_price_gross, net_total, vat_amount, gross_total)
            VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
            """,
            [
                (
                    invoice_id,
                    idx + 1,
                    item["name"],
                    item["quantity"],
                    item.get("unit"),
                    item.get("vat_code"),
                    item.get("vat_label"),
                    item["unit_price_net"],
                    item["unit_price_gross"],
                    item["net_total"],
                    item["vat_amount"],
                    item["gross_total"],
                )
                for idx, item in enumerate(invoice["items"])
            ],
        )

        cur.executemany(
            """
            INSERT INTO invoice_vat_summary (invoice_id, vat_label, net_total, vat_total, gross_total)
            VALUES (%s, %s, %s, %s, %s)
            """,
            [
                (
                    invoice_id,
                    row["vat_label"],
                    row["net_total"],
                    row["vat_total"],
                    row["gross_total"],
                )
                for row in invoice["summary"]
            ],
        )

        return invoice_id