- 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
This commit is contained in:
@@ -0,0 +1,6 @@
|
||||
FROM python:3.12-slim
|
||||
WORKDIR /app
|
||||
COPY requirements.txt .
|
||||
RUN pip install --no-cache-dir -r requirements.txt
|
||||
COPY . .
|
||||
CMD ["python", "-u", "main.py"]
|
||||
@@ -0,0 +1,80 @@
|
||||
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()
|
||||
@@ -0,0 +1,33 @@
|
||||
import aiosqlite
|
||||
import os
|
||||
|
||||
DB_PATH = os.environ.get("DB_PATH", "/app/data/chat_summary.db")
|
||||
|
||||
async def init_db():
|
||||
async with aiosqlite.connect(DB_PATH) as db:
|
||||
await db.execute("""
|
||||
CREATE TABLE IF NOT EXISTS messages (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
group_id INTEGER NOT NULL,
|
||||
group_name TEXT,
|
||||
sender_name TEXT,
|
||||
text TEXT,
|
||||
msg_id INTEGER,
|
||||
timestamp DATETIME NOT NULL
|
||||
)
|
||||
""")
|
||||
await db.execute("""
|
||||
CREATE TABLE IF NOT EXISTS summaries (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
date TEXT UNIQUE NOT NULL,
|
||||
content TEXT NOT NULL,
|
||||
created_at DATETIME DEFAULT CURRENT_TIMESTAMP
|
||||
)
|
||||
""")
|
||||
await db.execute("""
|
||||
CREATE TABLE IF NOT EXISTS checkpoints (
|
||||
group_id INTEGER PRIMARY KEY,
|
||||
last_msg_id INTEGER NOT NULL
|
||||
)
|
||||
""")
|
||||
await db.commit()
|
||||
@@ -0,0 +1,20 @@
|
||||
services:
|
||||
chat-summarizer:
|
||||
build: .
|
||||
container_name: chat-summarizer
|
||||
restart: unless-stopped
|
||||
volumes:
|
||||
- /volume1/docker/chat-summarizer/data:/app/data
|
||||
environment:
|
||||
- TG_API_ID=${TG_API_ID}
|
||||
- TG_API_HASH=${TG_API_HASH}
|
||||
- TG_SESSION_NAME=${TG_SESSION_NAME:-chat_summarizer}
|
||||
- TG_GROUP_IDS=${TG_GROUP_IDS}
|
||||
- ANTHROPIC_API_KEY=${ANTHROPIC_API_KEY}
|
||||
- ANTHROPIC_BASE_URL=${ANTHROPIC_BASE_URL:-https://www.bytecatcode.org}
|
||||
- SUMMARY_HOUR=${SUMMARY_HOUR:-23}
|
||||
logging:
|
||||
driver: json-file
|
||||
options:
|
||||
max-size: "10m"
|
||||
max-file: "3"
|
||||
@@ -0,0 +1,41 @@
|
||||
import asyncio
|
||||
import logging
|
||||
import os
|
||||
from datetime import datetime, timedelta, timezone
|
||||
|
||||
from db import init_db
|
||||
from collector import collect_messages
|
||||
from summarizer import summarize
|
||||
|
||||
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(name)s %(levelname)s %(message)s")
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
CST = timezone(timedelta(hours=8))
|
||||
SUMMARY_HOUR = int(os.environ.get("SUMMARY_HOUR", 23))
|
||||
COLLECT_INTERVAL = 300 # 5 minutes
|
||||
|
||||
async def main():
|
||||
await init_db()
|
||||
log.info("Chat summarizer started (summary hour=%d CST)", SUMMARY_HOUR)
|
||||
|
||||
last_summary_date = None
|
||||
|
||||
while True:
|
||||
try:
|
||||
await collect_messages()
|
||||
except Exception as e:
|
||||
log.error("Collection error: %s", e)
|
||||
|
||||
now = datetime.now(CST)
|
||||
today = now.strftime("%Y-%m-%d")
|
||||
if now.hour >= SUMMARY_HOUR and last_summary_date != today:
|
||||
try:
|
||||
await summarize(today)
|
||||
last_summary_date = today
|
||||
except Exception as e:
|
||||
log.error("Summary error: %s", e)
|
||||
|
||||
await asyncio.sleep(COLLECT_INTERVAL)
|
||||
|
||||
if __name__ == "__main__":
|
||||
asyncio.run(main())
|
||||
@@ -0,0 +1,4 @@
|
||||
telethon
|
||||
httpx
|
||||
aiosqlite
|
||||
cryptg
|
||||
@@ -0,0 +1,77 @@
|
||||
import os
|
||||
import logging
|
||||
from datetime import datetime, timedelta, timezone
|
||||
import aiosqlite
|
||||
import httpx
|
||||
from db import DB_PATH
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
API_KEY = os.environ.get("ANTHROPIC_API_KEY", "")
|
||||
BASE_URL = os.environ.get("ANTHROPIC_BASE_URL", "https://www.bytecatcode.org")
|
||||
MODEL = "claude-haiku-4-5-20251001"
|
||||
CST = timezone(timedelta(hours=8))
|
||||
|
||||
PROMPT = """You are a chat group digest assistant. Summarize the following chat messages into a concise daily digest.
|
||||
|
||||
Rules:
|
||||
- Group by topic/theme, not by chat group
|
||||
- Highlight key decisions, shared links, and action items
|
||||
- Skip greetings, small talk, and low-value chatter
|
||||
- Use markdown formatting
|
||||
- Write in the same language as the messages (likely Chinese)
|
||||
- Be concise
|
||||
|
||||
Messages:
|
||||
{messages}"""
|
||||
|
||||
async def summarize(date_str: str = None):
|
||||
now = datetime.now(CST)
|
||||
if not date_str:
|
||||
date_str = now.strftime("%Y-%m-%d")
|
||||
|
||||
start = datetime.strptime(date_str, "%Y-%m-%d").replace(tzinfo=CST)
|
||||
end = start + timedelta(days=1)
|
||||
|
||||
async with aiosqlite.connect(DB_PATH) as db:
|
||||
cursor = await db.execute(
|
||||
"SELECT group_name, sender_name, text, timestamp FROM messages WHERE timestamp >= ? AND timestamp < ? ORDER BY timestamp",
|
||||
(start.isoformat(), end.isoformat())
|
||||
)
|
||||
rows = await cursor.fetchall()
|
||||
|
||||
if not rows:
|
||||
log.info("No messages for %s, skipping summary", date_str)
|
||||
return None
|
||||
|
||||
# Format messages for the prompt
|
||||
formatted = []
|
||||
for group_name, sender, text, ts in rows:
|
||||
formatted.append(f"[{group_name}] {sender}: {text}")
|
||||
messages_text = "\n".join(formatted)
|
||||
|
||||
# Call Claude API
|
||||
async with httpx.AsyncClient(timeout=120) as client:
|
||||
resp = await client.post(
|
||||
f"{BASE_URL}/v1/messages",
|
||||
headers={"x-api-key": API_KEY, "anthropic-version": "2023-06-01", "content-type": "application/json"},
|
||||
json={
|
||||
"model": MODEL,
|
||||
"max_tokens": 4096,
|
||||
"messages": [{"role": "user", "content": PROMPT.format(messages=messages_text)}]
|
||||
}
|
||||
)
|
||||
resp.raise_for_status()
|
||||
data = resp.json()
|
||||
summary = data["content"][0]["text"]
|
||||
|
||||
# Store summary
|
||||
async with aiosqlite.connect(DB_PATH) as db:
|
||||
await db.execute(
|
||||
"INSERT OR REPLACE INTO summaries (date, content, created_at) VALUES (?,?,?)",
|
||||
(date_str, summary, now.isoformat())
|
||||
)
|
||||
await db.commit()
|
||||
|
||||
log.info("Generated summary for %s (%d messages)", date_str, len(rows))
|
||||
return summary
|
||||
Reference in New Issue
Block a user