| """Telegram notifications with product images for on-chain events.""" |
|
|
| import os |
| import logging |
| import httpx |
|
|
| logger = logging.getLogger(__name__) |
|
|
| TG_BOT_TOKEN = os.environ.get("TG_BOT_TOKEN", "") |
| TG_CHAT_ID = os.environ.get("TG_CHAT_ID", "") |
|
|
| _client = httpx.AsyncClient(timeout=15.0) |
| _base = "" |
|
|
|
|
| def _url(method: str) -> str: |
| return f"https://api.telegram.org/bot{TG_BOT_TOKEN}/{method}" |
|
|
|
|
| async def _send_photo(image_url: str, caption: str): |
| if not TG_BOT_TOKEN or not TG_CHAT_ID: |
| return |
| try: |
| await _client.post(_url("sendPhoto"), json={ |
| "chat_id": TG_CHAT_ID, |
| "photo": image_url, |
| "caption": caption, |
| "parse_mode": "Markdown", |
| }) |
| except Exception as e: |
| logger.error(f"TG sendPhoto failed: {e}") |
| await _send_text(caption) |
|
|
|
|
| async def _send_text(message: str): |
| if not TG_BOT_TOKEN or not TG_CHAT_ID: |
| return |
| try: |
| await _client.post(_url("sendMessage"), json={ |
| "chat_id": TG_CHAT_ID, |
| "text": message, |
| "parse_mode": "Markdown", |
| "disable_web_page_preview": True, |
| }) |
| except Exception as e: |
| logger.error(f"TG sendMessage failed: {e}") |
|
|
|
|
| async def product_registered(product: dict, tx_hash: str, explorer_url: str): |
| caption = ( |
| f"\U0001F3F7 *Product Registered On-Chain*\n\n" |
| f"*{product['name']}*\n" |
| f"NGN {product['price']:,} · {product.get('origin', '')}\n" |
| f"_{product['desc'][:100]}_\n\n" |
| f"\U0001F517 [View Transaction]({explorer_url})\n" |
| f"`{tx_hash}`" |
| ) |
| image = product.get("image") |
| if image: |
| await _send_photo(image, caption) |
| else: |
| await _send_text(caption) |
|
|
|
|
| async def sale_recorded(product: dict, tx_hash: str, explorer_url: str): |
| caption = ( |
| f"\U0001F4B0 *Sale Verified On-Chain*\n\n" |
| f"*{product['name']}*\n" |
| f"NGN {product['price']:,}\n" |
| f"Stock remaining: {product['stock']}\n\n" |
| f"\U0001F517 [View Transaction]({explorer_url})\n" |
| f"`{tx_hash}`" |
| ) |
| image = product.get("image") |
| if image: |
| await _send_photo(image, caption) |
| else: |
| await _send_text(caption) |
|
|
|
|
| async def bot_started(store_name: str, wallet: str, balance: float): |
| await _send_text( |
| f"\U0001F916 *StoreOS Online*\n\n" |
| f"Store: _{store_name}_\n" |
| f"Wallet: `{wallet}`\n" |
| f"Balance: {balance:.4f} 0G\n" |
| f"Chain: 0G Mainnet\n\n" |
| f"Robot is ready for customers." |
| ) |
|
|
|
|
| async def customer_spoke(text: str): |
| await _send_text(f"\U0001F3A4 *Customer said:*\n_{text}_") |
|
|
|
|
| async def assistant_replied(text: str): |
| await _send_text(f"\U0001F4AC *Robot replied:*\n_{text}_") |
|
|
