天才程序源写的TG频道转发。

import re

import asyncio

from telethon import TelegramClient, events

from telethon.utils import get_input_media



# ==================== 配置板块 ====================

API_ID = 66666          # 替换为你的 API ID

API_HASH = '666666666666666'  # 替换为你的 API HASH



FORWARD_RULES = [

    {

        "sources": [-1001111111111, -1001111111111, "source_channel_username"], 

        "destinations": [-1002222222222],                     

        "whitelist_keywords": [],      # 白名单：含有这些词才转发 (留空 [] 代表转发所有)  

        "blacklist_keywords": ["采集", "管理员", "上架"],      # 黑名单：含有这些词则不转发  

        "replacements": [                                      

            (r"@\w+", "@xxx_bot "), # 示例：替换原频道的 username 推广

            (r"http[s]?://\S+", "[链接已屏蔽]"),      # 示例：屏蔽所有网址           

            (r"关键词1", "替换词1"),                

            (r"关键词2", "替换词2"),                       

            (r"关键词3", "替换词3")  

        ]

    }

]

# ==================================================



for rule in FORWARD_RULES:

    rule["_compiled_replacements"] = [

        (re.compile(pattern, flags=re.IGNORECASE), repl) 

        for pattern, repl in rule["replacements"]

    ]



client = TelegramClient('session_forwarder', API_ID, API_HASH)

album_cache = {}



def process_text(text, rule):

    if any(word in text for word in rule["blacklist_keywords"]):

        return None, "blacklist"

    if rule["whitelist_keywords"] and not any(word in text for word in rule["whitelist_keywords"]):

        return None, "whitelist"

    

    new_text = text

    for pattern, repl in rule["_compiled_replacements"]:

        new_text = pattern.sub(repl, new_text)

    return new_text, "pass"



async def send_single_video_after_delay(dest, text, media, delay=35):

    """新增：专门用于单个视频延迟 35 秒发送的异步函数"""

    print(f"[排队] 单个视频已进入队列，将在 {delay} 秒后发送...")

    await asyncio.sleep(delay)

    try:

        await client.send_message(dest, text, file=media)

        print(f"[成功] 单个视频(已延迟 {delay} 秒)成功转发至 {dest}")

    except Exception as e:

        print(f"[错误] 延迟转发单个视频至 {dest} 失败: {e}")



async def send_album_after_delay(grouped_id, rule, delay=30):

    """等待 30 秒确保所有图片、视频和文字全部收齐"""

    await asyncio.sleep(delay)

    

    if grouped_id not in album_cache:

        return

        

    group_data = album_cache.pop(grouped_id)

    events_list = group_data["events"]

    

    # 1. 提取合并文案

    final_text = ""

    for ev in events_list:

        if ev.raw_text:

            final_text = ev.raw_text

            break

            

    # 过滤与替换处理

    processed_text, status = process_text(final_text, rule)

    if status != "pass":

        print(f"[拦截] 混合媒体组被跳过。原因: {status}")

        return



    # 2. 收集媒体对象

    media_files = []

    for ev in events_list:

        if ev.media:

            media_files.append(ev.media)



    if not media_files:

        return

    

    # 3. 转发到所有目的地

    for dest in rule["destinations"]:

        try:

            await client.send_file(dest, media_files, caption=processed_text)

            print(f"[成功] 混合媒体组(共 {len(media_files)} 个文件)及文字已完美转发至 {dest}")

        except Exception as e:

            print(f"[错误] 转发混合媒体组至 {dest} 失败: {e}")



@client.on(events.NewMessage)

async def handler(event):

    chat_id = event.chat_id

    

    matched_rule = None

    for rule in FORWARD_RULES:

        is_source = False

        for src in rule["sources"]:

            if isinstance(src, int) and chat_id == src:

                is_source = True

                break

            elif isinstance(src, str):

                sender = await event.get_chat()

                if getattr(sender, 'username', None) == src.strip('@'):

                    is_source = True

                    break

        if is_source:

            matched_rule = rule

            break

            

    if not matched_rule:

        return



    # 情况 A：属于媒体组（无论里面是纯图片、纯视频还是图视混合）

    if event.grouped_id is not None:

        g_id = event.grouped_id

        

        if g_id not in album_cache:

            album_cache[g_id] = {"events": [event]}

            asyncio.create_task(send_album_after_delay(g_id, matched_rule))

        else:

            album_cache[g_id]["events"].append(event)

            

    # 情况 B：普通的单条消息

    else:

        text = event.raw_text or ""

        processed_text, status = process_text(text, matched_rule)

        

        if status != "pass":

            return

            

        for dest in matched_rule["destinations"]:

            try:

                if event.media:

                    # 💡 核心修改：判断单条媒体消息是否为视频

                    if event.video:

                        # 启动一个后台异步任务进行延时发送，不阻塞其他消息的监听

                        asyncio.create_task(send_single_video_after_delay(dest, processed_text, event.media, 35))

                    else:

                        # 如果是单张图片或其他非视频媒体，依然立即发送

                        await client.send_message(dest, processed_text, file=event.media)

                        print(f"[成功] 单条非视频媒体消息成功转发至 {dest}")

                else:

                    # 纯文本，立即发送

                    await client.send_message(dest, processed_text)

                    print(f"[成功] 单条纯文本消息成功转发至 {dest}")

            except Exception as e:

                print(f"[错误] 单条消息处理失败: {e}")



async def main():

    print("正在启动终极完美版 Telegram 转发器(支持图视文一模一样混合)...")

    await client.start()

    print("登录成功！转发服务已运行。")

    await client.run_until_disconnected()



if __name__ == '__main__':

    asyncio.run(main())