How to Build a Telegram Lead Generation Bot in Python: Complete Tutorial 2026

async def classify_lead(text: str, group_name: str = "", post_topic: str = "") -> Classification:
"""Tiered classification: regex -> mini -> full."""
# Tier 1: Try regex first
result = classify_regex(text)
if result:
return result

# Tier 2: Try GPT-4o-mini (cheaper)
result = await classify_ai(text, group_name, post_topic, model="gpt-4o-mini")
if result.confidence >= 0.8:
return result

# Tier 3: Escalate to GPT-4o for ambiguous cases
if result.confidence < 0.7:
result = await classify_ai(text, group_name, post_topic, model="gpt-4o")

return result

<a name="auto-reply"></a>
9. Step 5: Auto-Reply Bot

Reply Templates

app/bot/templates.py

@dataclass
class ReplyTemplate:
name: str
text: str
follow_ups: list[str]
reply_rate: float

TEMPLATES = {
"direct": ReplyTemplate(
name="Direct Approach",
text="Hi {name}! I noticed your question about {topic}. We actually have a solution for exactly that. Want me to share some details?",
follow_ups=["Here is a quick overview:", "Would you like to see a demo?"],
reply_rate=0.40
),
"value_first": ReplyTemplate(
name="Value First",
text="Hey {name}! Regarding {topic} - we put together a free guide/checklist on this. Would you like me to send it over?",
follow_ups=["Great! I will send it now.", "Here it is! Hope you find it useful."],
reply_rate=0.55
),
"social_proof": ReplyTemplate(
name="Social Proof",
text="Hi {name}! We have helped dozens of companies with {topic}. If you are interested, I can show you a quick case study.",
follow_ups=["Here is the case study:", "Happy to walk you through how it applies to your situation."],
reply_rate=0.45
),
"question": ReplyTemplate(
name="Question Hook",
text="Hey {name}, great question about {topic}! Quick follow-up - are you currently using any solution for this, or looking to start fresh?",
follow_ups=["Makes sense! Here is what I would recommend:", "Let me know if you want to explore this further."],
reply_rate=0.42
),
}

def get_template(intent: str) -> ReplyTemplate:
"""Select best template based on lead intent."""
if intent == "HOT":
return random.choice([TEMPLATES["direct"], TEMPLATES["question"]])
else: # WARM
return random.choice([TEMPLATES["value_first"], TEMPLATES["social_proof"]])

def personalize(template: ReplyTemplate, lead_data: dict) -> str:
"""Fill template with lead-specific data."""
return template.text.format(
name=lead_data.get("first_name", "there"),
topic=lead_data.get("topic", "this"),
company_count=lead_data.get("company_count", "dozens of")
)

Auto-Reply Bot

app/bot/replier.py

from app.bot.templates import get_template, personalize
from app.db.database import get_pool
from app.db.sessions import load_session

load_dotenv()
logger = logging.getLogger(__name__)

class AutoReplier:
"""Send auto-replies to classified leads."""

def __init__(self):
self.api_id = int(os.getenv("TG_API_ID"))
self.api_hash = os.getenv("TG_API_HASH")
self.max_replies_per_hour = int(os.getenv("MAX_REPLIES_PER_HOUR", "30"))
self.db_pool = None
self._replies_sent = 0
self._hour_start = datetime.now()

async def initialize(self):
self.db_pool = await get_pool()

def _check_rate_limit(self) -> bool:
now = datetime.now()
if (now - self._hour_start).total_seconds() > 3600:
self._replies_sent = 0
self._hour_start = now
return self._replies_sent < self.max_replies_per_hour

async def process_leads(self, account_id: str):
if not self._check_rate_limit():
logger.info("Rate limit reached. Skipping.")
return

async with self.db_pool.acquire() as conn:
leads = await conn.fetch("""
SELECT id, user_id, username, first_name, text, group_name,
intent, topic, sentiment
FROM leads
WHERE account_id = $1
AND intent IN ('HOT', 'WARM')
AND replied = false
AND reply_attempted = false
ORDER BY
CASE intent WHEN 'HOT' THEN 1 WHEN 'WARM' THEN 2 END,
created_at DESC
LIMIT $2
""", account_id, self.max_replies_per_hour - self._replies_sent)

if not leads:
return

session_str = await load_session(self.db_pool, account_id)
client = TelegramClient(StringSession(session_str), self.api_id, self.api_hash)
await client.connect()

