Render Leo x402 checkout as Telegram button card
This commit is contained in:
parent
ac64fd4cf4
commit
c6d4376a51
3 changed files with 50 additions and 4 deletions
|
|
@ -35,7 +35,7 @@ from pathlib import Path
|
||||||
# Add pipeline lib to path for shared modules
|
# Add pipeline lib to path for shared modules
|
||||||
sys.path.insert(0, "/opt/teleo-eval/pipeline")
|
sys.path.insert(0, "/opt/teleo-eval/pipeline")
|
||||||
|
|
||||||
from telegram import Update
|
from telegram import InlineKeyboardButton, InlineKeyboardMarkup, Update
|
||||||
from telegram.ext import (
|
from telegram.ext import (
|
||||||
Application,
|
Application,
|
||||||
CommandHandler,
|
CommandHandler,
|
||||||
|
|
@ -198,11 +198,17 @@ async def _reply_checkout_qr_photo(msg, proxy_body: dict | None, *, do_quote: bo
|
||||||
card = extract_checkout_qr_photo_message(proxy_body)
|
card = extract_checkout_qr_photo_message(proxy_body)
|
||||||
if not card:
|
if not card:
|
||||||
return False
|
return False
|
||||||
|
reply_markup = None
|
||||||
|
if card.get("button_url"):
|
||||||
|
reply_markup = InlineKeyboardMarkup(
|
||||||
|
[[InlineKeyboardButton(card.get("button_text") or "Pay with x402", url=card["button_url"])]]
|
||||||
|
)
|
||||||
try:
|
try:
|
||||||
await msg.reply_photo(
|
await msg.reply_photo(
|
||||||
photo=card["photo_url"],
|
photo=card["photo_url"],
|
||||||
caption=card["caption_html"],
|
caption=card["caption_html"],
|
||||||
parse_mode=card.get("parse_mode") or "HTML",
|
parse_mode=card.get("parse_mode") or "HTML",
|
||||||
|
reply_markup=reply_markup,
|
||||||
do_quote=do_quote,
|
do_quote=do_quote,
|
||||||
)
|
)
|
||||||
return True
|
return True
|
||||||
|
|
|
||||||
|
|
@ -193,6 +193,28 @@ def extract_checkout_qr_url(payload: dict[str, Any] | None) -> str | None:
|
||||||
return url
|
return url
|
||||||
|
|
||||||
|
|
||||||
|
def extract_checkout_url(payload: dict[str, Any] | None) -> str | None:
|
||||||
|
"""Return a Telegram-safe Leo checkout URL for an inline payment button."""
|
||||||
|
if not isinstance(payload, dict):
|
||||||
|
return None
|
||||||
|
checkout = payload.get("checkout")
|
||||||
|
if not isinstance(checkout, dict):
|
||||||
|
return None
|
||||||
|
telegram = checkout.get("telegram")
|
||||||
|
raw_url = None
|
||||||
|
if isinstance(telegram, dict):
|
||||||
|
raw_url = telegram.get("buttonUrl") or telegram.get("button_url") or telegram.get("checkoutUrl")
|
||||||
|
raw_url = raw_url or checkout.get("checkoutUrl") or checkout.get("checkout_url")
|
||||||
|
if not isinstance(raw_url, str):
|
||||||
|
return None
|
||||||
|
url = raw_url.strip()
|
||||||
|
if len(url) > 2000:
|
||||||
|
return None
|
||||||
|
if not re.match(r"^https://leo\.livingip\.xyz/agents/leo/research/checkout\?", url):
|
||||||
|
return None
|
||||||
|
return url
|
||||||
|
|
||||||
|
|
||||||
def extract_checkout_qr_photo_message(payload: dict[str, Any] | None) -> dict[str, str] | None:
|
def extract_checkout_qr_photo_message(payload: dict[str, Any] | None) -> dict[str, str] | None:
|
||||||
"""Return a compact Telegram photo+caption payload for a Leo paid-research quote."""
|
"""Return a compact Telegram photo+caption payload for a Leo paid-research quote."""
|
||||||
qr_url = extract_checkout_qr_url(payload)
|
qr_url = extract_checkout_qr_url(payload)
|
||||||
|
|
@ -205,10 +227,14 @@ def extract_checkout_qr_photo_message(payload: dict[str, Any] | None) -> dict[st
|
||||||
|
|
||||||
telegram = checkout.get("telegram")
|
telegram = checkout.get("telegram")
|
||||||
caption_html = None
|
caption_html = None
|
||||||
|
button_text = "Pay with x402"
|
||||||
if isinstance(telegram, dict):
|
if isinstance(telegram, dict):
|
||||||
raw_caption = telegram.get("captionHtml") or telegram.get("caption_html")
|
raw_caption = telegram.get("captionHtml") or telegram.get("caption_html")
|
||||||
if isinstance(raw_caption, str) and 1 <= len(raw_caption) <= 1024:
|
if isinstance(raw_caption, str) and 1 <= len(raw_caption) <= 1024:
|
||||||
caption_html = raw_caption.strip()
|
caption_html = raw_caption.strip()
|
||||||
|
raw_button_text = telegram.get("buttonText") or telegram.get("button_text")
|
||||||
|
if isinstance(raw_button_text, str) and 1 <= len(raw_button_text.strip()) <= 40:
|
||||||
|
button_text = raw_button_text.strip()
|
||||||
|
|
||||||
if not caption_html:
|
if not caption_html:
|
||||||
price = str(checkout.get("priceUsd") or checkout.get("price_usd") or "0.07").strip()
|
price = str(checkout.get("priceUsd") or checkout.get("price_usd") or "0.07").strip()
|
||||||
|
|
@ -237,6 +263,8 @@ def extract_checkout_qr_photo_message(payload: dict[str, Any] | None) -> dict[st
|
||||||
"photo_url": qr_url,
|
"photo_url": qr_url,
|
||||||
"caption_html": caption_html,
|
"caption_html": caption_html,
|
||||||
"parse_mode": "HTML",
|
"parse_mode": "HTML",
|
||||||
|
"button_text": button_text,
|
||||||
|
"button_url": extract_checkout_url(payload) or "",
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -16,6 +16,7 @@ from http_chat_proxy import ( # noqa: E402
|
||||||
classify_smart_research_auto_resume_response,
|
classify_smart_research_auto_resume_response,
|
||||||
extract_auto_smart_research_followup_goal,
|
extract_auto_smart_research_followup_goal,
|
||||||
extract_auto_smart_research_goal,
|
extract_auto_smart_research_goal,
|
||||||
|
extract_checkout_url,
|
||||||
extract_checkout_qr_photo_message,
|
extract_checkout_qr_photo_message,
|
||||||
extract_checkout_qr_url,
|
extract_checkout_qr_url,
|
||||||
extract_paid_work_order_id,
|
extract_paid_work_order_id,
|
||||||
|
|
@ -308,8 +309,10 @@ def test_extract_checkout_qr_url(payload, expected):
|
||||||
|
|
||||||
|
|
||||||
def test_extract_checkout_qr_photo_message_prefers_structured_telegram_card():
|
def test_extract_checkout_qr_photo_message_prefers_structured_telegram_card():
|
||||||
|
checkout_url = "https://leo.livingip.xyz/agents/leo/research/checkout?q=test"
|
||||||
payload = {
|
payload = {
|
||||||
"checkout": {
|
"checkout": {
|
||||||
|
"checkoutUrl": checkout_url,
|
||||||
"checkoutQrUrl": (
|
"checkoutQrUrl": (
|
||||||
"https://leo.livingip.xyz/api/agents/leo/research/checkout-qr?"
|
"https://leo.livingip.xyz/api/agents/leo/research/checkout-qr?"
|
||||||
"url=https%3A%2F%2Fleo.livingip.xyz%2Fagents%2Fleo%2Fresearch%2Fcheckout%3Fq%3Dtest"
|
"url=https%3A%2F%2Fleo.livingip.xyz%2Fagents%2Fleo%2Fresearch%2Fcheckout%3Fq%3Dtest"
|
||||||
|
|
@ -319,10 +322,11 @@ def test_extract_checkout_qr_photo_message_prefers_structured_telegram_card():
|
||||||
"captionHtml": (
|
"captionHtml": (
|
||||||
"<b>Paid Leo research</b>\n"
|
"<b>Paid Leo research</b>\n"
|
||||||
"0.07 USDC on Solana mainnet.\n"
|
"0.07 USDC on Solana mainnet.\n"
|
||||||
'<a href="https://leo.livingip.xyz/agents/leo/research/checkout?q=test">'
|
"Pay with the button below or scan the QR.\n"
|
||||||
"Pay with x402</a>\n"
|
|
||||||
"Recipient: <code>8EgACpZ16XWEt7YjJPsh1ZheVRZUGmmwQ8nJdmA1o5w4</code>"
|
"Recipient: <code>8EgACpZ16XWEt7YjJPsh1ZheVRZUGmmwQ8nJdmA1o5w4</code>"
|
||||||
)
|
),
|
||||||
|
"buttonText": "Pay with x402",
|
||||||
|
"buttonUrl": checkout_url,
|
||||||
},
|
},
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
@ -333,7 +337,10 @@ def test_extract_checkout_qr_photo_message_prefers_structured_telegram_card():
|
||||||
"photo_url": payload["checkout"]["checkoutQrUrl"],
|
"photo_url": payload["checkout"]["checkoutQrUrl"],
|
||||||
"caption_html": payload["checkout"]["telegram"]["captionHtml"],
|
"caption_html": payload["checkout"]["telegram"]["captionHtml"],
|
||||||
"parse_mode": "HTML",
|
"parse_mode": "HTML",
|
||||||
|
"button_text": "Pay with x402",
|
||||||
|
"button_url": checkout_url,
|
||||||
}
|
}
|
||||||
|
assert checkout_url not in card["caption_html"]
|
||||||
assert "npx agentcash" not in card["caption_html"]
|
assert "npx agentcash" not in card["caption_html"]
|
||||||
assert "Human checkout:" not in card["caption_html"]
|
assert "Human checkout:" not in card["caption_html"]
|
||||||
assert "QR checkout:" not in card["caption_html"]
|
assert "QR checkout:" not in card["caption_html"]
|
||||||
|
|
@ -354,6 +361,11 @@ def test_extract_checkout_qr_photo_message_builds_address_fallback_caption():
|
||||||
assert card["photo_url"] == payload["checkout"]["checkoutQrUrl"]
|
assert card["photo_url"] == payload["checkout"]["checkoutQrUrl"]
|
||||||
assert "0.07 USDC on Solana mainnet" in card["caption_html"]
|
assert "0.07 USDC on Solana mainnet" in card["caption_html"]
|
||||||
assert "<code>8EgACpZ16XWEt7YjJPsh1ZheVRZUGmmwQ8nJdmA1o5w4</code>" in card["caption_html"]
|
assert "<code>8EgACpZ16XWEt7YjJPsh1ZheVRZUGmmwQ8nJdmA1o5w4</code>" in card["caption_html"]
|
||||||
|
assert card["button_url"] == ""
|
||||||
|
|
||||||
|
|
||||||
|
def test_extract_checkout_url_rejects_non_leo_checkout_url():
|
||||||
|
assert extract_checkout_url({"checkout": {"checkoutUrl": "https://example.com/pay"}}) is None
|
||||||
|
|
||||||
|
|
||||||
def test_extract_checkout_qr_photo_message_rejects_non_leo_qr_url():
|
def test_extract_checkout_qr_photo_message_rejects_non_leo_qr_url():
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue