tg-watchbot
-
Telegram 双向客服机器人 + Web/RSS 监控推送 + 可视化管理面板
-
双向对话 · 关键词监控 · 私聊广告拦截 · 多管理员 · 配置导入导出
+
Telegram 双向客服机器人 + Web/RSS 监控推送 + 频道媒体下载 + 可视化管理面板
+
双向对话 · 关键词监控 · 频道媒体下载 · 私聊广告拦截 · 多管理员 · 配置导入导出
AI 一句话安装 ·
Docker 安装 ·
@@ -13,11 +13,12 @@
## 简介：
-tg-watchbot 是一个轻量级 Python 服务，把 **Telegram 双向客服机器人** 和 **Web/RSS 监控推送** 合在一起：
+tg-watchbot 是一个轻量级 Python 服务，把 **Telegram 双向客服机器人**、**Web/RSS 监控推送** 和 **频道媒体下载** 合在一起：
- 普通用户私聊 Bot，消息会转发给管理员；
- 管理员可以直接回复、主动发文字/图片、封禁/备注用户；
- 后台定时监控 RSS 或网页，命中关键词、新条目、价格/库存变化后推送给管理员；
+- 使用 Telethon 用户账号自动下载频道/群组中的视频、文档等媒体文件；
- 自带一个 Web 管理面板，可配置监控目标、编辑 YAML、查看收件箱和日志。
项目为单文件应用，适合个人服务器、NAT 小鸡、轻量 VPS 直接用 systemd 跑。
@@ -29,6 +30,18 @@ tg-watchbot 是一个轻量级 Python 服务，把 **Telegram 双向客服机器
```
## 更新日志
+### 2026-05-28 更新
+
+- 新增「频道媒体下载」功能：使用 Telethon 用户账号登录 Telegram，可浏览已加入的群组/频道，选择性下载媒体文件。
+- 面板新增「频道媒体」页面：支持搜索群组、添加/暂停/恢复/删除监控、查看下载记录。
+- 支持断点续传：大文件下载中断后自动续传，不重复下载。
+- 支持并发下载控制：可设置同时下载数（默认 3），避免带宽打满。
+- 支持 SOCKS5/HTTP 代理：国内用户可配置代理访问 Telegram。
+- 支持按日期范围过滤：只下载指定时间段内的消息。
+- 支持关键词过滤、媒体类型过滤（视频/文档/图片/音频）、文件大小限制。
+- 下载完成可自动推送 Telegram 通知给管理员。
+- 需要在设置页填写 `TG_API_ID`、`TG_API_HASH`、`TG_API_SESSION` 后使用。
+
### 2026-05-22 更新
- TG 群监听功能增强：支持可视化配置监听规则、AI 总结参数与防刷屏策略。
@@ -101,6 +114,20 @@ tg-watchbot 是一个轻量级 Python 服务，把 **Telegram 双向客服机器
+### 频道媒体下载
+
+- 使用 Telethon 用户账号（非 Bot）登录 Telegram，可访问已加入的所有频道和群组。
+- 面板「频道媒体」页面支持搜索已加入的群组/频道，一键添加监控。
+- 支持暂停/恢复监控（保留配置）、删除监控。
+- 支持实时自动下载新消息中的媒体，也支持手动触发下载历史媒体。
+- 支持断点续传：大文件下载中断后自动续传，不重复下载。
+- 支持并发下载控制：可设置同时下载数（1-10，默认 3）。
+- 支持 SOCKS5/HTTP 代理，适合国内服务器使用。
+- 支持按日期范围过滤：只下载指定时间段内的消息。
+- 支持关键词过滤、媒体类型选择（视频/文档/图片/音频）、文件大小限制。
+- 下载完成可自动推送 Telegram 通知给管理员。
+- 需要在设置页填写 `TG_API_ID`、`TG_API_HASH`、`TG_API_SESSION`。
+
### Web 管理面板
- 登录页 + HttpOnly session cookie，不使用丑陋的浏览器 Basic Auth。
@@ -490,6 +517,11 @@ monitors:
| `/rules` | 私聊广告拦截规则 |
| `/replies` | 快捷回复模板 |
| `/monitor/events` | 监控推送历史 |
+| `/channel-media` | 频道媒体监控 |
+| `/channel-media/{id}/pause` | 暂停频道监控 |
+| `/channel-media/{id}/resume` | 恢复频道监控 |
+| `/channel-media/{id}/check` | 手动下载频道媒体 |
+| `/channel-media/{id}/download` | 查看下载记录 |
| `/config/export` | 导出 / 导入 `config.yaml` |
| `/logs` | 日志 |
| `/health` | 健康检查 |
diff --git a/app.py b/app.py
index 3711131..6a3e577 100644
--- a/app.py
+++ b/app.py
@@ -27,6 +27,7 @@ from pathlib import Path
from types import SimpleNamespace
from typing import Any
from urllib.parse import quote_plus, urljoin
+import os.path as ospath
import feedparser
import httpx
@@ -79,6 +80,7 @@ pending_sendpic: dict[int, dict[str, Any]] = {}
scheduler_ref: AsyncIOScheduler | None = None
user_session_listener_task: asyncio.Task | None = None
user_session_client: Any = None
+channel_media_clients: dict[str, Any] = {}
GROUP_SUMMARY_MAX_CHARS = 800
@@ -230,11 +232,51 @@ def init_db() -> None:
last_seen_at TEXT NOT NULL,
active INTEGER DEFAULT 1
);
+ CREATE TABLE IF NOT EXISTS channel_media_monitors (
+ id INTEGER PRIMARY KEY AUTOINCREMENT,
+ channel_id INTEGER NOT NULL,
+ channel_title TEXT NOT NULL,
+ channel_username TEXT DEFAULT '',
+ status TEXT DEFAULT 'active',
+ media_types TEXT DEFAULT 'video,document',
+ keywords TEXT DEFAULT '',
+ max_file_size_mb INTEGER DEFAULT 2000,
+ download_dir TEXT DEFAULT '',
+ last_message_id INTEGER DEFAULT 0,
+ total_downloaded INTEGER DEFAULT 0,
+ total_size_bytes INTEGER DEFAULT 0,
+ notify_telegram INTEGER DEFAULT 1,
+ proxy TEXT DEFAULT '',
+ date_from TEXT DEFAULT '',
+ date_to TEXT DEFAULT '',
+ max_concurrent INTEGER DEFAULT 3,
+ created_at TEXT NOT NULL,
+ updated_at TEXT NOT NULL
+ );
+ CREATE TABLE IF NOT EXISTS channel_media_downloads (
+ id INTEGER PRIMARY KEY AUTOINCREMENT,
+ monitor_id INTEGER NOT NULL,
+ channel_id INTEGER NOT NULL,
+ message_id INTEGER NOT NULL,
+ media_type TEXT NOT NULL,
+ file_name TEXT DEFAULT '',
+ file_path TEXT DEFAULT '',
+ file_size INTEGER DEFAULT 0,
+ caption TEXT DEFAULT '',
+ sender_id INTEGER DEFAULT 0,
+ status TEXT DEFAULT 'completed',
+ created_at TEXT NOT NULL,
+ FOREIGN KEY (monitor_id) REFERENCES channel_media_monitors(id)
+ );
"""
)
for sql in [
"ALTER TABLE inbox_messages ADD COLUMN direction TEXT DEFAULT 'in'",
"ALTER TABLE inbox_messages ADD COLUMN source TEXT DEFAULT 'user'",
+ "ALTER TABLE channel_media_monitors ADD COLUMN proxy TEXT DEFAULT ''",
+ "ALTER TABLE channel_media_monitors ADD COLUMN date_from TEXT DEFAULT ''",
+ "ALTER TABLE channel_media_monitors ADD COLUMN date_to TEXT DEFAULT ''",
+ "ALTER TABLE channel_media_monitors ADD COLUMN max_concurrent INTEGER DEFAULT 3",
]:
try:
conn.execute(sql)
@@ -538,6 +580,454 @@ def user_session_ready() -> bool:
return True
+# ---- Channel Media Download (Telethon user session) ----
+
+def get_or_create_channel_media_client(client_type: str = "channel_media", proxy: str = "") -> Any:
+ if TelegramClient is None or StringSession is None:
+ return None
+ if client_type in channel_media_clients:
+ c = channel_media_clients[client_type]
+ if c.is_connected():
+ return c
+ api_id, api_hash, session = user_session_config()
+ if not api_id or not api_hash or not session:
+ return None
+ try:
+ proxy_arg = None
+ if proxy and proxy.strip():
+ p = proxy.strip()
+ if p.startswith("socks5://") or p.startswith("socks4://"):
+ import socks
+ parts = p.replace("socks5://", "").replace("socks4://", "").split(":")
+ proxy_arg = (socks.SOCKS5 if "socks5" in p else socks.SOCKS4,
+ parts[0], int(parts[1]) if len(parts) > 1 else 1080)
+ elif p.startswith("http://") or p.startswith("https://"):
+ proxy_arg = p
+ else:
+ proxy_arg = p
+ client = TelegramClient(StringSession(session), int(api_id), api_hash, proxy=proxy_arg)
+ channel_media_clients[client_type] = client
+ return client
+ except Exception:
+ logger.exception("failed to create channel media client")
+ return None
+
+
+async def disconnect_channel_media_client(client_type: str = "channel_media") -> None:
+ client = channel_media_clients.pop(client_type, None)
+ if client:
+ try:
+ await client.disconnect()
+ except Exception:
+ pass
+
+
+def channel_media_monitors_all() -> list[dict[str, Any]]:
+ with closing(db()) as conn:
+ rows = conn.execute("SELECT * FROM channel_media_monitors ORDER BY id DESC").fetchall()
+ return [dict(r) for r in rows]
+
+
+def channel_media_monitor_get(monitor_id: int) -> dict[str, Any] | None:
+ with closing(db()) as conn:
+ row = conn.execute("SELECT * FROM channel_media_monitors WHERE id=?", (monitor_id,)).fetchone()
+ return dict(row) if row else None
+
+
+def channel_media_monitor_create(
+ channel_id: int,
+ channel_title: str,
+ channel_username: str = "",
+ media_types: str = "video,document",
+ keywords: str = "",
+ max_file_size_mb: int = 2000,
+ download_dir: str = "",
+ notify_telegram: bool = True,
+ proxy: str = "",
+ date_from: str = "",
+ date_to: str = "",
+ max_concurrent: int = 3,
+) -> int:
+ ts = now_iso()
+ with closing(db()) as conn:
+ cur = conn.execute(
+ """INSERT INTO channel_media_monitors
+ (channel_id, channel_title, channel_username, status, media_types, keywords,
+ max_file_size_mb, download_dir, notify_telegram, proxy, date_from, date_to,
+ max_concurrent, created_at, updated_at)
+ VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)""",
+ (channel_id, channel_title, channel_username, "active", media_types, keywords,
+ max_file_size_mb, download_dir, 1 if notify_telegram else 0,
+ proxy, date_from, date_to, max_concurrent, ts, ts),
+ )
+ conn.commit()
+ return int(cur.lastrowid)
+
+
+def channel_media_monitor_update(monitor_id: int, **kwargs: Any) -> None:
+ allowed = {"status", "media_types", "keywords", "max_file_size_mb", "download_dir",
+ "last_message_id", "notify_telegram", "total_downloaded", "total_size_bytes",
+ "proxy", "date_from", "date_to", "max_concurrent"}
+ updates = []
+ values = []
+ for k, v in kwargs.items():
+ if k in allowed:
+ updates.append(f"{k}=?")
+ values.append(v)
+ if not updates:
+ return
+ updates.append("updated_at=?")
+ values.append(now_iso())
+ values.append(monitor_id)
+ with closing(db()) as conn:
+ conn.execute(f"UPDATE channel_media_monitors SET {', '.join(updates)} WHERE id=?", values)
+ conn.commit()
+
+
+def channel_media_monitor_delete(monitor_id: int) -> None:
+ with closing(db()) as conn:
+ conn.execute("DELETE FROM channel_media_downloads WHERE monitor_id=?", (monitor_id,))
+ conn.execute("DELETE FROM channel_media_monitors WHERE id=?", (monitor_id,))
+ conn.commit()
+
+
+def channel_media_download_record(
+ monitor_id: int, channel_id: int, message_id: int,
+ media_type: str, file_name: str, file_path: str,
+ file_size: int, caption: str, sender_id: int, status: str = "completed",
+) -> int:
+ ts = now_iso()
+ with closing(db()) as conn:
+ cur = conn.execute(
+ """INSERT INTO channel_media_downloads
+ (monitor_id, channel_id, message_id, media_type, file_name, file_path,
+ file_size, caption, sender_id, status, created_at)
+ VALUES (?,?,?,?,?,?,?,?,?,?,?)""",
+ (monitor_id, channel_id, message_id, media_type, file_name, file_path,
+ file_size, caption, sender_id, status, ts),
+ )
+ conn.commit()
+ return int(cur.lastrowid)
+
+
+def channel_media_downloads_list(monitor_id: int | None = None, limit: int = 100) -> list[dict[str, Any]]:
+ with closing(db()) as conn:
+ if monitor_id:
+ rows = conn.execute(
+ "SELECT * FROM channel_media_downloads WHERE monitor_id=? ORDER BY id DESC LIMIT ?",
+ (monitor_id, limit),
+ ).fetchall()
+ else:
+ rows = conn.execute(
+ "SELECT * FROM channel_media_downloads ORDER BY id DESC LIMIT ?", (limit,)
+ ).fetchall()
+ return [dict(r) for r in rows]
+
+
+def channel_media_download_exists(channel_id: int, message_id: int) -> bool:
+ with closing(db()) as conn:
+ row = conn.execute(
+ "SELECT id FROM channel_media_downloads WHERE channel_id=? AND message_id=?",
+ (channel_id, message_id),
+ ).fetchone()
+ return row is not None
+
+
+async def telethon_list_dialogs(limit: int = 500) -> list[dict[str, Any]]:
+ client = get_or_create_channel_media_client()
+ if not client:
+ return []
+ try:
+ if not client.is_connected():
+ await client.connect()
+ if not await client.is_user_authorized():
+ return []
+ result = []
+ count = 0
+ async for dialog in client.iter_dialogs(limit=limit):
+ entity = dialog.entity
+ chat_type = "unknown"
+ if dialog.is_group:
+ chat_type = "group"
+ elif dialog.is_channel:
+ chat_type = "channel"
+ elif dialog.is_user:
+ chat_type = "user"
+ result.append({
+ "id": dialog.id,
+ "title": dialog.name or "",
+ "username": getattr(entity, "username", "") or "",
+ "type": chat_type,
+ "unread_count": dialog.unread_count,
+ })
+ count += 1
+ logger.info("listed %d dialogs", count)
+ return result
+ except Exception:
+ logger.exception("telethon_list_dialogs failed")
+ return []
+
+
+async def telethon_search_dialogs(query: str, limit: int = 100) -> list[dict[str, Any]]:
+ client = get_or_create_channel_media_client()
+ if not client:
+ return []
+ try:
+ if not client.is_connected():
+ await client.connect()
+ if not await client.is_user_authorized():
+ return []
+ result = []
+ q = query.strip().lower()
+ async for dialog in client.iter_dialogs(limit=500):
+ name = (dialog.name or "").lower()
+ username = str(getattr(dialog.entity, "username", "") or "").lower()
+ if q in name or q in username or query.strip() in str(dialog.id):
+ entity = dialog.entity
+ chat_type = "group" if dialog.is_group else ("channel" if dialog.is_channel else ("user" if dialog.is_user else "unknown"))
+ result.append({
+ "id": dialog.id,
+ "title": dialog.name or "",
+ "username": getattr(entity, "username", "") or "",
+ "type": chat_type,
+ "unread_count": dialog.unread_count,
+ })
+ if len(result) >= limit:
+ break
+ return result
+ except Exception:
+ logger.exception("telethon_search_dialogs failed")
+ return []
+
+
+def _detect_media_type(media: Any, allowed_types: set[str]) -> str:
+ if not media:
+ return ""
+ if hasattr(media, "video") or hasattr(media, "document')"):
+ pass
+ from telethon.tl.types import (
+ DocumentAttributeVideo,
+ DocumentAttributeAudio,
+ DocumentAttributeFilename,
+ )
+ if hasattr(media, "photo"):
+ if "photo" not in allowed_types:
+ return ""
+ return "photo"
+ if hasattr(media, "document"):
+ doc = media.document
+ if doc is None:
+ return ""
+ mime = str(getattr(doc, "mime_type", "") or "")
+ for attr in (getattr(doc, "attributes", None) or []):
+ if isinstance(attr, DocumentAttributeVideo):
+ if "video" not in allowed_types:
+ return ""
+ return "video"
+ if isinstance(attr, DocumentAttributeAudio):
+ if "audio" not in allowed_types:
+ return ""
+ return "audio"
+ if mime.startswith("video/"):
+ if "video" not in allowed_types:
+ return ""
+ return "video"
+ if mime.startswith("audio/"):
+ if "audio" not in allowed_types:
+ return ""
+ return "audio"
+ if "document" in allowed_types:
+ return "document"
+ return ""
+ if "document" in allowed_types:
+ return "document"
+ return ""
+
+
+def _get_media_size(media: Any) -> int:
+ if hasattr(media, "document") and media.document:
+ return int(getattr(media.document, "size", 0) or 0)
+ if hasattr(media, "photo"):
+ return 0
+ return 0
+
+
+async def channel_media_monitor_loop() -> None:
+ while True:
+ await asyncio.sleep(300)
+ try:
+ monitors = channel_media_monitors_all()
+ active_monitors = [m for m in monitors if m.get("status") == "active"]
+ if not active_monitors:
+ continue
+ for monitor in active_monitors:
+ try:
+ await telethon_download_from_channel(int(monitor["id"]))
+ except Exception:
+ logger.exception("channel media monitor failed id=%s", monitor.get("id"))
+ except Exception:
+ logger.exception("channel_media_monitor_loop error")
+ finally:
+ await disconnect_channel_media_client()
+
+
+async def telethon_download_from_channel(monitor_id: int, download_history: bool = False) -> int:
+ monitor = channel_media_monitor_get(monitor_id)
+ if not monitor:
+ return 0
+ proxy = str(monitor.get("proxy") or "").strip()
+ client = get_or_create_channel_media_client(proxy=proxy)
+ if not client:
+ logger.warning("channel media client not available")
+ return 0
+ try:
+ if not client.is_connected():
+ await client.connect()
+ if not await client.is_user_authorized():
+ logger.warning("channel media client not authorized")
+ return 0
+ channel_id = int(monitor["channel_id"])
+ try:
+ entity = await client.get_entity(channel_id)
+ except Exception:
+ logger.warning("cannot resolve channel entity id=%s", channel_id)
+ return 0
+ media_types_str = str(monitor.get("media_types") or "video,document")
+ allowed_types = {t.strip().lower() for t in media_types_str.split(",") if t.strip()}
+ keywords_str = str(monitor.get("keywords") or "").strip()
+ keywords_list = [k.strip() for k in keywords_str.split(",") if k.strip()] if keywords_str else []
+ max_size = int(monitor.get("max_file_size_mb") or 2000) * 1024 * 1024
+ base_dir = str(monitor.get("download_dir") or "").strip()
+ if not base_dir:
+ base_dir = str(BASE_DIR / "channel_downloads" / str(channel_id))
+ os.makedirs(base_dir, exist_ok=True)
+ last_msg_id = int(monitor.get("last_message_id") or 0)
+ offset_id = last_msg_id if not download_history else 0
+
+ # Date filtering
+ from datetime import datetime as dt_type
+ offset_date = None
+ date_from_str = str(monitor.get("date_from") or "").strip()
+ date_to_str = str(monitor.get("date_to") or "").strip()
+ if date_from_str:
+ try:
+ offset_date = dt_type.fromisoformat(date_from_str)
+ except Exception:
+ pass
+ # Collect messages to download
+ messages_to_download = []
+ async for message in client.iter_messages(
+ entity, limit=500, offset_id=offset_id,
+ offset_date=offset_date, reverse=bool(offset_date),
+ ):
+ if offset_id > 0 and message.id <= offset_id and not offset_date:
+ break
+ # Date range check
+ if date_to_str and message.date:
+ try:
+ dt_to = dt_type.fromisoformat(date_to_str)
+ if message.date.replace(tzinfo=None) > dt_to:
+ continue
+ except Exception:
+ pass
+ if channel_media_download_exists(channel_id, message.id):
+ continue
+ if keywords_list:
+ msg_text = (message.text or "") + " " + (message.caption or "")
+ if not any(k.lower() in msg_text.lower() for k in keywords_list):
+ continue
+ media = message.media
+ if not media:
+ continue
+ media_type = _detect_media_type(media, allowed_types)
+ if not media_type:
+ continue
+ file_size = _get_media_size(media)
+ if file_size and file_size > max_size:
+ continue
+ messages_to_download.append((message, media_type, file_size))
+
+ # Concurrent download with semaphore
+ max_concurrent = max(1, int(monitor.get("max_concurrent") or 3))
+ semaphore = asyncio.Semaphore(max_concurrent)
+ count = 0
+ total_size_added = 0
+
+ async def download_one(msg: Any, mt: str, fs: int) -> tuple[int, int]:
+ nonlocal count, total_size_added
+ async with semaphore:
+ caption = (msg.text or msg.caption or "")[:500]
+ sender_id = msg.sender_id or 0
+ file_name = ""
+ for attr in (getattr(getattr(msg.media, "document", None), "attributes", None) or []):
+ if hasattr(attr, "file_name") and attr.file_name:
+ file_name = attr.file_name
+ break
+ if not file_name:
+ file_name = f"{channel_id}_{msg.id}"
+ ext_map = {"video": ".mp4", "photo": ".jpg", "audio": ".mp3", "document": ".bin"}
+ file_name += ext_map.get(mt, ".bin")
+ file_path = ospath.join(base_dir, file_name)
+ if ospath.exists(file_path):
+ file_path = ospath.join(base_dir, f"{channel_id}_{msg.id}_{file_name}")
+ # Resume: use .part file
+ part_path = file_path + ".part"
+ try:
+ # Check for existing partial download
+ existing_size = ospath.getsize(part_path) if ospath.exists(part_path) else 0
+ if existing_size > 0 and fs and existing_size >= fs:
+ # Already fully downloaded as .part, just rename
+ os.rename(part_path, file_path)
+ else:
+ await client.download_media(msg, file=part_path)
+ if ospath.exists(part_path):
+ os.rename(part_path, file_path)
+ else:
+ return (0, 0)
+ actual_size = ospath.getsize(file_path) if ospath.exists(file_path) else fs
+ channel_media_download_record(
+ monitor_id, channel_id, msg.id,
+ mt, file_name, file_path,
+ actual_size, caption, sender_id,
+ )
+ logger.info("downloaded channel media: channel=%s msg=%s type=%s", channel_id, msg.id, mt)
+ return (1, actual_size)
+ except Exception:
+ logger.exception("download failed channel=%s msg=%s", channel_id, msg.id)
+ return (0, 0)
+
+ # Run all downloads concurrently
+ tasks = [download_one(msg, mt, fs) for msg, mt, fs in messages_to_download]
+ results = await asyncio.gather(*tasks, return_exceptions=True)
+ for r in results:
+ if isinstance(r, tuple):
+ c, s = r
+ count += c
+ total_size_added += s
+
+ new_total = int(monitor.get("total_downloaded") or 0) + count
+ new_size = int(monitor.get("total_size_bytes") or 0) + total_size_added
+ update_kwargs: dict[str, Any] = {
+ "total_downloaded": new_total,
+ "total_size_bytes": new_size,
+ }
+ async for last_msg in client.iter_messages(entity, limit=1):
+ update_kwargs["last_message_id"] = last_msg.id
+ break
+ channel_media_monitor_update(monitor_id, **update_kwargs)
+ if count > 0 and monitor.get("notify_telegram"):
+ title = monitor.get("channel_title") or str(channel_id)
+ await admin_send(
+ f"[频道媒体下载] {html_escape(title)}\n"
+ f"新增 {count} 个文件，共 {total_size_added // 1024 // 1024} MB\n"
+ f"累计：{new_total} 个文件，{new_size // 1024 // 1024} MB"
+ )
+ return count
+ except Exception:
+ logger.exception("telethon_download_from_channel failed monitor_id=%s", monitor_id)
+ return 0
+
+
def group_message_text(message: Message) -> str:
parts = [message.text or "", message.caption or ""]
if getattr(message, "reply_to_message", None):
@@ -2259,7 +2749,7 @@ pre{{white-space:pre-wrap;background:#121212;color:#fff;padding:13px;border:4px
@media (prefers-reduced-motion: reduce){{
*,*::before,*::after{{animation:none!important;transition:none!important}}
}}
-