try:
for lead in leads:
try:
await self._send_reply(client, dict(lead))
self._replies_sent += 1
delay = random.uniform(15, 45)
await asyncio.sleep(delay)
except Exception as e:
logger.error(f"Failed to reply to lead: {e}")
await self._mark_failed(lead["id"])
finally:
await client.disconnect()

async def _send_reply(self, client: TelegramClient, lead: dict):
template = get_template(lead["intent"])
message = personalize(template, {
"first_name": lead.get("first_name") or "there",
"topic": lead.get("topic") or "this",
})

await client.send_message(lead["user_id"], message)

async with self.db_pool.acquire() as conn:
await conn.execute("""
UPDATE leads
SET replied = true, reply_sent_at = NOW(), reply_template = $1
WHERE id = $2
""", template.name, lead["id"])

logger.info(f"Replied to {lead.get('username', lead['user_id'])}")

async def _mark_failed(self, lead_id: int):
async with self.db_pool.acquire() as conn:
await conn.execute("""
UPDATE leads
SET reply_attempted = true, reply_error = true
WHERE id = $1
""", lead_id)

<a name="database"></a>
10. Step 6: Database Storage

Schema

CREATE TABLE IF NOT EXISTS account_sessions (
account_id UUID PRIMARY KEY REFERENCES accounts(id),
session_data TEXT NOT NULL,
updated_at TIMESTAMPTZ DEFAULT NOW()
);

CREATE TABLE IF NOT EXISTS target_groups (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
account_id UUID REFERENCES accounts(id),
group_id BIGINT NOT NULL,
group_name VARCHAR(255),
group_username VARCHAR(255),
is_joined BOOLEAN DEFAULT false,
last_scraped_at TIMESTAMPTZ,
created_at TIMESTAMPTZ DEFAULT NOW(),
UNIQUE(account_id, group_id)
);

CREATE TABLE IF NOT EXISTS scraped_posts (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
account_id UUID REFERENCES accounts(id),
group_id BIGINT NOT NULL,
group_name VARCHAR(255),
post_id BIGINT NOT NULL,
text TEXT,
date TIMESTAMPTZ,
views INT DEFAULT 0,
comments_count INT DEFAULT 0,
created_at TIMESTAMPTZ DEFAULT NOW(),
UNIQUE(group_id, post_id)
);

CREATE TABLE IF NOT EXISTS scraped_comments (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
post_db_id UUID REFERENCES scraped_posts(id),
user_id BIGINT NOT NULL,
username VARCHAR(255),
first_name VARCHAR(255),
last_name VARCHAR(255),
text TEXT NOT NULL,
date TIMESTAMPTZ,
group_id BIGINT NOT NULL,
message_id BIGINT NOT NULL,
has_phone BOOLEAN DEFAULT false,
created_at TIMESTAMPTZ DEFAULT NOW(),
UNIQUE(group_id, message_id)
);

CREATE TABLE IF NOT EXISTS leads (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
comment_id UUID REFERENCES scraped_comments(id),
account_id UUID REFERENCES accounts(id),
user_id BIGINT NOT NULL,
username VARCHAR(255),
first_name VARCHAR(255),
last_name VARCHAR(255),
text TEXT NOT NULL,
group_name VARCHAR(255),
intent VARCHAR(20) NOT NULL,
confidence FLOAT,
topic VARCHAR(255),
sentiment VARCHAR(20),
action VARCHAR(20),
reason TEXT,
classification_tier VARCHAR(20),
replied BOOLEAN DEFAULT false,
reply_sent_at TIMESTAMPTZ,
reply_template VARCHAR(100),
reply_attempted BOOLEAN DEFAULT false,
reply_error BOOLEAN DEFAULT false,
created_at TIMESTAMPTZ DEFAULT NOW()
);

CREATE TABLE IF NOT EXISTS follow_ups (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
lead_id UUID REFERENCES leads(id),
step INT NOT NULL,
message TEXT NOT NULL,
sent_at TIMESTAMPTZ,
replied BOOLEAN DEFAULT false,
created_at TIMESTAMPTZ DEFAULT NOW()
);

CREATE INDEX IF NOT EXISTS idx_leads_intent ON leads(intent);
CREATE INDEX IF NOT EXISTS idx_leads_replied ON leads(replied);
CREATE INDEX IF NOT EXISTS idx_leads_account ON leads(account_id);
CREATE INDEX IF NOT EXISTS idx_leads_created ON leads(created_at);
CREATE INDEX IF NOT EXISTS idx_comments_group ON scraped_comments(group_id);
CREATE INDEX IF NOT EXISTS idx_comments_user ON scraped_comments(user_id);

