autoreg-skills-max/skills/telegram-automation/SKILL.md

name, description, category, tags, triggers

name	description	category	tags	triggers
telegram-automation	Telethon-based Telegram automation — spam, join, leave, session repair, flood handling. Use when building Telegram bots, mass messaging, chat management, or fixing broken Telethon sessions.	automation	telegram telethon spam flood sessions cron	рассылка спам Telegram spam telethon Telethon session session repair flood wait массовая рассылка выход из чатов join чаты авто-рассылка

Telegram Automation — Telethon Patterns

Repeatable patterns for Telegram mass-messaging, chat management, session repair, and flood-resilient cron loops.

Telethon Session Repair (SQLite Schema)

Telethon sessions are SQLite databases. Version mismatches between Telethon and session files cause schema errors:

Symptom: ValueError: too many values to unpack (expected 6)

Session has 7 columns but current Telethon expects 6. Check actual columns first — the 7-column schema varies:

Variant A: number column (Telethon v2, 2026+) Columns: dc_id, server_address, port, auth_key, takeout_id, tmp_auth_key, number

Fix — drop number column:

import sqlite3, shutil
fname = 'sessions/account.session'
shutil.copy(fname, fname + '.bak7cols')
db = sqlite3.connect(fname)
db.execute('''
    CREATE TABLE sessions_new (
        dc_id INT PRIMARY KEY, server_address TEXT, port INT,
        auth_key BLOB, takeout_id INTEGER, tmp_auth_key BLOB
    )
''')
db.execute('INSERT INTO sessions_new SELECT dc_id, server_address, port, auth_key, takeout_id, tmp_auth_key FROM sessions')
db.execute('DROP TABLE sessions')
db.execute('ALTER TABLE sessions_new RENAME TO sessions')
db.commit()
db.close()

Variant B: auth_key_id column (older) Columns: dc_id, server_address, port, auth_key, auth_key_id, takeout_id, tmp_auth_key

Fix — drop auth_key_id column:

import sqlite3, shutil
fname = 'sessions/account.session'
shutil.copy(fname, fname + '.bak7cols')
db = sqlite3.connect(fname)
db.execute('''
    CREATE TABLE sessions_new (
        dc_id INT, server_address TEXT, port INT,
        auth_key BLOB, takeout_id INTEGER, tmp_auth_key BLOB
    )
''')
db.execute('INSERT INTO sessions_new SELECT dc_id, server_address, port, auth_key, takeout_id, tmp_auth_key FROM sessions')
db.execute('DROP TABLE sessions')
db.execute('ALTER TABLE sessions_new RENAME TO sessions')
db.commit()
db.close()

Symptom: ValueError: not enough values to unpack (expected 6, got 5)

Session has 5 columns (old schema, missing auth_key_id and takeout_id). Fix by recreating with correct 6-column schema.

Symptom: TypeError: Strings must be encoded before hashing

Corrupted auth_key — session is dead. Mark as bad_session and skip.

Verification

import sqlite3
db = sqlite3.connect('session_file.session')
cur = db.execute('PRAGMA table_info(sessions)')
cols = [r[1] for r in cur.fetchall()]
assert len(cols) == 6, f"Expected 6 cols, got {len(cols)}: {cols}"
assert cols == ['dc_id', 'server_address', 'port', 'auth_key', 'auth_key_id', 'takeout_id']

Session File Discovery & Priority

When multiple session directories exist, prioritize by freshness:

def find_sessions():
    sessions = {}
    # Priority: sessions5 > sessions_clean (first scanned wins)
    for folder in ["sessions5", "sessions_clean"]:
        for f in Path(folder).glob("*.session"):
            if f.stem not in sessions:
                sessions[f.stem] = f
    return sessions

Identifying unique accounts via auth key hash

Different sessions may be the same account. Use auth key MD5 to deduplicate:

import sqlite3, hashlib, os

def get_key_hash(session_path):
    db = sqlite3.connect(session_path)
    row = db.execute('SELECT auth_key FROM sessions').fetchone()
    db.close()
    if row and row[0]:
        return hashlib.md5(row[0]).hexdigest()[:12]
    return None

# Group sessions by account
accounts = {}
for f in Path("sessions5").glob("*.session"):
    h = get_key_hash(str(f))
    if h:
        accounts.setdefault(h, []).append(f.name)

for h, files in accounts.items():
    print(f"Account {h}: {len(files)} session(s) — {', '.join(files)}")

