homelab-brain/chatlogger/chatlogger_bot.py

"""Arakawa Concierge — passive Telegram group chat logger.

Reads messages from configured groups, stores them in SQLite,
never replies to group messages. Provides /whichchat and /log_status
commands for the admin (private chat or in any chat).
"""
from __future__ import annotations

import asyncio
import json
import logging
import os
import signal
import sys
import time
from typing import Any

from telegram import Update
from telegram.constants import ChatType
from telegram.ext import (
    Application,
    CommandHandler,
    ContextTypes,
    MessageHandler,
    filters,
)

from store import Store, format_messages

LOG = logging.getLogger("chatlogger")


def _env(name: str, default: str | None = None) -> str | None:
    val = os.environ.get(name, default)
    return val.strip() if isinstance(val, str) else val


def _env_int(name: str, default: int = 0) -> int:
    val = _env(name)
    try:
        return int(val) if val else default
    except (TypeError, ValueError):
        return default


def _allowed_chats() -> set[int]:
    raw = _env("ARAKAWA_BOT_ALLOWED_CHATS", "") or ""
    out: set[int] = set()
    for part in raw.split(","):
        part = part.strip()
        if not part:
            continue
        try:
            out.add(int(part))
        except ValueError:
            LOG.warning("Ignoring non-integer chat id in ALLOWED_CHATS: %r", part)
    return out


CONFIG = {
    "instance": _env("CHATLOGGER_INSTANCE", "default"),
    "token": _env("ARAKAWA_BOT_TOKEN"),
    "db_path": _env("ARAKAWA_BOT_DB", "/opt/chatlogger/chats.db"),
    "admin_user_id": _env_int("ARAKAWA_BOT_ADMIN_USER_ID", 0),
    "tz_offset_hours": _env_int("ARAKAWA_BOT_TZ_OFFSET_HOURS", 7),
}


def _extract_forward_origin(message) -> str | None:
    """Return a short string describing where a forwarded message came from.

    python-telegram-bot 22.x uses MessageOrigin* classes via message.forward_origin.
    Keep this defensive — older fields may also exist depending on PTB version.
    """
    origin = getattr(message, "forward_origin", None)
    if origin is not None:
        sender_user = getattr(origin, "sender_user", None)
        if sender_user is not None:
            return getattr(sender_user, "full_name", None) or getattr(sender_user, "username", None)
        sender_chat = getattr(origin, "sender_chat", None) or getattr(origin, "chat", None)
        if sender_chat is not None:
            return getattr(sender_chat, "title", None) or getattr(sender_chat, "username", None)
        sender_user_name = getattr(origin, "sender_user_name", None)
        if sender_user_name:
            return sender_user_name
        return type(origin).__name__
    legacy_user = getattr(message, "forward_from", None)
    if legacy_user is not None:
        return getattr(legacy_user, "full_name", None)
    legacy_chat = getattr(message, "forward_from_chat", None)
    if legacy_chat is not None:
        return getattr(legacy_chat, "title", None) or getattr(legacy_chat, "username", None)
    return None


def detect_media_type(message) -> tuple[str | None, str | None]:
    """Return (media_type, file_id_or_summary) for the message, if any."""
    if message.photo:
        return "photo", message.photo[-1].file_id
    if message.document:
        return "document", message.document.file_id
    if message.voice:
        return "voice", message.voice.file_id
    if message.video:
        return "video", message.video.file_id
    if message.video_note:
        return "video_note", message.video_note.file_id
    if message.audio:
        return "audio", message.audio.file_id
    if message.sticker:
        return "sticker", message.sticker.file_id
    if message.animation:
        return "animation", message.animation.file_id
    if message.location:
        loc = message.location
        return "location", f"{loc.latitude},{loc.longitude}"
    if message.venue:
        v = message.venue
        return "venue", f"{v.title} | {v.address}"
    if message.contact:
        return "contact", message.contact.phone_number or ""
    if message.poll:
        return "poll", message.poll.question
    if message.dice:
        return "dice", str(message.dice.value)
    return None, None


