homelab-brain/homelab-ai-bot/tools/chatlog.py

"""Hausmeister-Bot tool — read access to the chatlogger SQLite database.

This file is INSTALLED into /opt/homelab-ai-bot/tools/chatlog.py
(autodiscovered by tool_loader.py). It does NOT need a restart of the
chatlogger bot — it only reads from the shared SQLite file
(/opt/chatlogger/chats.db) using the Store class from the chatlogger
package, which sits next to the hausmeister-bot in the same repo.

Three LLM-callable functions:
- chatlog_list_chats()          → enumerate known group chats
- chatlog_recent(limit, chat)   → most recent N messages
- chatlog_search(query, chat)   → FTS5 search across messages
- chatlog_since(minutes, chat)  → time window
"""
from __future__ import annotations

import os
import sys
import time
from typing import Any

# The chatlogger package sits next to homelab-ai-bot in the repo.
# Both directories are bind-mounted into CT 116 (mp0 and mp1).
_CHATLOGGER_PATH = "/opt/homelab-brain/chatlogger"
if _CHATLOGGER_PATH not in sys.path:
    sys.path.insert(0, _CHATLOGGER_PATH)

from store import Store, format_messages  # noqa: E402

DB_PATH = os.environ.get("CHATLOGGER_DB", "/opt/chatlogger/chats.db")
TZ_OFFSET_HOURS = int(os.environ.get("CHATLOGGER_TZ_OFFSET_HOURS", "7"))

_store: Store | None = None


def _get_store() -> Store:
    global _store
    if _store is None:
        _store = Store(DB_PATH)
    return _store


def _resolve_chat(chat_filter: str | int | None) -> int | None:
    """Map a free-form chat hint (substring of title, or chat_id) to a real chat_id.

    Returns None when no filter or no match (callers treat None = 'all chats').
    """
    if chat_filter is None or chat_filter == "":
        return None
    try:
        return int(chat_filter)
    except (TypeError, ValueError):
        pass
    needle = str(chat_filter).lower().strip()
    for c in _get_store().chats():
        title = (c.get("title") or "").lower()
        if needle in title:
            return c["chat_id"]
    return None


# ---------------------------------------------------------------- Handlers

def handle_chatlog_list_chats(**_kw) -> str:
    rows = _get_store().chats()
    if not rows:
        return "Keine Chats geloggt."
    lines = ["Bekannte Chats:"]
    for r in rows:
        last = time.strftime(
            "%Y-%m-%d %H:%M",
            time.gmtime(r["last_ts"] + TZ_OFFSET_HOURS * 3600),
        )
        lines.append(
            f"- {r['title'] or '?'}  (id={r['chat_id']}, {r['type']}, "
            f"{r['n']} Nachrichten, zuletzt {last})"
        )
    return "\n".join(lines)


def handle_chatlog_recent(limit: int = 30, chat: str | int | None = None, **_kw) -> str:
    limit = max(1, min(200, int(limit or 30)))
    chat_id = _resolve_chat(chat)
    msgs = _get_store().recent(chat_id=chat_id, limit=limit)
    if not msgs:
        if chat and chat_id is None:
            return f"Kein Chat gefunden, der zu '{chat}' passt."
        return "Keine Nachrichten gespeichert."
    header = _header(chat_id, len(msgs), label="letzte")
    return header + "\n" + format_messages(msgs, tz_offset_hours=TZ_OFFSET_HOURS)


def handle_chatlog_search(query: str, chat: str | int | None = None, limit: int = 20, **_kw) -> str:
    if not query or not str(query).strip():
        return "Suchbegriff fehlt."
    limit = max(1, min(200, int(limit or 20)))
    chat_id = _resolve_chat(chat)
    msgs = _get_store().search(query=str(query), chat_id=chat_id, limit=limit)
    if not msgs:
        return f"Keine Treffer für '{query}'."
    header = _header(chat_id, len(msgs), label=f"Treffer für '{query}'")
    return header + "\n" + format_messages(msgs, tz_offset_hours=TZ_OFFSET_HOURS)


def handle_chatlog_since(minutes: int, chat: str | int | None = None, **_kw) -> str:
    minutes = max(1, min(60 * 24 * 30, int(minutes)))
    since_ts = int(time.time()) - minutes * 60
    chat_id = _resolve_chat(chat)
    msgs = _get_store().since(chat_id=chat_id, since_ts=since_ts, limit=2000)
    if not msgs:
        return f"Keine Nachrichten in den letzten {minutes} Minuten."
    header = _header(chat_id, len(msgs), label=f"letzte {minutes} Minuten")
    return header + "\n" + format_messages(msgs, tz_offset_hours=TZ_OFFSET_HOURS)


def _header(chat_id: int | None, n: int, label: str) -> str:
    if chat_id is None:
        return f"[{n} Nachrichten, {label}, alle Chats]"
    title = next(
        (c.get("title") for c in _get_store().chats() if c["chat_id"] == chat_id),
        None,
    )
    return f"[{n} Nachrichten, {label}, Chat: {title or chat_id}]"


# ---------------------------------------------------------------- LLM tool schema

