wohnbot/telegram_bot.py

import os
import logging
import threading
import time
import requests
import asyncio
import httpx

# Configuration from environment
TELEGRAM_BOT_TOKEN = os.environ.get("TELEGRAM_BOT_TOKEN", "")
TELEGRAM_CHAT_ID = os.environ.get("TELEGRAM_CHAT_ID", "")
TELEGRAM_MAX_RETRIES = int(os.environ.get("TELEGRAM_MAX_RETRIES", 3))

logger = logging.getLogger(__name__)


class TelegramBot:

    async def _handle_help_command(self) -> None:
        """Send a help message with available commands."""
        help_text = (
            "<b>Available commands:</b>\n"
            "/start - Resume monitoring\n"
            "/stop - Pause monitoring\n"
            "/autopilot on|off - Enable/disable autopilot\n"
            "/status - Show current status\n"
            "/plot - Show weekly listing pattern plot\n"
            "/errorrate - Show autopilot error rate plot\n"
            "/retryfailed - Retry failed applications\n"
            "/resetlistings - Reset listings file\n"
            "/logs [n] - Show last n log lines (default 50)\n"
            "/help - Show this help message"
        )
        await self._send_message(help_text)

    async def _handle_unknown_command(self, text: str) -> None:
        """Handle unknown commands and notify the user."""
        cmd = text.split()[0] if text else text
        msg = (
            f"❓ Unknown command: <code>{cmd}</code>\n\nUse /help to see available commands."
        )
        await self._send_message(msg)

    async def _handle_start_command(self) -> None:
        """Resume monitoring for new listings."""
        self.monitor.state_manager.set_monitoring_enabled(True)
        await self._send_message(
            "▶️ <b>Monitoring RESUMED</b>\n\n"
            "Bot will now check for new listings and notify you."
        )
        logger.info("Monitoring resumed via /start command")

    async def _handle_stop_command(self) -> None:
        """Pause monitoring without stopping the bot."""
        self.monitor.state_manager.set_monitoring_enabled(False)
        await self._send_message(
            "⏸️ <b>Monitoring PAUSED</b>\n\n"
            "Bot will not check for new listings until you send /start.\n"
            "Commands like /status, /plot, and /retryfailed still work."
        )
        logger.info("Monitoring paused via /stop command")

    async def _handle_logs_command(self, num_lines: int = 50) -> None:
        """Send the last n lines of the log file."""
        log_file = "data/monitor.log"
        try:
            if not os.path.exists(log_file):
                await self._send_message("📋 No log file found.")
                return

            with open(log_file, "r", encoding="utf-8") as f:
                lines = f.readlines()

            last_lines = lines[-num_lines:] if len(lines) > num_lines else lines
            log_text = "".join(last_lines)

            # Truncate if too long for Telegram (4096 char limit)
            if len(log_text) > 4000:
                log_text = log_text[-4000:]
                log_text = "..." + log_text[log_text.find("\n") + 1:]  # Start from next newline

            message = f"📋 <b>Last {len(last_lines)} log lines:</b>\n\n<pre>{log_text}</pre>"
            await self._send_message(message)
            logger.info(f"Sent last {len(last_lines)} log lines via /logs command")
        except Exception as e:
            logger.error(f"Failed to read log file: {e}")
            await self._send_message(f"❌ Error reading logs: {e}")

    async def _handle_reset_listings_command(self) -> None:
        """Move listings.json to data/old/ with a timestamp, preserving statistics and application history."""
        import shutil
        from datetime import datetime
        try:
            listings_path = os.path.join("data", "listings.json")
            old_dir = os.path.join("data", "old")
            if os.path.exists(listings_path):
                # Ensure old_dir exists
                os.makedirs(old_dir, exist_ok=True)
                timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
                dest_path = os.path.join(
                    old_dir, f"listings_{timestamp}.json"
                )
                shutil.move(listings_path, dest_path)
                msg = (
                    f"🗑️ <b>Listings reset:</b>\n<code>listings.json</code> moved to "
                    f"<code>old/listings_{timestamp}.json</code>."
                )
            else:
                msg = "ℹ️ No listings file found to move."
            await self._send_message(msg)
        except Exception as e:
            logger.error(f"Error resetting listings: {e}")
            await self._send_message(f"[ERROR] Error resetting listings: {str(e)}")

    def __init__(self, monitor, bot_token: str | None = None, chat_id: str | None = None, event_loop=None) -> None:
        self.monitor = monitor
        self.bot_token = bot_token or TELEGRAM_BOT_TOKEN
        self.chat_id = chat_id or TELEGRAM_CHAT_ID
        self.last_update_id: int = 0
        self.running: bool = False

        # Add reference to application handler
        self.app_handler = monitor
        # Store the main event loop for thread-safe async calls
        self.event_loop = event_loop or asyncio.get_event_loop()

        # Initialize persistent httpx client with connection pooling
        self._http_client: httpx.AsyncClient | None = None

    async def _get_http_client(self) -> httpx.AsyncClient:
        """Get or create the persistent httpx client with connection pooling."""
        if self._http_client is None:
            self._http_client = httpx.AsyncClient(
                timeout=30,
                limits=httpx.Limits(max_keepalive_connections=5, max_connections=10)
            )
        return self._http_client

    async def close(self) -> None:
        """Close the httpx client gracefully."""
        if self._http_client:
            await self._http_client.aclose()
            self._http_client = None

    def start(self) -> None:
        if not self.bot_token:
            logger.warning("Telegram bot token not configured, commands disabled")
            return
        self.running = True
        thread = threading.Thread(target=self._poll_updates, daemon=True)
        thread.start()
        logger.info("Telegram command listener started")

    def stop(self) -> None:
        self.running = False

    def _poll_updates(self) -> None:
        while self.running:
            try:
                url = f"https://api.telegram.org/bot{self.bot_token}/getUpdates"
                params = {"offset": self.last_update_id + 1, "timeout": 30}
                response = requests.get(url, params=params, timeout=35)
                if response.ok:
                    data = response.json()
                    if data.get("ok") and data.get("result"):
                        for update in data["result"]:
                            self.last_update_id = update["update_id"]
                            self._handle_update(update)
            except requests.exceptions.Timeout:
                continue
            except Exception as e:
                logger.error(f"Telegram polling error: {e}")
                time.sleep(5)

    def _handle_update(self, update: dict) -> None:
        message = update.get("message", {})
        text = message.get("text", "")
        chat_id = str(message.get("chat", {}).get("id", ""))
        if chat_id != self.chat_id:
            logger.debug(f"Ignoring message from unknown chat: {chat_id}")
            return
        logger.info(f"Received Telegram command: {text}")
        loop = self.event_loop
        if text == "/start":
            asyncio.run_coroutine_threadsafe(self._handle_start_command(), loop)
        elif text == "/stop":
            asyncio.run_coroutine_threadsafe(self._handle_stop_command(), loop)
        elif text.startswith("/autopilot"):
            asyncio.run_coroutine_threadsafe(self._handle_autopilot_command(text), loop)
        elif text == "/status":
            asyncio.run_coroutine_threadsafe(self._handle_status_command(), loop)
        elif text == "/help":
            asyncio.run_coroutine_threadsafe(self._handle_help_command(), loop)
        elif text == "/plot":
            asyncio.run_coroutine_threadsafe(self._handle_plot_command(), loop)
        elif text == "/errorrate":
            asyncio.run_coroutine_threadsafe(self._handle_error_rate_command(), loop)
        elif text == "/retryfailed":
            fut = asyncio.run_coroutine_threadsafe(
                self._handle_retry_failed_command(max_retries=TELEGRAM_MAX_RETRIES),
                loop
            )
            try:
                fut.result()
            except Exception as e:
                logger.error(f"/retryfailed command failed: {e}")
        elif text == "/resetlistings":
            asyncio.run_coroutine_threadsafe(self._handle_reset_listings_command(), loop)
        elif text.startswith("/logs"):
            # Parse optional number parameter
            parts = text.split()
            num_lines = 50  # default
            if len(parts) > 1:
                try:
                    num_lines = int(parts[1])
                    num_lines = max(1, min(num_lines, 500))  # clamp between 1-500
                except ValueError:
                    pass  # use default if invalid
            asyncio.run_coroutine_threadsafe(self._handle_logs_command(num_lines), loop)
        elif text.startswith("/"):
            asyncio.run_coroutine_threadsafe(self._handle_unknown_command(text), loop)

    async def _handle_retry_failed_command(self, max_retries: int = 3) -> None:
        """Retry all failed applications up to max_retries."""
        # Ensure browser context is initialized
        if not hasattr(self.app_handler, 'context') or self.app_handler.context is None:
            if hasattr(self.app_handler, 'init_browser'):
                await self.app_handler.init_browser()
        # After (re-)init, propagate context to all sub-handlers (defensive)
        if hasattr(self.app_handler, 'context') and hasattr(self.app_handler, 'handlers'):
            for handler in self.app_handler.handlers.values():
                handler.context = self.app_handler.context
        applications = self.app_handler.load_applications()
        failed = [
            app for app in applications.values()
            if not app.get("success")
            and app.get("retries", 0) < max_retries
            and not app.get("deactivated", False)
        ]
        await self._send_message(f"[RETRY] Retrying {len(failed)} failed applications (max retries: {max_retries})...")
        if not failed:
            await self._send_message("[INFO] No failed applications to retry (or all reached max retries).")
            return
        results = {}
        details = []
        for app in failed:
            listing = {
                "id": app["listing_id"],
                "rooms": app.get("rooms", ""),
                "size": app.get("size", ""),
                "price": app.get("price", ""),
                "address": app.get("address", ""),
                "link": app.get("link", "")
            }
            retries = app.get("retries", 0) + 1
            result = await self.app_handler.apply(listing)
            result["retries"] = retries
            # Preserve original timestamp only if still failing; update on success
            if not result["success"]:
                result["timestamp"] = app.get("timestamp", result["timestamp"])
            self.app_handler.save_application(result)
            results[listing["id"]] = result
            status_emoji = "[SUCCESS]" if result["success"] else "[FAILED]"
            details.append(
                f"{status_emoji} <b>{result.get('address', '')}</b> ({result.get('company', '')})\n"
                f"<code>{result.get('link', '')}</code>\n"
                f"<i>{result.get('message', '')}</i>\n"
            )
        n_success = sum(1 for r in results.values() if r["success"])
        n_fail = sum(1 for r in results.values() if not r["success"])
        summary = f"[RETRY] Retried {len(results)} failed applications.\n[SUCCESS]: {n_success}\n[FAILED]: {n_fail}"
        if details:
            summary += "\n\n<b>Details:</b>\n" + "\n".join(details)
        await self._send_message(summary)

    async def _handle_autopilot_command(self, text: str) -> None:
        logger.info(f"Processing autopilot command: {text}")
        parts = text.split()
        if len(parts) < 2:
            await self._send_message("Usage: /autopilot on|off")
            return
        action = parts[1].lower()
        if action == "on":
            logger.info("Enabling autopilot mode")
            self.monitor.set_autopilot(True)
            await self._send_message("<b>Autopilot ENABLED</b>\n\nI will automatically apply to new listings!")
        elif action == "off":
            self.monitor.set_autopilot(False)
            await self._send_message("🛑 <b>Autopilot DISABLED</b>\n\nI will only notify you of new listings.")
        else:
            await self._send_message("Usage: /autopilot on|off")

    async def _handle_status_command(self) -> None:
        state = self.app_handler.load_state()
        autopilot = state.get("autopilot", False)
        monitoring = state.get("monitoring_enabled", True)
        applications = self.app_handler.load_applications()

        status = "<b>Monitoring:</b> " + ("▶️ RUNNING" if monitoring else "⏸️ PAUSED")
        status += "\n<b>Autopilot:</b> " + ("✅ ON" if autopilot else "🛑 OFF")
        status += f"\n📝 <b>Applications sent:</b> {len(applications)}"

        by_company: dict[str, int] = {}
        for app in applications.values():
            company = app.get("company", "unknown")
            by_company[company] = by_company.get(company, 0) + 1
        if by_company:
            status += "\n\n<b>By company:</b>"
            for company, count in sorted(by_company.items()):
                status += f"\n  • {company}: {count}"
        await self._send_message(status)

    async def _handle_plot_command(self) -> None:
        logger.info("Generating listing times plot...")
        try:
            plot_path = self.app_handler._generate_weekly_plot()
            await self._send_photo(plot_path, "\U0001f4ca <b>Weekly Listing Patterns</b>\n\nThis shows when new listings typically appear throughout the week.")
        except Exception as e:
            logger.error(f"Error generating plot: {e}")
            import traceback
            logger.error(traceback.format_exc())
            await self._send_message(f"\u274c Error generating plot: {str(e)}")

    async def _handle_error_rate_command(self) -> None:
        logger.info("Generating autopilot errorrate plot...")
        try:
            plot_path, summary = self.app_handler._generate_error_rate_plot()
            caption = "📉 <b>Autopilot Success vs Failure</b>\n\n" + summary
            await self._send_photo(plot_path, caption)
        except Exception as e:
            logger.error(f"Error generating errorrate plot: {e}")
            import traceback
            logger.error(traceback.format_exc())
            await self._send_message(f"[ERROR] Error generating errorrate plot: {str(e)}")


    async def _send_message(self, text: str) -> None:
        """Send a text message to the configured Telegram chat, with retry logic and detailed error logging (async)."""
        MAX_LENGTH = 4096  # Telegram message character limit
        if not self.bot_token or not self.chat_id:
            logger.warning("Telegram bot token or chat ID not configured, cannot send message")
            return

        # Clean text: remove invalid unicode surrogates
        text = text.encode('utf-8', errors='ignore').decode('utf-8')

        url = f"https://api.telegram.org/bot{self.bot_token}/sendMessage"

        # Split message into chunks if too long
        messages: list[str] = []
        if isinstance(text, str) and len(text) > MAX_LENGTH:
            # Try to split on line breaks for readability
            lines = text.split('\n')
            chunk = ""
            for line in lines:
                if len(chunk) + len(line) + 1 > MAX_LENGTH:
                    messages.append(chunk)
                    chunk = line
                else:
                    if chunk:
                        chunk += "\n"
                    chunk += line
            if chunk:
                messages.append(chunk)
        else:
            messages = [text]

        max_retries = 3
        retry_delay = 1  # Initial delay in seconds

        try:
            client = await self._get_http_client()
            for idx, msg in enumerate(messages):
                payload = {"chat_id": self.chat_id, "text": msg, "parse_mode": "HTML"}

                # Retry logic for each message chunk
                for attempt in range(max_retries):
                    try:
                        response = await client.post(url, json=payload)
                        if response.is_success:
                            logger.info(f"[TELEGRAM] Sent message part {idx+1}/{len(messages)}: status={response.status_code}")
                            break
                        else:
                            logger.warning(f"[TELEGRAM] Failed attempt {attempt+1}/{max_retries}: {response.status_code}")
                            if attempt < max_retries - 1:
                                wait_time = retry_delay * (2 ** attempt)
                                await asyncio.sleep(wait_time)
                            else:
                                logger.error(f"Failed to send Telegram message after {max_retries} attempts: {response.text}")
                    except httpx.TimeoutException as e:
                        logger.warning(f"[TELEGRAM] Timeout on attempt {attempt+1}/{max_retries}")
                        if attempt < max_retries - 1:
                            wait_time = retry_delay * (2 ** attempt)
                            await asyncio.sleep(wait_time)
                        else:
                            logger.error(f"Telegram message timed out after {max_retries} attempts")
                    except httpx.RequestError as e:
                        logger.warning(f"[TELEGRAM] Network error on attempt {attempt+1}/{max_retries}: {e}")
                        if attempt < max_retries - 1:
                            wait_time = retry_delay * (2 ** attempt)
                            await asyncio.sleep(wait_time)
                        else:
                            logger.error(f"Telegram message failed after {max_retries} attempts: {e}")
        except Exception as e:
            logger.error(f"Unexpected error while sending Telegram message: {e}")

    async def _send_photo(self, photo_path: str, caption: str) -> None:
        """Send a photo to the configured Telegram chat with retry logic (async)."""
        if not self.bot_token or not self.chat_id:
            logger.warning("Telegram bot token or chat ID not configured, cannot send photo")
            return

        # Clean caption: remove invalid unicode surrogates
        caption = caption.encode('utf-8', errors='ignore').decode('utf-8')

        url = f"https://api.telegram.org/bot{self.bot_token}/sendPhoto"
        max_retries = 3
        retry_delay = 1  # Initial delay in seconds

        for attempt in range(max_retries):
            try:
                with open(photo_path, "rb") as photo:
                    files = {"photo": (photo_path, photo, "image/jpeg")}
                    data = {"chat_id": self.chat_id, "caption": caption, "parse_mode": "HTML"}
                    client = await self._get_http_client()
                    response = await client.post(url, data=data, files=files)

                if response.is_success:
                    logger.info(f"[TELEGRAM] Sent photo: {photo_path}")
                    return
                else:
                    logger.warning(f"[TELEGRAM] Photo send attempt {attempt+1}/{max_retries} failed: {response.status_code}")
                    if attempt < max_retries - 1:
                        wait_time = retry_delay * (2 ** attempt)
                        await asyncio.sleep(wait_time)
                    else:
                        logger.error(f"Failed to send Telegram photo after {max_retries} attempts: {response.text}")
            except httpx.TimeoutException:
                logger.warning(f"[TELEGRAM] Photo timeout on attempt {attempt+1}/{max_retries}")
                if attempt < max_retries - 1:
                    wait_time = retry_delay * (2 ** attempt)
                    await asyncio.sleep(wait_time)
                else:
                    logger.error(f"Telegram photo timed out after {max_retries} attempts")
            except httpx.RequestError as e:
                logger.warning(f"[TELEGRAM] Network error on attempt {attempt+1}/{max_retries}: {e}")
                if attempt < max_retries - 1:
                    wait_time = retry_delay * (2 ** attempt)
                    await asyncio.sleep(wait_time)
                else:
                    logger.error(f"Telegram photo failed after {max_retries} attempts: {e}")
            except FileNotFoundError:
                logger.error(f"Photo file not found: {photo_path}")
                return  # No point retrying if file doesn't exist
            except Exception as e:
                logger.error(f"Unexpected error while sending Telegram photo: {e}")
                if attempt < max_retries - 1:
                    wait_time = retry_delay * (2 ** attempt)
                    await asyncio.sleep(wait_time)
                else:
                    return