129 lines
No EOL
5.3 KiB
Python
129 lines
No EOL
5.3 KiB
Python
"""Simbiot Harness — Telegram-инбокс.
|
||
|
||
Входящие сообщения Алекса → события MessageReceived в стор →
|
||
минутный уровень видит их в контексте и может ответить через
|
||
`cli.py tg-reply <msg_id> <text>` (событие 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() |