import asyncio
import os
import random
import re
import logging
from logging.handlers import RotatingFileHandler
from typing import List, Dict, Optional, Set
import aiosqlite
from telethon import TelegramClient
from telethon.tl.types import MessageMediaDocument, Message, DocumentAttributeVideo
from telethon.errors import FloodWaitError
from dotenv import load_dotenv
# ==================== 1. 基础配置与初始化 ====================
DB_FOLDER = "database"
SESSION_FOLDER = "session"
for folder in [DB_FOLDER, SESSION_FOLDER]:
if not os.path.exists(folder):
os.makedirs(folder)
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s - %(levelname)s - [%(name)s] - %(message)s",
handlers=[
RotatingFileHandler(os.path.join(DB_FOLDER, "bot_work.log"), maxBytes=10*1024*1024, backupCount=5, encoding="utf-8"),
logging.StreamHandler()
]
)
logger = logging.getLogger("SuperForwarder")
load_dotenv()
API_ID = int(os.getenv("API_ID", 0))
API_HASH = os.getenv("API_HASH", "")
PHONE_NUMBER = os.getenv("PHONE_NUMBER", "")
TWO_STEP_PASSWORD = os.getenv("TWO_STEP_PASSWORD", "")
TARGET_CHANNEL = int(os.getenv("TARGET_CHANNEL", 0))
SOURCE_CHANNELS = [int(x.strip()) for x in os.getenv("SOURCE_CHANNELS", "").split(",") if x.strip()]
MAX_WORKERS = int(os.getenv("MAX_WORKERS", 3))
MIN_INTERVAL = float(os.getenv("MIN_INTERVAL", 2.0))
MAX_INTERVAL = float(os.getenv("MAX_INTERVAL", 5.0))
ALBUM_WAIT_TIME = 4.0
# 视频过滤配置 (可在 .env 中自由配置，设置为 0 表示不限制)
MIN_SIZE_MB = float(os.getenv("MIN_SIZE_MB", 0)) # 最小体积 (MB)
MAX_SIZE_MB = float(os.getenv("MAX_SIZE_MB", 0)) # 最大体积 (MB)
MIN_HEIGHT = int(os.getenv("MIN_HEIGHT", 0)) # 最低垂直分辨率 (如 720)
MAX_HEIGHT = int(os.getenv("MAX_HEIGHT", 0)) # 最高垂直分辨率 (如 1080)
AD_PATTERNS = [
r"https?://\S+",
r"t\.me/\S+",
r"@\w+",
r"Via .*",
r"\[.*?\]\(https?://.*?\)"
]
# 用于防止实时与历史任务重复放入队列的内存集合
processing_msg_ids: Set[int] = set()
# ==================== 2. 数据库管理 ====================
class AsyncDB:
def __init__(self, path):
self.path = path
self.conn: Optional[aiosqlite.Connection] = None
async def connect(self):
self.conn = await aiosqlite.connect(self.path)
await self.conn.execute("PRAGMA journal_mode=WAL;")
await self.conn.execute("""
CREATE TABLE IF NOT EXISTS progress (
channel_id TEXT PRIMARY KEY,
last_msg_id INTEGER DEFAULT 0,
min_msg_id INTEGER DEFAULT 0
)""")
await self.conn.execute("CREATE TABLE IF NOT EXISTS videos (video_key TEXT PRIMARY KEY)")
await self.conn.execute("CREATE INDEX IF NOT EXISTS idx_vkey ON videos (video_key);")
await self.conn.commit()
async def close(self):
if self.conn:
await self.conn.close()
async def is_seen(self, key: str) -> bool:
async with self.conn.execute("SELECT 1 FROM videos WHERE video_key=?", (key,)) as cursor:
return await cursor.fetchone() is not None
async def mark_seen_batch(self, keys: List[str]):
if not keys: return
await self.conn.executemany(
"INSERT OR IGNORE INTO videos (video_key) VALUES (?)",
[(k,) for k in keys]
)
await self.conn.commit()
async def get_prog(self, cid: int):
async with self.conn.execute("SELECT last_msg_id, min_msg_id FROM progress WHERE channel_id=?", (str(cid),)) as cursor:
r = await cursor.fetchone()
return r if r else (0, 0)
async def update_prog(self, cid: int, last_id: Optional[int] = None, min_id: Optional[int] = None):
if last_id is not None:
await self.conn.execute(
"INSERT INTO progress (channel_id, last_msg_id) VALUES (?, ?) ON CONFLICT(channel_id) DO UPDATE SET last_msg_id=?",
(str(cid), last_id, last_id)
)
if min_id is not None:
await self.conn.execute(
"INSERT INTO progress (channel_id, min_msg_id) VALUES (?, ?) ON CONFLICT(channel_id) DO UPDATE SET min_msg_id=?",
(str(cid), min_id, min_id)
)
await self.conn.commit()
# ==================== 3. 工具与过滤逻辑 ====================
def clean_caption(text: str) -> str:
if not text: return ""
for p in AD_PATTERNS:
text = re.sub(p, "", text, flags=re.I)
return text.strip()
def is_video(msg: Message) -> bool:
if not msg or not msg.media or not isinstance(msg.media, MessageMediaDocument):
return False
doc = msg.media.document
if not doc.mime_type or not doc.mime_type.startswith("video"):
return False
# 1. 文件体积校验 (MB)
size_mb = doc.size / 1048576
if MIN_SIZE_MB > 0 and size_mb < MIN_SIZE_MB:
return False
if MAX_SIZE_MB > 0 and size_mb > MAX_SIZE_MB:
return False
# 2. 读取视频元数据并校验分辨率 (Height)
height = 0
if doc.attributes:
for attr in doc.attributes:
if isinstance(attr, DocumentAttributeVideo):
height = attr.h
break
if MIN_HEIGHT > 0 and height < MIN_HEIGHT:
return False
if MAX_HEIGHT > 0 and height > MAX_HEIGHT:
return False
return True
# ==================== 4. 相册防抖聚合逻辑 ====================
forward_queue = asyncio.Queue(maxsize=500)
pending_albums: Dict[int, List[Message]] = {}
album_tasks: Dict[int, asyncio.TimerHandle] = {}
async def _flush_album(grouped_id: int):
await asyncio.sleep(ALBUM_WAIT_TIME)
msgs = pending_albums.pop(grouped_id, None)
album_tasks.pop(grouped_id, None)
if msgs:
msgs.sort(key=lambda x: x.id)
await forward_queue.put(msgs)
async def handle_incoming(msg: Message):
if msg.id in processing_msg_ids:
return
processing_msg_ids.add(msg.id)
if len(processing_msg_ids) > 10000:
processing_msg_ids.clear()
if msg.grouped_id:
gid = msg.grouped_id
if gid not in pending_albums:
pending_albums[gid] = []
pending_albums[gid].append(msg)
# 防抖重置倒计时
if gid in album_tasks:
album_tasks[gid].cancel()
album_tasks[gid] = asyncio.create_task(_flush_album(gid))
else:
await forward_queue.put([msg])
# ==================== 5. 转发 Worker ====================
async def worker(wid: int):
while True:
batch = await forward_queue.get()
try:
to_send = []
for m in batch:
if not m.media or not hasattr(m.media, 'document'):
continue
v_key = str(m.media.document.id)
if not await db.is_seen(v_key):
to_send.append(m)
if to_send:
caption = ""
for m in batch:
if m.text:
caption = clean_caption(m.text)
break
files = [m.media for m in to_send]
await client.send_file(TARGET_CHANNEL, file=files, caption=caption, supports_streaming=True)
seen_keys = [str(m.media.document.id) for m in to_send]
await db.mark_seen_batch(seen_keys)
logger.info(f"Worker-{wid} | 成功转发 {len(to_send)} 个视频")
await asyncio.sleep(random.uniform(MIN_INTERVAL, MAX_INTERVAL))
except FloodWaitError as e:
logger.warning(f"Worker-{wid} 遇到 FloodWait，暂停 {e.seconds + 5} 秒")
await asyncio.sleep(e.seconds + 5)
except Exception as e:
logger.error(f"Worker-{wid} 转发异常: {e}", exc_info=True)
finally:
forward_queue.task_done()
# ==================== 6. 扫描与同步任务 ====================
async def scan_latest_task(cid: int):
last_id, _ = await db.get_prog(cid)
if last_id == 0:
async for msg in client.iter_messages(cid, limit=1):
last_id = msg.id
await db.update_prog(cid, last_id=last_id)
logger.info(f"频道 {cid} 实时监控已就绪，当前最新消息 ID: {last_id}")
while True:
try:
async for msg in client.iter_messages(cid, min_id=last_id, reverse=True):
if is_video(msg):
await handle_incoming(msg)
last_id = max(last_id, msg.id)
await db.update_prog(cid, last_id=last_id)
await asyncio.sleep(15)
except FloodWaitError as e:
await asyncio.sleep(e.seconds + 5)
except Exception as e:
logger.error(f"频道 {cid} 实时监控出错: {e}")
await asyncio.sleep(30)
async def backfill_history_task(cid: int):
logger.info(f"历史补全任务启动: {cid}")
_, min_id = await db.get_prog(cid)
if min_id == 0:
async for msg in client.iter_messages(cid, limit=1):
min_id = msg.id + 1
await db.update_prog(cid, min_id=min_id)
while True:
try:
if min_id <= 1:
logger.info(f"频道 {cid} 历史消息已全部补全完毕！")
break
if forward_queue.qsize() < 100:
fetched_count = 0
async for msg in client.iter_messages(cid, offset_id=min_id, limit=50):
fetched_count += 1
min_id = msg.id
if is_video(msg):
await handle_incoming(msg)
await db.update_prog(cid, min_id=min_id)
if fetched_count == 0:
logger.info(f"频道 {cid} 历史消息回溯完毕 (触底)。")
await db.update_prog(cid, min_id=1)
break
await asyncio.sleep(random.uniform(3.0, 7.0))
else:
await asyncio.sleep(10)
except FloodWaitError as e:
logger.warning(f"历史补全触发限制，等待 {e.seconds + 5} 秒")
await asyncio.sleep(e.seconds + 5)
except Exception as e:
logger.error(f"频道 {cid} 历史补全错误: {e}")
await asyncio.sleep(30)
# ==================== 7. 启动入口 ====================
s_tag = str(SOURCE_CHANNELS[0]) if SOURCE_CHANNELS else "unknown"
t_tag = str(TARGET_CHANNEL)
db_filename = f"{s_tag}to{t_tag}.db"
db_path = os.path.join(DB_FOLDER, db_filename)
session_path = os.path.join(SESSION_FOLDER, "forwarder_session")
client = TelegramClient(session_path, API_ID, API_HASH)
db = AsyncDB(db_path)
async def main():
await db.connect()
await client.start(PHONE_NUMBER, TWO_STEP_PASSWORD)
logger.info(f"--- 登录成功 | 数据库: {db_filename} ---")
workers = [asyncio.create_task(worker(i + 1)) for i in range(MAX_WORKERS)]
tasks = []
for cid in SOURCE_CHANNELS:
tasks.append(asyncio.create_task(scan_latest_task(cid)))
tasks.append(asyncio.create_task(backfill_history_task(cid)))
try:
await client.run_until_disconnected()
finally:
for t in tasks + workers:
t.cancel()
await db.close()
if __name__ == "__main__":
try:
asyncio.run(main())
except KeyboardInterrupt:
logger.info("程序已被用户手动终止")