<a name="dashboard"></a>
11. Step 7: Dashboard with FastAPI

API Endpoints

app/api/main.py

load_dotenv()

app = FastAPI(title="Lead Generation Dashboard", version="1.0.0")

app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_methods=["*"],
allow_headers=["*"],
)

DB_URL = os.getenv("DATABASE_URL")

async def get_conn():
return await asyncpg.connect(DB_URL)

@app.get("/api/stats")
async def get_stats():
conn = await get_conn()
try:
today = datetime.now().replace(hour=0, minute=0, second=0, microsecond=0)

stats = await conn.fetchrow("""
SELECT
COUNT(*) as total_leads,
COUNT(*) FILTER (WHERE intent = 'HOT') as hot_leads,
COUNT(*) FILTER (WHERE intent = 'WARM') as warm_leads,
COUNT(*) FILTER (WHERE intent = 'COLD') as cold_leads,
COUNT(*) FILTER (WHERE intent = 'SPAM') as spam_leads,
COUNT(*) FILTER (WHERE replied = true) as replied,
COUNT(*) FILTER (WHERE created_at >= $1) as today_leads
FROM leads
""", today)

return {
"total": stats["total_leads"],
"hot": stats["hot_leads"],
"warm": stats["warm_leads"],
"cold": stats["cold_leads"],
"spam": stats["spam_leads"],
"replied": stats["replied"],
"today": stats["today_leads"],
"conversion_rate": round(stats["replied"] / max(stats["total_leads"], 1) * 100, 1)
}
finally:
await conn.close()

@app.get("/api/leads")
async def list_leads(
intent: str = Query(None),
replied: bool = Query(None),
limit: int = Query(50, ge=1, le=200),
offset: int = Query(0, ge=0)
):
conn = await get_conn()
try:
conditions = []
params = []

if intent:
conditions.append(f"intent = ${len(params) + 1}")
params.append(intent)
if replied is not None:
conditions.append(f"replied = ${len(params) + 1}")
params.append(replied)

where = "WHERE " + " AND ".join(conditions) if conditions else ""

query = f"""
SELECT id, user_id, username, first_name, text, group_name,
intent, confidence, topic, sentiment, action,
replied, reply_sent_at, created_at
FROM leads
{where}
ORDER BY created_at DESC
LIMIT ${len(params) + 1} OFFSET ${len(params) + 2}
"""
params.extend([limit, offset])

leads = await conn.fetch(query, *params)

return [{
"id": str(l["id"]),
"user_id": l["user_id"],
"username": l["username"],
"name": l["first_name"],
"text": l["text"],
"group": l["group_name"],
"intent": l["intent"],
"confidence": l["confidence"],
"topic": l["topic"],
"replied": l["replied"],
"date": l["created_at"].isoformat() if l["created_at"] else None
} for l in leads]
finally:
await conn.close()

if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)

<a name="scheduling"></a>
12. Step 8: Scheduling and Automation

Scheduler

app/scheduler.py

from app.scraper.pipeline import ScrapingPipeline
from app.classifier.ai_classifier import classify_lead
from app.bot.replier import AutoReplier
from app.db.database import get_pool

logger = logging.getLogger(__name__)

scheduler = AsyncIOScheduler()

async def scraping_job():
logger.info("Starting scraping job...")
pipeline = ScrapingPipeline()
await pipeline.initialize()

pool = await get_pool()
async with pool.acquire() as conn:
accounts = await conn.fetch("SELECT id FROM accounts WHERE status = 'active'")
groups = await conn.fetch("SELECT group_id FROM target_groups WHERE is_joined = true")

for account in accounts:
try:
await pipeline.scrape_account(
str(account["id"]),
[g["group_id"] for g in groups]
)
except Exception as e:
logger.error(f"Scraping failed: {e}")

async def classification_job():
logger.info("Starting classification job...")
pool = await get_pool()

async with pool.acquire() as conn:
comments = await conn.fetch("""
SELECT sc.id, sc.text, sp.group_name, sp.text as post_text
FROM scraped_comments sc
JOIN scraped_posts sp ON sc.post_db_id = sp.id
LEFT JOIN leads l ON l.comment_id = sc.id
WHERE l.id IS NULL
ORDER BY sc.created_at DESC
LIMIT 100
""")