Quick authorization check

28KB sessions are often deauthorized (session file exists but auth expired). 400KB+ sessions are typically active:

async def is_authorized(session_path):
    try:
        c = TelegramClient(session_path, API_ID, API_HASH)
        await asyncio.wait_for(c.connect(), timeout=5)
        auth = await c.is_user_authorized()
        await c.disconnect()
        return auth
    except:
        return False

Rules:

• sessions5/ — primary, most recent
• sessions_clean/ — fallback, may have repaired sessions
• sessions_archive/, sessions_banned/, sessions_dead/ — do not use for active work
• Always check session validity before connecting (try/except around TelegramClient init)

Bot Interaction — Inline Keyboard Navigation

MCP tools (maintg, [REDACTED-NAME]) strip inline keyboard buttons from bot messages — they return empty text for bot replies with inline keyboards. Always use Telethon directly for bot interaction, never browser/playwright.

Reading bot menus

from telethon import TelegramClient

client = TelegramClient(session_path, API_ID, API_HASH)
await client.connect()
bot = await client.get_input_entity("BotUsername")

msgs = await client.get_messages(bot, limit=1)
msg = msgs[0]

print(f"TEXT: {msg.text}")
if msg.buttons:
    for i, row in enumerate(msg.buttons):
        for j, btn in enumerate(row):
            url = getattr(btn, 'url', None)
            print(f"  BTN[{i}][{j}]: '{btn.text}' data={btn.data} url={url}")

Clicking inline buttons

# Click button at row X, col Y
await msg.click(row, col)

# Wait for bot response
await asyncio.sleep(1)
response = await client.get_messages(bot, limit=1)

Common pitfalls

• Button indices shift after navigation — re-fetch the message and re-read buttons before each click
• URL buttons have data=None and url=https://... — detect with getattr(btn, 'url', None)
• Callback buttons have data=b'...' — use the data string for programmatic clicking
• Empty text + inline keyboard = MCP tools will show nothing. Use Telethon.

Core pattern for mass messaging with flood resilience:

PROGRESS = Path("spam_progress.json")
p = {"sent": {}, "failed": {}, "flood_until": {}, "stats": {"total_sent": 0, "total_failed": 0}}

async def spam(account_id, chats, session_path):
    done = set(p["sent"].get(account_id, []) + p["failed"].get(account_id, []))
    remaining = [c for c in chats if c["username"] not in done]
    if not remaining:
        return

    # Skip flooded accounts
    if p["flood_until"].get(account_id, 0) > time.time():
        return

    client = None
    try:
        client = TelegramClient(str(session_path), API_ID, API_HASH)
        await client.connect()
    except Exception as e:
        log(f"BAD SESSION {account_id}: {e}")
        p["flood_until"][account_id] = int(time.time()) + 3600
        save()
        return

    msg = random.choice(MESSAGES)
    for c in remaining:
        try:
            await client.send_message(f"@{c['username']}", msg)
            p["sent"].setdefault(account_id, []).append(c["username"])
            p["stats"]["total_sent"] += 1
            save()
            await asyncio.sleep(random.uniform(4, 10))
        except FloodWaitError as e:
            p["flood_until"][account_id] = int(time.time()) + e.seconds
            save()
            break  # Stop this account, cron will retry later
        except Exception as e:
            p["failed"].setdefault(account_id, []).append(c["username"])
            save()

    await client.disconnect()

Key design decisions:

• Never await asyncio.sleep(flood_seconds) — timeouts can be 20+ hours. Save flood_until and move on.
• Save progress every message — crash-safe, resume on restart.
• Cron retry every 2 hours — picks up accounts after flood clears.
• Random delays 4-10s between messages — stay under rate limits.
• One message per account per run — use the same message for all chats on one account (less detectable).

Exit/Chat Leave Loop

Same pattern, adapted for leaving chats:

async def leave_chats(account_id, chat_list, session_path):
    undone = [c for c in chat_list if c not in PROGRESS["done"].get(account_id, [])]
    if not undone:
        return

    if p["flood_until"].get(account_id, 0) > time.time():
        return

    client = await connect_safe(session_path)
    if not client:
        return

    for chat_id in undone:
        try:
            await client.delete_dialog(chat_id)
            p["done"].setdefault(account_id, []).append(chat_id)
            save()
            await asyncio.sleep(random.uniform(2, 6))
        except FloodWaitError as e:
            p["flood_until"][account_id] = int(time.time()) + e.seconds
            save()
            break

