9cd10256c4
busy в read model всегда False — истинная занятость живёт в памяти планировщика, событий для неё нет. Срез показывал «ожидает», пока часовой уровень держал running-задачу. TaskClaimed закрепляет уровень за задачей в проекции — теперь срез показывает «занят (running-задача)».
225 lines
No EOL
10 KiB
Python
225 lines
No EOL
10 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 sys
|
||
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 {}
|
||
# нетекстовые сообщения: помечаем, что это (стикер/фото/голос/…),
|
||
# иначе пустой text выглядит как «пустое сообщение» и триаж не понимает, что случилось
|
||
kind = "text"
|
||
if msg.get("sticker"):
|
||
kind = "sticker"
|
||
elif msg.get("photo"):
|
||
kind = "photo"
|
||
elif msg.get("voice"):
|
||
kind = "voice"
|
||
elif msg.get("video_note") or msg.get("video"):
|
||
kind = "video"
|
||
elif msg.get("audio"):
|
||
kind = "audio"
|
||
elif msg.get("document"):
|
||
kind = "document"
|
||
elif msg.get("animation"):
|
||
kind = "animation"
|
||
# file_id нужен, чтобы потом скачать файл (голосовое → транскрипция, фото → OCR)
|
||
file = (msg.get("voice") or msg.get("audio") or msg.get("document")
|
||
or msg.get("video") or msg.get("video_note") or {})
|
||
file_id = file.get("file_id")
|
||
duration = file.get("duration") or file.get("duration_seconds")
|
||
photo = msg.get("photo")
|
||
if photo and not file_id:
|
||
file_id = photo[-1].get("file_id") # самый большой размер
|
||
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 msg.get("caption") or "")[:4000],
|
||
"kind": kind,
|
||
"file_id": file_id,
|
||
"duration": duration,
|
||
"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)
|
||
|
||
|
||
def notify_chat(text: str):
|
||
"""Отправить сообщение в чат харнеса (без входящего-родителя — уведомления).
|
||
|
||
Чат берём из последнего входящего MessageReceived (это личка Алекса);
|
||
если входящих ещё не было — тихо пропускаем (некому показывать).
|
||
"""
|
||
_load_env()
|
||
if not TG_TOKEN:
|
||
return None
|
||
model = es.run_projections()
|
||
inbox = [m for m in (model.get("inbox") or {}).values()
|
||
if m.get("chat_id") and not m.get("ignored")]
|
||
if not inbox:
|
||
return None # некому: чат ещё не известен
|
||
chat_id = sorted(inbox, key=lambda m: m.get("ts") or "")[-1]["chat_id"]
|
||
try:
|
||
return send_message(chat_id, text)
|
||
except Exception:
|
||
return None # уведомление — best effort, не валить тик
|
||
|
||
|
||
def ladder_snapshot() -> str:
|
||
"""Срез лестницы для Telegram: уровни, занятость, задачи, эскалации.
|
||
|
||
Занятость: busy в read model всегда False (истинное состояние живёт
|
||
в памяти планировщика, событий для него нет), поэтому выводим её из
|
||
running-задач: TaskClaimed закрепляет уровень за задачей в проекции.
|
||
Уровень без задачи (чистый дайджест-тик) здесь выглядит как «ожидает» —
|
||
приемлемо: дайджест-тик короткий, а взятие задачи — главный кейс.
|
||
"""
|
||
model = es.run_projections()
|
||
cs = model.get("cycle_state") or {}
|
||
tasks_all = list((model.get("tasks") or {}).values())
|
||
running_levels = {t.get("level") for t in tasks_all if t.get("status") == "running"}
|
||
order = ["sec", "min", "hour", "day"]
|
||
lines = []
|
||
for name in order:
|
||
st = cs.get(name) or {}
|
||
if name in running_levels:
|
||
busy = "🔧 занят (running-задача)"
|
||
else:
|
||
busy = "ожидает"
|
||
lines.append(f"{'🔹' if name != 'sec' else '⚡'} {name}: {busy}, тиков {st.get('ticks', 0)}"
|
||
+ (f", ошибок {st['errors']}" if st.get("errors") else ""))
|
||
tasks = tasks_all
|
||
tasks.sort(key=lambda t: t.get("created") or "", reverse=True)
|
||
board = [t for t in tasks if t.get("status") in ("ready", "running")]
|
||
lines.append(f"📋 готово/в работе: {sum(1 for t in tasks if t['status'] == 'ready')}/"
|
||
f"{sum(1 for t in tasks if t['status'] == 'running')}")
|
||
for t in board[:8]:
|
||
lines.append(f" [{t['size']}] {t['status']}: {t['title'][:60]}")
|
||
open_esc = [e for e in (model.get("escalations") or {}).values()
|
||
if isinstance(e, dict) and e.get("status") == "open"]
|
||
lines.append(f"🚨 открытых эскалаций: {len(open_esc)}")
|
||
for e in open_esc[:5]:
|
||
lines.append(f" [{e.get('severity', 'info')}] {e.get('reason', '')[:80]}")
|
||
return "\n".join(lines)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
# Единственный легитимный поллер — systemd-сервис simbiot-harness-tg.
|
||
# `python -m harness.tg_inbox` с любыми аргументами (--help, list, …)
|
||
# раньше поднимал второго поллера → 409 Conflict в легитимном.
|
||
if len(sys.argv) > 1:
|
||
print("usage: python -m harness.tg_inbox # long-poll поллер (systemd)")
|
||
print(" cli.py ladder # срез лестницы (текст)")
|
||
print(" cli.py tg-notify <text> # отправка в чат харнеса")
|
||
sys.exit(0)
|
||
poll_loop() |