import json import logging from pathlib import Path from urllib.parse import urlencode from urllib.error import HTTPError, URLError from urllib.request import Request, urlopen from app.core.config import get_telegram_config logger = logging.getLogger(__name__) TELEGRAM_API_URL = "https://api.telegram.org" async def send_telegram_message(chat_id: str, text: str) -> dict: """Send a message via Telegram Bot API. Returns {success, error, status_code}.""" bot_token = (get_telegram_config().get("bot_token") or "").strip() chat_id = (chat_id or "").strip() if not bot_token: return { "success": False, "error": "Telegram bot token not configured", "status_code": 400, } if not chat_id: return { "success": False, "error": "Telegram chat_id not configured", "status_code": 400, } url = f"{TELEGRAM_API_URL}/bot{bot_token}/sendMessage" payload = json.dumps( { "chat_id": chat_id, "text": text, "parse_mode": "HTML", "disable_web_page_preview": True, } ).encode("utf-8") try: req = Request(url, data=payload, headers={"Content-Type": "application/json"}) with urlopen(req, timeout=30) as resp: data = json.loads(resp.read().decode("utf-8")) if data.get("ok"): return {"success": True, "error": None, "status_code": 200} return { "success": False, "error": data.get("description", "Unknown Telegram error"), "status_code": 400, } except HTTPError as e: description = f"Telegram API error: HTTP {e.code}" try: body = e.read().decode("utf-8") data = json.loads(body) description = data.get("description", description) except Exception: pass logger.error("Telegram send failed: %s", description) status_code = 400 if e.code in (400, 401, 403, 404) else 502 return {"success": False, "error": description, "status_code": status_code} except URLError as e: logger.error("Telegram send failed: %s", e) return {"success": False, "error": str(e), "status_code": 502} except Exception as e: logger.error("Telegram send failed: %s", e) return {"success": False, "error": str(e), "status_code": 500} def _read_telegram_json(url: str) -> dict: req = Request(url) with urlopen(req, timeout=30) as resp: return json.loads(resp.read().decode("utf-8")) def _telegram_api_url(method: str, params: dict | None = None) -> str: bot_token = (get_telegram_config().get("bot_token") or "").strip() if not bot_token: return "" url = f"{TELEGRAM_API_URL}/bot{bot_token}/{method}" if params: url = f"{url}?{urlencode(params)}" return url async def get_telegram_webhook_info() -> dict: """Read Telegram webhook status for diagnostics.""" url = _telegram_api_url("getWebhookInfo") if not url: return {"success": False, "error": "Telegram bot token not configured", "status_code": 400} try: data = _read_telegram_json(url) except HTTPError as e: return {"success": False, "error": f"Telegram API error: HTTP {e.code}", "status_code": 502} except URLError as e: return {"success": False, "error": str(e), "status_code": 502} if not data.get("ok"): return {"success": False, "error": data.get("description", "Telegram getWebhookInfo failed"), "status_code": 400} return {"success": True, "webhook": data.get("result") or {}, "status_code": 200} async def set_telegram_webhook(webhook_url: str) -> dict: """Set Telegram webhook URL. Telegram requires a public HTTPS URL.""" webhook_url = (webhook_url or "").strip() if not webhook_url.startswith("https://"): return {"success": False, "error": "Telegram webhook URL must be HTTPS", "status_code": 400} url = _telegram_api_url("setWebhook", {"url": webhook_url}) if not url: return {"success": False, "error": "Telegram bot token not configured", "status_code": 400} try: data = _read_telegram_json(url) except HTTPError as e: return {"success": False, "error": f"Telegram API error: HTTP {e.code}", "status_code": 502} except URLError as e: return {"success": False, "error": str(e), "status_code": 502} if not data.get("ok"): return {"success": False, "error": data.get("description", "Telegram setWebhook failed"), "status_code": 400} return {"success": True, "description": data.get("description", "Webhook was set"), "status_code": 200} async def get_telegram_file_metadata(file_id: str) -> dict: bot_token = (get_telegram_config().get("bot_token") or "").strip() if not bot_token: return { "success": False, "error": "Telegram bot token not configured", "status_code": 400, } try: data = _read_telegram_json(f"{TELEGRAM_API_URL}/bot{bot_token}/getFile?file_id={file_id}") except HTTPError as e: return {"success": False, "error": f"Telegram API error: HTTP {e.code}", "status_code": 502} except URLError as e: return {"success": False, "error": str(e), "status_code": 502} if not data.get("ok"): return { "success": False, "error": data.get("description", "Telegram getFile failed"), "status_code": 400, } return {"success": True, "file": data.get("result") or {}, "status_code": 200} async def download_telegram_file(file_id: str) -> dict: bot_token = (get_telegram_config().get("bot_token") or "").strip() metadata = await get_telegram_file_metadata(file_id) if not metadata.get("success"): return metadata file_path = (metadata.get("file") or {}).get("file_path") or "" if not file_path: return {"success": False, "error": "Telegram file path missing", "status_code": 400} url = f"{TELEGRAM_API_URL}/file/bot{bot_token}/{file_path}" try: with urlopen(Request(url), timeout=60) as resp: file_bytes = resp.read() content_type = resp.headers.get_content_type() or "application/octet-stream" except HTTPError as e: return {"success": False, "error": f"Telegram file download error: HTTP {e.code}", "status_code": 502} except URLError as e: return {"success": False, "error": str(e), "status_code": 502} filename = Path(file_path).name or f"{file_id}.bin" return { "success": True, "filename": filename, "content_type": content_type, "file_bytes": file_bytes, "status_code": 200, } def format_invoice_message(invoice) -> str: """Format an invoice as a Telegram message.""" status_value = str(invoice.status.value if hasattr(invoice.status, "value") else invoice.status) status_emoji = { "draft": "[DRAFT]", "issued": "[ISSUED]", "sent": "[SENT]", "partially_paid": "[PARTIAL]", "paid": "[PAID]", "overdue": "[OVERDUE]", "void": "[VOID]", } emoji = status_emoji.get(status_value, "[INFO]") return ( f"Invoice {invoice.invoice_number}\n" f"{emoji}\n" f"Customer: {getattr(invoice.customer, 'name', 'N/A')}\n" f"Amount: {invoice.currency} {float(invoice.total_amount):,.2f}\n" f"Status: {status_value.upper()}\n" f"Due Date: {invoice.due_date or 'Upon receipt'}" ) def format_reminder_message(reminder_type: str, entity_name: str, days_remaining: int, details: str = "") -> str: """Format a reminder notification as Telegram message.""" if reminder_type == "subscription_expiry": return ( "Subscription Expiry Reminder\n" f"Subscription: {entity_name}\n" f"Expires in: {days_remaining} days\n" f"{details}" ) if reminder_type == "invoice_overdue": return ( "Invoice Overdue\n" f"Invoice: {entity_name}\n" f"Days overdue: {days_remaining}\n" f"{details}" ) return ( "Reminder\n" f"Item: {entity_name}\n" f"{details}" )