From 56df23aba0c2e514f1e5d0352c5e320af4971463 Mon Sep 17 00:00:00 2001 From: twentyOne2x Date: Wed, 1 Jul 2026 00:26:00 +0200 Subject: [PATCH] Render Leo paid research as Telegram payment card --- telegram/bot.py | 19 ++++---- telegram/http_chat_proxy.py | 48 +++++++++++++++++++ tests/test_telegram_leo_x402_bridge.py | 64 ++++++++++++++++++++++++++ 3 files changed, 122 insertions(+), 9 deletions(-) diff --git a/telegram/bot.py b/telegram/bot.py index 998f95a..b1b5996 100644 --- a/telegram/bot.py +++ b/telegram/bot.py @@ -58,7 +58,7 @@ from http_chat_proxy import ( classify_smart_research_auto_resume_response, extract_auto_smart_research_followup_goal, extract_auto_smart_research_goal, - extract_checkout_qr_url, + extract_checkout_qr_photo_message, extract_paid_work_order_id, extract_smart_research_goal, post_chat_proxy, @@ -194,15 +194,16 @@ async def _reply_text_native(msg, text: str, *, do_quote: bool = True): first = False -async def _reply_checkout_qr_photo(msg, proxy_body: dict | None) -> bool: - qr_url = extract_checkout_qr_url(proxy_body) - if not qr_url: +async def _reply_checkout_qr_photo(msg, proxy_body: dict | None, *, do_quote: bool = True) -> bool: + card = extract_checkout_qr_photo_message(proxy_body) + if not card: return False try: await msg.reply_photo( - photo=qr_url, - caption="QR checkout for Leo paid research", - do_quote=False, + photo=card["photo_url"], + caption=card["caption_html"], + parse_mode=card.get("parse_mode") or "HTML", + do_quote=do_quote, ) return True except Exception as e: @@ -1435,8 +1436,8 @@ async def handle_tagged(update: Update, context: ContextTypes.DEFAULT_TYPE): ) return - await _reply_text_native(msg, proxy_reply, do_quote=True) - await _reply_checkout_qr_photo(msg, proxy_body) + if not await _reply_checkout_qr_photo(msg, proxy_body, do_quote=True): + await _reply_text_native(msg, proxy_reply, do_quote=True) if _should_start_smart_research_auto_resume_poll( paid_work_order_id=paid_work_order_id, diff --git a/telegram/http_chat_proxy.py b/telegram/http_chat_proxy.py index 3e29687..72d9549 100644 --- a/telegram/http_chat_proxy.py +++ b/telegram/http_chat_proxy.py @@ -3,6 +3,7 @@ from __future__ import annotations import re +from html import escape from typing import Any DEFAULT_SMART_RESEARCH_COMMAND_PREFIXES = ("/smart_research", "/paid_research") @@ -192,6 +193,53 @@ def extract_checkout_qr_url(payload: dict[str, Any] | None) -> str | None: return url +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.""" + qr_url = extract_checkout_qr_url(payload) + if not qr_url or not isinstance(payload, dict): + return None + + checkout = payload.get("checkout") + if not isinstance(checkout, dict): + return None + + telegram = checkout.get("telegram") + caption_html = None + if isinstance(telegram, dict): + raw_caption = telegram.get("captionHtml") or telegram.get("caption_html") + if isinstance(raw_caption, str) and 1 <= len(raw_caption) <= 1024: + caption_html = raw_caption.strip() + + if not caption_html: + price = str(checkout.get("priceUsd") or checkout.get("price_usd") or "0.07").strip() + network = str(checkout.get("networkLabel") or checkout.get("network") or "Solana mainnet").strip() + payment_address = checkout.get("paymentAddress") or checkout.get("payment_address") + address_line = ( + f"Recipient: {escape(str(payment_address), quote=False)}" + if isinstance(payment_address, str) and payment_address.strip() + else None + ) + caption_html = "\n".join( + line + for line in [ + "Paid Leo research", + f"{escape(price, quote=False)} USDC on {escape(network, quote=False)}.", + "Scan the QR to pay. I will continue here after payment.", + address_line, + ] + if line + ) + + if len(caption_html) > 1024: + caption_html = caption_html[:1000].rstrip() + "..." + + return { + "photo_url": qr_url, + "caption_html": caption_html, + "parse_mode": "HTML", + } + + def should_attach_structured_market_context(message: str) -> bool: """Return true only for explicit market-data questions, not social narrative research.""" text = message.strip() diff --git a/tests/test_telegram_leo_x402_bridge.py b/tests/test_telegram_leo_x402_bridge.py index 259f029..810b190 100644 --- a/tests/test_telegram_leo_x402_bridge.py +++ b/tests/test_telegram_leo_x402_bridge.py @@ -16,6 +16,7 @@ from http_chat_proxy import ( # noqa: E402 classify_smart_research_auto_resume_response, extract_auto_smart_research_followup_goal, extract_auto_smart_research_goal, + extract_checkout_qr_photo_message, extract_checkout_qr_url, extract_paid_work_order_id, extract_chat_proxy_reply, @@ -288,6 +289,69 @@ def test_extract_checkout_qr_url(payload, expected): assert extract_checkout_qr_url(payload) == expected +def test_extract_checkout_qr_photo_message_prefers_structured_telegram_card(): + payload = { + "checkout": { + "checkoutQrUrl": ( + "https://leo.livingip.xyz/api/agents/leo/research/checkout-qr?" + "url=https%3A%2F%2Fleo.livingip.xyz%2Fagents%2Fleo%2Fresearch%2Fcheckout%3Fq%3Dtest" + ), + "paymentAddress": "8EgACpZ16XWEt7YjJPsh1ZheVRZUGmmwQ8nJdmA1o5w4", + "telegram": { + "captionHtml": ( + "Paid Leo research\n" + "0.07 USDC on Solana mainnet.\n" + '' + "Pay with x402\n" + "Recipient: 8EgACpZ16XWEt7YjJPsh1ZheVRZUGmmwQ8nJdmA1o5w4" + ) + }, + } + } + + card = extract_checkout_qr_photo_message(payload) + + assert card == { + "photo_url": payload["checkout"]["checkoutQrUrl"], + "caption_html": payload["checkout"]["telegram"]["captionHtml"], + "parse_mode": "HTML", + } + assert "npx agentcash" not in card["caption_html"] + assert "Human checkout:" not in card["caption_html"] + assert "QR checkout:" not in card["caption_html"] + + +def test_extract_checkout_qr_photo_message_builds_address_fallback_caption(): + payload = { + "checkout": { + "checkoutQrUrl": "https://leo.livingip.xyz/api/agents/leo/research/checkout-qr?url=x", + "priceUsd": "0.07", + "networkLabel": "Solana mainnet", + "paymentAddress": "8EgACpZ16XWEt7YjJPsh1ZheVRZUGmmwQ8nJdmA1o5w4", + } + } + + card = extract_checkout_qr_photo_message(payload) + + assert card["photo_url"] == payload["checkout"]["checkoutQrUrl"] + assert "0.07 USDC on Solana mainnet" in card["caption_html"] + assert "8EgACpZ16XWEt7YjJPsh1ZheVRZUGmmwQ8nJdmA1o5w4" in card["caption_html"] + + +def test_extract_checkout_qr_photo_message_rejects_non_leo_qr_url(): + assert ( + extract_checkout_qr_photo_message( + { + "checkout": { + "checkoutQrUrl": "https://example.com/qr.png", + "telegram": {"captionHtml": "Paid Leo research"}, + } + } + ) + is None + ) + + @pytest.mark.parametrize( ("message", "expected"), [