import sys
import subprocess
import os
import asyncio
import random
import logging
import hashlib
from typing import List, Dict

==================== 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 aiosqlite
from telethon import TelegramClient
from telethon.tl.types import MessageMediaDocument, DocumentAttributeVideo, Message
from telethon.errors import (
FloodWaitError,
SecurityError,
FileReferenceExpiredError,
rpcerrorlist
)
from dotenv import load_dotenv

加载环境变量

load_dotenv()

==================== 📁 文件夹初始化 ====================

DB_FOLDER = “database”
SESSION_FOLDER = “session”

for folder in [DB_FOLDER, SESSION_FOLDER]:
if not os.path.exists(folder):
os.makedirs(folder)

==================== 🛠️ 配置读取帮助函数 ====================

def get_env_float(key, default=0.0):
val = os.getenv(key, “”)
if not val: return default
try:
return float(val)
except ValueError:
print(f”⚠️ 配置错误: {key} 必须是数字，当前为 ‘{val}’，已重置为 {default}”)
return default

def get_env_int(key, default=0):
val = os.getenv(key, “”)
if not val: return default
try:
return int(val)
except ValueError:
return default

==================== ⚙️ 基础配置 ====================

API_ID = get_env_int(“API_ID”)
API_HASH = os.getenv(“API_HASH”)
PHONE_NUMBER = os.getenv(“PHONE_NUMBER”)

MIN_SIZE_MB = get_env_float(“MIN_SIZE_MB”, 0)
MAX_SIZE_MB = get_env_float(“MAX_SIZE_MB”, 0)
MIN_DURATION = get_env_int(“MIN_DURATION”, 0)
MAX_DURATION = get_env_int(“MAX_DURATION”, 0)

STOP_AFTER_DUPLICATES = 100
MAX_SCAN_COUNT = 22000

MIN_INTERVAL = 2
MAX_INTERVAL = 5
ALBUM_WAIT_TIME = 3.0

==================== 📋 任务清单 ====================

TASK_CONFIG = [

]

==================== 日志初始化 ====================

logging.basicConfig(
level=logging.INFO,
format=”%(asctime)s – %(message)s”,
datefmt=”%H:%M:%S”
)
logger = logging.getLogger(“AutoBot”)

session_path = os.path.join(SESSION_FOLDER, “user_session”)
client = TelegramClient(session_path, API_ID, API_HASH)

全局状态变量

forward_queue = None
pending_albums = {}
album_timers = {}
db = None

==================== 🗄️ 数据库类 ====================

class AsyncDB:
def init(self, path):
self.path = path
self.conn = None
self.lock = asyncio.Lock()

async def connect(self):
    self.conn = await aiosqlite.connect(self.path)
    await self.conn.execute("CREATE TABLE IF NOT EXISTS videos (video_key TEXT PRIMARY KEY)")
    await self.conn.commit()

async def close(self):
    if self.conn: await self.conn.close()

async def seen(self, key):
    async with self.lock:
        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(self, key):
    async with self.lock:
        await self.conn.execute("INSERT OR IGNORE INTO videos (video_key) VALUES (?)", (key,))
        await self.conn.commit()

==================== 🛠️ 核心工具函数 ====================

def get_unique_key(msg: Message) -> str:
return f”{msg.chat_id}_{msg.media.document.id}”

def generate_safe_filename(sources: List[int], target: int) -> str:
sources_str = “_”.join([str(abs(s)) for s in sources])
target_str = str(abs(target))

if len(sources_str) > 50:
    hash_object = hashlib.md5(sources_str.encode())
    short_hash = hash_object.hexdigest()[:8]
    preview = "_".join([str(abs(s)) for s in sources[:2]])
    fname = f"{preview}_etc_{short_hash}_to_{target_str}.db"
else:
    fname = f"{sources_str}to{target_str}.db"

return os.path.join(DB_FOLDER, fname)

def is_video(msg: Message) -> bool:
if not msg.media or not isinstance(msg.media, MessageMediaDocument):
return False
doc = msg.media.document
if not getattr(doc, ‘mime_type’, None) or not doc.mime_type.startswith(“video”):
return False
video_attr = next((a for a in doc.attributes if isinstance(a, DocumentAttributeVideo)), None)
if getattr(video_attr, “round_message”, False):
return False
size_mb = doc.size / (1024.0 * 1024.0)
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
if video_attr:
duration = video_attr.duration
if MIN_DURATION > 0 and duration < MIN_DURATION: return False
if MAX_DURATION > 0 and duration > MAX_DURATION: return False
return True

==================== 🔄 异步逻辑 ====================

async def process_album_later(grouped_id):
try:
await asyncio.sleep(ALBUM_WAIT_TIME)
if grouped_id in pending_albums:
messages = pending_albums.pop(grouped_id)
album_timers.pop(grouped_id, None)
if messages:
messages.sort(key=lambda x: x.id)
await forward_queue.put(messages)
except asyncio.CancelledError:
pass

async def queue_message(message: Message):
if message.grouped_id:
gid = message.grouped_id
if gid not in pending_albums: pending_albums[gid] = []
pending_albums[gid].append(message)
if gid in album_timers: album_timers[gid].cancel()
album_timers[gid] = asyncio.create_task(process_album_later(gid))
else:
await forward_queue.put([message])

