#!/usr/bin/env python3 """Telegram bot — laisvas pokalbis su Claude (streaming output kaip Claude Code).""" import asyncio, json, os, re, subprocess, time, uuid from pathlib import Path from telegram import Update, constants from telegram.error import RetryAfter, BadRequest from telegram.ext import Application, MessageHandler, CommandHandler, filters, ContextTypes TOKEN = "8780969572:AAGP3udRX19hAhz-7g4j-uxJ8xSmjFz1JoI" OWNER_FILE = Path("/home/johnbarley/files/tv-bot-owner.json") SESSIONS_FILE = Path("/home/johnbarley/files/tv-bot-sessions.json") WORKDIR = "/home/johnbarley/files" CLAUDE_BIN = "/home/johnbarley/.local/bin/claude" EDIT_MIN_INTERVAL = 1.6 # sekundės tarp Telegram message edit'ų TELEGRAM_LIMIT = 4000 SYSTEM_PROMPT = """Tu esi mr.barley namų asistentas valdantis tinklo įrenginius per Pi4 (192.168.1.217). Bendrauji per Telegram — atsakyk LIETUVIŠKAI, TRUMPAI (1-3 sakiniai paprastai, daugiau tik jei tikrai būtina). NAMŲ ĮRENGINIAI: - Samsung Q60BA TV (192.168.1.163, MAC d0:c2:4e:52:f0:3d) — VALDOMA tyliai per: `python3 /home/johnbarley/files/tv-ctrl.py KEY_POWER` (KEY_VOLUP, KEY_VOLDOWN, KEY_MUTE, KEY_HOME, KEY_HDMI, KEY_MENU, KEY_RETURN, KEY_CHUP, KEY_CHDOWN, KEY_PLAY, KEY_PAUSE, KEY_UP/DOWN/LEFT/RIGHT, KEY_ENTER, KEY_GUIDE...) Galima paduoti kelias komandas: `tv-ctrl.py KEY_HOME KEY_RIGHT KEY_ENTER` TV pažadinimui iš deep sleep: `wakeonlan d0:c2:4e:52:f0:3d` (tada palauk 4s, paskui KEY_POWERON) - Samsung TV #2 (192.168.1.137) — neaktyvinta (reikia 1× Allow popup) - Telia STB 381 (192.168.1.97), STB 087 (192.168.1.242) — Android TV - Philips Hue Bridge (192.168.1.202) — kol kas be API key - Bambu Lab 3D printer (192.168.1.204) — kol kas be access code - Windows PC (192.168.1.128), laptop (192.168.1.162) KITOS GALIMOS KOMANDOS: - ping įrenginiui: `ping -c2 -W1 IP` - arp-scan tinkle: `sudo arp-scan --interface=wlan0 --localnet` - nmap portų skenavimas ELGESIO TAISYKLĖS: - Veiksmus vykdyk be klausimų (vartotojas suteikęs root prieigą) - Klausimus „kas vyksta" / „status" — atsakyk faktais (ping, ar TV pasiekiama, ir pan.) - Jei prašo kažko, ko negali — paaiškink kodėl trumpai - NIEKADA neperinstaliuok bot'o ir nemodifikuok savęs (`tv_telegram_bot.py`, `tv-bot.service`) - Visada kreipkis „mr.barley" """ def load_owner(): return json.loads(OWNER_FILE.read_text())["chat_id"] if OWNER_FILE.exists() else None def save_owner(chat_id): OWNER_FILE.write_text(json.dumps({"chat_id": chat_id})); OWNER_FILE.chmod(0o600) def load_sessions(): return json.loads(SESSIONS_FILE.read_text()) if SESSIONS_FILE.exists() else {} def save_sessions(s): SESSIONS_FILE.write_text(json.dumps(s)) def get_session(chat_id): sessions = load_sessions() key = str(chat_id) if key not in sessions: sessions[key] = str(uuid.uuid4()); save_sessions(sessions) return sessions[key], True return sessions[key], False def reset_session(chat_id): s = load_sessions(); s.pop(str(chat_id), None); save_sessions(s) async def stream_claude(prompt, session_id, is_new, on_update): """Spawn claude --print --output-format stream-json. Calls on_update(text, status) periodically. status: tuple (assistant_text, current_action_or_None). Returns final assistant text or None on retryable failure.""" args = [ CLAUDE_BIN, "-p", "--append-system-prompt", SYSTEM_PROMPT, "--allowed-tools", "Bash,Read", "--permission-mode", "bypassPermissions", "--max-budget-usd", "0.50", "--model", "sonnet", "--output-format", "stream-json", "--include-partial-messages", "--verbose", ] if is_new: args += ["--session-id", session_id] else: args += ["--resume", session_id] args.append(prompt) proc = await asyncio.create_subprocess_exec( *args, cwd=WORKDIR, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, ) text_so_far = "" current_action = None last_emit = 0.0 stderr_buf = b"" async def maybe_emit(force=False): nonlocal last_emit now = time.time() if force or (now - last_emit) >= EDIT_MIN_INTERVAL: try: await on_update(text_so_far, current_action) except Exception as e: print(f"[on_update err] {e}", flush=True) last_emit = now async def drain_stderr(): nonlocal stderr_buf while True: chunk = await proc.stderr.read(4096) if not chunk: break stderr_buf += chunk stderr_task = asyncio.create_task(drain_stderr()) try: async for raw in proc.stdout: line = raw.decode("utf-8", errors="replace").strip() if not line: continue try: evt = json.loads(line) except Exception: continue etype = evt.get("type") if etype == "stream_event": e = evt.get("event", {}) et = e.get("type") if et == "content_block_delta": d = e.get("delta", {}) if d.get("type") == "text_delta": text_so_far += d.get("text", "") await maybe_emit() elif et == "content_block_start": blk = e.get("content_block", {}) if blk.get("type") == "tool_use": name = blk.get("name", "?") current_action = f"🔧 {name}" await maybe_emit(force=True) elif et == "content_block_stop": pass elif etype == "assistant": msg = evt.get("message", {}) for b in msg.get("content", []): if b.get("type") == "tool_use": name = b.get("name", "?") inp = b.get("input", {}) cmd = inp.get("command") or inp.get("file_path") or json.dumps(inp)[:120] current_action = f"🔧 {name}: `{cmd[:200]}`" await maybe_emit(force=True) elif etype == "user": # tool result — clear action indicator current_action = None await maybe_emit(force=True) elif etype == "result": # final pass finally: await proc.wait() await stderr_task err = stderr_buf.decode("utf-8", errors="replace").strip() print(f"[stream done] rc={proc.returncode} text_len={len(text_so_far)} err={err[:300]!r}", flush=True) if proc.returncode != 0 and not text_so_far: if not is_new and "Session" in err: return None # signal: retry text_so_far = f"⚠️ Klaida (rc={proc.returncode}):\n{err[-400:] or '(no stderr)'}" await on_update(text_so_far, None) return text_so_far async def handle(update: Update, ctx: ContextTypes.DEFAULT_TYPE): msg = update.effective_message if not msg or not msg.text: return chat_id = update.effective_chat.id text = msg.text.strip() print(f"[MSG] chat={chat_id} text={text!r}", flush=True) owner = load_owner() if owner is None: save_owner(chat_id) await msg.reply_text( f"✅ Užregistruotas savininku (chat_id={chat_id}).\n" 'Rašyk laisvai. /reset = naujas pokalbis.' ) return if chat_id != owner: await msg.reply_text("❌ Access denied"); return if text.lower() in ("/reset", "/new"): reset_session(chat_id) await msg.reply_text("🧹 Sesija išvalyta."); return # Initial placeholder reply that we'll edit as Claude streams placeholder = await msg.reply_text("🤔 ...") last_text_sent = "🤔 ..." async def on_update(assistant_text: str, action: str | None): nonlocal last_text_sent body = assistant_text or "🤔" if action: body = f"{body}\n\n_{action}_" # Truncate for Telegram limit if len(body) > TELEGRAM_LIMIT: body = body[-TELEGRAM_LIMIT:] if body == last_text_sent: return try: await placeholder.edit_text(body, parse_mode="Markdown") last_text_sent = body except RetryAfter as e: await asyncio.sleep(e.retry_after) except BadRequest as e: # Markdown parse failure → send plain try: await placeholder.edit_text(body) last_text_sent = body except Exception: pass except Exception as e: print(f"[edit err] {e}", flush=True) session_id, is_new = get_session(chat_id) print(f"[CLAUDE] session={session_id} new={is_new}", flush=True) result = await stream_claude(text, session_id, is_new, on_update) if result is None: # resume failed — recreate print("[RETRY new session]", flush=True) reset_session(chat_id) new_id, _ = get_session(chat_id) await stream_claude(text, new_id, True, on_update) async def cmd_status(update: Update, ctx: ContextTypes.DEFAULT_TYPE): chat_id = update.effective_chat.id owner = load_owner() if owner and chat_id != owner: await update.message.reply_text("❌ Access denied"); return r = subprocess.run(["ping","-c1","-W1","192.168.1.163"], capture_output=True, text=True) on = "🟢 pasiekiama" if r.returncode == 0 else "🔴 nepasiekiama" s = load_sessions().get(str(chat_id), "(nėra)") await update.message.reply_text(f"TV 192.168.1.163: {on}\nSession: `{s}`", parse_mode="Markdown") def main(): app = Application.builder().token(TOKEN).build() app.add_handler(CommandHandler("status", cmd_status)) app.add_handler(CommandHandler("reset", handle)) app.add_handler(MessageHandler(filters.TEXT, handle)) print("Bot running (Claude streaming)...", flush=True) app.run_polling(allowed_updates=Update.ALL_TYPES) if __name__ == "__main__": main()