From c676c84bc1bc10738d9138701556e5679d3f6ab5 Mon Sep 17 00:00:00 2001 From: "Gan, Jimmy" Date: Fri, 27 Feb 2026 02:51:52 +0800 Subject: [PATCH] feat: add chat group daily summarizer - 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 --- chat-summarizer/Dockerfile | 6 + chat-summarizer/collector.py | 80 +++++++++++ chat-summarizer/db.py | 33 +++++ chat-summarizer/docker-compose.yml | 20 +++ chat-summarizer/main.py | 41 ++++++ chat-summarizer/requirements.txt | 4 + chat-summarizer/summarizer.py | 77 +++++++++++ dashboard/backend/config.py | 3 + dashboard/backend/main.py | 3 +- dashboard/backend/routers/chat_summary.py | 41 ++++++ dashboard/docker-compose.yml | 1 + dashboard/frontend/src/App.svelte | 3 + .../frontend/src/components/Sidebar.svelte | 5 + .../frontend/src/routes/ChatSummary.svelte | 125 ++++++++++++++++++ 14 files changed, 441 insertions(+), 1 deletion(-) create mode 100644 chat-summarizer/Dockerfile create mode 100644 chat-summarizer/collector.py create mode 100644 chat-summarizer/db.py create mode 100644 chat-summarizer/docker-compose.yml create mode 100644 chat-summarizer/main.py create mode 100644 chat-summarizer/requirements.txt create mode 100644 chat-summarizer/summarizer.py create mode 100644 dashboard/backend/routers/chat_summary.py create mode 100644 dashboard/frontend/src/routes/ChatSummary.svelte diff --git a/chat-summarizer/Dockerfile b/chat-summarizer/Dockerfile new file mode 100644 index 0000000..3613a77 --- /dev/null +++ b/chat-summarizer/Dockerfile @@ -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"] diff --git a/chat-summarizer/collector.py b/chat-summarizer/collector.py new file mode 100644 index 0000000..443b710 --- /dev/null +++ b/chat-summarizer/collector.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() diff --git a/chat-summarizer/db.py b/chat-summarizer/db.py new file mode 100644 index 0000000..bba0abc --- /dev/null +++ b/chat-summarizer/db.py @@ -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() diff --git a/chat-summarizer/docker-compose.yml b/chat-summarizer/docker-compose.yml new file mode 100644 index 0000000..3ccd86f --- /dev/null +++ b/chat-summarizer/docker-compose.yml @@ -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" diff --git a/chat-summarizer/main.py b/chat-summarizer/main.py new file mode 100644 index 0000000..f41d77f --- /dev/null +++ b/chat-summarizer/main.py @@ -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()) diff --git a/chat-summarizer/requirements.txt b/chat-summarizer/requirements.txt new file mode 100644 index 0000000..d7c5dd2 --- /dev/null +++ b/chat-summarizer/requirements.txt @@ -0,0 +1,4 @@ +telethon +httpx +aiosqlite +cryptg diff --git a/chat-summarizer/summarizer.py b/chat-summarizer/summarizer.py new file mode 100644 index 0000000..b8cd501 --- /dev/null +++ b/chat-summarizer/summarizer.py @@ -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 diff --git a/dashboard/backend/config.py b/dashboard/backend/config.py index f2d111b..c5ce0fc 100644 --- a/dashboard/backend/config.py +++ b/dashboard/backend/config.py @@ -41,5 +41,8 @@ WEBAUTHN_ORIGINS = os.environ.get("WEBAUTHN_ORIGIN", "https://nas.jimmygan.com,h # CORS CORS_ORIGINS = os.environ.get("CORS_ORIGINS", "https://nas.jimmygan.com,https://nas.jimmygan.com:8443").split(",") +# Chat Summary +CHAT_SUMMARY_DB = os.environ.get("CHAT_SUMMARY_DB", "/app/data/chat-summarizer/chat_summary.db") + # OpenClaw OPENCLAW_GATEWAY_TOKEN = os.environ.get("OPENCLAW_GATEWAY_TOKEN", "") diff --git a/dashboard/backend/main.py b/dashboard/backend/main.py index f57bd43..c684513 100644 --- a/dashboard/backend/main.py +++ b/dashboard/backend/main.py @@ -5,7 +5,7 @@ from fastapi.responses import JSONResponse from slowapi import Limiter from slowapi.util import get_remote_address from slowapi.errors import RateLimitExceeded -from routers import docker_router, gitea, files, terminal, system, auth +from routers import docker_router, gitea, files, terminal, system, auth, chat_summary import auth as auth_module import asyncio @@ -139,6 +139,7 @@ app.include_router(docker_router.router, prefix="/api/docker", dependencies=[Dep app.include_router(gitea.router, prefix="/api/gitea", dependencies=[Depends(auth_module.get_current_user)]) app.include_router(files.router, prefix="/api/files", dependencies=[Depends(auth_module.get_current_user)]) app.include_router(system.router, prefix="/api/system", dependencies=[Depends(auth_module.get_current_user)]) +app.include_router(chat_summary.router, prefix="/api/chat-summary", dependencies=[Depends(auth_module.get_current_user)]) # WebSocket endpoint (auth handled inside handler via Query param) app.add_api_websocket_route("/ws/terminal", terminal.ws_endpoint) diff --git a/dashboard/backend/routers/chat_summary.py b/dashboard/backend/routers/chat_summary.py new file mode 100644 index 0000000..fe143dc --- /dev/null +++ b/dashboard/backend/routers/chat_summary.py @@ -0,0 +1,41 @@ +from fastapi import APIRouter, HTTPException +import aiosqlite +from config import CHAT_SUMMARY_DB + +router = APIRouter() + +async def _db(): + return aiosqlite.connect(CHAT_SUMMARY_DB) + +@router.get("/dates") +async def list_dates(): + async with await _db() as db: + cursor = await db.execute("SELECT date FROM summaries ORDER BY date DESC") + return [r[0] for r in await cursor.fetchall()] + +@router.get("/summary/{date}") +async def get_summary(date: str): + async with await _db() as db: + cursor = await db.execute("SELECT content, created_at FROM summaries WHERE date=?", (date,)) + row = await cursor.fetchone() + if not row: + raise HTTPException(404, "No summary for this date") + return {"date": date, "content": row[0], "created_at": row[1]} + +@router.get("/messages/{date}") +async def get_messages(date: str): + async with await _db() as db: + cursor = await db.execute( + "SELECT group_name, sender_name, text, timestamp FROM messages WHERE date(timestamp)=? ORDER BY timestamp", + (date,) + ) + return [{"group": r[0], "sender": r[1], "text": r[2], "time": r[3]} for r in await cursor.fetchall()] + +@router.post("/trigger") +async def trigger_summary(): + import os + trigger_path = os.environ.get("CHAT_SUMMARY_TRIGGER", "/app/data/chat-summarizer/trigger") + os.makedirs(os.path.dirname(trigger_path), exist_ok=True) + with open(trigger_path, "w") as f: + f.write("1") + return {"status": "triggered"} diff --git a/dashboard/docker-compose.yml b/dashboard/docker-compose.yml index b7aa405..b0e2c85 100644 --- a/dashboard/docker-compose.yml +++ b/dashboard/docker-compose.yml @@ -45,6 +45,7 @@ services: - /volume1/docker/nas-dashboard/ssh/dashboard_terminal:/app/ssh/id_ed25519:ro - /volume1/docker/nas-dashboard/ssh/vps_terminal:/app/ssh/vps_id_ed25519:ro - /volume1/docker/nas-dashboard/ssh/known_hosts:/app/ssh/known_hosts:ro + - /volume1/docker/chat-summarizer/data:/app/data/chat-summarizer:ro healthcheck: test: ["CMD", "python", "-c", "import urllib.request; urllib.request.urlopen('http://localhost:4000/api/health')"] interval: 30s diff --git a/dashboard/frontend/src/App.svelte b/dashboard/frontend/src/App.svelte index cecfadc..4c263f1 100644 --- a/dashboard/frontend/src/App.svelte +++ b/dashboard/frontend/src/App.svelte @@ -7,6 +7,7 @@ import Terminal from "./routes/Terminal.svelte"; import OpenClaw from "./routes/OpenClaw.svelte"; import Settings from "./routes/Settings.svelte"; + import ChatSummary from "./routes/ChatSummary.svelte"; import Login from "./routes/Login.svelte"; import { onMount } from "svelte"; import { getToken, checkAuth, setToken, setRefreshToken } from "./lib/api.js"; @@ -92,6 +93,8 @@ {:else if page === "openclaw"} + {:else if page === "chat-digest"} + {:else if page === "settings"} {/if} diff --git a/dashboard/frontend/src/components/Sidebar.svelte b/dashboard/frontend/src/components/Sidebar.svelte index 7332d00..5e338b0 100644 --- a/dashboard/frontend/src/components/Sidebar.svelte +++ b/dashboard/frontend/src/components/Sidebar.svelte @@ -27,6 +27,7 @@ const tools = [ { id: "openclaw", label: "OpenClaw", icon: "openclaw" }, + { id: "chat-digest", label: "Chat Digest", icon: "chat" }, { id: "gitea", label: "Repos", icon: "git" }, { label: "Gitea Web", port: 3300, icon: "git", external: true }, { label: "Stirling PDF", port: 8030, icon: "pdf", external: true }, @@ -133,6 +134,8 @@ {#if item.icon === "openclaw"} + {:else if item.icon === "chat"} + {:else if item.icon === "git"} {:else if item.icon === "pdf"} @@ -160,6 +163,8 @@ {#if item.icon === "openclaw"} + {:else if item.icon === "chat"} + {:else if item.icon === "git"} {/if} diff --git a/dashboard/frontend/src/routes/ChatSummary.svelte b/dashboard/frontend/src/routes/ChatSummary.svelte new file mode 100644 index 0000000..6f629d9 --- /dev/null +++ b/dashboard/frontend/src/routes/ChatSummary.svelte @@ -0,0 +1,125 @@ + + +
+
+

Chat Digest

+ +
+ + {#if error} +
{error}
+ {/if} + +
+ {#each dates as d} + + {/each} + {#if !dates.length && !error} +

No summaries yet.

+ {/if} +
+ + {#if loading} +
+ {:else if summary} +
+ {@html markdownToHtml(summary.content)} +
+ + + + {#if showMessages} +
+ {#each messages as m} +
+ {m.time?.slice(11,16)} + [{m.group}] {m.sender}: + {m.text} +
+ {/each} +
+ {/if} + {/if} +
+ +