async def worker(target_channel_id):
processed_count = 0
while True:
try:
batch = await forward_queue.get()
if batch is None:
forward_queue.task_done()
break
files = [m.media for m in batch]
caption = batch[0].text or “”
try:
await client.send_file(target_channel_id, files, caption=caption)
processed_count += len(batch)
logger.info(f”✅ 发送成功 | 本次: {len(batch)} | 总计: {processed_count}”)
for m in batch:
await db.mark(get_unique_key(m))
await asyncio.sleep(random.uniform(MIN_INTERVAL, MAX_INTERVAL))
except FloodWaitError as e:
logger.warning(f”⏳ 触发流控: 暂停 {e.seconds} 秒”)
await asyncio.sleep(e.seconds + 2)
except FileReferenceExpiredError:
logger.error(“❌ 文件引用过期，跳过”)
except Exception as e:
logger.error(f”❌ 发送异常: {e}”)
forward_queue.task_done()
except Exception as e:
logger.error(f”Worker Error: {e}”)

async def scanner(source_channels, forward_limit):
all_collected_videos = []
for ch in source_channels:
channel_collected = 0
logger.info(f”🔍 正在扫描: {ch} (目标: {forward_limit})”)
consecutive_duplicates = 0
scanned_count = 0
async for msg in client.iter_messages(ch, limit=MAX_SCAN_COUNT):
scanned_count += 1
if scanned_count % 200 == 0:
print(f”\r …扫描深度: {scanned_count} 条”, end=””)
if not is_video(msg): continue
if await db.seen(get_unique_key(msg)):
consecutive_duplicates += 1
if consecutive_duplicates >= STOP_AFTER_DUPLICATES:
print(f”\n🛑 [频道 {ch}] 连续 {STOP_AFTER_DUPLICATES} 条重复，停止扫描。”)
break
else:
consecutive_duplicates = 0
all_collected_videos.append(msg)
channel_collected += 1
print(f”\r📦 [频道 {ch}] 命中: {channel_collected}/{forward_limit}”, end=””)
if channel_collected >= forward_limit:
print(f”\n✅ [频道 {ch}] 额度已满”)
break
print(“”)

if all_collected_videos:
    logger.info(f"📊 找到 {len(all_collected_videos)} 个新视频。排序入队...")
    all_collected_videos.sort(key=lambda x: x.date)
    for msg in all_collected_videos:
        await queue_message(msg)
    if pending_albums:
        logger.info("⏳ 等待相册分组...")
        while pending_albums or album_timers:
            finished = [k for k, t in album_timers.items() if t.done()]
            for k in finished: del album_timers[k]
            if not album_timers and not pending_albums: break
            await asyncio.sleep(0.5)
else:
    logger.info("⚠️ 没有发现任何新视频。")
await forward_queue.put(None)

async def run_single_task(task):
global db, forward_queue, pending_albums, album_timers

sources = task['sources']
target = task['target']
limit = task.get('limit', 200)

full_db_path = generate_safe_filename(sources, target)
logger.info(f"\n🚀 >>> 启动任务: {os.path.basename(full_db_path)} <<<")

forward_queue = asyncio.Queue()
pending_albums = {}
album_timers = {}

db = AsyncDB(full_db_path)
await db.connect()

worker_task = asyncio.create_task(worker(target))

try:
    await scanner(sources, limit)
    await forward_queue.join()
except Exception as e:
    logger.error(f"❌ 任务运行异常: {e}")
finally:
    if not worker_task.done():
        worker_task.cancel()
    await db.close()
    logger.info(f"🏁 >>> 任务结束: {os.path.basename(full_db_path)} <<<\n")

==================== 🚀 主程序入口 ====================

async def main():
print(“=”*50)
print(“🎥 Auto-Forwarder Pro (Task Sequencer)”)
print(f”📉 最小视频限制: {MIN_SIZE_MB} MB”)
print(f”📈 最大视频限制: {MAX_SIZE_MB if MAX_SIZE_MB > 0 else ‘不限’} MB”)
print(“=”*50)

if not TASK_CONFIG:
    logger.error("❌ TASK_CONFIG 为空，请在代码中配置任务列表！")
    return

await client.start(PHONE_NUMBER)

try:
    total_tasks = len(TASK_CONFIG)
    for index, task in enumerate(TASK_CONFIG):
        await run_single_task(task)
        
        if index < total_tasks - 1:
            wait_sec = 10
            logger.info(f"⏳ 任务 [{index + 1}/{total_tasks}] 已完成。")
            logger.info(f"⏳ 冷却中... {wait_sec} 秒后执行下一个任务...")
            await asyncio.sleep(wait_sec)
        else:
            logger.info("🎉 所有任务已按顺序执行完毕！")

except KeyboardInterrupt:
    print("\n👋 用户手动停止程序")
except Exception as e:
    logger.error(f"🔥 系统致命错误: {e}")
finally:
    await client.disconnect()
    logger.info("🔌 已安全退出并断开连接")

if name == “main“:
asyncio.run(main())