anti
48f486ac97
- Core Telegram monitoring pipeline (scraper, processor, notifier, downloaders) - Textual TUI frontend with thread-safe event bus - SQLite persistence, severity scoring, dedup cache - Fixed ULP parser: handles https:// truncation, port+path URLs, semicolon separator - Test suite: 88 tests across scorer, cache, database, processor
143 lines
4.6 KiB
Python
143 lines
4.6 KiB
Python
"""
|
|
main.py — Entry point for the ULP credential monitor.
|
|
|
|
Usage:
|
|
python main.py # TUI mode (default, requires textual)
|
|
python main.py --no-tui # Plain CLI mode
|
|
|
|
First run will prompt for your Telegram phone number and 2FA code
|
|
to create a session file. Subsequent runs are fully automatic.
|
|
"""
|
|
|
|
import asyncio
|
|
import logging
|
|
import sys
|
|
import shutil
|
|
import argparse
|
|
|
|
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()
|
|
|
|
|
|
# ─── 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)",
|
|
)
|
|
args = parser.parse_args()
|
|
|
|
if args.no_tui:
|
|
try:
|
|
asyncio.run(_cli_main())
|
|
except KeyboardInterrupt:
|
|
log.info("Interrupted by user.")
|
|
finally:
|
|
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.")
|
|
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:
|
|
log.info("Cleaning up tmp/...")
|
|
if config.TEMP_DIR.exists():
|
|
shutil.rmtree(config.TEMP_DIR, ignore_errors=True)
|
|
config.TEMP_DIR.mkdir()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|