lnbits-legend/lnbits/extensions/copilot/lnurl.py

import json
import hashlib
import math
from fastapi import Request
import hashlib
from http import HTTPStatus

from starlette.exceptions import HTTPException
from starlette.responses import HTMLResponse, JSONResponse  # type: ignore
import base64
from lnurl import (
    LnurlPayResponse,
    LnurlPayActionResponse,
    LnurlErrorResponse,
)  # type: ignore
from lnurl.types import LnurlPayMetadata
from lnbits.core.services import create_invoice
from .models import Copilots, CreateCopilotData
from . import copilot_ext
from .crud import get_copilot
from typing import Optional
from fastapi.params import Depends
from fastapi.param_functions import Query


@copilot_ext.get("/lnurl/{cp_id}", response_class=HTMLResponse)
async def lnurl_response(req: Request, cp_id: str = Query(None)):
    cp = await get_copilot(cp_id)
    if not cp:
        raise HTTPException(
            status_code=HTTPStatus.NOT_FOUND, detail="Copilot not found"
        )

    resp = LnurlPayResponse(
        callback=req.url_for("copilot.lnurl_callback", cp_id=cp_id, _external=True),
        min_sendable=10000,
        max_sendable=50000000,
        metadata=LnurlPayMetadata(json.dumps([["text/plain", str(cp.lnurl_title)]])),
    )

    params = resp.dict()
    if cp.show_message:
        params["commentAllowed"] = 300

    return params


@copilot_ext.get("/lnurl/cb/{cp_id}", response_class=HTMLResponse)
async def lnurl_callback(
    cp_id: str = Query(None), amount: str = Query(None), comment: str = Query(None)
):
    cp = await get_copilot(cp_id)
    if not cp:
        raise HTTPException(
            status_code=HTTPStatus.NOT_FOUND, detail="Copilot not found"
        )

    amount_received = int(amount)

    if amount_received < 10000:
        return LnurlErrorResponse(
            reason=f"Amount {round(amount_received / 1000)} is smaller than minimum 10 sats."
        ).dict()
    elif amount_received / 1000 > 10000000:
        return LnurlErrorResponse(
            reason=f"Amount {round(amount_received / 1000)} is greater than maximum 50000."
        ).dict()
    comment = ""
    if comment:
        if len(comment or "") > 300:
            return LnurlErrorResponse(
                reason=f"Got a comment with {len(comment)} characters, but can only accept 300"
            ).dict()
        if len(comment) < 1:
            comment = "none"

    payment_hash, payment_request = await create_invoice(
        wallet_id=cp.wallet,
        amount=int(amount_received / 1000),
        memo=cp.lnurl_title,
        description_hash=hashlib.sha256(
            (
                LnurlPayMetadata(json.dumps([["text/plain", str(cp.lnurl_title)]]))
            ).encode("utf-8")
        ).digest(),
        extra={"tag": "copilot", "copilot": cp.id, "comment": comment},
    )
    resp = LnurlPayActionResponse(
        pr=payment_request, success_action=None, disposable=False, routes=[]
    )
    return resp.dict()