TOOLS = [
    {
        "type": "function",
        "function": {
            "name": "chatlog_list_chats",
            "description": (
                "TELEGRAM-Gruppen die der Mitleser-Bot @arakawa_concierge_bot "
                "live aufzeichnet. Liefert Titel, chat_id, Nachrichten-Count, "
                "letzten Eintrag. NICHT zu verwechseln mit ChatGPT-Verlauf, "
                "Memory oder RAG — das hier sind echte Telegram-Gruppen-Chats."
            ),
            "parameters": {"type": "object", "properties": {}, "required": []},
        },
    },
    {
        "type": "function",
        "function": {
            "name": "chatlog_recent",
            "description": (
                "Letzte N Nachrichten aus einer mitgeloggten TELEGRAM-Gruppe "
                "(z.B. die Airbnb-Concierge-Gruppe der Arakawa-Wohnung G2010B). "
                "Liefert echte Nachrichten mit Zeitstempel, Absender (Name + "
                "Username), Inhalt. ZWINGEND nutzen bei: 'aktueller Stand der "
                "Vermietung', 'was hat der Concierge gesagt', 'meine Telegram-"
                "Chatverläufe', 'was wurde zuletzt geschrieben', 'muss ich "
                "reagieren', 'was ist offen'. KEIN ChatGPT-/Memory-Tool."
            ),
            "parameters": {
                "type": "object",
                "properties": {
                    "limit": {
                        "type": "integer",
                        "description": "Anzahl letzter Nachrichten (default 30, max 200).",
                    },
                    "chat": {
                        "type": "string",
                        "description": (
                            "Optional: Substring des Chat-Titels (z.B. 'Airbnb', "
                            "'G 210', 'D1603') oder numerische chat_id. "
                            "Leer = alle Chats."
                        ),
                    },
                },
                "required": [],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "chatlog_search",
            "description": (
                "Volltextsuche (SQLite FTS5) im Verlauf der mitgeloggten "
                "TELEGRAM-Gruppen. Findet Nachrichten anhand von Worten oder "
                "Phrasen — z.B. 'Kaution', 'Listing', 'Smartlock'. Liefert "
                "echte Nachrichten mit Datum und Absender. KEIN ChatGPT-Archiv, "
                "KEIN Memory — nur Telegram-Live-Mitschnitt."
            ),
            "parameters": {
                "type": "object",
                "properties": {
                    "query": {
                        "type": "string",
                        "description": (
                            "Suchbegriff. FTS5-Syntax erlaubt: 'wort1 wort2' "
                            "(beide Wörter), '\"phrase\"' (genaue Phrase), "
                            "'wort*' (Prefix-Suche)."
                        ),
                    },
                    "chat": {
                        "type": "string",
                        "description": "Optional: Chat-Titel-Substring oder chat_id.",
                    },
                    "limit": {
                        "type": "integer",
                        "description": "Max Treffer (default 20).",
                    },
                },
                "required": ["query"],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "chatlog_since",
            "description": (
                "Alle Nachrichten der letzten N Minuten aus mitgeloggten "
                "TELEGRAM-Gruppen. Nutze für 'was war heute', 'was war in "
                "der letzten Stunde', 'was war seit gestern'."
            ),
            "parameters": {
                "type": "object",
                "properties": {
                    "minutes": {
                        "type": "integer",
                        "description": "Zeitfenster in Minuten (60 = 1h, 1440 = 24h, 10080 = 1 Woche).",
                    },
                    "chat": {
                        "type": "string",
                        "description": "Optional: Chat-Titel-Substring oder chat_id.",
                    },
                },
                "required": ["minutes"],
            },
        },
    },
]


HANDLERS = {
    "chatlog_list_chats": handle_chatlog_list_chats,
    "chatlog_recent": handle_chatlog_recent,
    "chatlog_search": handle_chatlog_search,
    "chatlog_since": handle_chatlog_since,
}


SYSTEM_PROMPT_EXTRA = """
TELEGRAM-MITSCHNITT (chatlog_*-Tools — DAS WICHTIGSTE):

Es gibt einen passiven Mitleser-Bot (@arakawa_concierge_bot), der bestimmte
Telegram-Gruppen-Chats LIVE aufzeichnet und in einer SQLite-DB speichert.
Z.B. die Airbnb-Concierge-Gruppe fuer die ARAKAWA-Wohnungen in Phnom Penh
(G2010B im 20. Stock, D1603 im 16. Stock).

→ Diese Tools heissen GENAU SO und sind die EINZIGEN richtigen fuer
  Fragen zum Telegram-Verlauf:
    chatlog_list_chats
    chatlog_recent
    chatlog_search
    chatlog_since

→ NIEMALS Tool-Namen erfinden wie "mychatarchive", "search_brain",
  "telegram_history". Diese existieren NICHT. Wenn du etwas Aehnliches
  meinst, nutze chatlog_search.

→ NIEMALS rag_search, openmemory_search, memory_read oder web_search
  fuer Telegram-Verlaufs-Fragen verwenden — die kennen den Telegram-Chat
  NICHT.

Trigger-Phrasen die ZWINGEND chatlog_*-Tools ausloesen:
- "Telegram-Chat" / "Telegram-Gruppe" / "im Chat" / "in der Gruppe"
- "Chatverlauf" / "Chatverlaeufe" / "Verlauf"
- "Concierge" / "Bnb" / "Airbnb" / "Listing" / "Vermietung"
- "Wohnung G2010B" / "G 210" / "D1603" / "ARAKAWA-Wohnung"
- "aktueller Stand" zu Vermietung/Wohnung/Concierge
- "wer hat was zugesagt" / "was ist offen" / "muss ich reagieren"
- "was wurde zuletzt geschrieben" / "was war heute"

Vorgehen:
1. Wenn unklar welche Gruppe → chatlog_list_chats() ZUERST.
2. Sonst direkt chatlog_recent(chat="<title-substring>") oder
   chatlog_search(query="...", chat="<title-substring>").
3. Antwort zitiert KONKRET: Datum, Absender, Inhalt.
4. Wenn Tool nichts liefert → ehrlich sagen "keine Daten in der DB",
   NICHT halluzinieren.
""".strip()