1"""
2watcher.py — полный хендлер сообщений для Март 7.
3 
4- Хендлит ВСЕ входящие и исходящие сообщения (лички + группы до 1000 чел)
5- Каналы и чаты >1000 участников игнорирует
6- Записывает контекст в context/messages.jsonl
7- Триггер "клав" работает ТОЛЬКО от создателя (OWNER_ID)
8- OWNER_ID задаётся через переменную окружения TELEGRAM_OWNER_ID
9"""
10 
11import asyncio
12import os
13import subprocess
14import re
15import json
16import logging
17from datetime import datetime, timezone
18from pathlib import Path
19from dotenv import load_dotenv
20from telethon import TelegramClient, events
21from telethon.sessions import StringSession
22from telethon.tl.types import (
23    User, Chat, Channel,
24    PeerUser, PeerChat, PeerChannel
25)
26 
27load_dotenv()
28 
29API_ID = int(os.getenv("TELEGRAM_API_ID"))
30API_HASH = os.getenv("TELEGRAM_API_HASH")
31SESSION_NAME = os.getenv("TELEGRAM_SESSION_NAME", "my_session")
32SESSION_STRING = os.getenv("TELEGRAM_SESSION_STRING")
33 
34# ═══ СОЗДАТЕЛЬ — единственный кто может управлять ═══
35OWNER_ID = int(os.getenv("TELEGRAM_OWNER_ID", "0"))  # set in .env
36 
37TRIGGER = "клав"
38MAX_CHAT_MEMBERS = 1000  # не хендлим чаты больше этого числа
39 
40# Семафор — только один запрос к openclaw одновременно
41_openclaw_lock = asyncio.Semaphore(1)
42 
43# Кеш размеров чатов {chat_id: count}
44_chat_size_cache: dict = {}
45 
46# Хранилище контекста
47CONTEXT_DIR = Path(__file__).parent / "context"
48CONTEXT_DIR.mkdir(exist_ok=True)
49MESSAGES_FILE = CONTEXT_DIR / "messages.jsonl"
50 
51logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
52log = logging.getLogger("watcher")
53 
54if SESSION_STRING:
55    client = TelegramClient(StringSession(SESSION_STRING), API_ID, API_HASH)
56else:
57    client = TelegramClient(SESSION_NAME, API_ID, API_HASH)
58 
59 
60# ═══ Утилиты ═══
61 
62def get_chat_type(entity) -> str:
63    if isinstance(entity, User):
64        return "private"
65    if isinstance(entity, Chat):
66        return "group"
67    if isinstance(entity, Channel):
68        if getattr(entity, "megagroup", False):
69            return "group"
70        return "channel"
71    return "unknown"
72 
73 
74async def get_members_count(entity) -> int:
75    """Возвращает количество участников чата или 0 для личек."""
76    try:
77        if isinstance(entity, User):
78            return 2  # личка — всегда пропускаем
79        full = await client.get_entity(entity)
80        return getattr(full, "participants_count", 0) or 0
81    except Exception:
82        return 0
83 
84 
85def save_message(record: dict):
86    """Сохраняет запись в JSONL файл."""
87    with open(MESSAGES_FILE, "a", encoding="utf-8") as f:
88        f.write(json.dumps(record, ensure_ascii=False) + "\n")
89 
90 
91async def build_record(event, direction: str) -> dict | None:
92    """Строит запись для сохранения."""
93    try:
94        msg = event.message
95        chat = await event.get_chat()
96        chat_type = get_chat_type(chat)
97 
98        # Пропускаем каналы
99        if chat_type == "channel":
100            return None
101 
102        # Проверяем размер чата (с кешем)
103        if chat_type == "group":
104            cid = getattr(chat, "id", None)
105            if cid not in _chat_size_cache:
106                count = getattr(chat, "participants_count", None)
107                if not count:
108                    try:
109                        from telethon.tl.functions.channels import GetFullChannelRequest
110                        from telethon.tl.functions.messages import GetFullChatRequest
111                        if isinstance(chat, Channel):
112                            full = await client(GetFullChannelRequest(chat))
113                            count = full.full_chat.participants_count
114                        elif isinstance(chat, Chat):
115                            full = await client(GetFullChatRequest(chat.id))
116                            count = getattr(full.full_chat, "participants_count", 0) or 0
117                    except Exception as e:
118                        log.warning(f"Не удалось получить размер чата {cid}: {e}")
119                        count = 9999  # если ошибка — считаем большим, не пишем
120                _chat_size_cache[cid] = count or 0
121            if _chat_size_cache.get(cid, 0) > MAX_CHAT_MEMBERS:
122                return None
123 
124        # Данные отправителя
125        sender_id = None
126        sender_name = ""
127        sender_username = ""
128        try:
129            sender = await msg.get_sender()
130            # Для исходящих sender может быть None — берём кешированные данные
131            if sender is None and msg.out:
132                sender = _me
133            if sender:
134                sender_id = sender.id
135                sender_name = (getattr(sender, "first_name", "") or "") + \
136                              (" " + getattr(sender, "last_name", "") if getattr(sender, "last_name", "") else "")
137                sender_name = sender_name.strip()
138                # username может быть строкой или None
139                uname = getattr(sender, "username", None)
140                usernames = getattr(sender, "usernames", None)  # Telegram Premium multiple usernames
141                if uname:
142                    sender_username = uname
143                elif usernames:
144                    sender_username = ",".join(u.username for u in usernames if u.username)
145                else:
146                    sender_username = ""
147        except Exception:
148            pass
149 
150        # Данные чата
151        chat_id = msg.chat_id or msg.peer_id
152        chat_title = getattr(chat, "title", None) or \
153                     getattr(chat, "first_name", None) or str(chat_id)
154 
155        # Реплай
156        reply_to_msg_id = None
157        reply_to_text = ""
158        reply_to_user = ""
159        if msg.reply_to and getattr(msg.reply_to, "reply_to_msg_id", None):
160            reply_to_msg_id = msg.reply_to.reply_to_msg_id
161            try:
162                replied = await msg.get_reply_message()
163                if replied:
164                    reply_to_text = (replied.message or "")[:300]
165                    rs = await replied.get_sender()
166                    if rs:
167                        reply_to_user = (getattr(rs, "first_name", "") or "") + \
168                                       (" " + getattr(rs, "last_name", "") if getattr(rs, "last_name", "") else "")
169                        reply_to_user = reply_to_user.strip()
170            except Exception:
171                pass
172 
173        record = {
174            "ts": datetime.now(timezone.utc).isoformat(),
175            "direction": direction,  # "in" | "out"
176            "chat_type": chat_type,  # "private" | "group"
177            "chat_id": chat_id,
178            "chat_title": chat_title,
179            "msg_id": msg.id,
180            "sender_id": sender_id,
181            "sender_name": sender_name,
182            "sender_username": sender_username,
183            "text": msg.message or "",
184            "has_media": msg.media is not None,
185            "reply_to_msg_id": reply_to_msg_id,
186            "reply_to_text": reply_to_text,
187            "reply_to_user": reply_to_user,
188        }
189        return record
190    except Exception as e:
191        log.error(f"build_record error: {e}")
192        return None
193 
194 
195LIVE_MESSAGES_LIMIT = 30  # последних сообщений из messages.jsonl
196 
197def load_live_messages(chat_id: int, limit: int = LIVE_MESSAGES_LIMIT) -> str:
198    """Читает последние N сообщений этого чата прямо из messages.jsonl (live-данные)."""
199    if not MESSAGES_FILE.exists():
200        return ""
201    msgs = []
202    with open(MESSAGES_FILE, encoding="utf-8") as f:
203        for line in f:
204            try:
205                m = json.loads(line)
206                if m.get("chat_id") == chat_id and m.get("text","").strip():
207                    msgs.append(m)
208            except Exception:
209                pass
210    # берём последние N
211    recent = msgs[-limit:]
212    if not recent:
213        return ""
214    lines = []
215    for m in recent:
216        direction = "→" if m.get("sender_id") == OWNER_ID else "←"
217        sender = m.get("sender", "?")
218        text = m.get("text", "").replace("\n", " ")[:200]
219        date = m.get("date", "")[:16]
220        lines.append(f"[{date}] {direction} {sender}: {text}")
221    return "## Последние сообщения (live)\n" + "\n".join(lines)
222 
223 
224def load_chat_context(chat_id: int) -> str:
225    """Загружает саммари чата, последние live-сообщения и профили участников."""
226    parts = []
227    chat_file = CONTEXT_DIR / "chats" / f"{chat_id}.md"
228    if chat_file.exists():
229        parts.append(f"[Контекст чата]\n{chat_file.read_text(encoding='utf-8')[:1500]}")
230 
231    # Live-сообщения из messages.jsonl (реальное время)
232    live = load_live_messages(chat_id)
233    if live:
234        parts.append(live)
235 
236    # Профили людей из этого чата
237    people_dir = CONTEXT_DIR / "people"
238    if people_dir.exists():
239        seen_uids = set()
240        if MESSAGES_FILE.exists():
241            with open(MESSAGES_FILE, encoding="utf-8") as f:
242                for line in f:
243                    try:
244                        m = json.loads(line)
245                        if m.get("chat_id") == chat_id and m.get("sender_id") and m["sender_id"] != OWNER_ID:
246                            seen_uids.add(m["sender_id"])
247                    except Exception:
248                        pass
249        for uid in seen_uids:
250            pf = people_dir / f"{uid}.md"
251            if pf.exists():
252                parts.append(f"[Участник]\n{pf.read_text(encoding='utf-8')[:500]}")
253 
254    return "\n\n".join(parts)
255 
256 
257# Кеш session_id живой main-сессии
258_main_session_id: str | None = None
259 
260def _get_main_session_id() -> str | None:
261    """Читает session_id живой main-сессии из sessions.json (без subprocess)."""
262    global _main_session_id
263    if _main_session_id:
264        return _main_session_id
265    try:
266        import json as _j
267        path = os.path.expanduser("~/.openclaw/agents/main/sessions/sessions.json")
268        if os.path.exists(path):
269            data = _j.loads(open(path).read())
270            # Структура: {"agent:main:main": {"sessionId": "...", ...}, ...}
271            for key, val in data.items():
272                if key == "agent:main:main" and isinstance(val, dict):
273                    sid = val.get("sessionId", "")
274                    if sid:
275                        _main_session_id = sid
276                        log.info(f"[watcher] Найден session-id: {sid[:8]}...")
277                        return sid
278    except Exception as e:
279        log.debug(f"[watcher] _get_main_session_id error: {e}")
280    return None
281 
282 
283def ask_openclaw(query: str, chat_id: int) -> str:
284    """Отправляет запрос в OpenClaw — переиспользует живую main-сессию (быстро)."""
285    chat_context = load_chat_context(chat_id)
286    ctx_prefix = f"\n[Контекст из базы данных]\n{chat_context}\n\n" if chat_context else ""
287    context = f"{ctx_prefix}[Запрос из Telegram чата {chat_id}, отвечай кратко как в чате, без лишних слов]: {query}"
288 
289    # Пробуем переиспользовать живую сессию (без нового процесса)
290    session_id = _get_main_session_id()
291    if session_id:
292        cmd = ["openclaw", "agent", "--session-id", session_id, "--message", context, "--json"]
293        log.info(f"[watcher] Используем session-id={session_id}")
294    else:
295        # Фолбэк: создаём новый процесс (медленнее)
296        cmd = ["openclaw", "agent", "--agent", "main", "--message", context, "--json"]
297        log.info("[watcher] session-id не найден, фолбэк на --agent main")
298 
299    result = subprocess.run(cmd, capture_output=True, text=True, timeout=120)
300    if result.returncode != 0:
301        # Если session-id устарел — сбросить кеш и повторить
302        global _main_session_id
303        _main_session_id = None
304        return f"Ошибка: {result.stderr.strip()}"
305 
306    import json as _json
307    try:
308        data = _json.loads(result.stdout)
309        if "result" in data and "payloads" in data["result"]:
310            return data["result"]["payloads"][0]["text"]
311        elif "payloads" in data:
312            return data["payloads"][0]["text"]
313        elif "text" in data:
314            return data["text"]
315        else:
316            def find_text(obj):
317                if isinstance(obj, dict):
318                    if "text" in obj and isinstance(obj["text"], str) and len(obj["text"]) > 0:
319                        return obj["text"]
320                    for v in obj.values():
321                        r = find_text(v)
322                        if r:
323                            return r
324                elif isinstance(obj, list):
325                    for item in obj:
326                        r = find_text(item)
327                        if r:
328                            return r
329                return None
330            return find_text(data) or result.stdout.strip()
331    except Exception:
332        return result.stdout.strip()
333 
334 
335# ═══ Хендлеры ═══
336 
337@client.on(events.NewMessage(outgoing=True))
338async def handle_outgoing(event):
339    """Хендлит исходящие сообщения."""
340    text = event.raw_text or ""
341 
342    # Сохраняем в контекст
343    record = await build_record(event, "out")
344    if record:
345        save_message(record)
346 
347    # Триггер "клав" — только от создателя
348    if not re.match(rf"^{TRIGGER}\b", text, re.IGNORECASE):
349        return
350 
351    # Получаем chat_id
352    chat_id = event.chat_id
353 
354    # Извлекаем запрос
355    query = re.sub(rf"^{TRIGGER}[,\s]*", "", text, flags=re.IGNORECASE).strip()
356    if not query:
357        return
358 
359    log.info(f"[trigger] чат {chat_id}: {query}")
360 
361    # Добавляем контекст реплая
362    reply_context = ""
363    if event.reply_to and getattr(event.reply_to, "reply_to_msg_id", None):
364        try:
365            replied_msg = await event.get_reply_message()
366            if replied_msg:
367                sender_id = replied_msg.sender_id
368                sender_name = ""
369                try:
370                    sender = await replied_msg.get_sender()
371                    if sender:
372                        sender_name = (getattr(sender, "first_name", "") or "").strip()
373                except Exception:
374                    sender_name = str(sender_id)
375                msg_text = replied_msg.message or "[медиа]"
376                reply_context = f"\n[Реплай на сообщение от {sender_name} (user_id: {sender_id}, msg_id: {replied_msg.id}): \"{msg_text[:300]}\"]"
377        except Exception:
378            pass
379 
380    full_query = query + reply_context
381 
382    # Отправляем ⏳ как реплай
383    thinking_msg = await event.reply("⏳")
384 
385    # Запрос к OpenClaw — сериализуем через семафор
386    async with _openclaw_lock:
387        loop = asyncio.get_event_loop()
388        response = await loop.run_in_executor(None, ask_openclaw, full_query, chat_id)
389 
390    # Отправляем ответ
391    if response:
392        MAX_LEN = 4096
393        if len(response) <= MAX_LEN:
394            await thinking_msg.edit(response)
395        else:
396            await thinking_msg.delete()
397            for i in range(0, len(response), MAX_LEN):
398                await client.send_message(chat_id, response[i:i + MAX_LEN])
399    else:
400        await thinking_msg.delete()
401 
402 
403@client.on(events.NewMessage(incoming=True))
404async def handle_incoming(event):
405    """Хендлит входящие сообщения — только записывает контекст."""
406    record = await build_record(event, "in")
407    if record:
408        save_message(record)
409 
410 
411# Кеш данных о себе — загружается один раз при старте
412_me = None
413 
414async def refresh_me():
415    """Обновляет данные о себе через API. Вызывать по запросу."""
416    global _me
417    _me = await client.get_me()
418    uname = getattr(_me, "username", None)
419    usernames = getattr(_me, "usernames", None)
420    if usernames:
421        uname = ",".join(u.username for u in usernames if u.username)
422    log.info(f"[me] Обновлено: {_me.first_name} id={_me.id} username={uname}")
423    return _me
424 
425async def main():
426    log.info(f"[watcher] Запуск... триггер: '{TRIGGER}', только для owner: {OWNER_ID}")
427    log.info(f"[watcher] Контекст сохраняется в: {MESSAGES_FILE}")
428    await client.start()
429    me = await client.get_me()
430    await refresh_me()  # загружаем данные о себе один раз при старте
431    log.info(f"[watcher] Авторизован как {_me.first_name} (id: {_me.id}). Слушаю все сообщения...")
432    await client.run_until_disconnected()
433 
434 
435if __name__ == "__main__":
436    asyncio.run(main())
437