import sys
import subprocess
import os

==================== 0. 面板专属：环境依赖自动安装器 ====================

lib_path = os.path.join(os.path.dirname(os.path.abspath(file)), ‘libs’)
sys.path.insert(0, lib_path)

try:
import telethon
import aiosqlite
import dotenv
except ImportError:
print(“>>> 首次运行：检测到缺少依赖，正在局部目录(libs)自动安装，请稍候…”)
subprocess.check_call([
sys.executable, “-m”, “pip”, “install”, “-t”, lib_path,
“telethon”, “aiosqlite”, “python-dotenv”
])
print(“>>> 依赖安装完成！准备启动机器人…”)

确保安装后可以正常导入

import asyncio
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=1010241024, 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

MIN_SIZE_MB = float(os.getenv(“MIN_SIZE_MB”, 0))
MAX_SIZE_MB = float(os.getenv(“MAX_SIZE_MB”, 0))
MIN_HEIGHT = int(os.getenv(“MIN_HEIGHT”, 0))
MAX_HEIGHT = int(os.getenv(“MAX_HEIGHT”, 0))

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 getattr(doc, 'mime_type', None) is None or not doc.mime_type.startswith("video"): 
    return False

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

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(“程序已被用户手动终止”)