Gan, Jimmy
c676c84bc1
Deploy Dashboard / deploy (push) Failing after 38s
- Telegram collector (Telethon MTProto) with checkpoint-based message fetching - Claude API summarization via proxy (haiku model) - Dashboard page with date picker, markdown summary, raw message drill-down - Separate container lifecycle from dashboard for stable Telethon sessions - Shared SQLite DB mounted read-only into dashboard
81 lines
3.3 KiB
Python
81 lines
3.3 KiB
Python
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()
|