xmrcheckout/api/app/qr_codes.py

from __future__ import annotations

import base64
import os
import tempfile
from dataclasses import dataclass
from typing import Any

from fastapi import HTTPException, status
import qrcode
from qrcode.constants import ERROR_CORRECT_H
from PIL import Image

from .formatting import format_xmr_amount
from .models import Invoice


@dataclass(frozen=True)
class QrSettings:
    logo: str  # "monero" | "none" | "custom"
    logo_data_url: str | None = None


def invoice_qr_url(invoice_id: str) -> str:
    return f"/qr/{invoice_id}.png"


def ensure_invoice_qr_png(
    *,
    invoice: Invoice,
    storage_dir: str,
    settings: QrSettings,
) -> str:
    if not storage_dir:
        raise HTTPException(
            status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
            detail="QR storage is not configured",
        )
    os.makedirs(storage_dir, exist_ok=True)
    try:
        os.chmod(storage_dir, 0o755)
    except Exception:
        pass
    filename = f"{invoice.id}.png"
    path = os.path.join(storage_dir, filename)
    if os.path.exists(path):
        try:
            os.chmod(path, 0o644)
        except Exception:
            pass
        return path
    png_bytes = build_invoice_qr_png_bytes(invoice=invoice, settings=settings)
    _atomic_write(path, png_bytes)
    try:
        os.chmod(path, 0o644)
    except Exception:
        pass
    return path


def build_invoice_qr_png_bytes(*, invoice: Invoice, settings: QrSettings) -> bytes:
    uri = build_monero_uri(invoice)
    qr = qrcode.QRCode(
        error_correction=ERROR_CORRECT_H,
        border=2,
        box_size=10,
    )
    qr.add_data(uri)
    qr.make(fit=True)
    image = qr.make_image(fill_color="black", back_color="white").convert("RGBA")

    if settings.logo == "custom" and settings.logo_data_url:
        logo = _load_logo_from_data_url(settings.logo_data_url)
        image = _overlay_logo(image, logo)

    out = tempfile.SpooledTemporaryFile(max_size=1024 * 1024)
    image.save(out, format="PNG", optimize=True)
    out.seek(0)
    return out.read()


def build_monero_uri(invoice: Invoice) -> str:
    amount = format_xmr_amount(invoice.amount_xmr)
    params: dict[str, str] = {"tx_amount": amount}
    metadata = invoice.metadata_json or {}
    if isinstance(metadata, dict):
        recipient_name = metadata.get("recipient_name")
        description = metadata.get("description")
        if isinstance(recipient_name, str) and recipient_name.strip():
            params["recipient_name"] = recipient_name.strip()
        if isinstance(description, str) and description.strip():
            params["tx_description"] = description.strip()
    query = "&".join(f"{_url_escape(k)}={_url_escape(v)}" for k, v in params.items())
    return f"monero:{invoice.address}?{query}"


def resolve_qr_settings(invoice: Invoice) -> QrSettings:
    metadata: Any = invoice.metadata_json or {}
    if not isinstance(metadata, dict):
        return QrSettings(logo="monero", logo_data_url=None)
    qr_meta = metadata.get("qr")
    if not isinstance(qr_meta, dict):
        return QrSettings(logo="monero", logo_data_url=None)
    logo = qr_meta.get("logo")
    logo_data_url = qr_meta.get("logo_data_url")
    if not isinstance(logo, str) or logo not in ("monero", "none", "custom"):
        return QrSettings(logo="monero", logo_data_url=None)
    if logo == "custom" and isinstance(logo_data_url, str) and logo_data_url.strip():
        return QrSettings(logo="custom", logo_data_url=logo_data_url.strip())
    return QrSettings(logo=logo, logo_data_url=None)


def _atomic_write(path: str, data: bytes) -> None:
    directory = os.path.dirname(path) or "."
    os.makedirs(directory, exist_ok=True)
    fd, tmp_path = tempfile.mkstemp(prefix=".tmp-", dir=directory)
    try:
        with os.fdopen(fd, "wb") as tmp_file:
            tmp_file.write(data)
            tmp_file.flush()
            os.fsync(tmp_file.fileno())
        os.replace(tmp_path, path)
    finally:
        try:
            if os.path.exists(tmp_path):
                os.remove(tmp_path)
        except Exception:
            pass


def _load_logo_from_data_url(data_url: str) -> Image.Image:
    # Expected: data:image/png;base64,<...>
    try:
        header, encoded = data_url.split(",", 1)
    except ValueError as exc:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="QR logo must be a data URL",
        ) from exc
    if ";base64" not in header:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="QR logo must be base64 encoded",
        )
    try:
        raw = base64.b64decode(encoded, validate=True)
    except Exception as exc:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="QR logo could not be decoded",
        ) from exc
    try:
        from io import BytesIO

        logo = Image.open(BytesIO(raw)).convert("RGBA")
    except Exception as exc:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="QR logo image format is not supported",
        ) from exc
    return logo


def _overlay_logo(qr_image: Image.Image, logo: Image.Image) -> Image.Image:
    # Center a small logo with a white backing.
    size = min(qr_image.size)
    target_logo_size = int(size * 0.22)
    if target_logo_size < 24:
        return qr_image
    logo = logo.copy()
    logo.thumbnail((target_logo_size, target_logo_size), Image.LANCZOS)

    overlay = Image.new("RGBA", qr_image.size, (255, 255, 255, 0))
    box_size = int(size * 0.28)
    box_offset = int((size - box_size) / 2)
    box = Image.new("RGBA", (box_size, box_size), (255, 255, 255, 255))
    overlay.paste(box, (box_offset, box_offset))

    logo_offset = (int((size - logo.size[0]) / 2), int((size - logo.size[1]) / 2))
    overlay.paste(logo, logo_offset, mask=logo)

    return Image.alpha_composite(qr_image, overlay)


def _url_escape(value: str) -> str:
    from urllib.parse import quote

    return quote(value, safe="")