147 lines
5.1 KiB
Python
147 lines
5.1 KiB
Python
"""
|
||
Lead funnel auto-reply bot — human-like DM responder.
|
||
Shows typing indicator, realistic delays, 3-4 messages → redirect to target.
|
||
|
||
Copy to project, fill in SESSION_PATH, MY_USERNAME, TARGET_LINK, run.
|
||
"""
|
||
|
||
import asyncio, json, random, time
|
||
from datetime import datetime
|
||
from pathlib import Path
|
||
from telethon import TelegramClient, events
|
||
from telethon.tl.functions.messages import SetTypingRequest
|
||
from telethon.tl.types import SendMessageTypingAction
|
||
from telethon.errors import FloodWaitError
|
||
|
||
# ── Config ──────────────────────────────────────────────
|
||
API_ID = 2040
|
||
API_HASH = "[REDACTED-KEY]"
|
||
SESSION_PATH = "session_file.session"
|
||
MY_USERNAME = "my_username"
|
||
TARGET_LINK = "https://t.me/target"
|
||
|
||
STATE_FILE = Path("lead_funnel_state.json")
|
||
LOG_FILE = Path("lead_funnel_log.txt")
|
||
|
||
FUNNEL = [
|
||
("Привет! Да, всё актуально. Что именно интересует?", (3, 7)),
|
||
("Понял, есть в наличии. Цены адекватные. Что планируешь брать?", (5, 10)),
|
||
("Я сейчас редко тут сижу. Напиши партнёру — он всегда в сети: {target}", (4, 8)),
|
||
("Скажи что от @{me}, он подберёт под бюджет. Давай 🤝", (3, 6)),
|
||
]
|
||
|
||
REPLY_DELAY_MIN = 15
|
||
REPLY_DELAY_MAX = 60
|
||
|
||
BUY_KEYWORDS = [
|
||
"купить", "продаёшь", "цена", "сколько", "стоит", "прайс",
|
||
"хочу", "интересует", "нужен", "заказ", "готов", "возьму",
|
||
"беру", "покупаю", "приобрести", "каталог", "в наличии",
|
||
"есть", "почём", "расценки", "отзывы", "тема", "подробнее",
|
||
]
|
||
|
||
def load_state():
|
||
if STATE_FILE.exists():
|
||
return json.loads(STATE_FILE.read_text(encoding="utf-8"))
|
||
return {}
|
||
|
||
def save_state(state):
|
||
STATE_FILE.write_text(json.dumps(state, indent=2, ensure_ascii=False), encoding="utf-8")
|
||
|
||
def log(msg):
|
||
line = f"[{datetime.now().strftime('%H:%M:%S')}] {msg}"
|
||
print(line)
|
||
with open(LOG_FILE, "a", encoding="utf-8") as f:
|
||
f.write(line + "\n")
|
||
|
||
def is_buy_intent(text):
|
||
if not text:
|
||
return False
|
||
t = text.lower().strip()
|
||
skip = ["привет", "здравствуй", "хай", "hello", "hi", "как дела", "салам", "ку", "здарова"]
|
||
if t in skip:
|
||
return False
|
||
return any(kw in t for kw in BUY_KEYWORDS)
|
||
|
||
async def type_and_send(client, user_id, message, typing_range):
|
||
secs = random.uniform(*typing_range)
|
||
try:
|
||
await client(SetTypingRequest(peer=user_id, action=SendMessageTypingAction()))
|
||
except Exception:
|
||
pass
|
||
await asyncio.sleep(secs)
|
||
await client.send_message(user_id, message)
|
||
log(f"SENT: {message[:60]}...")
|
||
|
||
async def process_user(client, user_id, state):
|
||
user_key = str(user_id)
|
||
ud = state.get(user_key, {})
|
||
step = ud.get("step", 0)
|
||
if ud.get("done") or step >= len(FUNNEL):
|
||
state[user_key]["done"] = True
|
||
save_state(state)
|
||
return
|
||
|
||
if step == 0:
|
||
await asyncio.sleep(random.uniform(REPLY_DELAY_MIN, REPLY_DELAY_MAX))
|
||
|
||
msg, trange = FUNNEL[step]
|
||
msg = msg.format(target=TARGET_LINK, me=MY_USERNAME)
|
||
await type_and_send(client, user_id, msg, trange)
|
||
|
||
state[user_key]["step"] = step + 1
|
||
state[user_key]["last_msg_time"] = time.time()
|
||
save_state(state)
|
||
|
||
async def funnel_timer(client):
|
||
while True:
|
||
await asyncio.sleep(30)
|
||
state = load_state()
|
||
for user_key, ud in list(state.items()):
|
||
if ud.get("done") or ud.get("step", 0) == 0:
|
||
continue
|
||
if ud.get("step", 0) >= len(FUNNEL):
|
||
state[user_key]["done"] = True
|
||
save_state(state)
|
||
continue
|
||
if time.time() - ud.get("last_msg_time", 0) < random.uniform(120, 240):
|
||
continue
|
||
try:
|
||
await process_user(client, int(user_key), state)
|
||
except FloodWaitError:
|
||
pass
|
||
except Exception as e:
|
||
log(f"Timer error: {e}")
|
||
|
||
async def main():
|
||
client = TelegramClient(SESSION_PATH, API_ID, API_HASH)
|
||
await client.start()
|
||
me = await client.get_me()
|
||
log(f"STARTED: @{me.username} ({me.first_name})")
|
||
|
||
state = load_state()
|
||
|
||
@client.on(events.NewMessage(incoming=True))
|
||
async def handler(event):
|
||
if not event.is_private or event.out:
|
||
return
|
||
sender = await event.get_sender()
|
||
uid, text = event.sender_id, event.raw_text or ""
|
||
uk = str(uid)
|
||
ud = state.get(uk, {})
|
||
|
||
if is_buy_intent(text) or ud.get("step", 0) > 0:
|
||
if uk not in state:
|
||
state[uk] = {"step": 0, "last_msg_time": 0, "first_seen": time.time()}
|
||
save_state(state)
|
||
try:
|
||
await process_user(client, uid, state)
|
||
except FloodWaitError:
|
||
pass
|
||
|
||
asyncio.create_task(funnel_timer(client))
|
||
log("Listening...")
|
||
await client.run_until_disconnected()
|
||
|
||
if __name__ == "__main__":
|
||
asyncio.run(main()) |