stealergram/main.py

"""
main.py - Entry point for the ULP credential monitor.

Usage:
    python main.py               # TUI mode (default)
    python main.py --no-tui      # Plain CLI mode
    python main.py --web         # TUI + web frontend (port 8080)
    python main.py --no-tui --web # CLI + web frontend
"""

import asyncio
import logging
import sys
import shutil
import argparse
import threading

import config
from utils.database import init_db


# ─── Logging setup ────────────────────────────────────────────────────────────

config.LOG_FILE.parent.mkdir(parents=True, exist_ok=True)
config.TEMP_DIR.mkdir(parents=True, exist_ok=True)

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
    handlers=[
        logging.FileHandler(config.LOG_FILE, encoding="utf-8"),
    ],
)
log = logging.getLogger(__name__)

init_db()


# ─── Web thread ───────────────────────────────────────────────────────────────

def _run_web(host: str, port: int) -> None:
    """Start uvicorn in its own thread with its own asyncio loop."""
    import uvicorn
    from web.app import create_app
    uvicorn.run(create_app(), host=host, port=port, log_level="warning")


def _start_web_thread(host: str, port: int) -> threading.Thread:
    t = threading.Thread(target=_run_web, args=(host, port), daemon=True, name="web")
    t.start()
    log.info(f"Web frontend started at http://{host}:{port}")
    return t


# ─── Plain CLI mode ───────────────────────────────────────────────────────────

async def _cli_main():
    """Original asyncio main - runs without the TUI."""
    logging.getLogger().addHandler(logging.StreamHandler(sys.stdout))

    from telethon import TelegramClient
    from core.processor import compile_patterns
    from core.notifier import send_status
    from core.scraper import backfill_all, register_handlers, warm_entity_cache

    log.info("=" * 60)
    log.info("  ULP Credential Monitor - CLI mode")
    log.info("=" * 60)

    patterns = compile_patterns(config.TARGET_KEYWORDS)
    log.info(f"Loaded {len(patterns)} keyword pattern(s)")
    log.info(f"Watching {len(config.WATCHED_CHANNELS)} channel(s)")

    user_client = TelegramClient(
        config.SESSION_NAME, config.API_ID, config.API_HASH,
        connection_retries=5, auto_reconnect=True, request_retries=5,
    )
    bot_client = TelegramClient(
        "bot_session", config.API_ID, config.API_HASH,
    )

    async with user_client, bot_client:
        await bot_client.start(bot_token=config.BOT_TOKEN)
        log.info("Bot client connected.")

        await user_client.start()
        me = await user_client.get_me()
        log.info(f"User client connected as: {me.first_name} (@{me.username})")

        await send_status(
            bot_client,
            f"✅ *Monitor started*\n"
            f"User: `{me.first_name}`\n"
            f"Channels: `{len(config.WATCHED_CHANNELS)}`\n"
            f"Patterns: `{len(patterns)}`\n"
            f"Backfill: `{config.BACKFILL_LIMIT} msg/channel`",
        )

        await warm_entity_cache(user_client)
        register_handlers(user_client, bot_client, patterns)
        log.info("Live listener registered.")

        await backfill_all(user_client, bot_client, patterns)

        log.info("Listening for new messages... (Ctrl+C to stop)")
        await user_client.run_until_disconnected()

    log.info("Monitor stopped.")


# ─── Entry point ──────────────────────────────────────────────────────────────

def main():
    parser = argparse.ArgumentParser(description="ULP Credential Monitor")
    parser.add_argument("--no-tui",    action="store_true", help="Run in plain CLI mode (no Textual TUI)")
    parser.add_argument("--web",       action="store_true", help="Start web frontend")
    parser.add_argument("--web-host",  default="127.0.0.1", help="Web frontend bind host (default: 127.0.0.1)")
    parser.add_argument("--web-port",  type=int, default=8080, help="Web frontend port (default: 8080)")
    args = parser.parse_args()

    if args.web:
        _start_web_thread(args.web_host, args.web_port)

    def _cleanup():
        log.info("Cleaning up tmp/...")
        if config.TEMP_DIR.exists():
            shutil.rmtree(config.TEMP_DIR, ignore_errors=True)
            config.TEMP_DIR.mkdir()
        log.info("Done.")

    if args.no_tui:
        try:
            asyncio.run(_cli_main())
        except KeyboardInterrupt:
            log.info("Interrupted by user.")
        finally:
            _cleanup()
    else:
        try:
            from tui.app import run_tui
        except ImportError:
            print(
                "⚠  Textual is not installed. Install it with:\n"
                "     pip install textual\n"
                "Or run in plain CLI mode:\n"
                "     python main.py --no-tui",
                file=sys.stderr,
            )
            sys.exit(1)

        try:
            run_tui()
        except KeyboardInterrupt:
            pass
        finally:
            _cleanup()


if __name__ == "__main__":
    main()