diff --git a/telegram/bot.py b/telegram/bot.py index 280792d..998f95a 100644 --- a/telegram/bot.py +++ b/telegram/bot.py @@ -58,6 +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_paid_work_order_id, extract_smart_research_goal, post_chat_proxy, @@ -193,6 +194,22 @@ 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: + return False + try: + await msg.reply_photo( + photo=qr_url, + caption="QR checkout for Leo paid research", + do_quote=False, + ) + return True + except Exception as e: + logger.warning("Telegram checkout QR photo reply failed: %s", e) + return False + + async def _typing_keepalive(chat, stop_event: asyncio.Event, interval_seconds: float = 4.0) -> None: while not stop_event.is_set(): try: @@ -1419,6 +1436,7 @@ 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 _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 481af11..3e29687 100644 --- a/telegram/http_chat_proxy.py +++ b/telegram/http_chat_proxy.py @@ -174,6 +174,24 @@ def extract_paid_work_order_id(message: str) -> str | None: return match.group(1) +def extract_checkout_qr_url(payload: dict[str, Any] | None) -> str | None: + """Return a Telegram-safe checkout QR image URL from a Leo research response.""" + if not isinstance(payload, dict): + return None + checkout = payload.get("checkout") + if not isinstance(checkout, dict): + return None + raw_url = checkout.get("checkoutQrUrl") or checkout.get("checkout_qr_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/api/agents/leo/research/checkout-qr\?", url): + return None + return url + + 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 62e2da6..259f029 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_url, extract_paid_work_order_id, extract_chat_proxy_reply, extract_smart_research_goal, @@ -238,6 +239,55 @@ def test_extract_paid_work_order_id(message, expected): assert extract_paid_work_order_id(message) == expected +@pytest.mark.parametrize( + ("payload", "expected"), + [ + ( + { + "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" + ) + } + }, + ( + "https://leo.livingip.xyz/api/agents/leo/research/checkout-qr?" + "url=https%3A%2F%2Fleo.livingip.xyz%2Fagents%2Fleo%2Fresearch%2Fcheckout%3Fq%3Dtest" + ), + ), + ( + { + "checkout": { + "checkout_qr_url": ( + "https://leo.livingip.xyz/api/agents/leo/research/checkout-qr?url=x" + ) + } + }, + "https://leo.livingip.xyz/api/agents/leo/research/checkout-qr?url=x", + ), + ({"checkout": {"checkoutQrUrl": "https://example.com/qr.png"}}, None), + ({"checkout": {"checkoutQrUrl": "http://leo.livingip.xyz/api/agents/leo/research/checkout-qr?url=x"}}, None), + ({"checkout": {"checkoutQrUrl": "https://leo.livingip.xyz/not-qr?url=x"}}, None), + ( + { + "checkout": { + "checkoutQrUrl": ( + "https://leo.livingip.xyz/api/agents/leo/research/checkout-qr?" + + ("x" * 2100) + ) + } + }, + None, + ), + ({}, None), + (None, None), + ], +) +def test_extract_checkout_qr_url(payload, expected): + assert extract_checkout_qr_url(payload) == expected + + @pytest.mark.parametrize( ("message", "expected"), [