for comment in comments:
try:
result = await classify_lead(
comment["text"],
comment["group_name"] or "",
(comment["post_text"] or "")[:200]
)

await conn.execute("""
INSERT INTO leads
(comment_id, user_id, username, first_name, last_name,
text, group_name, intent, confidence, topic, sentiment,
action, reason, classification_tier)
SELECT
sc.id, sc.user_id, sc.username, sc.first_name, sc.last_name,
sc.text, $1, $2, $3, $4, $5, $6, $7, $8
FROM scraped_comments sc
WHERE sc.id = $9
""",
comment["group_name"],
result.intent.value,
result.confidence,
result.topic,
result.sentiment,
result.action,
result.reason,
result.tier,
comment["id"]
)
except Exception as e:
logger.error(f"Classification failed: {e}")

async def auto_reply_job():
logger.info("Starting auto-reply job...")
replier = AutoReplier()
await replier.initialize()

pool = await get_pool()
async with pool.acquire() as conn:
accounts = await conn.fetch("SELECT id FROM accounts WHERE status = 'active'")

for account in accounts:
try:
await replier.process_leads(str(account["id"]))
except Exception as e:
logger.error(f"Auto-reply failed: {e}")

def setup_scheduler():
scheduler.add_job(scraping_job, IntervalTrigger(minutes=15), id="scraping", replace_existing=True)
scheduler.add_job(classification_job, IntervalTrigger(minutes=5), id="classification", replace_existing=True)
scheduler.add_job(auto_reply_job, IntervalTrigger(minutes=10), id="auto_reply", replace_existing=True)
scheduler.start()
logger.info("Scheduler started with 3 jobs")

<a name="monitoring"></a>
13. Step 9: Monitoring and Anti-Detection

Rate Limiter

app/utils/rate_limiter.py

class RateLimiter:
def __init__(self):
self._buckets = defaultdict(lambda: {"tokens": 30, "last_refill": time.time()})
self._lock = asyncio.Lock()

async def acquire(self, account_id: str, tokens: int = 1) -> bool:
async with self._lock:
bucket = self._buckets[account_id]
now = time.time()
elapsed = now - bucket["last_refill"]
bucket["tokens"] = min(30, bucket["tokens"] + elapsed * 0.5)
bucket["last_refill"] = now

if bucket["tokens"] >= tokens:
bucket["tokens"] -= tokens
return True
return False

async def wait_for_slot(self, account_id: str):
while not await self.acquire(account_id):
await asyncio.sleep(5)

rate_limiter = RateLimiter()

<a name="deployment"></a>
14. Step 10: Deployment

Docker Compose

docker-compose.yml

services:
postgres:
image: postgres:16-alpine
environment:
POSTGRES_DB: leadgen
POSTGRES_USER: leadgen_user
POSTGRES_PASSWORD: ${DB_PASSWORD}
volumes:
- pgdata:/var/lib/postgresql/data

api:
build: .
command: uvicorn app.api.main:app --host 0.0.0.0 --port 8000
environment:
DATABASE_URL: postgresql://leadgen_user:${DB_PASSWORD}@postgres:5432/leadgen
TG_API_ID: ${TG_API_ID}
TG_API_HASH: ${TG_API_HASH}
OPENAI_API_KEY: ${OPENAI_API_KEY}
ports:
- "8000:8000"
depends_on:
- postgres

worker:
build: .
command: python -m app.scheduler
environment:
DATABASE_URL: postgresql://leadgen_user:${DB_PASSWORD}@postgres:5432/leadgen
TG_API_ID: ${TG_API_ID}
TG_API_HASH: ${TG_API_HASH}
OPENAI_API_KEY: ${OPENAI_API_KEY}
depends_on:
- postgres
restart: unless-stopped

volumes:
pgdata:

Launch

<a name="advanced"></a>
15. Advanced Features

Follow-Up Sequences

<a name="troubleshooting"></a>
16. Troubleshooting

Common Issues

Problem: AuthKeyUnregisteredError
Solution: Re-generate session file

Problem: Low classification accuracy
Solution: Improve prompt, use tiered approach (regex -> mini -> full)

Problem: Replies not being delivered
Solution: Check user privacy settings

Performance Tips

1. Use connection pooling for database (asyncpg.Pool)


2. Rotate between 2-3 accounts for scraping


3. Cache Telegram entities


4. Batch database inserts (executemany)


5. Use Redis for caching frequent queries


6. Monitor rate limits actively


7. Use async/await throughout