Cron Wrapper Pattern (Lock File)

From eni-orchestrator-suite — use for all no_agent cron jobs:

#!/usr/bin/env bash
set -u
cd /path/to/project || exit 1
LOCK="script.lock"
LOG="script_cron.log"

if [ -f "$LOCK" ]; then
  pid=$(cat "$LOCK" 2>/dev/null || true)
  if [ -n "$pid" ] && kill -0 "$pid" 2>/dev/null; then
    exit 0
  fi
fi
echo $$ > "$LOCK"
trap 'rm -f "$LOCK"' EXIT

echo "===== $(date) tick =====" >> "$LOG"
python -u script.py >> "$LOG" 2>&1

Cron registration:

cronjob(action='create', name='job-name', schedule='every 2h',
         script='name.sh', no_agent=true, deliver='origin')

Windows Telethon Invocation (Hermes Venv Contamination)

Hermes venv injects its own Telethon into sys.path. System Python 3.14+ breaks Telethon SQLite sessions. Always use:

PYTHONPATH="" /c/Users/User/AppData/Local/Programs/Python/Python311/python.exe -c "..."

Rules:

• PYTHONPATH="" mandatory — otherwise venv's Telethon contaminates sys.path
• Python 3.11 (not 3.14) — 3.14 breaks SQLite session handling in Telethon
• Session path as raw string — r'C:\Users\User\.orca\sessions\[REDACTED-NAME]' (backslashes, not forward slashes)
• API_ID=23926822, API_HASH=[REDACTED-KEY] — [REDACTED-NAME] session
• AlStack channel ID: -1004420835850 — use parse_mode='html', link_preview=False
• MessageTooLongError — TG limit 4096 chars. Split into parts with (1/3) markers.

Pitfalls

• ❌ PYTHONPATH contamination — Hermes venv injects its own Telethon. Always PYTHONPATH="" when running Telethon scripts.
• ❌ Python 3.14 breaks Telethon — sqlite3.OperationalError: unable to open database file. Use Python 3.11.
• ❌ MCP tools strip inline keyboards — maintg and [REDACTED-NAME] return empty text for bot messages with inline keyboards. Use Telethon directly for any bot interaction.
• ❌ Don't use browser/playwright for Telegram — Telethon is faster, more reliable, and doesn't need browser auth. The user expects Telethon, not browser tools.
• ❌ Don't wait for FloodWait — save flood_until and let cron retry. Waiting blocks the whole run.
• ❌ Don't use session/ directory for active work — it may contain stale/banned sessions. Use sessions5/ or sessions_clean/.
• ❌ Don't crash on bad sessions — try/except around TelegramClient init, mark as flooded for 1h, continue to next account.
• ❌ Don't spam without progress tracking — if the script crashes, you lose all state. Save after every message.
• ❌ Don't use the same message for all accounts — rotate 5+ variants. One account = one message variant per run.
• ❌ Don't run spam without a cron fallback — if all accounts are flooded, cron retries automatically. Manual runs require manual retries.
• ❌ Don't use spam_progress.json and spam_live_progress.json interchangeably — they track different account sets. Keep separate progress files for separate scripts.

Verification

# Check session health
python -c "
import sqlite3
from pathlib import Path
for f in Path('sessions5').glob('*.session'):
    db = sqlite3.connect(str(f))
    cols = [r[1] for r in db.execute('PRAGMA table_info(sessions)')]
    ok = 'OK' if len(cols) == 6 else f'BAD({len(cols)})'
    print(f'{f.stem}: {ok}')
"

# Check flood status
python -c "
import json, time
p = json.load(open('spam_progress.json'))
now = time.time()
flooded = sum(1 for v in p.get('flood_until', {}).values() if v > now)
print(f'Flooded: {flooded}, Ready: {len(p[\"sent\"]) - flooded}')
"

References

• references/session-repair-recipes.md — copy-paste scripts for fixing Telethon sessions (7→6 cols, 5→6 cols, bulk health check, corrupted auth_key detection)
• references/mcp-fallback-direct-telethon.md — bypass [REDACTED-NAME] MCP when Telethon is down by connecting directly to the same SQLite session file. Includes the AI intel classification regex patterns, dedup via state file, and full cron job adaptation flow.

Templates

• templates/spam_live_template.py — canonical flood-resilient spam script. Copy to project, fill in API_ID/API_HASH/MESSAGES/BASE path, run.