import os import logging import aiosqlite from telethon import TelegramClient from db import DB_PATH, init_db log = logging.getLogger(__name__) API_ID = int(os.environ.get("TG_API_ID", 0)) API_HASH = os.environ.get("TG_API_HASH", "") SESSION_NAME = os.environ.get("TG_SESSION_NAME", "chat_summarizer") GROUP_IDS = [int(g) for g in os.environ.get("TG_GROUP_IDS", "").split(",") if g.strip()] def get_client(): return TelegramClient(f"/app/data/{SESSION_NAME}", API_ID, API_HASH) async def auth(): """Interactive one-time auth. Run: docker exec -it chat-summarizer python3 -c 'import asyncio; from collector import auth; asyncio.run(auth())'""" client = get_client() await client.start() me = await client.get_me() print(f"Logged in as {me.first_name} (id={me.id})") await client.disconnect() async def collect_messages(): if not GROUP_IDS: log.warning("No TG_GROUP_IDS configured, skipping collection") return client = get_client() await client.connect() if not await client.is_user_authorized(): log.error("Not authorized — run interactive auth first") return try: async with aiosqlite.connect(DB_PATH) as db: for gid in GROUP_IDS: try: # Get checkpoint cursor = await db.execute("SELECT last_msg_id FROM checkpoints WHERE group_id=?", (gid,)) row = await cursor.fetchone() min_id = row[0] if row else 0 entity = await client.get_entity(gid) group_name = getattr(entity, 'title', str(gid)) count = 0 async for msg in client.iter_messages(entity, min_id=min_id, limit=1000): if not msg.text: continue sender = "" if msg.sender: sender = getattr(msg.sender, 'first_name', '') or getattr(msg.sender, 'title', '') or '' last = getattr(msg.sender, 'last_name', '') if last: sender += f" {last}" await db.execute( "INSERT INTO messages (group_id, group_name, sender_name, text, msg_id, timestamp) VALUES (?,?,?,?,?,?)", (gid, group_name, sender, msg.text, msg.id, msg.date.isoformat()) ) count += 1 if count > 0: # Update checkpoint to max msg_id cursor = await db.execute("SELECT MAX(msg_id) FROM messages WHERE group_id=?", (gid,)) max_row = await cursor.fetchone() if max_row and max_row[0]: await db.execute( "INSERT OR REPLACE INTO checkpoints (group_id, last_msg_id) VALUES (?,?)", (gid, max_row[0]) ) log.info("Collected %d messages from %s", count, group_name) await db.commit() except Exception as e: log.error("Error collecting from group %s: %s", gid, e) finally: await client.disconnect()