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

---
name: telegram-automation
description: 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.
category: automation
tags: [telegram, telethon, spam, flood, sessions, cron]
triggers:
  - рассылка
  - спам
  - 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:**

```python
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:**

```python
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

```python
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:

```python
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:

```python
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:

```python
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

```python
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

```python
# 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:

```python
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:

```python
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:

```bash
#!/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:

```bash
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

```bash
# 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.