ImpTune/imptune/api/printers.py

"""Printer CRUD API — POST /printers, DELETE /printers/{id}, PATCH /printers/{id}."""
from __future__ import annotations

from pathlib import Path

from fastapi import APIRouter, Form, Request
from fastapi.responses import HTMLResponse, RedirectResponse, Response
from fastapi.templating import Jinja2Templates
from peewee import JOIN

from imptune.api.pages import group_printers_by_client
from imptune.db.models import Client, Driver, Printer

router = APIRouter(prefix="/printers")

templates = Jinja2Templates(
    directory=str(Path(__file__).parent.parent / "templates")
)

_VALID_DUPLEX = {"OneSided", "LongEdge", "ShortEdge"}
_VALID_PAPER = {"A4", "Letter", "Legal"}


def _error_response(message: str, status_code: int = 400) -> HTMLResponse:
    """Return an HTMX-friendly error fragment swapped into #printer-list."""
    return HTMLResponse(
        content=f"<div id='printer-list' class='error'><p>{message}</p></div>",
        status_code=status_code,
    )


def _validate_fields(
    name: str, ip_address: str, port_name: str, duplex_mode: str, paper_size: str
) -> HTMLResponse | None:
    """Shared field validation for create and update — None when everything is valid."""
    if not name:
        return _error_response("Printer name is required.")
    if not ip_address:
        return _error_response("IP address is required.")
    if not port_name:
        return _error_response("Port name is required.")
    if duplex_mode not in _VALID_DUPLEX:
        return _error_response(f"Invalid duplex mode: {duplex_mode}.")
    if paper_size not in _VALID_PAPER:
        return _error_response(f"Invalid paper size: {paper_size}.")
    return None


def _resolve_client(raw: str, owner) -> tuple[int | None, HTMLResponse | None]:
    """Resolve the optional client_id form field to an owned Client id.

    The form value is attacker-controlled text: a non-numeric value used to
    raise ValueError (HTTP 500) instead of the 400 the HTMX form can render.
    """
    raw = raw.strip()
    if not raw:
        return None, None
    try:
        client_fk = int(raw)
    except ValueError:
        return None, _error_response(f"Invalid client id: {raw}.")
    if Client.get_or_none((Client.id == client_fk) & (Client.owner == owner)) is None:
        return None, _error_response(f"Client {client_fk} not found.", status_code=404)
    return client_fk, None


def _resolve_driver(raw: str) -> tuple[int | None, HTMLResponse | None]:
    """Resolve the optional driver_id form field. Drivers are global/shared.

    Existence is checked here because an unknown id otherwise reaches SQLite as
    a FOREIGN KEY violation — an IntegrityError (HTTP 500) rather than a 404.
    """
    raw = raw.strip()
    if not raw:
        return None, None
    try:
        driver_fk = int(raw)
    except ValueError:
        return None, _error_response(f"Invalid driver id: {raw}.")
    if Driver.get_or_none(Driver.id == driver_fk) is None:
        return None, _error_response(f"Driver {driver_fk} not found.", status_code=404)
    return driver_fk, None


def _render_printer_list(request: Request) -> HTMLResponse:
    """Query printers with LEFT JOIN on client and render grouped partial."""
    import json

    owner = request.state.owner
    query = (
        Printer.select(Printer, Client)
        .join(Client, JOIN.LEFT_OUTER)
        .where(Printer.owner == owner)
        .order_by(Client.name, Printer.name)
    )
    grouped = group_printers_by_client(query)

    clients = list(Client.select().where(Client.owner == owner).order_by(Client.name))
    all_drivers = list(Driver.select().order_by(Driver.uploaded_at.desc()))
    driver_data = [
        {"driver": d, "names": json.loads(d.driver_desc) if d.driver_desc else []}
        for d in all_drivers
    ]

    return templates.TemplateResponse(
        request=request,
        name="partials/printer_list.html",
        context={"grouped": grouped, "clients": clients, "driver_data": driver_data},
    )


@router.post("", response_class=HTMLResponse)
def create_printer(
    request: Request,
    name: str = Form(...),
    ip_address: str = Form(...),
    port_name: str = Form(...),
    duplex_mode: str = Form("OneSided"),
    color_mode: str = Form(""),
    paper_size: str = Form("A4"),
    collate: str = Form(""),
    client_id: str = Form(""),
    driver_id: str = Form(""),
) -> Response:
    """Create a new printer configuration.

    Redirects to /printers on success; returns an inline HTMX error fragment
    (400/404) on validation failure.
    """
    name = name.strip()
    ip_address = ip_address.strip()
    port_name = port_name.strip()

    invalid = _validate_fields(name, ip_address, port_name, duplex_mode, paper_size)
    if invalid is not None:
        return invalid

    owner = request.state.owner

    # Resolve optional FK IDs — client must belong to this owner
    client_fk, error = _resolve_client(client_id, owner)
    if error is not None:
        return error
    driver_fk, error = _resolve_driver(driver_id)
    if error is not None:
        return error

    Printer.create(
        name=name,
        ip_address=ip_address,
        port_name=port_name,
        duplex_mode=duplex_mode,
        # HTML checkbox convention: "on" = True, absent/empty = False
        color_mode=color_mode == "on",
        paper_size=paper_size,
        collate=collate == "on",
        owner=owner,
        client=client_fk,
        driver=driver_fk,
    )

    return RedirectResponse(url="/printers", status_code=303)


@router.delete("/{printer_id}", response_class=HTMLResponse)
def delete_printer(request: Request, printer_id: int) -> HTMLResponse:
    """Delete a printer by ID. Returns updated printer list partial."""
    deleted = (
        Printer.delete()
        .where((Printer.id == printer_id) & (Printer.owner == request.state.owner))
        .execute()
    )
    if not deleted:
        return _error_response(f"Printer {printer_id} not found.", status_code=404)

    return _render_printer_list(request)


@router.patch("/{printer_id}", response_class=HTMLResponse)
def update_printer(
    request: Request,
    printer_id: int,
    name: str = Form(...),
    ip_address: str = Form(""),
    port_name: str = Form(""),
    duplex_mode: str = Form("OneSided"),
    color_mode: str = Form(""),
    paper_size: str = Form("A4"),
    collate: str = Form(""),
    client_id: str = Form(""),
    driver_id: str = Form(""),
) -> HTMLResponse:
    """Update an existing printer configuration in-place."""
    from datetime import UTC, datetime

    owner = request.state.owner
    printer = Printer.get_or_none((Printer.id == printer_id) & (Printer.owner == owner))
    if printer is None:
        return _error_response(f"Printer {printer_id} not found.", status_code=404)

    name = name.strip()
    ip_address = ip_address.strip() or printer.ip_address
    port_name = port_name.strip() or printer.port_name

    invalid = _validate_fields(name, ip_address, port_name, duplex_mode, paper_size)
    if invalid is not None:
        return invalid

    client_fk, error = _resolve_client(client_id, owner)
    if error is not None:
        return error
    driver_fk, error = _resolve_driver(driver_id)
    if error is not None:
        return error

    printer.name = name
    printer.ip_address = ip_address
    printer.port_name = port_name
    printer.duplex_mode = duplex_mode
    printer.color_mode = color_mode == "on"
    printer.paper_size = paper_size
    printer.collate = collate == "on"
    printer.client = client_fk
    printer.driver = driver_fk
    printer.updated_at = datetime.now(UTC).replace(tzinfo=None)
    printer.save()

    return _render_printer_list(request)