async def on_message(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    msg = update.effective_message
    chat = update.effective_chat
    if msg is None or chat is None:
        return

    allowed = context.bot_data["allowed_chats"]
    if chat.id not in allowed:
        return

    user = msg.from_user
    media_type, media_payload = detect_media_type(msg)

    forward_from = _extract_forward_origin(msg)

    is_edit = bool(update.edited_message)
    store: Store = context.bot_data["store"]

    try:
        raw_json = msg.to_json()
    except Exception:
        raw_json = "{}"

    if is_edit:
        edit_ts = int(msg.edit_date.timestamp()) if msg.edit_date else int(time.time())
        ok = store.update_message(
            chat_id=chat.id,
            message_id=msg.message_id,
            new_text=msg.text,
            new_caption=msg.caption,
            edit_timestamp_utc=edit_ts,
        )
        if not ok:
            store.add_message(
                chat_id=chat.id,
                chat_title=chat.title,
                chat_type=str(chat.type),
                message_id=msg.message_id,
                user_id=user.id if user else None,
                user_first_name=user.first_name if user else None,
                user_last_name=user.last_name if user else None,
                user_username=user.username if user else None,
                text=msg.text,
                media_type=media_type,
                media_caption=msg.caption,
                media_file_id=media_payload,
                reply_to_message_id=msg.reply_to_message.message_id if msg.reply_to_message else None,
                forward_from=forward_from,
                timestamp_utc=edit_ts,
                raw_json=raw_json,
            )
        LOG.info("EDIT  chat=%s msg=%s user=%s", chat.id, msg.message_id, user.id if user else None)
        return

    ts = int(msg.date.timestamp()) if msg.date else int(time.time())
    inserted = store.add_message(
        chat_id=chat.id,
        chat_title=chat.title,
        chat_type=str(chat.type),
        message_id=msg.message_id,
        user_id=user.id if user else None,
        user_first_name=user.first_name if user else None,
        user_last_name=user.last_name if user else None,
        user_username=user.username if user else None,
        text=msg.text,
        media_type=media_type,
        media_caption=msg.caption,
        media_file_id=media_payload,
        reply_to_message_id=msg.reply_to_message.message_id if msg.reply_to_message else None,
        forward_from=forward_from,
        timestamp_utc=ts,
        raw_json=raw_json,
    )
    LOG.info(
        "%s chat=%s msg=%s user=%s media=%s text=%r",
        "STORE" if inserted else "DUP  ",
        chat.id,
        msg.message_id,
        user.id if user else None,
        media_type,
        (msg.text or msg.caption or "")[:80],
    )


async def cmd_whichchat(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    """Reply with current chat id + title. Open to anyone — useful for setup.

    The reply itself is the only message this bot ever sends in a group chat.
    Restricted to admin user to avoid noise from random people typing /whichchat.
    """
    chat = update.effective_chat
    user = update.effective_user
    if chat is None or user is None:
        return
    admin = context.bot_data["admin_user_id"]
    if admin and user.id != admin:
        return

    allowed = context.bot_data["allowed_chats"]
    status = "AKTIV (wird geloggt)" if chat.id in allowed else "NICHT konfiguriert"
    text = (
        f"chat_id: <code>{chat.id}</code>\n"
        f"type: {chat.type}\n"
        f"title: {chat.title or '(privat)'}\n"
        f"status: {status}"
    )
    await context.bot.send_message(chat_id=chat.id, text=text, parse_mode="HTML")


async def cmd_log_status(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    user = update.effective_user
    chat = update.effective_chat
    if chat is None or user is None:
        return
    admin = context.bot_data["admin_user_id"]
    if admin and user.id != admin:
        return

    store: Store = context.bot_data["store"]
    s = store.stats()
    lines = [
        f"📒 <b>Arakawa Concierge — Status</b>",
        f"DB: <code>{store.db_path}</code> ({s['db_size_bytes'] / 1024:.1f} KB)",
        f"Nachrichten gesamt: <b>{s['total_messages']}</b>",
        f"Edits: {s['total_edits']}",
        "",
        "<b>Pro Chat:</b>",
    ]
    for c in s["chats"]:
        last = time.strftime("%Y-%m-%d %H:%M", time.gmtime(c["last_ts"] + CONFIG["tz_offset_hours"] * 3600))
        lines.append(f"  • <code>{c['chat_id']}</code> {c['title'] or '?'}: {c['n']} (letzter {last})")
    if not s["chats"]:
        lines.append("  (noch keine Nachrichten)")
    await context.bot.send_message(chat_id=chat.id, text="\n".join(lines), parse_mode="HTML")


async def cmd_recent(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    user = update.effective_user
    chat = update.effective_chat
    if chat is None or user is None:
        return
    admin = context.bot_data["admin_user_id"]
    if admin and user.id != admin:
        return

    args = context.args or []
    n = 20
    if args:
        try:
            n = max(1, min(200, int(args[0])))
        except ValueError:
            pass
    store: Store = context.bot_data["store"]
    target_chat = chat.id if chat.type != ChatType.PRIVATE else None
    msgs = store.recent(chat_id=target_chat, limit=n)
    if not msgs:
        await context.bot.send_message(chat_id=chat.id, text="(keine Nachrichten gespeichert)")
        return
    text = format_messages(msgs, tz_offset_hours=CONFIG["tz_offset_hours"])
    for chunk in _chunks(text, 3500):
        await context.bot.send_message(chat_id=chat.id, text=f"<pre>{_html_escape(chunk)}</pre>", parse_mode="HTML")


async def cmd_search(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    user = update.effective_user
    chat = update.effective_chat
    if chat is None or user is None:
        return
    admin = context.bot_data["admin_user_id"]
    if admin and user.id != admin:
        return

    args = context.args or []
    if not args:
        await context.bot.send_message(chat_id=chat.id, text="Nutzung: /search &lt;begriff&gt;", parse_mode="HTML")
        return
    query = " ".join(args)
    store: Store = context.bot_data["store"]
    target_chat = chat.id if chat.type != ChatType.PRIVATE else None
    msgs = store.search(query=query, chat_id=target_chat, limit=30)
    if not msgs:
        await context.bot.send_message(chat_id=chat.id, text=f"(keine Treffer für „{query}“)")
        return
    text = format_messages(msgs, tz_offset_hours=CONFIG["tz_offset_hours"])
    for chunk in _chunks(text, 3500):
        await context.bot.send_message(chat_id=chat.id, text=f"<pre>{_html_escape(chunk)}</pre>", parse_mode="HTML")


def _chunks(text: str, max_len: int):
    if len(text) <= max_len:
        yield text
        return
    cur = []
    cur_len = 0
    for line in text.split("\n"):
        if cur_len + len(line) + 1 > max_len and cur:
            yield "\n".join(cur)
            cur = [line]
            cur_len = len(line) + 1
        else:
            cur.append(line)
            cur_len += len(line) + 1
    if cur:
        yield "\n".join(cur)


def _html_escape(s: str) -> str:
    return s.replace("&", "&amp;").replace("<", "&lt;").replace(">", "&gt;")


def build_app() -> Application:
    if not CONFIG["token"]:
        raise SystemExit("ARAKAWA_BOT_TOKEN missing in environment")

    store = Store(CONFIG["db_path"])
    allowed = _allowed_chats()
    LOG.info(
        "Starting chatlogger[%s]: db=%s allowed_chats=%s admin=%s",
        CONFIG["instance"], CONFIG["db_path"], sorted(allowed), CONFIG["admin_user_id"],
    )

    app = Application.builder().token(CONFIG["token"]).build()
    app.bot_data["store"] = store
    app.bot_data["allowed_chats"] = allowed
    app.bot_data["admin_user_id"] = CONFIG["admin_user_id"]

    app.add_handler(CommandHandler("whichchat", cmd_whichchat))
    app.add_handler(CommandHandler("log_status", cmd_log_status))
    app.add_handler(CommandHandler("recent", cmd_recent))
    app.add_handler(CommandHandler("search", cmd_search))

    msg_filter = (
        filters.ChatType.GROUPS | filters.ChatType.SUPERGROUP | filters.ChatType.PRIVATE
    ) & ~filters.COMMAND
    app.add_handler(MessageHandler(msg_filter, on_message))
    app.add_handler(MessageHandler(filters.UpdateType.EDITED_MESSAGE & ~filters.COMMAND, on_message))

    return app


def main() -> None:
    logging.basicConfig(
        level=logging.INFO,
        format="%(asctime)s %(levelname)s %(name)s: %(message)s",
        stream=sys.stdout,
    )
    logging.getLogger("httpx").setLevel(logging.WARNING)
    logging.getLogger("telegram").setLevel(logging.WARNING)

    app = build_app()
    app.run_polling(
        allowed_updates=["message", "edited_message", "channel_post", "my_chat_member"],
        drop_pending_updates=False,
    )


if __name__ == "__main__":
    main()