Real-Time Crypto Scraping & Alerts Bot with Gemini AI Key Rotation and Urdu Neural TTS
In the fast-moving cryptocurrency ecosystem, information is everything. A new Binance listing, launchpool announcement, staking campaign, airdrop, or project update can dramatically impact prices within minutes. Traders who receive information first often gain a significant advantage over those who discover news later.
Manually monitoring dozens of Telegram channels throughout the day is inefficient and time-consuming. Furthermore, many announcements appear in multiple languages and contain technical terminology that requires interpretation before action can be taken.
To solve these challenges, this project introduces a fully automated Crypto Alert Bot built with Python. The system monitors Telegram channels in real time, processes announcements through Google Gemini AI, translates content into Urdu, generates natural Urdu voice notes, categorizes announcements automatically, and distributes alerts through Telegram and email.
Core Features
Real-Time Telegram Monitoring
The bot uses Telethon to connect directly to Telegram through MTProto. Unlike simple web scraping approaches, Telethon receives messages immediately after publication, making it ideal for real-time crypto monitoring.
Gemini AI Processing
Each incoming announcement is analyzed using Google’s Gemini AI model. The AI detects the original language, translates content when necessary, generates a clean Urdu version for voice synthesis, and classifies the announcement into predefined categories.
Multi-Key Rotation System
Gemini API limits can become a bottleneck during periods of heavy activity. To avoid interruptions, the bot supports multiple Gemini API keys and automatically rotates between them whenever rate limits or temporary failures occur.
Urdu Neural Voice Notes
Microsoft Edge Neural TTS converts translated Urdu text into natural voice notes using the ur-PK-UzmaNeural voice model. This allows users to listen to important announcements without reading lengthy messages.
Email Notifications
In addition to Telegram delivery, alerts can be distributed through Gmail SMTP. Email reports include category information, original message content, and optional MP3 attachments.
Concurrent Processing
Instead of processing messages sequentially, the system uses multiple asynchronous workers. This architecture allows several announcements to be translated, classified, and delivered simultaneously.
System Architecture
- Telegram channels publish new announcements.
- Telethon receives messages instantly.
- Messages are pushed into an asyncio queue.
- Worker processes extract and clean content.
- Gemini AI translates and classifies announcements.
- Urdu TTS generates voice notes.
- Alerts are delivered through Telegram and Email.
Installation
Create a file named requirements.txt and install dependencies.
pip install -r requirements.txtconfig.py
# =========================
# TELEGRAM API
# =========================
api_id = 12345678
api_hash = "YOUR_TELEGRAM_API_HASH"
# =========================
# CHANNELS TO MONITOR
# =========================
TELEGRAM_CHANNELS = [
"binance_wallet_announcements",
"binance_announcements",
"techearningsupport",
"GaCryptOfficial",
"AirdropAlertDaily",
"freemejankari"
]
# =========================
# TELEGRAM RECIPIENTS
# =========================
ALERT_USERS = [
"@your_username"
]
# =========================
# EMAIL SETTINGS
# =========================
EMAIL_ENABLED = True
EMAIL_SENDER = "your_email@gmail.com"
EMAIL_PASSWORD = "your_app_password"
SMTP_SERVER = "smtp.gmail.com"
SMTP_PORT = 587
ALERT_EMAILS = [
"your_email@gmail.com"
]
# =========================
# GEMINI API KEYS
# =========================
GEMINI_KEYS = [
"YOUR_GEMINI_KEY_1",
"YOUR_GEMINI_KEY_2",
"YOUR_GEMINI_KEY_3"
]
GEMINI_MODEL = "gemini-2.5-flash"
GEMINI_ENABLED = Truemonitor.py
import asyncio
import logging
import os
import tempfile
import re
import time
import json
import smtplib
from datetime import datetime
from collections import deque
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
from email.mime.audio import MIMEAudio
import edge_tts
from telethon import TelegramClient
from telethon import events
from google import genai
from google.genai import types
import config
# =========================
# LOGGING
# =========================
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(__name__)
# =========================
# TELEGRAM CLIENT
# =========================
client = TelegramClient(
"monitor_session",
config.api_id,
config.api_hash
)
# =========================
# QUEUE
# =========================
queue = asyncio.Queue()
processed = deque(
maxlen=1000
)
# =========================
# CATEGORY MAP
# =========================
CATEGORY_EMOJIS = {
"airdrop": "🎁 Airdrop",
"booster": "⚡ Booster",
"campaign": "📅 Campaign",
"competition": "🏆 Competition",
"splash": "💦 Splash",
"boost": "🚀 Boost",
"update": "🔄 Update",
"alpha": "🧪 Alpha",
"claim": "💰 Claim",
"general": "📢 General"
}
def classify_text_by_keywords(text):
text_lower = text.lower()
if any(
k in text_lower
for k in [
"airdrop",
"drop",
"free token",
"distribution"
]
):
return "airdrop"
if any(
k in text_lower
for k in [
"booster",
"boosted",
"multiplier"
]
):
return "booster"
if any(
k in text_lower
for k in [
"campaign",
"event",
"quest",
"task"
]
):
return "campaign"
if any(
k in text_lower
for k in [
"competition",
"contest",
"tournament",
"race"
]
):
return "competition"
if any(
k in text_lower
for k in [
"launchpool",
"listing",
"deposit"
]
):
return "splash"
return "general"# =========================
# GEMINI KEY MANAGER
# =========================
class GeminiKeyManager:
def __init__(
self,
keys,
delay=60
):
self.keys = [
{
"key": k,
"last_used": 0,
"failed_until": 0
}
for k in keys
]
self.delay = delay
self.lock = asyncio.Lock()
async def get_available_key(self):
async with self.lock:
while True:
now = time.time()
active_keys = [
k for k in self.keys
if now >= k["failed_until"]
]
if not active_keys:
logger.warning(
"All Gemini keys failed. Resetting."
)
for k in self.keys:
k["failed_until"] = 0
active_keys = self.keys
ready_key = None
min_wait = float("inf")
for k in active_keys:
elapsed = now - k["last_used"]
if elapsed >= self.delay:
ready_key = k
break
wait_needed = (
self.delay - elapsed
)
if wait_needed < min_wait:
min_wait = wait_needed
if ready_key:
ready_key["last_used"] = time.time()
return ready_key["key"]
logger.info(
f"Waiting {min_wait:.1f}s for API key"
)
await asyncio.sleep(min_wait)
def mark_failed(
self,
key_str,
duration=300
):
for k in self.keys:
if k["key"] == key_str:
k["failed_until"] = (
time.time() + duration
)
logger.warning(
f"Key failed for {duration}s"
)
break
key_manager = GeminiKeyManager(
config.GEMINI_KEYS,
delay=60
)
# =========================
# CLEAN TEXT
# =========================
def clean_text(text):
if not text:
return ""
text = re.sub(
r"http\S+",
"",
text
)
text = re.sub(
r"@\w+",
"",
text
)
text = re.sub(
r"[^\w\s\u0600-\u06FF.,!?()%:/+-]",
"",
text
)
text = re.sub(
r"\s+",
" ",
text
)
return text.strip()
# =========================
# GEMINI PROCESSING
# =========================
async def process_with_gemini(text):
if not text:
return (
"general",
"",
text
)
max_retries = len(
config.GEMINI_KEYS
)
for attempt in range(
max_retries
):
key = await (
key_manager.get_available_key()
)
try:
ai_client = genai.Client(
api_key=key
)
prompt = f"""
Analyze this crypto news.
1. Detect language.
2. Translate to English.
3. Translate to Urdu.
4. Categorize as:
- airdrop
- booster
- campaign
- competition
- splash
- boost
- update
- alpha
- claim
- general
Return JSON:
{{
"detected_language":"",
"english_text":"",
"urdu_translation":"",
"category":""
}}
TEXT:
{text}
"""
response = await (
ai_client.aio.models.generate_content(
model=config.GEMINI_MODEL,
contents=prompt,
config=types.GenerateContentConfig(
response_mime_type=
"application/json"
)
)
)
result = (
response.text.strip()
)
try:
data = json.loads(
result
)
category = data.get(
"category",
"general"
).strip().lower()
urdu_tts = data.get(
"urdu_translation",
""
).strip()
english_text = data.get(
"english_text",
text
).strip()
urdu_tts = clean_text(
urdu_tts
)
return (
category,
urdu_tts,
english_text
)
except Exception as parse_error:
logger.error(
f"JSON Error: {parse_error}"
)
return (
classify_text_by_keywords(
text
),
clean_text(result),
text
)
except Exception as e:
logger.error(
f"Gemini Error: {e}"
)
key_manager.mark_failed(
key,
duration=300
)
continue
logger.error(
"All Gemini Keys Failed"
)
return (
classify_text_by_keywords(text),
text,
text
)# =========================
# EMAIL ALERT
# =========================
def send_email_alert(
subject,
html_body,
audio_path=None
):
if not config.EMAIL_ENABLED:
return
try:
msg = MIMEMultipart()
msg["From"] = (
config.EMAIL_SENDER
)
msg["To"] = ", ".join(
config.ALERT_EMAILS
)
msg["Subject"] = subject
msg.attach(
MIMEText(
html_body,
"html"
)
)
if (
audio_path and
os.path.exists(audio_path)
):
with open(
audio_path,
"rb"
) as f:
audio_data = f.read()
audio_attachment = (
MIMEAudio(
audio_data,
_subtype="mp3"
)
)
audio_attachment.add_header(
"Content-Disposition",
"attachment",
filename=os.path.basename(
audio_path
)
)
msg.attach(
audio_attachment
)
with smtplib.SMTP(
config.SMTP_SERVER,
config.SMTP_PORT
) as server:
server.starttls()
server.login(
config.EMAIL_SENDER,
config.EMAIL_PASSWORD
)
server.sendmail(
config.EMAIL_SENDER,
config.ALERT_EMAILS,
msg.as_string()
)
logger.info(
"Email Sent Successfully"
)
except Exception as e:
logger.error(
f"Email Error: {e}"
)
# =========================
# URDU TTS
# =========================
async def generate_tts(
text,
path
):
try:
communicate = (
edge_tts.Communicate(
text=text,
voice="ur-PK-UzmaNeural",
rate="+0%"
)
)
await communicate.save(
path
)
return True
except Exception as e:
logger.error(
f"TTS Error: {e}"
)
return False
# =========================
# WORKER
# =========================
async def worker(
worker_id=1
):
logger.info(
f"Worker {worker_id} Started"
)
while True:
event = await queue.get()
try:
text = getattr(
event,
"raw_text",
""
)
if not text:
queue.task_done()
continue
unique_id = (
f"{event.chat_id}_{event.id}"
)
if unique_id in processed:
queue.task_done()
continue
processed.append(
unique_id
)
try:
chat = await (
event.get_chat()
)
channel = getattr(
chat,
"title",
"Telegram"
)
if getattr(
chat,
"username",
None
):
link = (
f"https://t.me/"
f"{chat.username}/"
f"{event.id}"
)
else:
link = (
"Private Channel"
)
except Exception:
channel = "Telegram"
link = (
"Private Channel"
)
logger.info(
f"Processing {channel}"
)
summary = clean_text(
text
)
if len(summary) > 1000:
summary = (
summary[:1000]
)
category = (
"general"
)
urdu_tts = (
summary
)
alert_text = (
summary
)
if config.GEMINI_ENABLED:
(
category,
urdu_tts,
alert_text
) = await (
process_with_gemini(
summary
)
)
else:
category = (
classify_text_by_keywords(
summary
)
)
urdu_tts = clean_text(
urdu_tts
)
if len(
urdu_tts.strip()
) < 5:
urdu_tts = (
"معلومات دستیاب نہیں"
)
category_display = (
CATEGORY_EMOJIS.get(
category,
"📢 General"
)
)
message = f"""
Alert from HadiW3B.com
📌 Category:
{category_display}
📢 Channel:
{channel}
📝 Message:
{alert_text}
🔗 Link:
{link}
"""
logger.info(
f"Prepared Alert"
)
with tempfile.NamedTemporaryFile(
suffix=".mp3",
delete=False
) as f:
path = f.name
ok = await (
generate_tts(
urdu_tts,
path
)
)
if ok:
for user in (
config.ALERT_USERS
):
try:
entity = await (
client.get_entity(
user
)
)
await (
client.send_file(
entity,
path,
voice_note=True,
caption=message[:1024]
)
)
logger.info(
f"Sent To {user}"
)
except Exception as e:
logger.error(
f"Send Error: {e}"
) # =========================
# EMAIL ALERT
# =========================
if config.EMAIL_ENABLED:
time_str = (
datetime.now()
.strftime(
"%Y-%m-%d %H:%M:%S"
)
)
subject = (
f"Crypto Alert "
f"[{category.upper()}]"
)
html_body = f"""
<html>
<body>
<h2>
Crypto Alert
</h2>
<p>
<b>Channel:</b>
{channel}
</p>
<p>
<b>Category:</b>
{category_display}
</p>
<p>
<b>Time:</b>
{time_str}
</p>
<hr>
<h3>
Original Post
</h3>
<p>
{alert_text}
</p>
<p>
<a href="{link}">
View Original Post
</a>
</p>
</body>
</html>
"""
asyncio.create_task(
asyncio.to_thread(
send_email_alert,
subject,
html_body,
path if ok else None
)
)
# =========================
# TEMP FILE CLEANUP
# =========================
async def safe_remove(
filepath
):
await asyncio.sleep(
15
)
try:
if os.path.exists(
filepath
):
os.remove(
filepath
)
logger.info(
f"Removed {filepath}"
)
except Exception as e:
logger.error(
f"Remove Error: {e}"
)
asyncio.create_task(
safe_remove(path)
)
except Exception as e:
logger.error(
f"Worker Error: {e}"
)
finally:
queue.task_done()
# =========================
# MESSAGE HANDLER
# =========================
@client.on(
events.NewMessage(
chats=config.TELEGRAM_CHANNELS
)
)
async def handler(event):
await queue.put(
event
)
# =========================
# STARTUP TEST
# =========================
async def startup():
logger.info(
"Startup Test"
)
sent = False
for ch in (
config.TELEGRAM_CHANNELS
):
if sent:
break
try:
async for msg in (
client.iter_messages(
ch,
limit=1
)
):
await queue.put(
msg
)
sent = True
logger.info(
"Startup Message Queued"
)
break
except Exception as e:
logger.error(
f"Startup Error: {e}"
)
# =========================
# MAIN
# =========================
async def main():
logger.info(
"Bot Starting..."
)
await client.start()
logger.info(
"Telegram Connected"
)
for i in range(
1,
6
):
asyncio.create_task(
worker(i)
)
await startup()
logger.info(
"Monitoring Started"
)
await (
client.run_until_disconnected()
)
# =========================
# ENTRY POINT
# =========================
if __name__ == "__main__":
asyncio.run(
main()
)requirements.txt
telethon
requests
beautifulsoup4
lxml
edge-tts
static-ffmpeg
google-genai
cloudscraper
urllib3
certifi
websocket-client
PySocks
sniffio
outcome
sortedcontainers
h11
wsproto
trio
trio-websocketUsage
After creating config.py, monitor.py, and requirements.txt, install dependencies and run the bot:
pip install -r requirements.txt
python monitor.pyOn the first launch, Telethon will request your Telegram phone number and verification code. Once authenticated, a local session file will be created automatically. The bot will then continuously monitor configured channels, translate announcements, generate Urdu voice notes, and deliver alerts through Telegram and email.
Conclusion
This project demonstrates how Telethon, Gemini AI, Edge TTS, asyncio workers, Telegram delivery, and SMTP email notifications can be combined into a complete crypto intelligence system. The architecture is scalable, efficient, and capable of monitoring multiple announcement channels simultaneously while providing multilingual notifications and voice summaries.