"""Simbiot Harness — Telegram-инбокс. Входящие сообщения Алекса → события MessageReceived в стор → минутный уровень видит их в контексте и может ответить через `cli.py tg-reply ` (событие ReplySent закрывает входящее). Поллер — отдельный процесс (systemd simbiot-harness-tg), long polling getUpdates offset=last+1. Идемпотентность: update_id → aggregate_id. """ import json import os import time import urllib.request import urllib.parse from . import es _HARNESS_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) _ENV_PATH = os.path.join(_HARNESS_DIR, ".env") TG_TOKEN = None ALLOWED = set() def _load_env(): global TG_TOKEN, ALLOWED if os.path.exists(_ENV_PATH): with open(_ENV_PATH) as f: for line in f: line = line.strip() if line.startswith("SIMBIOT_TG_TOKEN="): TG_TOKEN = line.split("=", 1)[1].strip() elif line.startswith("SIMBIOT_TG_ALLOWED="): ALLOWED = {x.strip() for x in line.split("=", 1)[1].split(",") if x.strip()} def _api(method: str, params: dict, timeout: int = 35): url = f"https://api.telegram.org/bot{TG_TOKEN}/{method}" data = urllib.parse.urlencode(params).encode() req = urllib.request.Request(url, data=data) with urllib.request.urlopen(req, timeout=timeout) as r: return json.loads(r.read().decode()) def send_message(chat_id, text: str, reply_to=None): """Отправка сообщения (используется tg-reply и алертами).""" _load_env() params = {"chat_id": chat_id, "text": text, "parse_mode": "HTML"} if reply_to: params["reply_to_message_id"] = reply_to return _api("sendMessage", params) def receive_message(update: dict): """MessageReceived в стор. Идемпотентно по update_id.""" msg = update.get("message") or update.get("edited_message") or {} chat = msg.get("chat") or {} user = msg.get("from") or {} data = { "update_id": update["update_id"], "chat_id": chat.get("id"), "user_id": user.get("id"), "username": user.get("username"), "first_name": user.get("first_name"), "text": (msg.get("text") or "")[:4000], "date": msg.get("date"), } es.append_event("inbox", f"m-{update['update_id']}", "MessageReceived", data) return f"m-{update['update_id']}" def send_reply(msg_id: str, text: str): """Ответ на входящее: Telegram-отправка + событие ReplySent (закрывает входящее). Перед отправкой шлём sendChatAction typing — в TG видно «печатает…». sendChatAction живёт ~5с и НЕ обновляется сам — этого достаточно для минутного пульса (типичный лаг ответа < 5с после начала отправки). """ model = es.run_projections() m = (model.get("inbox") or {}).get(msg_id) if m is None: raise ValueError(f"входящее {msg_id} не найдено") if m.get("replied"): raise ValueError(f"на {msg_id} уже отвечено") try: # индикатор — best effort: недоступность TG-метода не мешает ответу _api("sendChatAction", {"chat_id": m["chat_id"], "action": "typing"}, timeout=10) except Exception: pass r = send_message(m["chat_id"], text, reply_to=None) es.append_event("inbox", msg_id, "ReplySent", { "to_chat": m["chat_id"], "text": text[:2000], "tg_ok": bool(r.get("ok")), }) return msg_id def poll_loop(): """Long polling getUpdates → MessageReceived. Отдельный процесс.""" _load_env() if not TG_TOKEN: raise SystemExit("SIMBIOT_TG_TOKEN не задан в .env") es.init_db() offset = 0 print("[tg] инбокс-поллер запущен", flush=True) while True: try: updates = _api("getUpdates", {"offset": offset, "timeout": 30, "limit": 16}) for u in updates.get("result", []): offset = max(offset, u["update_id"] + 1) msg = u.get("message") or u.get("edited_message") if not msg: continue user = (msg.get("from") or {}).get("id") if ALLOWED and str(user) not in ALLOWED: # чужой: пишем событие, но не отвечаем (потенциально спам) es.append_event("inbox", f"m-{u['update_id']}", "MessageIgnored", { "user_id": user, "username": (msg.get("from") or {}).get("username"), "reason": "not_allowed", "text": (msg.get("text") or "")[:500], }) continue mid = receive_message(u) print(f"[tg] {mid} от {(msg.get('from') or {}).get('username') or user}: " f"{(msg.get('text') or '')[:60]}", flush=True) except Exception as e: print(f"[tg] ошибка поллинга: {e}", flush=True) time.sleep(5) if __name__ == "__main__": poll_loop()